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::chat::{is_valid_room, room_subject};

use super::AppState;

/// GET /sse/{room} — stream the room's 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"));
    }

    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"),
    ))
}