chore: Release cnats version 0.2.4

Every video tile is built client-side via document.createElement, never
parsed from HTML - browsers only seed the live .muted property from the
muted attribute for parser-inserted elements, so the attribute alone
never actually silenced the local preview. Result: your own mic played
back through your own speakers even with no other peers in the call.
Now sets el.set_muted(true) directly whenever the preview's srcObject
is (re)assigned.

.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.4"
 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.4"
 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.4
 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,206 @@ 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.
+        //
+        // Also force `.muted` via the JS property here, not just the
+        // `muted` attribute on the element below: every video tile in this
+        // app is built client-side via `document.createElement` (never
+        // parsed from HTML), and browsers only seed the live `.muted`
+        // property from the `muted` *attribute* for parser-inserted
+        // elements. Without this, the local preview plays back the user's
+        // own mic through their speakers - audible as feedback in a
+        // solo call.
+        Effect::new(move |_| {
+            let stream = call_state.get_value().local_stream().get();
+            if let Some(el) = local_video_ref.get() {
+                el.set_muted(true);
+                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)] = &[
+/// The third field is the Kanidm group (via the `groups` OIDC claim,
-    ("lobby", "general traffic"),
+/// see `oauth2 update-claim-map`) required to read/post in that room -
-    ("dev", "build & ship"),
+/// `None` means open to anyone in `cnats_users`.
-    ("ops", "incidents & infra"),
+pub const ROOMS: &[(&str, &str, Option<&str>)] = &[
-    ("random", "off the record"),
+    ("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
+    authorize_room_stream(&room, &session).await?;
-        .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"));
-    }
 
     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 {