chore: Release cnats version 0.2.3

The SSR branch rendered an empty call-panel div while the hydrate
branch expected a child button node. On any full load/refresh of the
lobby room, the mismatched hydration cursor hit tachys's
unreachable!() panic path, trapping the wasm instance and killing all
reactivity (routing, room switching, SSE) for the rest of the page
load, while the already-rendered SSR HTML stayed visually intact.

SSR now renders the same default join-button markup hydrate expects.

.gitea/workflows/release.yml

@@ -36,8 +36,15 @@ jobs:
             ~/.cargo/bin
           key: ${{ runner.os }}-${{ matrix.arch }}-cargo-leptos-${{ hashFiles('Cargo.lock') }}
 
+      - name: Install cargo-binstall
+        run: |
+          command -v cargo-binstall || \
+            curl -L --proto '=https' --tlsv1.2 -sSf \
+              https://raw.githubusercontent.com/cargo-bins/cargo-binstall/main/install-from-binstall-release.sh \
+              | bash
+
       - name: Install cargo-leptos
-        run: command -v cargo-leptos || cargo install cargo-leptos --locked
+        run: command -v cargo-leptos || cargo binstall cargo-leptos --locked --no-confirm
 
       - name: Build
         run: cargo leptos build --release

Cargo.lock

@@ -345,15 +345,17 @@ dependencies = [
 
 [[package]]
 name = "cnats"
-version = "0.1.0"
+version = "0.2.3"
 dependencies = [
  "anyhow",
  "async-nats",
  "axum",
+ "base64 0.22.1",
  "chrono",
  "console_error_panic_hook",
  "dotenvy",
  "futures",
+ "js-sys",
  "leptos",
  "leptos_axum",
  "leptos_meta",
@@ -368,8 +370,10 @@ dependencies = [
  "tower-sessions",
  "tracing",
  "tracing-subscriber",
+ "url",
  "uuid",
  "wasm-bindgen",
+ "wasm-bindgen-futures",
  "web-sys",
 ]
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cnats"
-version = "0.1.0"
+version = "0.2.3"
 edition = "2021"
 
 [lib]
@@ -21,6 +21,7 @@ tower = { version = "0.5", optional = true }
 tower-http = { version = "0.6", features = ["fs", "trace"], optional = true }
 tower-sessions = { version = "0.14", optional = true }
 async-nats = { version = "0.38", optional = true }
+url = { version = "2", optional = true }
 sqlx = { version = "0.8", default-features = false, features = [
     "runtime-tokio",
     "tls-rustls",
@@ -28,6 +29,12 @@ sqlx = { version = "0.8", default-features = false, features = [
     "macros",
 ], optional = true }
 openidconnect = { version = "4", optional = true }
+# For pulling the `groups` custom claim out of the already-verified ID
+# token's raw JWT payload - openidconnect's Core* type aliases default to
+# EmptyAdditionalClaims, and reworking that generic stack for one extra
+# field isn't worth it. The token's signature is already checked by
+# id_token.claims(...) before this ever runs.
+base64 = { version = "0.22", optional = true }
 futures = { version = "0.3", optional = true }
 chrono = { version = "0.4", features = ["serde"], optional = true }
 uuid = { version = "1", features = ["v4"], optional = true }
@@ -38,12 +45,33 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"], optional = tr
 
 # --- browser only ---
 wasm-bindgen = { version = "0.2", optional = true }
+wasm-bindgen-futures = { version = "0.4", optional = true }
+js-sys = { version = "0.3", optional = true }
 console_error_panic_hook = { version = "0.1", optional = true }
 web-sys = { version = "0.3", features = [
     "EventSource",
     "MessageEvent",
     "HtmlElement",
     "Element",
+    # --- WebRTC mesh calling ---
+    "RtcPeerConnection",
+    "RtcConfiguration",
+    "RtcIceServer",
+    "RtcSdpType",
+    "RtcSessionDescriptionInit",
+    "RtcIceCandidate",
+    "RtcIceCandidateInit",
+    "RtcPeerConnectionIceEvent",
+    "RtcRtpSender",
+    "RtcTrackEvent",
+    "RtcRtpTransceiver",
+    "RtcOfferOptions",
+    "MediaStream",
+    "MediaStreamConstraints",
+    "MediaStreamTrack",
+    "MediaDevices",
+    "Navigator",
+    "HtmlVideoElement",
 ], optional = true }
 
 [features]
@@ -51,6 +79,8 @@ default = []
 hydrate = [
     "leptos/hydrate",
     "dep:wasm-bindgen",
+    "dep:wasm-bindgen-futures",
+    "dep:js-sys",
     "dep:console_error_panic_hook",
     "dep:web-sys",
 ]
@@ -65,8 +95,10 @@ ssr = [
     "dep:tower-http",
     "dep:tower-sessions",
     "dep:async-nats",
+    "dep:url",
     "dep:sqlx",
     "dep:openidconnect",
+    "dep:base64",
     "dep:futures",
     "dep:chrono",
     "dep:uuid",

aur/PKGBUILD

@@ -1,6 +1,6 @@
 # Maintainer: Bendik Aagaard Lynghaug <bendik.lynghaug@gmail.com>
 pkgname=cnats
-pkgver=0.1.0
+pkgver=0.2.3
 pkgrel=1
 pkgdesc="Web chat over NATS subjects with Kanidm SSO (Leptos SSR)"
 arch=('x86_64' 'aarch64')

release.toml

@@ -0,0 +1,12 @@
+publish = false
+allow-branch = ["main"]
+
+[[pre-release-replacements]]
+file = "aur/PKGBUILD"
+search = "pkgver=.*"
+replace = "pkgver={{version}}"
+
+[[pre-release-replacements]]
+file = "aur/PKGBUILD"
+search = "pkgrel=.*"
+replace = "pkgrel=1"

src/app.rs

@@ -7,7 +7,10 @@ use leptos_router::{
 };
 
 use crate::auth::{current_user, User};
-use crate::chat::{is_valid_room, room_subject, ChatMessage, SendMessage, DEFAULT_ROOM, ROOMS};
+use crate::chat::{
+    is_authorized_for_room, is_valid_room, room_subject, ChatMessage, SendMessage, DEFAULT_ROOM,
+    ROOMS,
+};
 #[cfg(feature = "hydrate")]
 use crate::chat::room_history;
 
@@ -75,7 +78,23 @@ fn ChatPage() -> impl IntoView {
             {move || {
                 user.get()
                     .map(|res| match res {
-                        Ok(Some(u)) => view! { <ChatShell room user=u/> }.into_any(),
+                        Ok(Some(u)) => {
+                            // `room`'s own Memo only knows about is_valid_room (no
+                            // user context available that early) - fall back to
+                            // DEFAULT_ROOM here too if this user isn't authorized
+                            // for the room the URL asked for, same as an unknown
+                            // room name already does.
+                            let u2 = u.clone();
+                            let effective_room = Memo::new(move |_| {
+                                let r = room.get();
+                                if is_authorized_for_room(&u2, &r) {
+                                    r
+                                } else {
+                                    DEFAULT_ROOM.to_string()
+                                }
+                            });
+                            view! { <ChatShell room=effective_room user=u/> }.into_any()
+                        }
                         _ => view! { <LoginGate/> }.into_any(),
                     })
             }}
@@ -83,6 +102,28 @@ fn ChatPage() -> impl IntoView {
     }
 }
 
+// ---------------------------------------------------------------------------
+// brand mark - the "voice pulse" ornament, currentColor so it always
+// matches whatever text color surrounds it (sidebar wordmark vs. the much
+// larger gate title)
+// ---------------------------------------------------------------------------
+
+#[component]
+fn PulseMark() -> impl IntoView {
+    view! {
+        <svg
+            class="pulse-mark"
+            viewBox="-15 -70 430 140"
+            fill="none"
+            stroke="currentColor"
+            stroke-width="18"
+            stroke-linecap="round"
+        >
+            <path d="M0,0 H60 Q80,-24 100,0 Q120,24 140,0 Q162,-52 184,0 Q206,52 228,0 Q250,-24 270,0 Q290,24 310,0 H400"></path>
+        </svg>
+    }
+}
+
 // ---------------------------------------------------------------------------
 // unauthenticated: the gate
 // ---------------------------------------------------------------------------
@@ -94,8 +135,10 @@ fn LoginGate() -> impl IntoView {
             <div class="gate-card">
                 <div class="gate-badge">"MESSAGE BUS · AUTH REQUIRED"</div>
                 <h1 class="gate-title">
+                    <PulseMark/>
                     "CN" <span class="gate-title-accent">"ATS"</span>
                 </h1>
+                <p class="wordmark-sub gate-tagline">"chat over the bus"</p>
                 <p class="gate-sub">
                     "Realtime chat carried on NATS subjects. Identity issued by your Kanidm realm — no separate passwords, no local accounts."
                 </p>
@@ -210,11 +253,13 @@ fn ChatShell(room: Memo<String>, user: User) -> impl IntoView {
     };
 
     let me = user.username.clone();
+    let call_me = me.clone();
 
     view! {
         <div class="console">
             <nav class="rail">
                 <div class="wordmark">
+                    <PulseMark/>
                     "CN" <span class="wordmark-accent">"ATS"</span>
                     <span class="wordmark-sub">"chat over the bus"</span>
                 </div>
@@ -223,7 +268,8 @@ fn ChatShell(room: Memo<String>, user: User) -> impl IntoView {
                 <div class="rooms">
                     {ROOMS
                         .iter()
-                        .map(|(name, desc)| {
+                        .filter(|(name, _, _)| is_authorized_for_room(&user, name))
+                        .map(|(name, desc, _)| {
                             let name = *name;
                             let desc = *desc;
                             view! {
@@ -276,6 +322,10 @@ fn ChatShell(room: Memo<String>, user: User) -> impl IntoView {
                     </div>
                 </header>
 
+                <Show when=move || room.get() == "lobby">
+                    <CallPanel room=room me=call_me.clone()/>
+                </Show>
+
                 <div class="stream" node_ref=list_ref>
                     <Show when=move || messages.with(|m| m.is_empty())>
                         <div class="stream-empty">
@@ -357,6 +407,196 @@ fn open_event_source(
     Some(es)
 }
 
+/// Minimal mesh call entry point, scoped to the `lobby` room only (see
+/// `ChatShell`'s `<Show when=move || room.get() == "lobby">`) - the
+/// signaling itself (`call.rs`, `/call-sse/{room}`) is already generic
+/// per-room, so widening this later is a one-line UI change, not an
+/// architectural one. `CallState` (browser-only: it holds `web_sys`
+/// types) can't exist in the `ssr` build at all, so the two targets get
+/// entirely separate bodies rather than sharing signals across the gate.
+#[component]
+fn CallPanel(room: Memo<String>, me: String) -> impl IntoView {
+    #[cfg(feature = "hydrate")]
+    {
+        use crate::webrtc::CallState;
+
+        let call_state = StoredValue::new_local(CallState::new(room.get_untracked(), me.clone()));
+        let in_call = call_state.get_value().in_call;
+        let local_video_ref = NodeRef::<leptos::html::Video>::new();
+
+        let es_handle = StoredValue::new_local(None::<web_sys::EventSource>);
+        Effect::new(move |_| {
+            let room_name = room.get();
+            es_handle.update_value(|es| {
+                if let Some(es) = es.take() {
+                    es.close();
+                }
+            });
+            es_handle.set_value(open_call_event_source(&room_name, call_state.get_value()));
+        });
+        on_cleanup(move || {
+            es_handle.update_value(|es| {
+                if let Some(es) = es.take() {
+                    es.close();
+                }
+            });
+            let cs = call_state.get_value();
+            if cs.in_call.get_untracked() {
+                leptos::task::spawn_local(async move {
+                    cs.leave().await;
+                });
+            }
+        });
+
+        // Mirror the local MediaStream into the preview <video> element -
+        // `srcObject` has no HTML attribute form, has to be set via JS.
+        Effect::new(move |_| {
+            let stream = call_state.get_value().local_stream().get();
+            if let Some(el) = local_video_ref.get() {
+                el.set_src_object(stream.as_ref());
+            }
+        });
+
+        let on_join = move |_| {
+            let cs = call_state.get_value();
+            leptos::task::spawn_local(async move {
+                cs.join().await;
+            });
+        };
+        let on_leave = move |_| {
+            let cs = call_state.get_value();
+            leptos::task::spawn_local(async move {
+                cs.leave().await;
+            });
+        };
+
+        view! {
+            <div class="call-panel">
+                {move || {
+                    if in_call.get() {
+                        view! {
+                            <CallActive
+                                local_video_ref=local_video_ref
+                                call_state=call_state.get_value()
+                                on_leave=on_leave
+                            />
+                        }
+                            .into_any()
+                    } else {
+                        view! {
+                            <button class="call-join" on:click=on_join>
+                                "☎ join call"
+                            </button>
+                        }
+                            .into_any()
+                    }
+                }}
+            </div>
+        }
+            .into_any()
+    }
+    // Must mirror the hydrate branch's default (not-in-call) markup exactly -
+    // hydration reconciles this SSR output against what the hydrate branch
+    // above expects to find, and an empty div here (vs. the button hydrate
+    // wants) is a hydration mismatch that panics and traps the whole wasm
+    // instance, killing all reactivity on the page.
+    #[cfg(not(feature = "hydrate"))]
+    {
+        let _ = (room, me);
+        view! {
+            <div class="call-panel">
+                <button class="call-join" disabled=true>
+                    "☎ join call"
+                </button>
+            </div>
+        }
+            .into_any()
+    }
+}
+
+/// The in-call subtree (video grid + leave button), split out of
+/// `CallPanel` so its `<For>`-over-peers view doesn't get inlined as a type
+/// parameter of `CallPanel`'s own `if`/`else` branch - that inlining is what
+/// was blowing the compiler's query recursion limit once mesh calling's
+/// nested `Show`/`For` landed inside `ChatShell`'s own `Show`.
+#[cfg(feature = "hydrate")]
+#[component]
+fn CallActive(
+    local_video_ref: NodeRef<leptos::html::Video>,
+    call_state: crate::webrtc::CallState,
+    on_leave: impl Fn(leptos::ev::MouseEvent) + 'static,
+) -> impl IntoView {
+    view! {
+        <div class="call-active">
+            <div class="video-grid">
+                <video
+                    class="video-tile video-tile-local"
+                    node_ref=local_video_ref
+                    autoplay=true
+                    muted=true
+                    playsinline=true
+                ></video>
+                <For
+                    each=move || call_state.peer_streams()
+                    key=|(id, _)| id.clone()
+                    children=move |(_id, stream)| {
+                        view! { <PeerVideoTile stream=stream/> }
+                    }
+                />
+            </div>
+            <button class="call-leave" on:click=on_leave>
+                "⏏ leave call"
+            </button>
+        </div>
+    }
+}
+
+/// One remote participant's video tile - a plain child component so each
+/// tile gets its own `NodeRef`/effect pair instead of trying to juggle a
+/// `Vec` of node refs by hand in the parent.
+#[cfg(feature = "hydrate")]
+#[component]
+fn PeerVideoTile(stream: RwSignal<Option<web_sys::MediaStream>>) -> impl IntoView {
+    let video_ref = NodeRef::<leptos::html::Video>::new();
+    Effect::new(move |_| {
+        let s = stream.get();
+        if let Some(el) = video_ref.get() {
+            el.set_src_object(s.as_ref());
+        }
+    });
+    view! { <video class="video-tile" node_ref=video_ref autoplay=true playsinline=true></video> }
+}
+
+/// Bridges `/call-sse/{room}` into `CallState::handle_signal`. A separate
+/// function from `open_event_source` (rather than a shared generic) since
+/// the event name differs: `sse::call_events` emits a custom `"signal"`
+/// SSE event, not the default unnamed one, so this needs
+/// `add_event_listener_with_callback` instead of `set_onmessage` (which
+/// only fires for the default event type).
+#[cfg(feature = "hydrate")]
+fn open_call_event_source(
+    room: &str,
+    call_state: crate::webrtc::CallState,
+) -> Option<web_sys::EventSource> {
+    use wasm_bindgen::{prelude::Closure, JsCast};
+    use web_sys::{EventSource, MessageEvent};
+
+    let es = EventSource::new(&format!("/call-sse/{room}")).ok()?;
+
+    let on_signal = Closure::<dyn FnMut(MessageEvent)>::new(move |ev: MessageEvent| {
+        if let Some(data) = ev.data().as_string() {
+            if let Ok(signal) = serde_json::from_str::<crate::call::CallSignal>(&data) {
+                call_state.handle_signal(signal.from, signal.to, signal.kind);
+            }
+        }
+    });
+    es.add_event_listener_with_callback("signal", on_signal.as_ref().unchecked_ref())
+        .ok()?;
+    on_signal.forget();
+
+    Some(es)
+}
+
 #[component]
 fn NotFound() -> impl IntoView {
     view! {

src/auth.rs

@@ -8,6 +8,11 @@ pub struct User {
     pub sub: String,
     pub username: String,
     pub display_name: String,
+    /// Kanidm group membership, from the `groups` OIDC claim (see
+    /// `oauth2 update-claim-map`). Fixed at login time - not re-checked
+    /// live, so a group change only takes effect on the next login.
+    #[serde(default)]
+    pub groups: Vec<String>,
 }
 
 pub const SESSION_USER_KEY: &str = "user";

src/call.rs

@@ -0,0 +1,86 @@
+use leptos::prelude::*;
+use serde::{Deserialize, Serialize};
+
+/// Call signaling for a room, kept entirely separate from `chat::ChatMessage`
+/// - deliberately a different NATS subject namespace (`call.room.<room>`,
+/// not `chat.room.<room>`) so `server::store`'s JetStream/Postgres archive
+/// (scoped to `chat.room.*` only) never sees it. Ephemeral SDP/ICE has no
+/// business being durably stored.
+pub fn call_subject(room: &str) -> String {
+    format!("call.room.{room}")
+}
+
+/// `from`/`to` are peer ids - currently just the signed-in username (same
+/// identity chat messages use). `to: None` is a room-wide broadcast (only
+/// `Join`/`Leave` use this); everything else is directed at one peer, with
+/// every other browser in the room ignoring it client-side. Mesh calls at
+/// this scale (~4 people) don't need per-peer NATS subjects - broadcast +
+/// client-side filter is the simplest thing that works.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct CallSignal {
+    pub room: String,
+    pub from: String,
+    pub to: Option<String>,
+    pub kind: CallSignalKind,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "kind", content = "data")]
+pub enum CallSignalKind {
+    /// Announces presence to the room; existing participants respond by
+    /// initiating an offer to the new peer.
+    Join,
+    Leave,
+    Offer(String),
+    Answer(String),
+    /// A single trickled ICE candidate, JSON-encoded
+    /// (`RTCIceCandidateInit`, produced client-side).
+    IceCandidate(String),
+}
+
+/// Publishes a call-signaling message to the room's signaling subject.
+/// Requires a signed-in, room-authorized session - same two checks as
+/// `chat::send_message`, and deliberately not shared via a helper since
+/// the existing chat functions already establish that each enforcement
+/// point re-does this small check inline rather than factoring it out.
+#[server]
+pub async fn send_signal(
+    room: String,
+    to: Option<String>,
+    kind: CallSignalKind,
+) -> Result<(), ServerFnError> {
+    use crate::auth::{User, SESSION_USER_KEY};
+    use crate::chat::is_authorized_for_room;
+    use crate::server::AppState;
+
+    if !crate::chat::is_valid_room(&room) {
+        return Err(ServerFnError::new("unknown room"));
+    }
+
+    let session: tower_sessions::Session = leptos_axum::extract().await?;
+    let Some(user) = session
+        .get::<User>(SESSION_USER_KEY)
+        .await
+        .map_err(|e| ServerFnError::new(e.to_string()))?
+    else {
+        return Err(ServerFnError::new("not signed in"));
+    };
+    if !is_authorized_for_room(&user, &room) {
+        return Err(ServerFnError::new("not authorized for this room"));
+    }
+
+    let state = expect_context::<AppState>();
+    let signal = CallSignal {
+        room: room.clone(),
+        from: user.username,
+        to,
+        kind,
+    };
+    let payload = serde_json::to_vec(&signal).map_err(|e| ServerFnError::new(e.to_string()))?;
+    state
+        .nats
+        .publish(call_subject(&room), payload.into())
+        .await
+        .map_err(|e| ServerFnError::new(format!("nats publish failed: {e}")))?;
+    Ok(())
+}

src/chat.rs

@@ -3,23 +3,42 @@ use serde::{Deserialize, Serialize};
 
 /// Rooms available in the UI. Each maps to the NATS subject
 /// `chat.room.<name>`, so any other NATS client on the bus can join in.
-pub const ROOMS: &[(&str, &str)] = &[
-    ("lobby", "general traffic"),
-    ("dev", "build & ship"),
-    ("ops", "incidents & infra"),
-    ("random", "off the record"),
+/// The third field is the Kanidm group (via the `groups` OIDC claim,
+/// see `oauth2 update-claim-map`) required to read/post in that room -
+/// `None` means open to anyone in `cnats_users`.
+pub const ROOMS: &[(&str, &str, Option<&str>)] = &[
+    ("lobby", "general traffic", None),
+    ("dev", "build & ship", Some("developers")),
+    ("ops", "incidents & infra", Some("developers")),
+    ("random", "off the record", None),
 ];
 
 pub const DEFAULT_ROOM: &str = "lobby";
 
 pub fn is_valid_room(room: &str) -> bool {
-    ROOMS.iter().any(|(name, _)| *name == room)
+    ROOMS.iter().any(|(name, _, _)| *name == room)
 }
 
 pub fn room_subject(room: &str) -> String {
     format!("chat.room.{room}")
 }
 
+/// Whether `user` may read/post in `room`. `false` for an unknown room -
+/// callers should check `is_valid_room` separately if they need to tell
+/// "unknown room" and "not authorized" apart in the error they return.
+/// Synchronous and I/O-free: the user's groups are already baked into
+/// their session (from the `groups` OIDC claim at login), so this never
+/// needs a live Kanidm round-trip - and never gets more current than
+/// that login until they sign in again.
+pub fn is_authorized_for_room(user: &crate::auth::User, room: &str) -> bool {
+    ROOMS
+        .iter()
+        .find(|(name, _, _)| *name == room)
+        .is_some_and(|(_, _, required_group)| {
+            required_group.is_none_or(|g| user.groups.iter().any(|ug| ug == g))
+        })
+}
+
 /// A single chat message as it travels over NATS (JSON-encoded payload).
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ChatMessage {
@@ -62,6 +81,9 @@ pub async fn send_message(room: String, text: String) -> Result<(), ServerFnErro
     else {
         return Err(ServerFnError::new("not signed in"));
     };
+    if !is_authorized_for_room(&user, &room) {
+        return Err(ServerFnError::new("not authorized for this room"));
+    }
 
     let state = expect_context::<AppState>();
     let now = chrono::Utc::now();
@@ -94,13 +116,15 @@ pub async fn room_history(room: String) -> Result<Vec<ChatMessage>, ServerFnErro
         return Err(ServerFnError::new("unknown room"));
     }
     let session: tower_sessions::Session = leptos_axum::extract().await?;
-    if session
+    let Some(user) = session
         .get::<User>(SESSION_USER_KEY)
         .await
         .map_err(|e| ServerFnError::new(e.to_string()))?
-        .is_none()
-    {
+    else {
         return Err(ServerFnError::new("not signed in"));
+    };
+    if !is_authorized_for_room(&user, &room) {
+        return Err(ServerFnError::new("not authorized for this room"));
     }
 
     let state = expect_context::<AppState>();

src/lib.rs

@@ -1,10 +1,14 @@
 pub mod app;
 pub mod auth;
+pub mod call;
 pub mod chat;
 
 #[cfg(feature = "ssr")]
 pub mod server;
 
+#[cfg(feature = "hydrate")]
+pub mod webrtc;
+
 #[cfg(feature = "hydrate")]
 #[wasm_bindgen::prelude::wasm_bindgen]
 pub fn hydrate() {

src/main.rs

@@ -32,7 +32,17 @@ async fn main() -> anyhow::Result<()> {
     let nats_url =
         std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string());
     tracing::info!(%nats_url, "connecting to NATS");
-    let nats = async_nats::connect(&nats_url).await?;
+    // async-nats does not honor userinfo embedded in the URL, so pass any
+    // credentials explicitly via ConnectOptions.
+    let parsed = url::Url::parse(&nats_url)?;
+    let mut nats_opts = async_nats::ConnectOptions::new();
+    if !parsed.username().is_empty() {
+        nats_opts = nats_opts.user_and_password(
+            parsed.username().to_string(),
+            parsed.password().unwrap_or_default().to_string(),
+        );
+    }
+    let nats = nats_opts.connect(&nats_url).await?;
 
     let database_url = std::env::var("DATABASE_URL")
         .unwrap_or_else(|_| "postgres://cnats:cnats@127.0.0.1:5432/cnats".to_string());
@@ -80,6 +90,7 @@ async fn main() -> anyhow::Result<()> {
         .route("/auth/callback", get(oidc::callback))
         .route("/auth/logout", get(oidc::logout))
         .route("/sse/{room}", get(sse::room_events))
+        .route("/call-sse/{room}", get(sse::call_events))
         .route("/api/{*fn_name}", any(server_fn_handler))
         .leptos_routes_with_context(
             &state,

src/server/oidc.rs

@@ -185,10 +185,20 @@ pub async fn callback(
         .map(|n| n.as_str().to_string())
         .unwrap_or_else(|| username.clone());
 
+    // `groups` is a custom claim (Kanidm `oauth2 update-claim-map`), not
+    // something the Core* typed claims struct above knows about. The
+    // signature is already verified by `id_token.claims(...)` above, so
+    // re-reading the same payload's raw JSON for one more field is safe -
+    // just a plain field extraction, not a second verification step.
+    // IdToken's Serialize impl (not Display - it has none) produces the
+    // raw compact JWT string "header.payload.signature".
+    let groups = extract_groups_claim(&id_token);
+
     let user = User {
         sub: claims.subject().as_str().to_string(),
         username,
         display_name,
+        groups,
     };
 
     // Rotate the session id on privilege change, then store the user.
@@ -207,3 +217,34 @@ pub async fn logout(session: Session) -> Result<Redirect, HandlerError> {
     session.flush().await.map_err(internal)?;
     Ok(Redirect::to("/"))
 }
+
+/// Pulls the `groups` custom claim (Kanidm `oauth2 update-claim-map`) out
+/// of an ID token's raw JWT payload. `IdToken`'s `Serialize` impl (it has
+/// no `Display`) produces the compact "header.payload.signature" string,
+/// which is where this reads from - the signature itself is never
+/// re-checked here, that already happened via `id_token.claims(...)`
+/// before this is called. Defensive by design: any parse failure (no
+/// claim, wrong shape) just yields no groups rather than failing login.
+fn extract_groups_claim<T: serde::Serialize>(id_token: &T) -> Vec<String> {
+    use base64::Engine;
+
+    let Ok(serde_json::Value::String(compact)) = serde_json::to_value(id_token) else {
+        return Vec::new();
+    };
+    let Some(payload_b64) = compact.split('.').nth(1) else {
+        return Vec::new();
+    };
+    let Ok(payload_bytes) = base64::engine::general_purpose::URL_SAFE_NO_PAD.decode(payload_b64)
+    else {
+        return Vec::new();
+    };
+    let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&payload_bytes) else {
+        return Vec::new();
+    };
+
+    payload
+        .get("groups")
+        .and_then(|g| g.as_array())
+        .map(|arr| arr.iter().filter_map(|v| v.as_str().map(String::from)).collect())
+        .unwrap_or_default()
+}

src/server/sse.rs

@@ -12,28 +12,45 @@ use futures::{Stream, StreamExt};
 use tower_sessions::Session;
 
 use crate::auth::{User, SESSION_USER_KEY};
-use crate::chat::{is_valid_room, room_subject};
+use crate::call::call_subject;
+use crate::chat::{is_authorized_for_room, is_valid_room, room_subject};
 
 use super::AppState;
 
-/// GET /sse/{room} — stream the room's NATS subject to the browser.
+/// Shared by `room_events` and `call_events`: signed in, valid room, and
+/// authorized for it (Kanidm group gate, checked against the session's
+/// own `groups` - see `chat::is_authorized_for_room`). Note this is only
+/// checked once, at connect time - a long-lived SSE stream doesn't get
+/// re-checked if the user's groups change mid-connection (same kind of
+/// staleness the "still signed in at all" check already has).
+async fn authorize_room_stream(
+    room: &str,
+    session: &Session,
+) -> Result<(), (StatusCode, &'static str)> {
+    let user = session
+        .get::<User>(SESSION_USER_KEY)
+        .await
+        .ok()
+        .flatten();
+    let Some(user) = user else {
+        return Err((StatusCode::UNAUTHORIZED, "sign in first"));
+    };
+    if !is_valid_room(room) {
+        return Err((StatusCode::NOT_FOUND, "unknown room"));
+    }
+    if !is_authorized_for_room(&user, room) {
+        return Err((StatusCode::FORBIDDEN, "not authorized for this room"));
+    }
+    Ok(())
+}
+
+/// GET /sse/{room} — stream the room's chat NATS subject to the browser.
 pub async fn room_events(
     Path(room): Path<String>,
     State(state): State<AppState>,
     session: Session,
 ) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, &'static str)> {
-    let signed_in = session
-        .get::<User>(SESSION_USER_KEY)
-        .await
-        .ok()
-        .flatten()
-        .is_some();
-    if !signed_in {
-        return Err((StatusCode::UNAUTHORIZED, "sign in first"));
-    }
-    if !is_valid_room(&room) {
-        return Err((StatusCode::NOT_FOUND, "unknown room"));
-    }
+    authorize_room_stream(&room, &session).await?;
 
     let subscriber = state
         .nats
@@ -56,3 +73,37 @@ pub async fn room_events(
             .text("ping"),
     ))
 }
+
+/// GET /call-sse/{room} — stream the room's call-signaling NATS subject
+/// (SDP offers/answers, ICE candidates). Deliberately a separate subject
+/// namespace (`call.room.*`, not `chat.room.*`) so this never touches the
+/// JetStream/Postgres chat archive - ephemeral signaling has no business
+/// being durably stored.
+pub async fn call_events(
+    Path(room): Path<String>,
+    State(state): State<AppState>,
+    session: Session,
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, &'static str)> {
+    authorize_room_stream(&room, &session).await?;
+
+    let subscriber = state
+        .nats
+        .subscribe(call_subject(&room))
+        .await
+        .map_err(|e| {
+            tracing::error!("nats subscribe failed: {e}");
+            (StatusCode::BAD_GATEWAY, "message bus unavailable")
+        })?;
+
+    let stream = subscriber.map(|msg| {
+        Ok(Event::default()
+            .event("signal")
+            .data(String::from_utf8_lossy(&msg.payload).into_owned()))
+    });
+
+    Ok(Sse::new(stream).keep_alive(
+        KeepAlive::new()
+            .interval(Duration::from_secs(15))
+            .text("ping"),
+    ))
+}

src/webrtc.rs

@@ -0,0 +1,328 @@
+//! Minimal mesh WebRTC calling, browser-only. Public STUN, no TURN - calls
+//! across hostile NATs (symmetric NAT, restrictive corporate networks)
+//! simply won't connect. That's a known, accepted limitation, not a bug to
+//! fix later: real NAT traversal needs a TURN relay, which is real
+//! infrastructure this pass deliberately isn't standing up. Mesh topology
+//! (every pair of peers connects directly) is fine at the ~4-person scale
+//! this is scoped for; it does not scale further than that.
+#![cfg(feature = "hydrate")]
+
+use std::collections::HashMap;
+
+use js_sys::{Array, Reflect};
+use leptos::prelude::*;
+use wasm_bindgen::{prelude::*, JsCast};
+use wasm_bindgen_futures::JsFuture;
+use web_sys::{
+    MediaStream, MediaStreamConstraints, RtcConfiguration, RtcIceCandidateInit,
+    RtcIceServer, RtcPeerConnection, RtcSdpType, RtcSessionDescriptionInit,
+};
+
+use crate::call::{send_signal, CallSignalKind};
+
+const STUN_URL: &str = "stun:stun.l.google.com:19302";
+
+/// One remote participant: their peer connection plus the remote stream
+/// their video tile renders once `ontrack` fires.
+struct Peer {
+    conn: RtcPeerConnection,
+    stream: RwSignal<Option<MediaStream>>,
+}
+
+/// Call state for one room. Lives for as long as the user is in the call;
+/// dropped (and everything torn down) on "leave".
+#[derive(Clone)]
+pub struct CallState {
+    room: String,
+    me: String,
+    local_stream: RwSignal<Option<MediaStream>>,
+    peers: StoredValue<HashMap<String, Peer>, LocalStorage>,
+    pub in_call: RwSignal<bool>,
+}
+
+fn new_peer_connection() -> Result<RtcPeerConnection, JsValue> {
+    let config = RtcConfiguration::new();
+    let ice_server = RtcIceServer::new();
+    ice_server.set_urls(&JsValue::from_str(STUN_URL));
+    let servers = Array::new();
+    servers.push(&ice_server);
+    config.set_ice_servers(&servers);
+    RtcPeerConnection::new_with_configuration(&config)
+}
+
+async fn get_local_stream() -> Result<MediaStream, JsValue> {
+    let window = web_sys::window().ok_or("no window")?;
+    let media_devices = window.navigator().media_devices()?;
+    let constraints = MediaStreamConstraints::new();
+    constraints.set_video(&JsValue::TRUE);
+    constraints.set_audio(&JsValue::TRUE);
+    let promise = media_devices.get_user_media_with_constraints(&constraints)?;
+    let stream = JsFuture::from(promise).await?;
+    stream.dyn_into::<MediaStream>()
+}
+
+fn attach_local_tracks(pc: &RtcPeerConnection, stream: &MediaStream) {
+    for track in stream.get_tracks().iter() {
+        if let Ok(track) = track.dyn_into::<web_sys::MediaStreamTrack>() {
+            pc.add_track_0(&track, stream);
+        }
+    }
+}
+
+/// Reads the `sdp` field off whatever `create_offer`/`create_answer`
+/// resolved to, and builds a fresh `RtcSessionDescriptionInit` from it -
+/// simpler and more reliable than trying to cast the resolved JsValue
+/// directly, since its concrete type varies by browser. Returns the sdp
+/// string alongside the desc (not `desc.get_sdp()` afterwards - that
+/// returns `Option<String>`, and we already have it as a plain `String`
+/// right here).
+fn session_description_from_resolved(
+    resolved: &JsValue,
+    sdp_type: RtcSdpType,
+) -> Result<(RtcSessionDescriptionInit, String), JsValue> {
+    let sdp = Reflect::get(resolved, &JsValue::from_str("sdp"))?
+        .as_string()
+        .ok_or("resolved session description had no sdp field")?;
+    let desc = RtcSessionDescriptionInit::new(sdp_type);
+    desc.set_sdp(&sdp);
+    Ok((desc, sdp))
+}
+
+impl CallState {
+    pub fn new(room: String, me: String) -> Self {
+        Self {
+            room,
+            me,
+            local_stream: RwSignal::new(None),
+            peers: StoredValue::new_local(HashMap::new()),
+            in_call: RwSignal::new(false),
+        }
+    }
+
+    pub fn local_stream(&self) -> ReadSignal<Option<MediaStream>> {
+        self.local_stream.read_only()
+    }
+
+    /// Streams for currently-known peers, keyed by their peer id (username).
+    /// Recomputed each call - fine at mesh scale (~4 peers).
+    pub fn peer_streams(&self) -> Vec<(String, RwSignal<Option<MediaStream>>)> {
+        self.peers
+            .with_value(|p| p.iter().map(|(id, peer)| (id.clone(), peer.stream)).collect())
+    }
+
+    /// getUserMedia, then broadcast Join so existing participants know to
+    /// offer us a connection.
+    pub async fn join(&self) {
+        match get_local_stream().await {
+            Ok(stream) => self.local_stream.set(Some(stream)),
+            Err(e) => {
+                leptos::logging::error!("getUserMedia failed: {e:?}");
+                return;
+            }
+        }
+        self.in_call.set(true);
+        let _ = send_signal(self.room.clone(), None, CallSignalKind::Join).await;
+    }
+
+    /// Tears down every peer connection, stops all local tracks (releases
+    /// the camera/mic), and tells the room we're gone.
+    pub async fn leave(&self) {
+        self.peers.update_value(|peers| {
+            for (_, peer) in peers.drain() {
+                peer.conn.close();
+            }
+        });
+        if let Some(stream) = self.local_stream.get_untracked() {
+            for track in stream.get_tracks().iter() {
+                if let Ok(track) = track.dyn_into::<web_sys::MediaStreamTrack>() {
+                    track.stop();
+                }
+            }
+        }
+        self.local_stream.set(None);
+        self.in_call.set(false);
+        let _ = send_signal(self.room.clone(), None, CallSignalKind::Leave).await;
+    }
+
+    /// One incoming signal from `/call-sse/{room}`. Ignores our own
+    /// broadcasts and anything not addressed to us (directed messages are
+    /// broadcast NATS-wide and filtered client-side - see call.rs).
+    pub fn handle_signal(&self, from: String, to: Option<String>, kind: CallSignalKind) {
+        if from == self.me || !self.in_call.get_untracked() {
+            return;
+        }
+        if let Some(to) = &to {
+            if *to != self.me {
+                return;
+            }
+        }
+
+        match kind {
+            CallSignalKind::Join => {
+                // A new peer announced themselves - we initiate the offer.
+                self.start_offer(from);
+            }
+            CallSignalKind::Leave => {
+                self.peers.update_value(|peers| {
+                    if let Some(peer) = peers.remove(&from) {
+                        peer.conn.close();
+                    }
+                });
+            }
+            CallSignalKind::Offer(sdp) => self.handle_offer(from, sdp),
+            CallSignalKind::Answer(sdp) => self.handle_answer(from, sdp),
+            CallSignalKind::IceCandidate(candidate_json) => {
+                self.handle_ice_candidate(from, candidate_json)
+            }
+        }
+    }
+
+    fn ensure_peer(&self, peer_id: &str) -> Option<RtcPeerConnection> {
+        if let Some(pc) = self
+            .peers
+            .with_value(|peers| peers.get(peer_id).map(|p| p.conn.clone()))
+        {
+            return Some(pc);
+        }
+
+        let pc = new_peer_connection().ok()?;
+        let Some(local) = self.local_stream.get_untracked() else {
+            return None;
+        };
+        attach_local_tracks(&pc, &local);
+
+        let remote_stream = RwSignal::new(None::<MediaStream>);
+        {
+            let remote_stream = remote_stream;
+            let ontrack = Closure::<dyn FnMut(web_sys::RtcTrackEvent)>::new(move |ev: web_sys::RtcTrackEvent| {
+                remote_stream.set(Some(ev.streams().get(0).dyn_into().unwrap()));
+            });
+            pc.set_ontrack(Some(ontrack.as_ref().unchecked_ref()));
+            ontrack.forget();
+        }
+
+        {
+            let room = self.room.clone();
+            let peer_id = peer_id.to_string();
+            let onicecandidate =
+                Closure::<dyn FnMut(web_sys::RtcPeerConnectionIceEvent)>::new(move |ev: web_sys::RtcPeerConnectionIceEvent| {
+                    let Some(candidate) = ev.candidate() else {
+                        return;
+                    };
+                    let Ok(candidate_json) = js_sys::JSON::stringify(&candidate.to_json())
+                        .map(|s| s.as_string().unwrap_or_default())
+                    else {
+                        return;
+                    };
+                    let room = room.clone();
+                    let peer_id = peer_id.clone();
+                    leptos::task::spawn_local(async move {
+                        let _ = send_signal(
+                            room,
+                            Some(peer_id),
+                            CallSignalKind::IceCandidate(candidate_json),
+                        )
+                        .await;
+                    });
+                });
+            pc.set_onicecandidate(Some(onicecandidate.as_ref().unchecked_ref()));
+            onicecandidate.forget();
+        }
+
+        self.peers.update_value(|peers| {
+            peers.insert(
+                peer_id.to_string(),
+                Peer {
+                    conn: pc.clone(),
+                    stream: remote_stream,
+                },
+            );
+        });
+        Some(pc)
+    }
+
+    fn start_offer(&self, peer_id: String) {
+        let Some(pc) = self.ensure_peer(&peer_id) else {
+            return;
+        };
+        let room = self.room.clone();
+        leptos::task::spawn_local(async move {
+            let Ok(resolved) = JsFuture::from(pc.create_offer()).await else {
+                return;
+            };
+            let Ok((desc, sdp)) = session_description_from_resolved(&resolved, RtcSdpType::Offer)
+            else {
+                return;
+            };
+            if JsFuture::from(pc.set_local_description(&desc)).await.is_err() {
+                return;
+            }
+            let _ = send_signal(room, Some(peer_id), CallSignalKind::Offer(sdp)).await;
+        });
+    }
+
+    fn handle_offer(&self, from: String, sdp: String) {
+        let Some(pc) = self.ensure_peer(&from) else {
+            return;
+        };
+        let room = self.room.clone();
+        leptos::task::spawn_local(async move {
+            let remote_desc = RtcSessionDescriptionInit::new(RtcSdpType::Offer);
+            remote_desc.set_sdp(&sdp);
+            if JsFuture::from(pc.set_remote_description(&remote_desc))
+                .await
+                .is_err()
+            {
+                return;
+            }
+            let Ok(resolved) = JsFuture::from(pc.create_answer()).await else {
+                return;
+            };
+            let Ok((answer_desc, answer_sdp)) =
+                session_description_from_resolved(&resolved, RtcSdpType::Answer)
+            else {
+                return;
+            };
+            if JsFuture::from(pc.set_local_description(&answer_desc))
+                .await
+                .is_err()
+            {
+                return;
+            }
+            let _ = send_signal(room, Some(from), CallSignalKind::Answer(answer_sdp)).await;
+        });
+    }
+
+    fn handle_answer(&self, from: String, sdp: String) {
+        let Some(pc) = self
+            .peers
+            .with_value(|peers| peers.get(&from).map(|p| p.conn.clone()))
+        else {
+            return;
+        };
+        leptos::task::spawn_local(async move {
+            let remote_desc = RtcSessionDescriptionInit::new(RtcSdpType::Answer);
+            remote_desc.set_sdp(&sdp);
+            let _ = JsFuture::from(pc.set_remote_description(&remote_desc)).await;
+        });
+    }
+
+    fn handle_ice_candidate(&self, from: String, candidate_json: String) {
+        let Some(pc) = self
+            .peers
+            .with_value(|peers| peers.get(&from).map(|p| p.conn.clone()))
+        else {
+            return;
+        };
+        let Ok(parsed) = js_sys::JSON::parse(&candidate_json) else {
+            return;
+        };
+        let init: RtcIceCandidateInit = parsed.unchecked_into();
+        leptos::task::spawn_local(async move {
+            let _ = JsFuture::from(
+                pc.add_ice_candidate_with_opt_rtc_ice_candidate_init(Some(&init)),
+            )
+            .await;
+        });
+    }
+}

style/main.css

@@ -197,6 +197,22 @@ body::before {
   text-transform: uppercase;
 }
 
+/* brand mark - height:1em scales it to whatever font-size the surrounding
+   heading uses, so the same element/class works at sidebar (1.6rem) and
+   gate-title (up to 4.5rem) scale with no per-placement sizing rules. */
+.pulse-mark {
+  display: block;
+  height: 1em;
+  width: auto;
+  color: var(--signal);
+  margin-bottom: 0.3em;
+}
+
+.gate-tagline {
+  font-size: 0.75rem;
+  margin-top: 0.6rem;
+}
+
 .rail-label {
   font-family: var(--mono);
   font-size: 0.6rem;
@@ -480,6 +496,63 @@ body::before {
 .composer-send:hover { filter: brightness(1.15); }
 .composer-send:disabled { filter: grayscale(0.6) brightness(0.7); cursor: wait; }
 
+/* call */
+
+.call-panel {
+  padding: 0.9rem 1.6rem;
+  border-bottom: 1px solid var(--line);
+  background: var(--ink-1);
+}
+
+.call-join {
+  font-family: var(--mono);
+  font-weight: 600;
+  font-size: 0.78rem;
+  letter-spacing: 0.08em;
+  color: var(--ink-0);
+  background: var(--signal);
+  border: none;
+  padding: 0.55rem 1rem;
+  cursor: pointer;
+  transition: filter 120ms;
+}
+
+.call-join:hover { filter: brightness(1.15); }
+
+.call-active { display: flex; flex-direction: column; gap: 0.8rem; }
+
+.video-grid {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 0.6rem;
+}
+
+.video-tile {
+  width: 200px;
+  height: 150px;
+  background: var(--ink-0);
+  border: 1px solid var(--line-hot);
+  object-fit: cover;
+}
+
+.video-tile-local { border-color: var(--signal-dim); }
+
+.call-leave {
+  align-self: flex-start;
+  font-family: var(--mono);
+  font-weight: 600;
+  font-size: 0.78rem;
+  letter-spacing: 0.08em;
+  color: var(--text);
+  background: none;
+  border: 1px solid var(--alarm);
+  padding: 0.5rem 0.9rem;
+  cursor: pointer;
+  transition: background 120ms;
+}
+
+.call-leave:hover { background: rgba(255, 90, 90, 0.12); }
+
 /* motion */
 
 @keyframes rise {