chore: Release cnats version 0.3.0

Screen sharing swaps the outgoing video track on every peer's sender with
replaceTrack (same m-line, no renegotiation); the camera returns when the
share stops, including via the browser's own "Stop sharing" bar. Mic and
camera toggles flip track.enabled. A new Status signal kind broadcasts
mic/cam/screen state so tiles can show badges; newcomers get it directed
when their Join arrives.

Tiles carry a name label, click to pin as a stage (a peer's screen share
pins itself), double-click for fullscreen. The call bar has an elapsed
timer and a participant count. Controls are icon-only inline SVGs on
currentColor: plain with a strike when off, a soft glow and ticking LED
when live. Red is reserved for leave.

Also fixed along the way: the peer map was a non-reactive StoredValue, so
the tile list never re-rendered when someone joined; ICE candidates that
arrive before the remote description are now buffered instead of
rejected; a peer whose connection fails or closes is dropped.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L3sD29ozDvqA7jozeZXZoB

.gitea/workflows/release.yml

@@ -15,29 +15,59 @@ jobs:
         include:
           - arch: x86_64
             runs-on: ubuntu-latest
+            sccache_dir: /sccache
           - arch: aarch64
             runs-on: aarch64
+            sccache_dir: /var/lib/gitea-runner/sccache
 
     runs-on: ${{ matrix.runs-on }}
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: builtin:checkout
 
       - name: Install Rust stable
         uses: dtolnay/rust-toolchain@stable
         with:
           targets: wasm32-unknown-unknown
 
-      - name: Cache cargo registry and tools
+      # Compiled-crate cache via sccache. x86_64 uses /sccache (a persistent
-        uses: actions/cache@v3
+      # host dir bind-mounted into the job container by the runner config);
-        with:
+      # aarch64 runs on the klokka host and uses a runner-owned dir. Both
-          path: |
+      # persist across releases, unlike the (unreachable) Gitea cache service.
-            ~/.cargo/registry
+      - name: Set up sccache
-            ~/.cargo/bin
+        run: |
-          key: ${{ runner.os }}-${{ matrix.arch }}-cargo-leptos-${{ hashFiles('Cargo.lock') }}
+          echo "RUSTC_WRAPPER=sccache" >> "$GITHUB_ENV"
+          echo "SCCACHE_DIR=${{ matrix.sccache_dir }}" >> "$GITHUB_ENV"
+          mkdir -p "${{ matrix.sccache_dir }}"
+          if command -v sccache >/dev/null 2>&1; then sccache --version; exit 0; fi
+          # Bare runners (aarch64 = klokka host) are pre-provisioned; CI must not
+          # install onto them (and this tarball is x86_64-only). Only the
+          # ephemeral x86_64 container installs; a bare host missing it fails loud.
+          if [ "${{ matrix.arch }}" != x86_64 ]; then
+            echo "::error::sccache missing on the bare ${{ matrix.arch }} runner — provision klokka; CI must not install on bare hosts"; exit 1
+          fi
+          V=0.8.2
+          curl -sSL "https://github.com/mozilla/sccache/releases/download/v${V}/sccache-v${V}-x86_64-unknown-linux-musl.tar.gz" | tar -xz
+          sudo install -m0755 "sccache-v${V}-x86_64-unknown-linux-musl/sccache" /usr/local/bin/sccache
+          sccache --version
+
+      - name: Install cargo-binstall
+        run: |
+          if command -v cargo-binstall >/dev/null 2>&1; then exit 0; fi
+          if [ "${{ matrix.arch }}" != x86_64 ]; then
+            echo "::error::cargo-binstall missing on the bare ${{ matrix.arch }} runner — provision klokka"; exit 1
+          fi
+          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: |
+          if command -v cargo-leptos >/dev/null 2>&1; then exit 0; fi
+          if [ "${{ matrix.arch }}" != x86_64 ]; then
+            echo "::error::cargo-leptos missing on the bare ${{ matrix.arch }} runner — provision klokka"; exit 1
+          fi
+          cargo binstall cargo-leptos --locked --no-confirm
 
       - name: Build
         run: cargo leptos build --release
@@ -58,7 +88,7 @@ jobs:
       - name: Create release
         run: |
           curl -sX POST \
-            -H "Authorization: token ${{ secrets.GITEA_TOKEN }}" \
+            -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
             -H "Content-Type: application/json" \
             "${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases" \
             -d "{\"tag_name\":\"${{ gitea.ref_name }}\",\"name\":\"${{ gitea.ref_name }}\"}" \
@@ -67,24 +97,24 @@ jobs:
       - name: Upload assets
         run: |
           RELEASE_ID=$(curl -s \
-            -H "Authorization: token ${{ secrets.GITEA_TOKEN }}" \
+            -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
             "${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases/tags/${{ gitea.ref_name }}" \
             | jq -r '.id')
 
           for FILE in "${{ env.TARBALL }}" "${{ env.TARBALL }}.sha256"; do
             # Remove any existing asset with the same name so re-runs stay clean
             EXISTING=$(curl -s \
-              -H "Authorization: token ${{ secrets.GITEA_TOKEN }}" \
+              -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
               "${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases/${RELEASE_ID}/assets" \
               | jq -r ".[] | select(.name == \"${FILE}\") | .id")
             for AID in $EXISTING; do
               curl -sX DELETE \
-                -H "Authorization: token ${{ secrets.GITEA_TOKEN }}" \
+                -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
                 "${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases/${RELEASE_ID}/assets/${AID}"
             done
 
             curl -sX POST \
-              -H "Authorization: token ${{ secrets.GITEA_TOKEN }}" \
+              -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
               -H "Content-Type: application/octet-stream" \
               "${{ gitea.server_url }}/api/v1/repos/${{ gitea.repository }}/releases/${RELEASE_ID}/assets?name=${FILE}" \
               --data-binary "@${FILE}" --fail-with-body
@@ -94,7 +124,7 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: builtin:checkout
 
       - name: Log in to Docker Hub
         run: echo "${{ secrets.DOCKERHUB_TOKEN }}" | docker login -u bendik --password-stdin
@@ -111,7 +141,7 @@ jobs:
     runs-on: aarch64
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: builtin:checkout
 
       - name: Compute checksums and update PKGBUILD
         run: |
@@ -125,6 +155,32 @@ jobs:
           sed -i "s/sha256sums_x86_64=('.*')/sha256sums_x86_64=('${SUM_X86}')/" aur/PKGBUILD
           sed -i "s/sha256sums_aarch64=('.*')/sha256sums_aarch64=('${SUM_AARCH}')/" aur/PKGBUILD
 
