cnats/src/server/sse.rs

//! Server-Sent Events bridge: one NATS subscription per connected browser,
//! scoped to a single room subject.

use std::{convert::Infallible, time::Duration};

use axum::{
    extract::{Path, State},
    http::StatusCode,
    response::sse::{Event, KeepAlive, Sse},
};
use futures::{Stream, StreamExt};
use tower_sessions::Session;

use crate::auth::{User, SESSION_USER_KEY};
use crate::call::call_subject;
use crate::chat::{is_authorized_for_room, is_valid_room, room_subject};

use super::AppState;

/// 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)> {
    authorize_room_stream(&room, &session).await?;

    let subscriber = state
        .nats
        .subscribe(room_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("message")
            .data(String::from_utf8_lossy(&msg.payload).into_owned()))
    });

    Ok(Sse::new(stream).keep_alive(
        KeepAlive::new()
            .interval(Duration::from_secs(15))
            .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"),
    ))
}