4 Commits

Author SHA1 Message Date
bl 4cc0103448 chore: Release cnats version 0.2.0
Release / build (x86_64, ubuntu-latest) (push) Failing after 17m18s
Release / docker (push) Failing after 18m48s
Release / build (aarch64, aarch64) (push) Failing after 1h17m53s
Release / update-aur (push) Has been skipped
2026-07-27 23:55:00 +02:00
bl 650ed50c21 Kanidm group-gated rooms and minimal mesh calling
Rooms can now require a Kanidm group (via the `groups` OIDC claim,
mapped by `oauth2 update-claim-map` server-side) - dev/ops require
`developers`, enforced at every message path (send, history, SSE).

Adds a minimal WebRTC mesh call feature scoped to the lobby room,
signaled over a separate `call.room.*` NATS subject kept out of the
chat archive: public STUN only, no TURN, no SFU - small groups on
friendly networks, by design.
2026-07-27 23:52:27 +02:00
bl 46bdad1629 Pass NATS credentials explicitly: async-nats ignores userinfo in the URL
async_nats::connect() silently drops user:pass embedded in NATS_URL,
so the server rejected every connection with an authorization violation.
Parse the URL and feed credentials through ConnectOptions instead.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-18 22:33:20 +02:00
bl b4dfe0c5ea Add cargo-release config matching gpupaper release flow
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-09 14:44:03 +02:00
14 changed files with 904 additions and 30 deletions
Generated
+5 -1
View File
@@ -345,15 +345,17 @@ dependencies = [
[[package]]
name = "cnats"
version = "0.1.0"
version = "0.2.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",
]
+33 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "cnats"
version = "0.1.0"
version = "0.2.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,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",
+1 -1
View File
@@ -1,6 +1,6 @@
# Maintainer: Bendik Aagaard Lynghaug <bendik.lynghaug@gmail.com>
pkgname=cnats
pkgver=0.1.0
pkgver=0.2.0
pkgrel=1
pkgdesc="Web chat over NATS subjects with Kanidm SSO (Leptos SSR)"
arch=('x86_64' 'aarch64')
+12
View File
@@ -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"
+206 -3
View File
@@ -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,159 @@ fn open_event_source(
Some(es)
}
/// Minimal mesh call entry point, scoped to the `lobby` room only (see
/// `ChatShell`'s `<Show when=move || room.get() == "lobby">`) - the
/// signaling itself (`call.rs`, `/call-sse/{room}`) is already generic
/// per-room, so widening this later is a one-line UI change, not an
/// architectural one. `CallState` (browser-only: it holds `web_sys`
/// types) can't exist in the `ssr` build at all, so the two targets get
/// entirely separate bodies rather than sharing signals across the gate.
#[component]
fn CallPanel(room: Memo<String>, me: String) -> impl IntoView {
#[cfg(feature = "hydrate")]
{
use crate::webrtc::CallState;
let call_state = StoredValue::new_local(CallState::new(room.get_untracked(), me.clone()));
let in_call = call_state.get_value().in_call;
let local_video_ref = NodeRef::<leptos::html::Video>::new();
let es_handle = StoredValue::new_local(None::<web_sys::EventSource>);
Effect::new(move |_| {
let room_name = room.get();
es_handle.update_value(|es| {
if let Some(es) = es.take() {
es.close();
}
});
es_handle.set_value(open_call_event_source(&room_name, call_state.get_value()));
});
on_cleanup(move || {
es_handle.update_value(|es| {
if let Some(es) = es.take() {
es.close();
}
});
let cs = call_state.get_value();
if cs.in_call.get_untracked() {
leptos::task::spawn_local(async move {
cs.leave().await;
});
}
});
// Mirror the local MediaStream into the preview <video> element -
// `srcObject` has no HTML attribute form, has to be set via JS.
Effect::new(move |_| {
let stream = call_state.get_value().local_stream().get();
if let Some(el) = local_video_ref.get() {
el.set_src_object(stream.as_ref());
}
});
let on_join = move |_| {
let cs = call_state.get_value();
leptos::task::spawn_local(async move {
cs.join().await;
});
};
let on_leave = move |_| {
let cs = call_state.get_value();
leptos::task::spawn_local(async move {
cs.leave().await;
});
};
view! {
<div class="call-panel">
<Show
when=move || in_call.get()
fallback=move || {
view! {
<button class="call-join" on:click=on_join>
"☎ join call"
</button>
}
}
>
<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.get_value().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>
</Show>
</div>
}
}
#[cfg(not(feature = "hydrate"))]
{
let _ = (room, me);
view! { <div class="call-panel"></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! {
+5
View File
@@ -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";
+86
View File
@@ -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(())
}
+33 -9
View File
@@ -3,23 +3,42 @@ use serde::{Deserialize, Serialize};
/// Rooms available in the UI. Each maps to the NATS subject
/// `chat.room.<name>`, so any other NATS client on the bus can join in.
pub const ROOMS: &[(&str, &str)] = &[
("lobby", "general traffic"),
("dev", "build & ship"),
("ops", "incidents & infra"),
("random", "off the record"),
/// The third field is the Kanidm group (via the `groups` OIDC claim,
/// see `oauth2 update-claim-map`) required to read/post in that room -
/// `None` means open to anyone in `cnats_users`.
pub const ROOMS: &[(&str, &str, Option<&str>)] = &[
("lobby", "general traffic", None),
("dev", "build & ship", Some("developers")),
("ops", "incidents & infra", Some("developers")),
("random", "off the record", None),
];
pub const DEFAULT_ROOM: &str = "lobby";
pub fn is_valid_room(room: &str) -> bool {
ROOMS.iter().any(|(name, _)| *name == room)
ROOMS.iter().any(|(name, _, _)| *name == room)
}
pub fn room_subject(room: &str) -> String {
format!("chat.room.{room}")
}
/// Whether `user` may read/post in `room`. `false` for an unknown room -
/// callers should check `is_valid_room` separately if they need to tell
/// "unknown room" and "not authorized" apart in the error they return.
/// Synchronous and I/O-free: the user's groups are already baked into
/// their session (from the `groups` OIDC claim at login), so this never
/// needs a live Kanidm round-trip - and never gets more current than
/// that login until they sign in again.
pub fn is_authorized_for_room(user: &crate::auth::User, room: &str) -> bool {
ROOMS
.iter()
.find(|(name, _, _)| *name == room)
.is_some_and(|(_, _, required_group)| {
required_group.is_none_or(|g| user.groups.iter().any(|ug| ug == g))
})
}
/// A single chat message as it travels over NATS (JSON-encoded payload).
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ChatMessage {
@@ -62,6 +81,9 @@ pub async fn send_message(room: String, text: String) -> Result<(), ServerFnErro
else {
return Err(ServerFnError::new("not signed in"));
};
if !is_authorized_for_room(&user, &room) {
return Err(ServerFnError::new("not authorized for this room"));
}
let state = expect_context::<AppState>();
let now = chrono::Utc::now();
@@ -94,13 +116,15 @@ pub async fn room_history(room: String) -> Result<Vec<ChatMessage>, ServerFnErro
return Err(ServerFnError::new("unknown room"));
}
let session: tower_sessions::Session = leptos_axum::extract().await?;
if session
let Some(user) = session
.get::<User>(SESSION_USER_KEY)
.await
.map_err(|e| ServerFnError::new(e.to_string()))?
.is_none()
{
else {
return Err(ServerFnError::new("not signed in"));
};
if !is_authorized_for_room(&user, &room) {
return Err(ServerFnError::new("not authorized for this room"));
}
let state = expect_context::<AppState>();
+4
View File
@@ -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() {
+12 -1
View File
@@ -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,
+41
View File
@@ -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()
}
+65 -14
View File
@@ -12,28 +12,45 @@ use futures::{Stream, StreamExt};
use tower_sessions::Session;
use crate::auth::{User, SESSION_USER_KEY};
use crate::chat::{is_valid_room, room_subject};
use crate::call::call_subject;
use crate::chat::{is_authorized_for_room, is_valid_room, room_subject};
use super::AppState;
/// GET /sse/{room} — stream the room's NATS subject to the browser.
/// Shared by `room_events` and `call_events`: signed in, valid room, and
/// authorized for it (Kanidm group gate, checked against the session's
/// own `groups` - see `chat::is_authorized_for_room`). Note this is only
/// checked once, at connect time - a long-lived SSE stream doesn't get
/// re-checked if the user's groups change mid-connection (same kind of
/// staleness the "still signed in at all" check already has).
async fn authorize_room_stream(
room: &str,
session: &Session,
) -> Result<(), (StatusCode, &'static str)> {
let user = session
.get::<User>(SESSION_USER_KEY)
.await
.ok()
.flatten();
let Some(user) = user else {
return Err((StatusCode::UNAUTHORIZED, "sign in first"));
};
if !is_valid_room(room) {
return Err((StatusCode::NOT_FOUND, "unknown room"));
}
if !is_authorized_for_room(&user, room) {
return Err((StatusCode::FORBIDDEN, "not authorized for this room"));
}
Ok(())
}
/// GET /sse/{room} — stream the room's chat NATS subject to the browser.
pub async fn room_events(
Path(room): Path<String>,
State(state): State<AppState>,
session: Session,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, &'static str)> {
let signed_in = session
.get::<User>(SESSION_USER_KEY)
.await
.ok()
.flatten()
.is_some();
if !signed_in {
return Err((StatusCode::UNAUTHORIZED, "sign in first"));
}
if !is_valid_room(&room) {
return Err((StatusCode::NOT_FOUND, "unknown room"));
}
authorize_room_stream(&room, &session).await?;
let subscriber = state
.nats
@@ -56,3 +73,37 @@ pub async fn room_events(
.text("ping"),
))
}
/// GET /call-sse/{room} — stream the room's call-signaling NATS subject
/// (SDP offers/answers, ICE candidates). Deliberately a separate subject
/// namespace (`call.room.*`, not `chat.room.*`) so this never touches the
/// JetStream/Postgres chat archive - ephemeral signaling has no business
/// being durably stored.
pub async fn call_events(
Path(room): Path<String>,
State(state): State<AppState>,
session: Session,
) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, &'static str)> {
authorize_room_stream(&room, &session).await?;
let subscriber = state
.nats
.subscribe(call_subject(&room))
.await
.map_err(|e| {
tracing::error!("nats subscribe failed: {e}");
(StatusCode::BAD_GATEWAY, "message bus unavailable")
})?;
let stream = subscriber.map(|msg| {
Ok(Event::default()
.event("signal")
.data(String::from_utf8_lossy(&msg.payload).into_owned()))
});
Ok(Sse::new(stream).keep_alive(
KeepAlive::new()
.interval(Duration::from_secs(15))
.text("ping"),
))
}
+328
View File
@@ -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;
});
}
}
+73
View File
@@ -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 {