+      # Also publish the built packages to this instance's Arch registry
+      # (docs.gitea.com/usage/packages/arch). The PKGBUILD only repacks the
+      # release tarballs, so CARCH can produce both architectures from this
+      # one host. Consumers: see the infrastructure README.
+      # Best-effort mirror to the instance Arch registry. The ephemeral
+      # GITHUB_TOKEN is not accepted as a package-write credential, so this
+      # uses a dedicated REGISTRY_TOKEN secret (a write:package token for bl);
+      # if it is unset the step is skipped, and continue-on-error keeps a
+      # registry hiccup from failing the release or the AUR push.
+      - name: Publish to the Arch package registry
+        continue-on-error: true
+        run: |
+          set -euo pipefail
+          if [ -z "${{ secrets.REGISTRY_TOKEN }}" ]; then
+            echo "::warning::REGISTRY_TOKEN not set — skipping Arch registry publish"
+            exit 0
+          fi
+          cd aur
+          for carch in aarch64 x86_64; do
+            pkgfile=$(CARCH="$carch" makepkg --packagelist | tail -1)
+            CARCH="$carch" makepkg -f --nodeps --noconfirm --skipinteg
+            curl --fail-with-body --user "bl:${{ secrets.REGISTRY_TOKEN }}" \
+              --upload-file "$pkgfile" \
+              "${{ gitea.server_url }}/api/packages/${{ gitea.repository_owner }}/arch/uhhm"
+          done
+
       - name: Push to AUR
         env:
           AUR_SSH_KEY: ${{ secrets.AUR_SSH_KEY }}

Cargo.lock

@@ -345,15 +345,17 @@ dependencies = [
 
 [[package]]
 name = "cnats"
-version = "0.1.0"
+version = "0.3.0"
 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.3.0"
 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,35 @@ 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",
+    "RtcPeerConnectionState",
+    "DisplayMediaStreamConstraints",
+    "RtcTrackEvent",
+    "RtcRtpTransceiver",
+    "RtcOfferOptions",
+    "MediaStream",
+    "MediaStreamConstraints",
+    "MediaStreamTrack",
+    "MediaDevices",
+    "Navigator",
+    "HtmlVideoElement",
 ], optional = true }
 
 [features]
@@ -51,6 +81,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 +97,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",

README.md

@@ -110,6 +110,8 @@ src/
   app.rs           Leptos UI (login gate + chat console)
   auth.rs          shared User type + current_user server fn
   chat.rs          shared ChatMessage/rooms + send_message server fn (publishes to NATS)
+  call.rs          call signaling types + send_signal server fn (call.room.*, never archived)
+  webrtc.rs        browser-only mesh WebRTC: peers, mic/cam toggles, screen share
   server/
     oidc.rs        Kanidm OIDC login/callback/logout handlers
     sse.rs         NATS → browser SSE bridge (one subscription per client)
