DUNGEONS-34: implement Webka local streaming app

Co-authored-by: multica-agent <github@multica.ai>

backend/src/main.rs

@@ -0,0 +1,767 @@
+use argon2::{
+    Argon2, PasswordHash, PasswordHasher, PasswordVerifier,
+    password_hash::{SaltString, rand_core::OsRng},
+};
+use axum::{
+    Json, Router,
+    extract::{
+        ConnectInfo, Path, Query, Request, State,
+        ws::{Message, WebSocket, WebSocketUpgrade},
+    },
+    http::{HeaderMap, StatusCode, header},
+    middleware::{self, Next},
+    response::{IntoResponse, Response},
+    routing::{delete, get, post},
+};
+use futures_util::{SinkExt, StreamExt};
+use hmac::{Hmac, Mac};
+use serde::Deserialize;
+use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
+use sqlx::{Row, SqlitePool, sqlite::SqlitePoolOptions};
+use std::{
+    collections::HashMap,
+    net::SocketAddr,
+    path::PathBuf,
+    sync::Arc,
+    time::{SystemTime, UNIX_EPOCH},
+};
+use tokio::sync::{Mutex, mpsc};
+use tower_http::services::{ServeDir, ServeFile};
+use uuid::Uuid;
+
+const IDLE: i64 = 30 * 60;
+const ABSOLUTE: i64 = 12 * 60 * 60;
+fn now() -> i64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap()
+        .as_secs() as i64
+}
+fn id() -> String {
+    Uuid::new_v4().to_string()
+}
+fn digest(s: &str) -> String {
+    hex::encode(Sha256::digest(s.as_bytes()))
+}
+type ApiResult<T> = Result<T, ApiError>;
+#[derive(Debug)]
+struct ApiError(StatusCode, &'static str);
+impl IntoResponse for ApiError {
+    fn into_response(self) -> Response {
+        (self.0, Json(json!({"error": self.1}))).into_response()
+    }
+}
+impl From<sqlx::Error> for ApiError {
+    fn from(e: sqlx::Error) -> Self {
+        eprintln!("database: {e}");
+        Self(StatusCode::INTERNAL_SERVER_ERROR, "server")
+    }
+}
+fn bad(code: &'static str) -> ApiError {
+    ApiError(StatusCode::BAD_REQUEST, code)
+}
+fn unauthorized() -> ApiError {
+    ApiError(StatusCode::UNAUTHORIZED, "unauthorized")
+}
+
+#[derive(Clone)]
+struct App {
+    db: SqlitePool,
+    key: Arc<Vec<u8>>,
+    setup: Arc<String>,
+    origin: Arc<String>,
+    secure: bool,
+    peers: Arc<Mutex<Peers>>,
+    limits: Arc<Mutex<HashMap<String, (i64, u32)>>>,
+    hashes: Arc<tokio::sync::Semaphore>,
+}
+#[derive(Default)]
+struct Peers {
+    clients: HashMap<String, Client>,
+    host: Option<String>,
+    since: Option<i64>,
+}
+struct Client {
+    session: String,
+    user: String,
+    name: String,
+    mode: String,
+    tx: mpsc::Sender<Message>,
+    shutdown: tokio::sync::watch::Sender<bool>,
+}
+#[derive(Clone)]
+struct Auth {
+    session: String,
+    user: String,
+    name: String,
+    role: String,
+}
+impl Auth {
+    fn admin(&self) -> ApiResult<()> {
+        if self.role == "admin" {
+            Ok(())
+        } else {
+            Err(ApiError(StatusCode::FORBIDDEN, "forbidden"))
+        }
+    }
+}
+
+async fn database(url: &str) -> Result<SqlitePool, sqlx::Error> {
+    let db = SqlitePoolOptions::new()
+        .max_connections(5)
+        .connect_with(
+            url.parse::<sqlx::sqlite::SqliteConnectOptions>()?
+                .create_if_missing(true)
+                .foreign_keys(true)
+                .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
+                .busy_timeout(std::time::Duration::from_secs(5)),
+        )
+        .await?;
+    sqlx::raw_sql("CREATE TABLE IF NOT EXISTS accounts(id TEXT PRIMARY KEY, username TEXT NOT NULL UNIQUE COLLATE NOCASE, password TEXT NOT NULL, role TEXT NOT NULL CHECK(role IN ('admin','viewer')), last_login INTEGER);
+    CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY, token_hash TEXT NOT NULL UNIQUE, user_id TEXT NOT NULL REFERENCES accounts(id) ON DELETE CASCADE, created INTEGER NOT NULL, last_seen INTEGER NOT NULL);
+    CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
+    CREATE TABLE IF NOT EXISTS settings(key TEXT PRIMARY KEY, value TEXT NOT NULL);
+    PRAGMA optimize;").execute(&db).await?;
+    Ok(db)
+}
+fn password_valid(p: &str) -> bool {
+    p.chars().count() >= 12
+        && p.len() <= 256
+        && p.chars().any(char::is_alphabetic)
+        && p.chars().any(|c| c.is_ascii_digit())
+}
+fn credentials(c: &Credentials) -> ApiResult<()> {
+    if c.username.len() < 3
+        || c.username.len() > 40
+        || !c
+            .username
+            .chars()
+            .all(|c| c.is_ascii_alphanumeric() || "_.-".contains(c))
+    {
+        return Err(bad("username"));
+    }
+    if !password_valid(&c.password) {
+        return Err(bad("password"));
+    }
+    Ok(())
+}
+async fn hash_password(app: &App, password: String) -> ApiResult<String> {
+    let permit = app
+        .hashes
+        .clone()
+        .acquire_owned()
+        .await
+        .map_err(|_| bad("server"))?;
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        Argon2::default()
+            .hash_password(password.as_bytes(), &SaltString::generate(&mut OsRng))
+            .map(|h| h.to_string())
+            .map_err(|_| bad("server"))
+    })
+    .await
+    .map_err(|_| bad("server"))?
+}
+async fn verify_password(app: &App, password: String, hash: String) -> bool {
+    let Ok(permit) = app.hashes.clone().acquire_owned().await else {
+        return false;
+    };
+    tokio::task::spawn_blocking(move || {
+        let _permit = permit;
+        PasswordHash::new(&hash)
+            .map(|h| {
+                Argon2::default()
+                    .verify_password(password.as_bytes(), &h)
+                    .is_ok()
+            })
+            .unwrap_or(false)
+    })
+    .await
+    .unwrap_or(false)
+}
+fn sign(key: &[u8], token: &str) -> String {
+    let mut mac = Hmac::<Sha256>::new_from_slice(key).unwrap();
+    mac.update(token.as_bytes());
+    format!("{token}.{}", hex::encode(mac.finalize().into_bytes()))
+}
+fn cookie_token(headers: &HeaderMap, key: &[u8]) -> ApiResult<String> {
+    let raw = headers
+        .get(header::COOKIE)
+        .and_then(|h| h.to_str().ok())
+        .and_then(|s| {
+            s.split(';')
+                .find_map(|c| c.trim().strip_prefix("webka_session="))
+        })
+        .ok_or_else(unauthorized)?;
+    let (token, sig) = raw.split_once('.').ok_or_else(unauthorized)?;
+    let mut mac = Hmac::<Sha256>::new_from_slice(key).unwrap();
+    mac.update(token.as_bytes());
+    mac.verify_slice(&hex::decode(sig).map_err(|_| unauthorized())?)
+        .map_err(|_| unauthorized())?;
+    Ok(token.to_owned())
+}
+async fn session_by_id(app: &App, session: &str) -> ApiResult<Auth> {
+    let row = sqlx::query("SELECT s.id, a.id AS user_id, a.username, a.role FROM sessions s JOIN accounts a ON a.id=s.user_id WHERE s.id=? AND s.last_seen>? AND s.created>?").bind(session).bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_optional(&app.db).await?.ok_or_else(unauthorized)?;
+    Ok(Auth {
+        session: row.get("id"),
+        user: row.get("user_id"),
+        name: row.get("username"),
+        role: row.get("role"),
+    })
+}
+async fn auth(app: &App, headers: &HeaderMap, touch: bool) -> ApiResult<Auth> {
+    let token = cookie_token(headers, &app.key)?;
+    let session: Option<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE token_hash=?")
+        .bind(digest(&token))
+        .fetch_optional(&app.db)
+        .await?;
+    let a = session_by_id(app, &session.ok_or_else(unauthorized)?).await?;
+    if touch {
+        sqlx::query("UPDATE sessions SET last_seen=? WHERE id=?")
+            .bind(now())
+            .bind(&a.session)
+            .execute(&app.db)
+            .await?;
+    }
+    Ok(a)
+}
+async fn guard(State(app): State<App>, req: Request, next: Next) -> Response {
+    let mutation = !matches!(
+        *req.method(),
+        axum::http::Method::GET | axum::http::Method::HEAD | axum::http::Method::OPTIONS
+    );
+    if (mutation || req.uri().path() == "/api/signal")
+        && req
+            .headers()
+            .get(header::ORIGIN)
+            .and_then(|v| v.to_str().ok())
+            != Some(app.origin.as_str())
+    {
+        return ApiError(StatusCode::FORBIDDEN, "origin").into_response();
+    }
+    let mut res = next.run(req).await;
+    res.headers_mut()
+        .insert(header::CACHE_CONTROL, "no-store".parse().unwrap());
+    res.headers_mut()
+        .insert(header::X_CONTENT_TYPE_OPTIONS, "nosniff".parse().unwrap());
+    res.headers_mut()
+        .insert(header::REFERRER_POLICY, "no-referrer".parse().unwrap());
+    res.headers_mut().insert(header::CONTENT_SECURITY_POLICY, "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'; img-src 'self' data: blob:; media-src 'self' blob:; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self'".parse().unwrap());
+    res
+}
+async fn throttle(app: &App, key: String) -> ApiResult<()> {
+    let mut limits = app.limits.lock().await;
+    limits.retain(|_, (start, _)| now() - *start < 60);
+    if limits.len() >= 4096 && !limits.contains_key(&key) {
+        return Err(ApiError(StatusCode::TOO_MANY_REQUESTS, "rateLimited"));
+    }
+    let entry = limits.entry(key).or_insert((now(), 0));
+    entry.1 += 1;
+    if entry.1 > 10 {
+        Err(ApiError(StatusCode::TOO_MANY_REQUESTS, "rateLimited"))
+    } else {
+        Ok(())
+    }
+}
+#[derive(Deserialize)]
+struct Credentials {
+    username: String,
+    password: String,
+    setup_token: Option<String>,
+}
+async fn bootstrap_status(State(app): State<App>) -> ApiResult<Json<Value>> {
+    let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM accounts")
+        .fetch_one(&app.db)
+        .await?;
+    Ok(Json(json!({"needs_setup": count == 0})))
+}
+async fn bootstrap(
+    State(app): State<App>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    Json(c): Json<Credentials>,
+) -> ApiResult<Response> {
+    throttle(&app, addr.ip().to_string()).await?;
+    if c.setup_token.as_deref().map(digest) != Some(digest(&app.setup)) {
+        return Err(ApiError(StatusCode::FORBIDDEN, "setupToken"));
+    }
+    credentials(&c)?;
+    let hash = hash_password(&app, c.password).await?;
+    let result = sqlx::query("INSERT INTO accounts(id,username,password,role) SELECT ?,?,?,'admin' WHERE NOT EXISTS (SELECT 1 FROM accounts)").bind(id()).bind(&c.username).bind(hash).execute(&app.db).await?;
+    if result.rows_affected() == 0 {
+        return Err(ApiError(StatusCode::CONFLICT, "alreadySetup"));
+    }
+    Ok(StatusCode::CREATED.into_response())
+}
+async fn login(
+    State(app): State<App>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    Json(c): Json<Credentials>,
+) -> ApiResult<Response> {
+    throttle(&app, addr.ip().to_string()).await?;
+    if c.password.len() > 256 || c.username.len() > 40 {
+        return Err(unauthorized());
+    }
+    let row = sqlx::query("SELECT id,password FROM accounts WHERE username=? COLLATE NOCASE")
+        .bind(&c.username)
+        .fetch_optional(&app.db)
+        .await?;
+    // Unknown users perform the same expensive Argon2 operation as existing users.
+    let dummy: String = sqlx::query_scalar("SELECT value FROM settings WHERE key='dummy_hash'")
+        .fetch_one(&app.db)
+        .await?;
+    let hash = row
+        .as_ref()
+        .map(|r| r.get::<String, _>("password"))
+        .unwrap_or(dummy);
+    if !verify_password(&app, c.password, hash.clone()).await || row.is_none() {
+        return Err(unauthorized());
+    }
+    let row = row.unwrap();
+    let uid: String = row.get("id");
+    let token = format!("{}{}", id(), id());
+    sqlx::query("DELETE FROM sessions WHERE last_seen<=? OR created<=?")
+        .bind(now() - IDLE)
+        .bind(now() - ABSOLUTE)
+        .execute(&app.db)
+        .await?;
+    // A reset/delete that won the race with password verification must invalidate this login.
+    let mut transaction = app.db.begin().await?;
+    let inserted = sqlx::query("INSERT INTO sessions(id,token_hash,user_id,created,last_seen) SELECT ?,?,id,?,? FROM accounts WHERE id=? AND password=?")
+        .bind(id()).bind(digest(&token)).bind(now()).bind(now()).bind(&uid).bind(&hash)
+        .execute(&mut *transaction).await?;
+    if inserted.rows_affected() == 0 {
+        return Err(unauthorized());
+    }
+    sqlx::query("UPDATE accounts SET last_login=? WHERE id=?")
+        .bind(now())
+        .bind(uid)
+        .execute(&mut *transaction)
+        .await?;
+    transaction.commit().await?;
+    let cookie = format!(
+        "webka_session={}; Path=/; HttpOnly; SameSite=Strict; Max-Age={ABSOLUTE}{}",
+        sign(&app.key, &token),
+        if app.secure { "; Secure" } else { "" }
+    );
+    Ok(([(header::SET_COOKIE, cookie)], Json(json!({"ok":true}))).into_response())
+}
+async fn me(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
+    let a = auth(&app, &headers, true).await?;
+    Ok(Json(
+        json!({"id":a.user,"username":a.name,"role":a.role,"session_id":a.session}),
+    ))
+}
+async fn logout(State(app): State<App>, headers: HeaderMap) -> ApiResult<Response> {
+    let a = auth(&app, &headers, false).await?;
+    revoke(&app, &a.session).await?;
+    Ok((
+        [(
+            header::SET_COOKIE,
+            "webka_session=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0",
+        )],
+        StatusCode::NO_CONTENT,
+    )
+        .into_response())
+}
+async fn accounts(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
+    auth(&app, &headers, true).await?.admin()?;
+    let rows = sqlx::query("SELECT a.id,a.username,a.role,a.last_login,(SELECT COUNT(*) FROM sessions s WHERE s.user_id=a.id AND s.last_seen>? AND s.created>?) AS active_sessions FROM accounts a ORDER BY a.role,a.username").bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_all(&app.db).await?;
+    Ok(Json(json!(rows.iter().map(|r| json!({"id":r.get::<String,_>("id"),"username":r.get::<String,_>("username"),"role":r.get::<String,_>("role"),"last_login":r.get::<Option<i64>,_>("last_login"),"active_sessions":r.get::<i64,_>("active_sessions")})).collect::<Vec<_>>())))
+}
+async fn create_account(
+    State(app): State<App>,
+    headers: HeaderMap,
+    Json(c): Json<Credentials>,
+) -> ApiResult<StatusCode> {
+    auth(&app, &headers, true).await?.admin()?;
+    credentials(&c)?;
+    let hash = hash_password(&app, c.password).await?;
+    let result =
+        sqlx::query("INSERT INTO accounts(id,username,password,role) VALUES(?,?,?,'viewer')")
+            .bind(id())
+            .bind(c.username)
+            .bind(hash)
+            .execute(&app.db)
+            .await;
+    match result {
+        Ok(_) => Ok(StatusCode::CREATED),
+        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
+            Err(ApiError(StatusCode::CONFLICT, "duplicate"))
+        }
+        Err(e) => Err(e.into()),
+    }
+}
+async fn delete_account(
+    State(app): State<App>,
+    headers: HeaderMap,
+    Path(uid): Path<String>,
+) -> ApiResult<StatusCode> {
+    auth(&app, &headers, true).await?.admin()?;
+    let mut tx = app.db.begin().await?;
+    let sessions: Vec<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE user_id=?")
+        .bind(&uid)
+        .fetch_all(&mut *tx)
+        .await?;
+    let result = sqlx::query("DELETE FROM accounts WHERE id=? AND role='viewer'")
+        .bind(&uid)
+        .execute(&mut *tx)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(bad("protectedAccount"));
+    }
+    tx.commit().await?;
+    for sid in sessions {
+        disconnect_session(&app, &sid).await;
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+#[derive(Deserialize)]
+struct PasswordChange {
+    password: String,
+}
+async fn reset_password(
+    State(app): State<App>,
+    headers: HeaderMap,
+    Path(uid): Path<String>,
+    Json(c): Json<PasswordChange>,
+) -> ApiResult<StatusCode> {
+    auth(&app, &headers, true).await?.admin()?;
+    if !password_valid(&c.password) {
+        return Err(bad("password"));
+    }
+    let hash = hash_password(&app, c.password).await?;
+    let mut tx = app.db.begin().await?;
+    let result = sqlx::query("UPDATE accounts SET password=? WHERE id=?")
+        .bind(hash)
+        .bind(&uid)
+        .execute(&mut *tx)
+        .await?;
+    if result.rows_affected() == 0 {
+        return Err(bad("notFound"));
+    }
+    let sessions: Vec<String> = sqlx::query_scalar("SELECT id FROM sessions WHERE user_id=?")
+        .bind(&uid)
+        .fetch_all(&mut *tx)
+        .await?;
+    sqlx::query("DELETE FROM sessions WHERE user_id=?")
+        .bind(&uid)
+        .execute(&mut *tx)
+        .await?;
+    tx.commit().await?;
+    for sid in sessions {
+        disconnect_session(&app, &sid).await;
+    }
+    Ok(StatusCode::NO_CONTENT)
+}
+async fn sessions(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
+    auth(&app, &headers, true).await?.admin()?;
+    let rows = sqlx::query("SELECT s.id,s.created,s.last_seen,a.username FROM sessions s JOIN accounts a ON a.id=s.user_id WHERE s.last_seen>? AND s.created>? ORDER BY s.last_seen DESC").bind(now()-IDLE).bind(now()-ABSOLUTE).fetch_all(&app.db).await?;
+    Ok(Json(json!(rows.iter().map(|r| json!({"id":r.get::<String,_>("id"),"username":r.get::<String,_>("username"),"created":r.get::<i64,_>("created"),"last_seen":r.get::<i64,_>("last_seen")})).collect::<Vec<_>>())))
+}
+async fn revoke(app: &App, sid: &str) -> ApiResult<()> {
+    sqlx::query("DELETE FROM sessions WHERE id=?")
+        .bind(sid)
+        .execute(&app.db)
+        .await?;
+    disconnect_session(app, sid).await;
+    Ok(())
+}
+async fn revoke_session(
+    State(app): State<App>,
+    headers: HeaderMap,
+    Path(sid): Path<String>,
+) -> ApiResult<StatusCode> {
+    auth(&app, &headers, true).await?.admin()?;
+    revoke(&app, &sid).await?;
+    Ok(StatusCode::NO_CONTENT)
+}
+fn send(client: &Client, value: Value) {
+    if client
+        .tx
+        .try_send(Message::Text(value.to_string().into()))
+        .is_err()
+    {
+        let _ = client.shutdown.send(true);
+    }
+}
+async fn disconnect_session(app: &App, sid: &str) {
+    let mut peers = app.peers.lock().await;
+    let ids: Vec<_> = peers
+        .clients
+        .iter()
+        .filter(|(_, c)| c.session == sid)
+        .map(|(id, _)| id.clone())
+        .collect();
+    for cid in ids {
+        remove_peer(&mut peers, &cid);
+    }
+}
+fn remove_peer(peers: &mut Peers, cid: &str) {
+    let Some(client) = peers.clients.remove(cid) else {
+        return;
+    };
+    let _ = client.tx.try_send(Message::Close(None));
+    let _ = client.shutdown.send(true);
+    if peers.host.as_deref() == Some(cid) {
+        peers.host = None;
+        peers.since = None;
+        for c in peers.clients.values() {
+            send(c, json!({"type":"host-offline"}));
+        }
+    } else if let Some(host) = peers.host.as_ref().and_then(|h| peers.clients.get(h)) {
+        send(host, json!({"type":"viewer-left","peer":cid}));
+    }
+}
+async fn status(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
+    let a = auth(&app, &headers, true).await?;
+    let peers = app.peers.lock().await;
+    let viewers: Vec<_> = peers
+        .clients
+        .iter()
+        .filter(|(_, c)| c.mode == "viewer")
+        .map(|(id, c)| {
+            if a.role == "admin" {
+                json!({"id":id,"session_id":c.session,"username":c.name,"user_id":c.user})
+            } else {
+                json!({})
+            }
+        })
+        .collect();
+    Ok(Json(
+        json!({"live":peers.host.is_some(),"since":peers.since,"viewer_count":viewers.len(),"viewers":if a.role=="admin" {viewers} else {vec![]},"local_url":format!("{}/watch",app.origin)}),
+    ))
+}
+#[derive(Deserialize)]
+struct SignalQuery {
+    mode: Option<String>,
+}
+async fn signal(
+    State(app): State<App>,
+    headers: HeaderMap,
+    Query(q): Query<SignalQuery>,
+    ws: WebSocketUpgrade,
+) -> ApiResult<Response> {
+    let a = auth(&app, &headers, true).await?;
+    let mode = q.mode.unwrap_or("viewer".into());
+    if mode != "host" && mode != "viewer" {
+        return Err(bad("mode"));
+    }
+    if mode == "host" {
+        a.admin()?;
+    }
+    Ok(ws
+        .max_message_size(64 * 1024)
+        .max_frame_size(64 * 1024)
+        .on_upgrade(move |socket| connection(app, a, mode, socket)))
+}
+async fn connection(app: App, a: Auth, mode: String, socket: WebSocket) {
+    let cid = id();
+    let (tx, mut rx) = mpsc::channel(64);
+    let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
+    let (mut sink, mut source) = socket.split();
+    {
+        let mut peers = app.peers.lock().await;
+        if peers.clients.len() >= 32
+            || (mode == "host" && peers.host.is_some())
+            || peers.clients.values().any(|c| c.session == a.session)
+        {
+            drop(peers);
+            let _ = tokio::time::timeout(
+                std::time::Duration::from_secs(2),
+                sink.send(Message::Text(
+                    json!({"type":"error","error":"hostBusy"})
+                        .to_string()
+                        .into(),
+                )),
+            )
+            .await;
+            return;
+        }
+        peers.clients.insert(
+            cid.clone(),
+            Client {
+                session: a.session.clone(),
+                user: a.user,
+                name: a.name,
+                mode: mode.clone(),
+                tx,
+                shutdown: shutdown_tx,
+            },
+        );
+        if mode == "host" {
+            peers.host = Some(cid.clone());
+            peers.since = Some(now());
+            for (vid, c) in &peers.clients {
+                if c.mode == "viewer" {
+                    send(
+                        peers.clients.get(&cid).unwrap(),
+                        json!({"type":"viewer-joined","peer":vid}),
+                    );
+                    send(c, json!({"type":"host-online"}));
+                }
+            }
+        } else if let Some(host) = peers.host.as_ref().and_then(|h| peers.clients.get(h)) {
+            send(host, json!({"type":"viewer-joined","peer":cid}));
+        }
+        send(
+            peers.clients.get(&cid).unwrap(),
+            json!({"type":"welcome","peer":cid,"live":peers.host.is_some()}),
+        );
+    }
+    let mut tick = tokio::time::interval(std::time::Duration::from_secs(2));
+    let mut count = 0;
+    let mut window = now();
+    loop {
+        tokio::select! {
+            _ = shutdown_rx.changed() => break,
+            _ = tick.tick() => { if session_by_id(&app,&a.session).await.is_err() { break; } }
+            msg = rx.recv() => { match msg { Some(msg) => { let close = matches!(msg, Message::Close(_)); if !tokio::time::timeout(std::time::Duration::from_secs(2), sink.send(msg)).await.is_ok_and(|result| result.is_ok()) || close { break; } }, None => break } }
+            msg = source.next() => {
+                match msg {
+                    Some(Ok(Message::Text(text))) => {
+                        if now() != window { window = now(); count = 0; } count += 1; if count > 120 { break; }
+                        if session_by_id(&app,&a.session).await.is_err() { break; }
+                        let Ok(v) = serde_json::from_str::<Value>(&text) else { continue; };
+                        let Some(kind) = v["type"].as_str() else { continue; };
+                        let peers = app.peers.lock().await;
+                        if !peers.clients.contains_key(&cid) { break; }
+                        let Some(target_id) = v["target"].as_str() else { continue; };
+                        let Some(target) = peers.clients.get(target_id) else { continue; };
+                        let allowed = if mode == "host" { peers.host.as_deref() == Some(&cid) && target.mode=="viewer" && matches!(kind,"offer"|"ice") } else { peers.host.as_deref()==Some(target_id) && matches!(kind,"answer"|"ice"|"renegotiate") };
+                        if allowed { send(target,json!({"type":kind,"peer":cid,"data":v["data"]})); }
+                    }
+                    Some(Ok(Message::Ping(data))) => { if !tokio::time::timeout(std::time::Duration::from_secs(2), sink.send(Message::Pong(data))).await.is_ok_and(|result| result.is_ok()) { break; } }
+                    Some(Ok(Message::Close(_))) | None | Some(Err(_)) => break,
+                    _ => {}
+                }
+            }
+        }
+    }
+    remove_peer(&mut *app.peers.lock().await, &cid);
+    let _ = tokio::time::timeout(std::time::Duration::from_secs(2), sink.close()).await;
+}
+fn router(app: App, assets: PathBuf) -> Router {
+    Router::new()
+        .route("/api/bootstrap", get(bootstrap_status).post(bootstrap))
+        .route("/api/login", post(login))
+        .route("/api/logout", post(logout))
+        .route("/api/session", get(me))
+        .route("/api/accounts", get(accounts).post(create_account))
+        .route("/api/accounts/{id}", delete(delete_account))
+        .route("/api/accounts/{id}/password", post(reset_password))
+        .route("/api/sessions", get(sessions))
+        .route("/api/sessions/{id}", delete(revoke_session))
+        .route("/api/status", get(status))
+        .route("/api/signal", get(signal))
+        .fallback_service(
+            ServeDir::new(&assets).not_found_service(ServeFile::new(assets.join("index.html"))),
+        )
+        .layer(axum::extract::DefaultBodyLimit::max(8192))
+        .layer(middleware::from_fn_with_state(app.clone(), guard))
+        .with_state(app)
+}
+fn secret_file(path: &std::path::Path) -> std::io::Result<String> {
+    use std::io::Write;
+    if path.exists() {
+        return std::fs::read_to_string(path).map(|s| s.trim().to_owned());
+    }
+    let mut options = std::fs::OpenOptions::new();
+    options.write(true).create_new(true);
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::OpenOptionsExt;
+        options.mode(0o600);
+    }
+    let mut file = options.open(path)?;
+    let secret = format!("{}{}", id(), id());
+    file.write_all(secret.as_bytes())?;
+    Ok(secret)
+}
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let data = PathBuf::from(std::env::var("WEBKA_DATA").unwrap_or("data".into()));
+    std::fs::create_dir_all(&data)?;
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        std::fs::set_permissions(&data, std::fs::Permissions::from_mode(0o700))?;
+    }
+    let key = secret_file(&data.join("session-key"))?;
+    let setup = secret_file(&data.join("setup-token"))?;
+    let origin = std::env::var("WEBKA_ORIGIN")
+        .unwrap_or("http://localhost:3000".into())
+        .trim_end_matches('/')
+        .to_owned();
+    let addr: SocketAddr = std::env::var("WEBKA_BIND")
+        .unwrap_or("127.0.0.1:3001".into())
+        .parse()?;
+    let app = App {
+        db: database(&format!("sqlite://{}", data.join("webka.db").display())).await?,
+        key: Arc::new(key.into_bytes()),
+        setup: Arc::new(setup),
+        secure: origin.starts_with("https://"),
+        origin: Arc::new(origin),
+        peers: Default::default(),
+        limits: Default::default(),
+        hashes: Arc::new(tokio::sync::Semaphore::new(2)),
+    };
+    let dummy = hash_password(&app, id())
+        .await
+        .map_err(|_| "hash initialization failed")?;
+    sqlx::query("INSERT OR IGNORE INTO settings(key,value) VALUES('dummy_hash',?)")
+        .bind(dummy)
+        .execute(&app.db)
+        .await?;
+    let mdns = if std::env::var("WEBKA_MDNS").as_deref() == Ok("1") {
+        match (mdns_sd::ServiceDaemon::new(), local_ip_address::local_ip()) {
+            (Ok(daemon), Ok(ip)) => {
+                let port: u16 = std::env::var("WEBKA_PUBLIC_PORT")
+                    .unwrap_or(if app.secure { "443" } else { "3000" }.into())
+                    .parse()?;
+                let service = mdns_sd::ServiceInfo::new(
+                    "_webka._tcp.local.",
+                    "Webka",
+                    "webka.local.",
+                    ip,
+                    port,
+                    [
+                        ("path", "/watch"),
+                        ("scheme", if app.secure { "https" } else { "http" }),
+                    ]
+                    .as_slice(),
+                )?;
+                daemon.register(service)?;
+                Some(daemon)
+            }
+            _ => {
+                eprintln!("mDNS unavailable; use the configured local URL.");
+                None
+            }
+        }
+    } else {
+        None
+    };
+    let listener = tokio::net::TcpListener::bind(addr).await?;
+    println!(
+        "Webka listening at {addr}. Open {}. First-run setup token: {}",
+        app.origin,
+        data.join("setup-token").display()
+    );
+    let assets = std::env::var("WEBKA_ASSETS").unwrap_or("frontend/.output/public".into());
+    axum::serve(
+        listener,
+        router(app, assets.into()).into_make_service_with_connect_info::<SocketAddr>(),
+    )
+    .with_graceful_shutdown(async {
+        let _ = tokio::signal::ctrl_c().await;
+    })
+    .await?;
+    if let Some(daemon) = mdns {
+        let _ = daemon.shutdown();
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests;

backend/src/tests.rs

@@ -0,0 +1,558 @@
+use super::*;
+use axum::{body::Body, http::Request};
+use tower::ServiceExt;
+
+async fn app() -> (App, tempfile::TempDir) {
+    let dir = tempfile::tempdir().unwrap();
+    let app = App {
+        db: database(&format!(
+            "sqlite://{}",
+            dir.path().join("test.db").display()
+        ))
+        .await
+        .unwrap(),
+        key: Arc::new(vec![42; 32]),
+        setup: Arc::new("setup-test".into()),
+        origin: Arc::new("http://localhost:3000".into()),
+        secure: false,
+        peers: Default::default(),
+        limits: Default::default(),
+        hashes: Arc::new(tokio::sync::Semaphore::new(2)),
+    };
+    let dummy = hash_password(&app, "dummy".into()).await.unwrap();
+    sqlx::query("INSERT INTO settings(key,value) VALUES('dummy_hash',?)")
+        .bind(dummy)
+        .execute(&app.db)
+        .await
+        .unwrap();
+    (app, dir)
+}
+async fn request(app: &App, method: &str, path: &str, cookie: &str, data: Value) -> Response {
+    let mut req = Request::builder()
+        .method(method)
+        .uri(path)
+        .header("origin", "http://localhost:3000")
+        .header("content-type", "application/json")
+        .header("cookie", cookie)
+        .body(Body::from(data.to_string()))
+        .unwrap();
+    req.extensions_mut()
+        .insert(ConnectInfo("127.0.0.1:1234".parse::<SocketAddr>().unwrap()));
+    router(app.clone(), "/nonexistent".into())
+        .oneshot(req)
+        .await
+        .unwrap()
+}
+async fn setup(app: &App) -> String {
+    let response = request(
+        app,
+        "POST",
+        "/api/bootstrap",
+        "",
+        json!({"username":"owner","password":"strong-password123","setup_token":"setup-test"}),
+    )
+    .await;
+    assert_eq!(response.status(), StatusCode::CREATED);
+    login_as(app, "owner", "strong-password123").await
+}
+async fn login_as(app: &App, username: &str, password: &str) -> String {
+    let response = request(
+        app,
+        "POST",
+        "/api/login",
+        "",
+        json!({"username":username,"password":password}),
+    )
+    .await;
+    assert_eq!(response.status(), StatusCode::OK);
+    let cookie = response.headers()[header::SET_COOKIE].to_str().unwrap();
+    assert!(cookie.contains("HttpOnly") && cookie.contains("SameSite=Strict"));
+    cookie.split(';').next().unwrap().into()
+}
+async fn json_body(response: Response) -> Value {
+    serde_json::from_slice(
+        &axum::body::to_bytes(response.into_body(), 1024 * 1024)
+            .await
+            .unwrap(),
+    )
+    .unwrap()
+}
+
+#[tokio::test]
+async fn argon2id_hashes_are_salted_and_verify() {
+    let (app, _dir) = app().await;
+    let a = hash_password(&app, "a long password 123".into())
+        .await
+        .unwrap();
+    let b = hash_password(&app, "a long password 123".into())
+        .await
+        .unwrap();
+    assert!(a.starts_with("$argon2id$"));
+    assert_ne!(a, b);
+    assert!(verify_password(&app, "a long password 123".into(), a.clone()).await);
+    assert!(!verify_password(&app, "wrong".into(), a).await);
+    assert!(!password_valid("short1"));
+    assert!(!password_valid("longwithoutnumbers"));
+    assert!(password_valid("длинныйпароль123"));
+}
+#[tokio::test]
+async fn bootstrap_is_token_protected_and_one_time() {
+    let (app, _dir) = app().await;
+    assert_eq!(
+        request(
+            &app,
+            "POST",
+            "/api/bootstrap",
+            "",
+            json!({"username":"owner","password":"strong-password123","setup_token":"wrong"})
+        )
+        .await
+        .status(),
+        StatusCode::FORBIDDEN
+    );
+    setup(&app).await;
+    assert_eq!(
+        request(
+            &app,
+            "POST",
+            "/api/bootstrap",
+            "",
+            json!({"username":"second","password":"strong-password123","setup_token":"setup-test"})
+        )
+        .await
+        .status(),
+        StatusCode::CONFLICT
+    );
+    assert_eq!(
+        json_body(request(&app, "GET", "/api/bootstrap", "", json!(null)).await).await["needs_setup"],
+        false
+    );
+}
+#[tokio::test]
+async fn sessions_expire_on_idle_and_absolute_deadlines_and_reject_tampering() {
+    let (app, _dir) = app().await;
+    let cookie = setup(&app).await;
+    assert_eq!(
+        request(
+            &app,
+            "GET",
+            "/api/session",
+            &format!("{cookie}0"),
+            json!(null)
+        )
+        .await
+        .status(),
+        StatusCode::UNAUTHORIZED
+    );
+    sqlx::query("UPDATE sessions SET last_seen=?")
+        .bind(now() - IDLE)
+        .execute(&app.db)
+        .await
+        .unwrap();
+    assert_eq!(
+        request(&app, "GET", "/api/session", &cookie, json!(null))
+            .await
+            .status(),
+        StatusCode::UNAUTHORIZED
+    );
+    let cookie = login_as(&app, "owner", "strong-password123").await;
+    sqlx::query("UPDATE sessions SET created=?")
+        .bind(now() - ABSOLUTE)
+        .execute(&app.db)
+        .await
+        .unwrap();
+    assert_eq!(
+        request(&app, "GET", "/api/session", &cookie, json!(null))
+            .await
+            .status(),
+        StatusCode::UNAUTHORIZED
+    );
+}
+#[tokio::test]
+async fn accounts_enforce_roles_and_reset_revokes_sessions() {
+    let (app, _dir) = app().await;
+    let owner = setup(&app).await;
+    assert_eq!(
+        request(
+            &app,
+            "POST",
+            "/api/accounts",
+            &owner,
+            json!({"username":"viewer","password":"viewer-password123"})
+        )
+        .await
+        .status(),
+        StatusCode::CREATED
+    );
+    assert_eq!(
+        request(
+            &app,
+            "POST",
+            "/api/accounts",
+            &owner,
+            json!({"username":"VIEWER","password":"viewer-password123"})
+        )
+        .await
+        .status(),
+        StatusCode::CONFLICT
+    );
+    let viewer = login_as(&app, "viewer", "viewer-password123").await;
+    assert_eq!(
+        request(&app, "GET", "/api/accounts", &viewer, json!(null))
+            .await
+            .status(),
+        StatusCode::FORBIDDEN
+    );
+    assert_eq!(
+        request(&app, "GET", "/api/sessions", &viewer, json!(null))
+            .await
+            .status(),
+        StatusCode::FORBIDDEN
+    );
+    assert_eq!(
+        request(
+            &app,
+            "POST",
+            "/api/accounts",
+            &viewer,
+            json!({"username":"intruder","password":"viewer-password123"})
+        )
+        .await
+        .status(),
+        StatusCode::FORBIDDEN
+    );
+    let me = json_body(request(&app, "GET", "/api/session", &viewer, json!(null)).await).await;
+    let uid = me["id"].as_str().unwrap();
+    assert_eq!(
+        request(
+            &app,
+            "POST",
+            &format!("/api/accounts/{uid}/password"),
+            &owner,
+            json!({"password":"replacement-pass123"})
+        )
+        .await
+        .status(),
+        StatusCode::NO_CONTENT
+    );
+    assert_eq!(
+        request(&app, "GET", "/api/session", &viewer, json!(null))
+            .await
+            .status(),
+        StatusCode::UNAUTHORIZED
+    );
+    let viewer = login_as(&app, "viewer", "replacement-pass123").await;
+    assert_eq!(
+        request(
+            &app,
+            "DELETE",
+            &format!("/api/accounts/{uid}"),
+            &owner,
+            json!(null)
+        )
+        .await
+        .status(),
+        StatusCode::NO_CONTENT
+    );
+    assert_eq!(
+        request(&app, "GET", "/api/session", &viewer, json!(null))
+            .await
+            .status(),
+        StatusCode::UNAUTHORIZED
+    );
+    let owner_me = json_body(request(&app, "GET", "/api/session", &owner, json!(null)).await).await;
+    assert_eq!(
+        request(
+            &app,
+            "DELETE",
+            &format!("/api/accounts/{}", owner_me["id"].as_str().unwrap()),
+            &owner,
+            json!(null)
+        )
+        .await
+        .status(),
+        StatusCode::BAD_REQUEST
+    );
+}
+#[tokio::test]
+async fn origins_and_bruteforce_are_rejected() {
+    let (app, _dir) = app().await;
+    let req = Request::builder()
+        .method("POST")
+        .uri("/api/login")
+        .header("origin", "https://evil.example")
+        .body(Body::empty())
+        .unwrap();
+    assert_eq!(
+        router(app.clone(), "/none".into())
+            .oneshot(req)
+            .await
+            .unwrap()
+            .status(),
+        StatusCode::FORBIDDEN
+    );
+    let req = Request::builder()
+        .uri("/api/signal")
+        .header("origin", "https://evil.example")
+        .body(Body::empty())
+        .unwrap();
+    assert_eq!(
+        router(app.clone(), "/none".into())
+            .oneshot(req)
+            .await
+            .unwrap()
+            .status(),
+        StatusCode::FORBIDDEN
+    );
+    for _ in 0..10 {
+        assert!(throttle(&app, "test".into()).await.is_ok());
+    }
+    assert_eq!(
+        throttle(&app, "test".into()).await.unwrap_err().0,
+        StatusCode::TOO_MANY_REQUESTS
+    );
+}
+#[tokio::test]
+async fn logout_revokes_the_cookie() {
+    let (app, _dir) = app().await;
+    let owner = setup(&app).await;
+    assert_eq!(
+        request(&app, "POST", "/api/logout", &owner, json!(null))
+            .await
+            .status(),
+        StatusCode::NO_CONTENT
+    );
+    assert_eq!(
+        request(&app, "GET", "/api/session", &owner, json!(null))
+            .await
+            .status(),
+        StatusCode::UNAUTHORIZED
+    );
+}
+#[tokio::test]
+async fn revoking_viewer_closes_transport_and_notifies_host() {
+    let (app, _dir) = app().await;
+    let (host_tx, mut host_rx) = mpsc::channel(16);
+    let (viewer_tx, mut viewer_rx) = mpsc::channel(16);
+    {
+        let mut peers = app.peers.lock().await;
+        peers.host = Some("host".into());
+        peers.clients.insert(
+            "host".into(),
+            Client {
+                session: "hs".into(),
+                user: "h".into(),
+                name: "h".into(),
+                mode: "host".into(),
+                tx: host_tx,
+                shutdown: tokio::sync::watch::channel(false).0,
+            },
+        );
+        peers.clients.insert(
+            "viewer".into(),
+            Client {
+                session: "vs".into(),
+                user: "v".into(),
+                name: "v".into(),
+                mode: "viewer".into(),
+                tx: viewer_tx,
+                shutdown: tokio::sync::watch::channel(false).0,
+            },
+        );
+    }
+    disconnect_session(&app, "vs").await;
+    assert!(matches!(viewer_rx.recv().await, Some(Message::Close(_))));
+    let Some(Message::Text(msg)) = host_rx.recv().await else {
+        panic!("missing leave notification")
+    };
+    assert_eq!(
+        serde_json::from_str::<Value>(&msg).unwrap()["type"],
+        "viewer-left"
+    );
+    assert!(!app.peers.lock().await.clients.contains_key("viewer"));
+}
+#[tokio::test]
+async fn revoking_host_notifies_viewers_to_stop_media() {
+    let (app, _dir) = app().await;
+    let (host_tx, _host_rx) = mpsc::channel(16);
+    let (viewer_tx, mut viewer_rx) = mpsc::channel(16);
+    {
+        let mut peers = app.peers.lock().await;
+        peers.host = Some("host".into());
+        peers.since = Some(now());
+        peers.clients.insert(
+            "host".into(),
+            Client {
+                session: "hs".into(),
+                user: "h".into(),
+                name: "h".into(),
+                mode: "host".into(),
+                tx: host_tx,
+                shutdown: tokio::sync::watch::channel(false).0,
+            },
+        );
+        peers.clients.insert(
+            "viewer".into(),
+            Client {
+                session: "vs".into(),
+                user: "v".into(),
+                name: "v".into(),
+                mode: "viewer".into(),
+                tx: viewer_tx,
+                shutdown: tokio::sync::watch::channel(false).0,
+            },
+        );
+    }
+    disconnect_session(&app, "hs").await;
+    let Some(Message::Text(msg)) = viewer_rx.recv().await else {
+        panic!("missing offline notification")
+    };
+    assert_eq!(
+        serde_json::from_str::<Value>(&msg).unwrap()["type"],
+        "host-offline"
+    );
+    assert!(app.peers.lock().await.host.is_none());
+}
+
+#[tokio::test]
+async fn websocket_routes_only_authorized_peers_and_closes_on_revocation() {
+    use tokio_tungstenite::{
+        connect_async,
+        tungstenite::{Message as WsMessage, client::IntoClientRequest},
+    };
+    let (app, _dir) = app().await;
+    let host_cookie = setup(&app).await;
+    request(
+        &app,
+        "POST",
+        "/api/accounts",
+        &host_cookie,
+        json!({"username":"viewer","password":"viewer-password123"}),
+    )
+    .await;
+    let viewer_cookie = login_as(&app, "viewer", "viewer-password123").await;
+    let viewer_me =
+        json_body(request(&app, "GET", "/api/session", &viewer_cookie, json!(null)).await).await;
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
+    let service = router(app.clone(), "/nonexistent".into())
+        .into_make_service_with_connect_info::<SocketAddr>();
+    let server = tokio::spawn(async move {
+        axum::serve(listener, service)
+            .with_graceful_shutdown(async {
+                let _ = shutdown_rx.await;
+            })
+            .await
+            .unwrap();
+    });
+    let ws_request = |mode: &str, cookie: &str| {
+        let mut req = format!("ws://{addr}/api/signal?mode={mode}")
+            .into_client_request()
+            .unwrap();
+        req.headers_mut()
+            .insert("origin", "http://localhost:3000".parse().unwrap());
+        req.headers_mut().insert("cookie", cookie.parse().unwrap());
+        req
+    };
+    let forbidden = connect_async(ws_request("host", &viewer_cookie))
+        .await
+        .unwrap_err();
+    assert!(
+        matches!(forbidden, tokio_tungstenite::tungstenite::Error::Http(ref r) if r.status()==StatusCode::FORBIDDEN)
+    );
+    let anonymous = connect_async(ws_request("viewer", "")).await.unwrap_err();
+    assert!(
+        matches!(anonymous, tokio_tungstenite::tungstenite::Error::Http(ref r) if r.status()==StatusCode::UNAUTHORIZED)
+    );
+    let (mut host, _) = connect_async(ws_request("host", &host_cookie))
+        .await
+        .unwrap();
+    let host_welcome = host.next().await.unwrap().unwrap().into_text().unwrap();
+    let host_id = serde_json::from_str::<Value>(&host_welcome).unwrap()["peer"]
+        .as_str()
+        .unwrap()
+        .to_owned();
+    let (mut viewer, _) = connect_async(ws_request("viewer", &viewer_cookie))
+        .await
+        .unwrap();
+    let viewer_welcome = viewer.next().await.unwrap().unwrap().into_text().unwrap();
+    let viewer_id = serde_json::from_str::<Value>(&viewer_welcome).unwrap()["peer"]
+        .as_str()
+        .unwrap()
+        .to_owned();
+    let joined = host.next().await.unwrap().unwrap().into_text().unwrap();
+    assert_eq!(
+        serde_json::from_str::<Value>(&joined).unwrap()["type"],
+        "viewer-joined"
+    );
+    host.send(WsMessage::Text(json!({"type":"offer","target":viewer_id,"peer":"forged","data":{"type":"offer","sdp":"test"}}).to_string().into())).await.unwrap();
+    let offer = viewer.next().await.unwrap().unwrap().into_text().unwrap();
+    let offer: Value = serde_json::from_str(&offer).unwrap();
+    assert_eq!(offer["peer"], host_id);
+    assert_eq!(offer["type"], "offer");
+    viewer
+        .send(WsMessage::Text(
+            json!({"type":"offer","target":host_id,"data":{}})
+                .to_string()
+                .into(),
+        ))
+        .await
+        .unwrap();
+    assert!(
+        tokio::time::timeout(std::time::Duration::from_millis(100), host.next())
+            .await
+            .is_err()
+    );
+    viewer
+        .send(WsMessage::Text(
+            json!({"type":"answer","target":host_id,"data":{"type":"answer","sdp":"test"}})
+                .to_string()
+                .into(),
+        ))
+        .await
+        .unwrap();
+    let answer = host.next().await.unwrap().unwrap().into_text().unwrap();
+    assert_eq!(
+        serde_json::from_str::<Value>(&answer).unwrap()["peer"],
+        viewer_id
+    );
+    revoke(&app, viewer_me["session_id"].as_str().unwrap())
+        .await
+        .unwrap();
+    let closed = tokio::time::timeout(std::time::Duration::from_secs(3), viewer.next())
+        .await
+        .unwrap();
+    assert!(matches!(closed, Some(Ok(WsMessage::Close(_)))));
+    let left = host.next().await.unwrap().unwrap().into_text().unwrap();
+    assert_eq!(
+        serde_json::from_str::<Value>(&left).unwrap()["type"],
+        "viewer-left"
+    );
+    host.close(None).await.unwrap();
+    drop(viewer);
+    drop(host);
+    let _ = shutdown_tx.send(());
+    tokio::time::timeout(std::time::Duration::from_secs(3), server)
+        .await
+        .unwrap()
+        .unwrap();
+}
+
+#[tokio::test]
+async fn saturated_signaling_queue_fails_closed() {
+    let (tx, _rx) = mpsc::channel(1);
+    let (shutdown, rx) = tokio::sync::watch::channel(false);
+    let client = Client {
+        session: "s".into(),
+        user: "u".into(),
+        name: "u".into(),
+        mode: "host".into(),
+        tx,
+        shutdown,
+    };
+    send(&client, json!({"type":"offer"}));
+    assert!(!*rx.borrow());
+    send(&client, json!({"type":"viewer-left"}));
+    assert!(*rx.borrow());
+}