@@ -117,6 +119,30 @@ src/
 style/main.css     the console theme
 ```
 
+## Calls
+
+The `lobby` room has a mesh WebRTC call (every pair of browsers connects
+directly; fine for a handful of people, not more). Signaling rides NATS on
+`call.room.<room>` via `/call-sse/<room>` and the `send_signal` server fn, so
+it is never archived. Only a public STUN server is configured: calls across
+symmetric NATs will not connect without a TURN relay, which this app does not
+run.
+
+In a call:
+
+- **MIC / CAM** toggle your mic and camera (the track stays attached and sends
+  silence/black, so toggling is instant and needs no renegotiation).
+- **SCR ▶ share** shares a screen, window, or tab. It swaps the outgoing video
+  track on every peer with `replaceTrack`; your camera comes back when you
+  stop, or when the browser's own "Stop sharing" bar is used.
+- Tiles show the participant's name and MIC ✕ / CAM ✕ / SCR badges, kept in
+  sync by a `Status` signal on the same subject.
+- **Click** a tile to pin it as the stage (a shared screen pins itself);
+  **double-click** for fullscreen.
+- A peer whose tab closes without leaving is dropped once their connection
+  fails, and ICE candidates that arrive before the offer are buffered rather
+  than lost.
+
 ## Notes & production hardening
 
 - Sessions are in-memory (`tower-sessions` `MemoryStore`): restart logs

aur/PKGBUILD

@@ -1,10 +1,11 @@
 # Maintainer: Bendik Aagaard Lynghaug <bendik.lynghaug@gmail.com>
 pkgname=cnats
-pkgver=0.1.0
+pkgver=0.3.0
 pkgrel=1
 pkgdesc="Web chat over NATS subjects with Kanidm SSO (Leptos SSR)"
 arch=('x86_64' 'aarch64')
-url="https://prosjekt.klingenbergbygg.no/bl/cnats"
+options=('!strip')
+url="https://project.uhhm.no/bl/cnats"
 license=('MIT')
 depends=('glibc' 'gcc-libs')
 optdepends=(
@@ -14,8 +15,8 @@ optdepends=(
 provides=('cnats')
 conflicts=('cnats-git' 'cnats-bin')
 backup=('etc/cnats/env')
-source_x86_64=("cnats-v${pkgver}-x86_64.tar.gz::https://prosjekt.klingenbergbygg.no/bl/cnats/releases/download/v${pkgver}/cnats-v${pkgver}-x86_64.tar.gz")
+source_x86_64=("cnats-v${pkgver}-x86_64.tar.gz::https://project.uhhm.no/bl/cnats/releases/download/v${pkgver}/cnats-v${pkgver}-x86_64.tar.gz")
-source_aarch64=("cnats-v${pkgver}-aarch64.tar.gz::https://prosjekt.klingenbergbygg.no/bl/cnats/releases/download/v${pkgver}/cnats-v${pkgver}-aarch64.tar.gz")
+source_aarch64=("cnats-v${pkgver}-aarch64.tar.gz::https://project.uhhm.no/bl/cnats/releases/download/v${pkgver}/cnats-v${pkgver}-aarch64.tar.gz")
 sha256sums_x86_64=('SKIP')
 sha256sums_aarch64=('SKIP')
 

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,456 @@ 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 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;
+                });
+            }
+        });
+
+        let on_join = move |_| {
+            let cs = call_state.get_value();
+            leptos::task::spawn_local(async move {
+                cs.join().await;
+            });
+        };
+
+        view! {
+            <div class="call-panel">
+                {move || {
+                    if in_call.get() {
+                        view! { <CallActive call_state=call_state.get_value()/> }.into_any()
+                    } else {
+                        view! {
+                            <button class="call-join" title="join call" aria-label="join call" on:click=on_join>
+                                "☎"
+                            </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" title="join call" aria-label="join call" disabled=true>
+                    "☎"
+                </button>
+            </div>
+        }
+            .into_any()
+    }
+}
+
+/// The in-call subtree (video grid + control bar), 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`.
+///
+/// Layout: a flat grid of equal tiles, or - once a tile is pinned (click)
+/// - that tile as a big "stage" with the rest in a strip below. A peer
+/// starting a screen share gets pinned automatically; nobody shares a
+/// screen to have it shown at thumbnail size.
+#[cfg(feature = "hydrate")]
+#[component]
+fn CallActive(call_state: crate::webrtc::CallState) -> impl IntoView {
+    let me = call_state.me().to_string();
+    let status = call_state.status;
+    let pinned = RwSignal::new(None::<String>);
+    let cs = StoredValue::new_local(call_state.clone());
+
+    // Call timer. The interval is cleared when this subtree unmounts (leave).
+    let elapsed = RwSignal::new(0u64);
+    {
+        let joined_at = call_state.joined_at;
+        let handle = set_interval_with_handle(
+            move || {
+                let secs = ((js_sys::Date::now() - joined_at.get_untracked()) / 1000.0).max(0.0);
+                elapsed.set(secs as u64);
+            },
+            std::time::Duration::from_secs(1),
+        );
+        on_cleanup(move || {
+            if let Ok(h) = handle {
+                h.clear();
+            }
+        });
+    }
+
+    // Auto-pin whoever starts sharing, and release that pin (only that
+    // one - a manual pin is left alone) when they stop.
+    let auto_pinned = RwSignal::new(false);
+    {
+        let cs2 = call_state.clone();
+        Effect::new(move |_| {
+            let sharing = cs2
+                .peer_views()
+                .into_iter()
+                .find(|p| p.status.get().screen)
+                .map(|p| p.id);
+            match sharing {
+                Some(id) => {
+                    if pinned.get_untracked().as_deref() != Some(id.as_str()) {
+                        pinned.set(Some(id));
+                        auto_pinned.set(true);
+                    }
+                }
+                None => {
+                    if auto_pinned.get_untracked() {
+                        pinned.set(None);
+                        auto_pinned.set(false);
+                    }
+                }
+            }
+        });
+    }
+    // Drop a pin pointing at a peer who has left.
+    {
+        let cs2 = call_state.clone();
+        Effect::new(move |_| {
+            let ids: Vec<String> = cs2.peer_views().into_iter().map(|p| p.id).collect();
+            if let Some(p) = pinned.get_untracked() {
+                if p != me && !ids.contains(&p) {
+                    pinned.set(None);
+                }
+            }
+        });
+    }
+
+    let participants = Memo::new(move |_| cs.get_value().peer_views().len() + 1);
+    let me_id = call_state.me().to_string();
+    let me_for_tile = me_id.clone();
+
+    view! {
+        <div class="call-active">
+            <div class="video-grid" class:staged=move || pinned.get().is_some()>
+                <VideoTile
+                    id=me_for_tile
+                    label="you".to_string()
+                    local=true
+                    stream=Signal::derive(move || cs.get_value().preview_stream())
+                    status=status.into()
+                    pinned=pinned
+                    auto_pinned=auto_pinned
+                />
+                <For
+                    each=move || cs.get_value().peer_views()
+                    key=|p| p.id.clone()
+                    children=move |p| {
+                        view! {
+                            <VideoTile
+                                id=p.id.clone()
+                                label=format!("@{}", p.id)
+                                local=false
+                                stream=p.stream.into()
+                                status=p.status.into()
+                                pinned=pinned
+                                auto_pinned=auto_pinned
+                            />
+                        }
+                    }
+                />
+            </div>
+            <div class="call-bar">
+                <span class="call-meta">
+                    <span class="led live"></span>
+                    <span class="call-timer">{move || format_elapsed(elapsed.get())}</span>
+                    <span class="call-count" title="participants">
+                        <CallIcon kind=CallIconKind::People off=Signal::derive(|| false)/>
+                        {move || participants.get().to_string()}
+                    </span>
+                </span>
+                <span class="call-ctls">
+                    <button
+                        class="call-ctl"
+                        class:live=move || status.get().mic
+                        title="microphone"
+                        aria-label="microphone"
+                        aria-pressed=move || status.get().mic.to_string()
+                        on:click=move |_| cs.get_value().toggle_mic()
+                    >
+                        <span class="led"></span>
+                        <CallIcon kind=CallIconKind::Mic off=Signal::derive(move || !status.get().mic)/>
+                    </button>
+                    <button
+                        class="call-ctl"
+                        class:live=move || status.get().cam
+                        title="camera"
+                        aria-label="camera"
+                        aria-pressed=move || status.get().cam.to_string()
+                        on:click=move |_| cs.get_value().toggle_cam()
+                    >
+                        <span class="led"></span>
+                        <CallIcon kind=CallIconKind::Cam off=Signal::derive(move || !status.get().cam)/>
+                    </button>
+                    <button
+                        class="call-ctl"
+                        class:on=move || status.get().screen
+                        title="share screen"
+                        aria-label="share screen"
+                        aria-pressed=move || status.get().screen.to_string()
+                        on:click=move |_| cs.get_value().toggle_screen_share()
+                    >
+                        <CallIcon kind=CallIconKind::Screen off=Signal::derive(|| false)/>
+                    </button>
+                    <button
+                        class="call-leave"
+                        title="leave call"
+                        aria-label="leave call"
+                        on:click=move |_| {
+                            let cs = cs.get_value();
+                            leptos::task::spawn_local(async move {
+                                cs.leave().await;
+                            });
+                        }
+                    >
+                        "⏏"
+                    </button>
+                </span>
+            </div>
+        </div>
+    }
+}
+
+#[cfg(feature = "hydrate")]
+fn format_elapsed(secs: u64) -> String {
+    let (h, m, s) = (secs / 3600, (secs % 3600) / 60, secs % 60);
+    if h > 0 {
+        format!("{h}:{m:02}:{s:02}")
+    } else {
+        format!("{m:02}:{s:02}")
+    }
+}
+
+#[cfg(feature = "hydrate")]
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum CallIconKind {
+    Mic,
+    Cam,
+    Screen,
+    People,
+}
+
+/// Inline SVG glyphs for the call controls and tile badges - stroked in
+/// `currentColor` like `PulseMark`, so they take whatever colour the
+/// surrounding control has. `off` draws a diagonal strike through the
+/// glyph (muted / camera off).
+#[cfg(feature = "hydrate")]
+#[component]
+fn CallIcon(kind: CallIconKind, off: Signal<bool>) -> impl IntoView {
+    let body = match kind {
+        CallIconKind::Mic => view! {
+            <rect x="9" y="3" width="6" height="11" rx="3"></rect>
+            <path d="M5 11a7 7 0 0 0 14 0M12 18v3M9 21h6"></path>
+        }
+        .into_any(),
+        CallIconKind::Cam => view! {
+            <rect x="3" y="7" width="13" height="10" rx="2"></rect>
+            <path d="M16 11l5-3v8l-5-3z"></path>
+        }
+        .into_any(),
+        CallIconKind::Screen => view! {
+            <rect x="3" y="4" width="18" height="12" rx="2"></rect>
+            <path d="M8 20h8M12 16v4"></path>
+        }
+        .into_any(),
+        CallIconKind::People => view! {
+            <circle cx="9" cy="8" r="3.5"></circle>
+            <path d="M3 20v-1a6 6 0 0 1 12 0v1M16 5a3.5 3.5 0 0 1 0 7M21 20v-1a6 6 0 0 0-4-5.6"></path>
+        }
+        .into_any(),
+    };
+    view! {
+        <svg
+            class="call-icon"
+            class:off=move || off.get()
+            viewBox="0 0 24 24"
+            fill="none"
+            stroke="currentColor"
+            stroke-width="2"
+            stroke-linecap="round"
+            stroke-linejoin="round"
+            aria-hidden="true"
+        >
+            {body}
+            <Show when=move || off.get()>
+                <path class="call-icon-strike-bg" d="M3 3l18 18"></path>
+                <path class="call-icon-strike" d="M3 3l18 18"></path>
+            </Show>
+        </svg>
+    }
+}
+
+/// One participant's tile: the `<video>` plus a name label and mic/cam/
+/// screen badges. Click pins it as the stage; double-click goes
+/// fullscreen (handy for reading a shared screen). A child component so
+/// each tile owns its `NodeRef`/effect pair instead of the parent juggling
+/// a `Vec` of node refs by hand.
+#[cfg(feature = "hydrate")]
+#[component]
+fn VideoTile(
+    id: String,
+    label: String,
+    local: bool,
+    stream: Signal<Option<web_sys::MediaStream>>,
+    status: Signal<crate::call::MediaStatus>,
+    pinned: RwSignal<Option<String>>,
+    auto_pinned: RwSignal<bool>,
+) -> impl IntoView {
+    let video_ref = NodeRef::<leptos::html::Video>::new();
+
+    // Mirror the MediaStream into the <video> - `srcObject` has no HTML
+    // attribute form, has to be set via JS.
+    //
+    // For the local tile also force `.muted` via the JS property, not just
+    // the `muted` attribute: every tile here is built client-side via
+    // `document.createElement` (never parsed from HTML), and browsers only
+    // seed the live `.muted` property from the attribute for
+    // parser-inserted elements. Without this the local preview plays the
+    // user's own mic back through their speakers.
+    Effect::new(move |_| {
+        let s = stream.get();
+        if let Some(el) = video_ref.get() {
+            if local {
+                el.set_muted(true);
+            }
+            el.set_src_object(s.as_ref());
+        }
+    });
+
+    let pin_id = id.clone();
+    let is_pinned = Memo::new(move |_| pinned.get().as_deref() == Some(pin_id.as_str()));
+    let click_id = id.clone();
+    let on_click = move |_| {
+        auto_pinned.set(false);
+        pinned.update(|p| {
+            if p.as_deref() == Some(click_id.as_str()) {
+                *p = None;
+            } else {
+                *p = Some(click_id.clone());
+            }
+        });
+    };
+    let on_dblclick = move |_| {
+        if let Some(el) = video_ref.get() {
+            let _ = el.request_fullscreen();
+        }
+    };
+
+    view! {
+        <div
+            class="tile"
+            class:local=local
+            class:pinned=move || is_pinned.get()
+            class:screen=move || status.get().screen
+            class:cam-off=move || !status.get().cam && !status.get().screen
+            on:click=on_click
+            on:dblclick=on_dblclick
+            title="click to pin · double-click for fullscreen"
+        >
+            <video
+                class="video-tile"
+                node_ref=video_ref
+                autoplay=true
+                muted=local
+                playsinline=true
+            ></video>
+            <span class="tile-label">{label}</span>
+            <span class="tile-flags">
+                <Show when=move || status.get().screen>
+                    <span class="flag on" title="sharing screen">
+                        <CallIcon kind=CallIconKind::Screen off=Signal::derive(|| false)/>
+                    </span>
+                </Show>
+                <Show when=move || !status.get().mic>
+                    <span class="flag" title="muted">
+                        <CallIcon kind=CallIconKind::Mic off=Signal::derive(|| true)/>
+                    </span>
+                </Show>
+                <Show when=move || !status.get().cam && !status.get().screen>
+                    <span class="flag" title="camera off">
+                        <CallIcon kind=CallIconKind::Cam off=Signal::derive(|| true)/>
+                    </span>
+                </Show>
+            </span>
+        </div>
+    }
+}
+
+/// 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,113 @@
+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),
+    /// Mic/camera/screen state, so other tiles can show "muted" or "sharing
+    /// screen" without sniffing tracks. Broadcast on every change, and sent
+    /// directly to each newcomer when their `Join` arrives (they don't
+    /// know anything about us yet).
+    Status(MediaStatus),
+}
+
+/// What a participant is currently sending. `mic`/`cam` are the
+/// `MediaStreamTrack.enabled` flags (muted tracks keep flowing as silence /
+/// black, which is what lets un-muting be instant and renegotiation-free);
+/// `screen` means their outgoing video track is a display capture rather
+/// than the camera.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct MediaStatus {
+    pub mic: bool,
+    pub cam: bool,
+    pub screen: bool,
+}
+
+impl Default for MediaStatus {
+    fn default() -> Self {
+        Self {
+            mic: true,
+            cam: true,
+            screen: false,
+        }
+    }
+}
+
+/// 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/server/store.rs

@@ -2,7 +2,7 @@
 //! into Postgres, so history survives restarts and includes messages
 //! published by any client on the bus (not just this app).
 
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use async_nats::jetstream;
 use futures::StreamExt;
@@ -38,11 +38,26 @@ pub async fn init_schema(pool: &PgPool) -> anyhow::Result<()> {
 /// Runs forever; (re)creates the stream/consumer and retries on any failure,
 /// so a NATS or Postgres outage never takes the chat server down.
 pub async fn run_consumer(nats: async_nats::Client, pool: PgPool) {
+    const MIN_BACKOFF: Duration = Duration::from_secs(5);
+    const MAX_BACKOFF: Duration = Duration::from_secs(60);
+    let mut backoff = MIN_BACKOFF;
     loop {
+        let started = Instant::now();
         if let Err(err) = consume(&nats, &pool).await {
-            tracing::error!("archive consumer failed: {err:#}; retrying in 5s");
+            // A failure after a long healthy run is a fresh incident, not an
+            // escalating one - reset the backoff so we retry promptly.
+            if started.elapsed() >= MAX_BACKOFF {
+                backoff = MIN_BACKOFF;
+            }
+            tracing::error!(
+                "archive consumer failed after {:?}: {err:#}; retrying in {}s",
+                started.elapsed(),
+                backoff.as_secs()
+            );
+            tokio::time::sleep(backoff).await;
+            // Cap the backoff so a persistent outage doesn't hammer NATS/PG.
+            backoff = (backoff * 2).min(MAX_BACKOFF);
         }
-        tokio::time::sleep(Duration::from_secs(5)).await;
     }
 }
 

src/webrtc.rs

@@ -0,0 +1,613 @@
+//! 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.
+//!
+//! Screen sharing swaps the outgoing video track on every peer's
+//! `RTCRtpSender` via `replaceTrack` - same m-line, no renegotiation, so
+//! the signaling path never sees it. Peers only learn about it through the
+//! `Status` broadcast (for the "sharing" badge); the video just changes.
+#![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::{
+    DisplayMediaStreamConstraints, MediaStream, MediaStreamConstraints, MediaStreamTrack,
+    RtcConfiguration, RtcIceCandidateInit, RtcIceServer, RtcPeerConnection,
+    RtcPeerConnectionState, RtcRtpSender, RtcSdpType, RtcSessionDescriptionInit,
+};
+
+use crate::call::{send_signal, CallSignalKind, MediaStatus};
+
+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>>,
+    status: RwSignal<MediaStatus>,
+    /// ICE candidates that arrived before `setRemoteDescription` resolved.
+    /// Offer and candidates travel as separate server-fn POSTs, so nothing
+    /// guarantees their order on arrival; `addIceCandidate` before a remote
+    /// description throws, and the candidate would be lost - a call that
+    /// then "just doesn't connect". Buffered here, flushed once the remote
+    /// description is in.
+    pending_ice: Vec<RtcIceCandidateInit>,
+    remote_description_set: bool,
+}
+
+/// A remote peer as the UI sees it: id (username), stream, status.
+#[derive(Clone)]
+pub struct PeerView {
+    pub id: String,
+    pub stream: RwSignal<Option<MediaStream>>,
+    pub status: RwSignal<MediaStatus>,
+}
+
+/// 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,
+    /// Camera + mic from `getUserMedia`. Also the stream every outgoing
+    /// track is grouped under (`addTrack(track, stream)`), including the
+    /// screen track, so the remote's `ontrack` always sees one stream.
+    local_stream: RwSignal<Option<MediaStream>>,
+    /// Display capture while sharing; `None` otherwise.
+    screen_stream: RwSignal<Option<MediaStream>>,
+    /// `RwSignal` (not `StoredValue`) so the tile list re-renders when
+    /// peers come and go. `LocalStorage` because `RtcPeerConnection` is
+    /// `!Send`.
+    peers: RwSignal<HashMap<String, Peer>, LocalStorage>,
+    pub in_call: RwSignal<bool>,
+    pub status: RwSignal<MediaStatus>,
+    /// `Date.now()` at join, for the call timer.
+    pub joined_at: RwSignal<f64>,
+}
+
+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>()
+}
+
+async fn get_display_stream() -> Result<MediaStream, JsValue> {
+    let window = web_sys::window().ok_or("no window")?;
+    let media_devices = window.navigator().media_devices()?;
+    let constraints = DisplayMediaStreamConstraints::new();
+    constraints.set_video(&JsValue::TRUE);
+    let promise = media_devices.get_display_media_with_constraints(&constraints)?;
+    let stream = JsFuture::from(promise).await?;
+    stream.dyn_into::<MediaStream>()
+}
+
+fn tracks(stream: &MediaStream) -> Vec<MediaStreamTrack> {
+    stream
+        .get_tracks()
+        .iter()
+        .filter_map(|t| t.dyn_into::<MediaStreamTrack>().ok())
+        .collect()
+}
+
+fn first_video_track(stream: &MediaStream) -> Option<MediaStreamTrack> {
+    tracks(stream).into_iter().find(|t| t.kind() == "video")
+}
+
+fn stop_all(stream: &MediaStream) {
+    for track in tracks(stream) {
+        track.stop();
+    }
+}
+
+/// The sender carrying our outgoing video on this connection, if any.
+fn video_sender(pc: &RtcPeerConnection) -> Option<RtcRtpSender> {
+    pc.get_senders()
+        .iter()
+        .filter_map(|s| s.dyn_into::<RtcRtpSender>().ok())
+        .find(|s| s.track().map(|t| t.kind() == "video").unwrap_or(false))
+}
+
+/// 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),
+            screen_stream: RwSignal::new(None),
+            peers: RwSignal::new_local(HashMap::new()),
+            in_call: RwSignal::new(false),
+            status: RwSignal::new(MediaStatus::default()),
+            joined_at: RwSignal::new(0.0),
+        }
+    }
+
+    pub fn me(&self) -> &str {
+        &self.me
+    }
+
+    /// What the local preview tile should show: the screen while sharing,
+    /// the camera otherwise.
+    pub fn preview_stream(&self) -> Option<MediaStream> {
+        self.screen_stream.get().or_else(|| self.local_stream.get())
+    }
+
+    /// Currently-known peers, keyed by their peer id (username). Tracked:
+    /// re-runs whoever reads it when a peer joins or leaves.
+    pub fn peer_views(&self) -> Vec<PeerView> {
+        let mut views: Vec<PeerView> = self.peers.with(|p| {
+            p.iter()
+                .map(|(id, peer)| PeerView {
+                    id: id.clone(),
+                    stream: peer.stream,
+                    status: peer.status,
+                })
+                .collect()
+        });
+        // HashMap order is arbitrary; keep tiles from shuffling on re-render.
+        views.sort_by(|a, b| a.id.cmp(&b.id));
+        views
+    }
+
+    /// 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.status.set(MediaStatus::default());
+        self.joined_at.set(js_sys::Date::now());
+        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/screen), and tells the room we're gone.
+    pub async fn leave(&self) {
+        self.peers.update(|peers| {
+            for (_, peer) in peers.drain() {
+                peer.conn.close();
+            }
+        });
+        if let Some(stream) = self.screen_stream.get_untracked() {
+            stop_all(&stream);
+        }
+        if let Some(stream) = self.local_stream.get_untracked() {
+            stop_all(&stream);
+        }
+        self.screen_stream.set(None);
+        self.local_stream.set(None);
+        self.in_call.set(false);
+        let _ = send_signal(self.room.clone(), None, CallSignalKind::Leave).await;
+    }
+
+    // -- local media controls ------------------------------------------------
+
+    fn broadcast_status(&self, to: Option<String>) {
+        let room = self.room.clone();
+        let status = self.status.get_untracked();
+        leptos::task::spawn_local(async move {
+            let _ = send_signal(room, to, CallSignalKind::Status(status)).await;
+        });
+    }
+
+    fn set_local_tracks_enabled(&self, kind: &str, enabled: bool) {
+        if let Some(stream) = self.local_stream.get_untracked() {
+            for track in tracks(&stream).into_iter().filter(|t| t.kind() == kind) {
+                track.set_enabled(enabled);
+            }
+        }
+    }
+
+    /// Mute/unmute the mic. Flips `enabled` on the audio track (sends
+    /// silence) rather than removing it - instant, and no renegotiation.
+    pub fn toggle_mic(&self) {
+        let mic = !self.status.get_untracked().mic;
+        self.set_local_tracks_enabled("audio", mic);
+        self.status.update(|s| s.mic = mic);
+        self.broadcast_status(None);
+    }
+
+    /// Camera on/off. Only affects the camera track - while screen sharing
+    /// it's a no-op for peers until sharing stops, which is what you'd
+    /// expect ("my camera is off" shouldn't black out the slides).
+    pub fn toggle_cam(&self) {
+        let cam = !self.status.get_untracked().cam;
+        self.set_local_tracks_enabled("video", cam);
+        self.status.update(|s| s.cam = cam);
+        self.broadcast_status(None);
+    }
+
+    pub fn toggle_screen_share(&self) {
+        if self.status.get_untracked().screen {
+            self.stop_screen_share();
+        } else {
+            let this = self.clone();
+            leptos::task::spawn_local(async move {
+                this.start_screen_share().await;
+            });
+        }
+    }
+
+    /// Swap every peer's outgoing video track for a track from `stream`
+    /// (or back to the camera when `stream` is the camera stream).
+    fn replace_outgoing_video(&self, track: Option<&MediaStreamTrack>) {
+        self.peers.with_untracked(|peers| {
+            for peer in peers.values() {
+                if let Some(sender) = video_sender(&peer.conn) {
+                    let _ = sender.replace_track(track);
+                }
+            }
+        });
+    }
+
+    async fn start_screen_share(&self) {
+        let stream = match get_display_stream().await {
+            Ok(s) => s,
+            // Most often: the user dismissed the picker. Not an error worth
+            // surfacing.
+            Err(e) => {
+                leptos::logging::log!("getDisplayMedia declined: {e:?}");
+                return;
+            }
+        };
+        // User may have left while the picker was up.
+        if !self.in_call.get_untracked() {
+            stop_all(&stream);
+            return;
+        }
+        let Some(track) = first_video_track(&stream) else {
+            stop_all(&stream);
+            return;
+        };
+
+        // Browsers put their own "Stop sharing" affordance on screen
+        // capture; when the user hits it the track ends under us and we
+        // have to fall back to the camera ourselves.
+        {
+            let this = self.clone();
+            let onended = Closure::<dyn FnMut()>::new(move || {
+                if this.status.get_untracked().screen {
+                    this.stop_screen_share();
+                }
+            });
+            track.set_onended(Some(onended.as_ref().unchecked_ref()));
+            onended.forget();
+        }
+
+        self.replace_outgoing_video(Some(&track));
+        self.screen_stream.set(Some(stream));
+        self.status.update(|s| s.screen = true);
+        self.broadcast_status(None);
+    }
+
+    fn stop_screen_share(&self) {
+        let camera_track = self
+            .local_stream
+            .get_untracked()
+            .as_ref()
+            .and_then(first_video_track);
+        self.replace_outgoing_video(camera_track.as_ref());
+        if let Some(stream) = self.screen_stream.get_untracked() {
+            stop_all(&stream);
+        }
+        self.screen_stream.set(None);
+        self.status.update(|s| s.screen = false);
+        self.broadcast_status(None);
+    }
+
+    // -- signaling -----------------------------------------------------------
+
+    /// 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,
+                // and tell them what we're sending (they know nothing yet).
+                self.start_offer(from.clone());
+                self.broadcast_status(Some(from));
+            }
+            CallSignalKind::Leave => self.remove_peer(&from),
+            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)
+            }
+            CallSignalKind::Status(status) => {
+                // May arrive before their offer does - `ensure_peer` is
+                // idempotent, so just create the slot early.
+                if self.ensure_peer(&from).is_some() {
+                    self.peers.with_untracked(|peers| {
+                        if let Some(peer) = peers.get(&from) {
+                            peer.status.set(status);
+                        }
+                    });
+                }
+            }
+        }
+    }
+
+    fn remove_peer(&self, peer_id: &str) {
+        self.peers.update(|peers| {
+            if let Some(peer) = peers.remove(peer_id) {
+                peer.conn.close();
+            }
+        });
+    }
+
+    fn ensure_peer(&self, peer_id: &str) -> Option<RtcPeerConnection> {
+        if let Some(pc) = self
+            .peers
+            .with_untracked(|peers| peers.get(peer_id).map(|p| p.conn.clone()))
+        {
+            return Some(pc);
+        }
+
+        let pc = new_peer_connection().ok()?;
+        let local = self.local_stream.get_untracked()?;
+
+        // Audio from the camera stream; video is whatever we're currently
+        // showing - the screen if a share is in progress, else the camera.
+        for track in tracks(&local).into_iter().filter(|t| t.kind() == "audio") {
+            pc.add_track_0(&track, &local);
+        }
+        let video = self
+            .screen_stream
+            .get_untracked()
+            .as_ref()
+            .and_then(first_video_track)
+            .or_else(|| first_video_track(&local));
+        if let Some(track) = video {
+            pc.add_track_0(&track, &local);
+        }
+
+        let remote_stream = RwSignal::new(None::<MediaStream>);
+        {
+            let ontrack = Closure::<dyn FnMut(web_sys::RtcTrackEvent)>::new(move |ev: web_sys::RtcTrackEvent| {
+                if let Ok(stream) = ev.streams().get(0).dyn_into::<MediaStream>() {
+                    remote_stream.set(Some(stream));
+                }
+            });
+            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();
+        }
+
+        // A peer whose tab just closed never sends `Leave`; drop them when
+        // the transport gives up so their tile doesn't linger forever.
+        {
+            let this = self.clone();
+            let pc2 = pc.clone();
+            let peer_id = peer_id.to_string();
+            let onstate = Closure::<dyn FnMut()>::new(move || {
+                if matches!(
+                    pc2.connection_state(),
+                    RtcPeerConnectionState::Failed | RtcPeerConnectionState::Closed
+                ) {
+                    this.remove_peer(&peer_id);
+                }
+            });
+            pc.set_onconnectionstatechange(Some(onstate.as_ref().unchecked_ref()));
+            onstate.forget();
+        }
+
+        self.peers.update(|peers| {
+            peers.insert(
+                peer_id.to_string(),
+                Peer {
+                    conn: pc.clone(),
+                    stream: remote_stream,
+                    status: RwSignal::new(MediaStatus::default()),
+                    pending_ice: Vec::new(),
+                    remote_description_set: false,
+                },
+            );
+        });
+        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;
+        });
+    }
+
+    /// Marks the peer's remote description as set and replays any ICE
+    /// candidates that arrived too early.
+    fn flush_pending_ice(&self, peer_id: &str, pc: &RtcPeerConnection) {
+        let pending = self.peers.try_update_untracked(|peers| {
+            peers.get_mut(peer_id).map(|peer| {
+                peer.remote_description_set = true;
+                std::mem::take(&mut peer.pending_ice)
+            })
+        });
+        for init in pending.flatten().unwrap_or_default() {
+            let pc = pc.clone();
+            leptos::task::spawn_local(async move {
+                let _ = JsFuture::from(
+                    pc.add_ice_candidate_with_opt_rtc_ice_candidate_init(Some(&init)),
+                )
+                .await;
+            });
+        }
+    }
+
+    fn handle_offer(&self, from: String, sdp: String) {
+        let Some(pc) = self.ensure_peer(&from) else {
+            return;
+        };
+        let this = self.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;
+            }
+            this.flush_pending_ice(&from, &pc);
+            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(this.room.clone(), Some(from), CallSignalKind::Answer(answer_sdp)).await;
+        });
+    }
+
+    fn handle_answer(&self, from: String, sdp: String) {
+        let Some(pc) = self
+            .peers
+            .with_untracked(|peers| peers.get(&from).map(|p| p.conn.clone()))
+        else {
+            return;
+        };
+        let this = self.clone();
+        leptos::task::spawn_local(async move {
+            let remote_desc = RtcSessionDescriptionInit::new(RtcSdpType::Answer);
+            remote_desc.set_sdp(&sdp);
+            if JsFuture::from(pc.set_remote_description(&remote_desc))
+                .await
+                .is_ok()
+            {
+                this.flush_pending_ice(&from, &pc);
+            }
+        });
+    }
+
+    fn handle_ice_candidate(&self, from: String, candidate_json: String) {
+        let Ok(parsed) = js_sys::JSON::parse(&candidate_json) else {
+            return;
+        };
+        let init: RtcIceCandidateInit = parsed.unchecked_into();
+
+        // Either queue it (remote description not in yet) or hand back the
+        // connection to add it to right now.
+        let ready = self.peers.try_update_untracked(|peers| {
+            let peer = peers.get_mut(&from)?;
+            if peer.remote_description_set {
+                Some(peer.conn.clone())
+            } else {
+                peer.pending_ice.push(init.clone());
+                None
+            }
+        });
+        let Some(pc) = ready.flatten() else {
+            return;
+        };
+        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

@@ -1,8 +1,10 @@
 /* ── cnats · message-bus console ─────────────────────────────────────────
-   dark phosphor terminal: deep green-black ground, mint signal, amber id.
+   phosphor terminal in two prints: dark (green-black ground, mint signal)
+   and light (green-tinted paper, forest signal), following the OS scheme.
    type: Archivo (UI voice) + IBM Plex Mono (wire voice). */
 
 :root {
+  color-scheme: light dark;
   --ink-0: #060a09;
   --ink-1: #0b1210;
   --ink-2: #101a17;
@@ -16,10 +18,34 @@
   --signal-dim: #2a8f6c;
   --amber: #ffb454;
   --alarm: #ff6b6b;
+  --scanline: rgba(255, 255, 255, 0.015);
+  --vignette: rgba(0, 0, 0, 0.45);
+  --card-shadow: rgba(0, 0, 0, 0.55);
   --mono: "IBM Plex Mono", ui-monospace, monospace;
   --sans: "Archivo", system-ui, sans-serif;
 }
 
+@media (prefers-color-scheme: light) {
+  :root {
+    --ink-0: #f3f6f4;
+    --ink-1: #eaf0ec;
+    --ink-2: #dfe8e2;
+    --ink-3: #d2ded6;
+    --line: #c3d2c9;
+    --line-hot: #a3bcae;
+    --text: #14211c;
+    --text-dim: #46584f;
+    --text-faint: #74887e;
+    --signal: #0b7a52;
+    --signal-dim: #2f8a66;
+    --amber: #a85f00;
+    --alarm: #c73f3f;
+    --scanline: rgba(6, 10, 9, 0.02);
+    --vignette: rgba(6, 10, 9, 0.06);
+    --card-shadow: rgba(20, 33, 28, 0.18);
+  }
+}
+
 * { margin: 0; padding: 0; box-sizing: border-box; }
 
 html, body { height: 100%; }
@@ -39,8 +65,8 @@ body::before {
   pointer-events: none;
   z-index: 999;
   background:
-    repeating-linear-gradient(0deg, rgba(255, 255, 255, 0.015) 0 1px, transparent 1px 3px),
+    repeating-linear-gradient(0deg, var(--scanline) 0 1px, transparent 1px 3px),
-    radial-gradient(ellipse 120% 90% at 50% 40%, transparent 55%, rgba(0, 0, 0, 0.45));
+    radial-gradient(ellipse 120% 90% at 50% 40%, transparent 55%, var(--vignette));
 }
 
 ::selection { background: var(--signal); color: var(--ink-0); }
@@ -53,7 +79,7 @@ body::before {
   place-items: center;
   padding: 2rem;
   background:
-    radial-gradient(ellipse 60% 45% at 50% 0%, rgba(78, 240, 177, 0.07), transparent 70%),
+    radial-gradient(ellipse 60% 45% at 50% 0%, color-mix(in srgb, var(--signal) 7%, transparent), transparent 70%),
     linear-gradient(var(--ink-0), var(--ink-1));
 }
 
@@ -69,7 +95,7 @@ body::before {
   background: linear-gradient(160deg, var(--ink-2), var(--ink-1) 60%);
   padding: 3rem 2.75rem 2.5rem;
   position: relative;
-  box-shadow: 0 40px 80px rgba(0, 0, 0, 0.55);
+  box-shadow: 0 40px 80px var(--card-shadow);
   animation: rise 0.5s cubic-bezier(0.2, 0.9, 0.3, 1) both;
 }
 
@@ -129,7 +155,8 @@ body::before {
 .gate-btn:hover {
   background: transparent;
   color: var(--signal);
-  box-shadow: 0 0 24px rgba(78, 240, 177, 0.25), inset 0 0 12px rgba(78, 240, 177, 0.08);
+  box-shadow: 0 0 24px color-mix(in srgb, var(--signal) 25%, transparent),
+    inset 0 0 12px color-mix(in srgb, var(--signal) 8%, transparent);
 }
 
 .gate-btn-glyph { font-size: 1.05rem; }
@@ -197,6 +224,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;
@@ -442,7 +485,7 @@ body::before {
 
 .composer:focus-within {
   border-color: var(--signal-dim);
-  box-shadow: 0 0 0 1px var(--signal-dim), 0 0 30px rgba(78, 240, 177, 0.08);
+  box-shadow: 0 0 0 1px var(--signal-dim), 0 0 30px color-mix(in srgb, var(--signal) 8%, transparent);
 }
 
 .composer-prompt {
@@ -480,6 +523,193 @@ 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: 1rem;
+  line-height: 1;
+  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;
+}
+
+.tile {
+  position: relative;
+  width: 200px;
+  height: 150px;
+  background: var(--ink-0);
+  border: 1px solid var(--line-hot);
+  cursor: pointer;
+  user-select: none;
+  transition: border-color 120ms;
+}
+
+.tile:hover { border-color: var(--signal-dim); }
+.tile.local { border-color: var(--signal-dim); }
+.tile.pinned { border-color: var(--signal); box-shadow: 0 0 0 1px var(--signal); }
+
+.video-tile {
+  display: block;
+  width: 100%;
+  height: 100%;
+  object-fit: cover;
+  background: var(--ink-0);
+}
+
+/* a shared screen is letterboxed, never cropped */
+.tile.screen .video-tile { object-fit: contain; }
+.tile.cam-off .video-tile { opacity: 0.25; }
+
+/* stage mode: the pinned tile fills the row, everyone else in a strip */
+.video-grid.staged .tile.pinned {
+  order: -1;
+  flex: 1 0 100%;
+  width: 100%;
+  height: auto;
+  aspect-ratio: 16 / 9;
+  max-height: 60vh;
+}
+
+.video-grid.staged .tile.pinned .video-tile { object-fit: contain; }
+
+.video-grid.staged .tile:not(.pinned) { width: 140px; height: 105px; }
+
+.tile-label,
+.tile-flags {
+  position: absolute;
+  font-family: var(--mono);
+  font-size: 0.66rem;
+  letter-spacing: 0.06em;
+  line-height: 1;
+  pointer-events: none;
+}
+
+.tile-label {
+  left: 0;
+  bottom: 0;
+  padding: 0.3rem 0.45rem;
+  color: var(--text);
+  background: color-mix(in srgb, var(--ink-0) 75%, transparent);
+}
+
+.tile-flags {
+  top: 0.3rem;
+  right: 0.3rem;
+  display: flex;
+  gap: 0.25rem;
+}
+
+.flag {
+  display: inline-flex;
+  padding: 0.22rem;
+  color: var(--text-dim);
+  background: color-mix(in srgb, var(--ink-0) 80%, transparent);
+  border: 1px solid var(--line-hot);
+}
+
+.flag .call-icon { width: 12px; height: 12px; }
+
+.flag.on { color: var(--ink-0); background: var(--signal); border-color: var(--signal); }
+
+/* control bar */
+
+.call-bar {
+  display: flex;
+  flex-wrap: wrap;
+  align-items: center;
+  justify-content: space-between;
+  gap: 0.6rem 1rem;
+  font-family: var(--mono);
+  font-size: 0.78rem;
+}
+
+.call-meta {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.6rem;
+  color: var(--text-dim);
+  letter-spacing: 0.06em;
+}
+
+.call-timer { color: var(--text); font-variant-numeric: tabular-nums; }
+.call-count { display: inline-flex; align-items: center; gap: 0.3rem; font-variant-numeric: tabular-nums; }
+.call-count .call-icon { width: 14px; height: 14px; }
+
+.call-ctls { display: inline-flex; flex-wrap: wrap; gap: 0.4rem; }
+
+.call-ctl {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.45rem;
+  color: var(--text);
+  background: none;
+  border: 1px solid var(--line-hot);
+  padding: 0.5rem 0.7rem;
+  cursor: pointer;
+  transition: background 120ms, border-color 120ms, color 120ms;
+}
+
+.call-ctl:hover { border-color: var(--signal-dim); }
+
+/* capture controls: plain when off, and when live a soft glow with the
+   same ticking LED as the bus status - "this is recording", not an alarm */
+.call-ctl .led { width: 6px; height: 6px; }
+.call-ctl.live {
+  border-color: var(--signal-dim);
+  box-shadow: 0 0 10px color-mix(in srgb, var(--signal) 30%, transparent);
+}
+.call-ctl.live .led {
+  background: var(--signal);
+  box-shadow: 0 0 8px var(--signal);
+  animation: blink 1.4s steps(1) infinite;
+}
+
+.call-ctl.on { color: var(--ink-0); background: var(--signal); border-color: var(--signal); }
+
+.call-icon { display: block; width: 18px; height: 18px; }
+
+/* strike: a background stroke in the control's own fill colour first, so
+   the slash reads as cutting the glyph rather than overlapping it */
+.call-icon-strike-bg { stroke: var(--ink-1); stroke-width: 5; }
+.call-ctl.on .call-icon-strike-bg { stroke: var(--signal); }
+.flag .call-icon-strike-bg { stroke: var(--ink-0); }
+
+.call-leave {
+  font-family: var(--mono);
+  font-weight: 600;
+  font-size: 0.95rem;
+  line-height: 18px;
+  color: var(--text);
+  background: none;
+  border: 1px solid var(--alarm);
+  padding: 0.5rem 0.8rem;
+  cursor: pointer;
+  transition: background 120ms;
+}
+
+.call-leave:hover { background: color-mix(in srgb, var(--alarm) 12%, transparent); }
+
 /* motion */
 
 @keyframes rise {
@@ -518,6 +748,8 @@ body::before {
   .room-link { border-left: 0; border-bottom: 2px solid transparent; }
   .room-link.active { border-bottom-color: var(--signal); }
   .msg { grid-template-columns: auto 1fr; grid-template-rows: auto auto; }
+  .tile, .video-grid.staged .tile:not(.pinned) { width: calc(50% - 0.3rem); height: auto; aspect-ratio: 4 / 3; }
+  .video-grid.staged .tile.pinned { width: 100%; }
   .msg-time { grid-row: 1; order: 2; justify-self: end; }
   .msg-text { grid-column: 1 / -1; }
 }