dahua-camera-server 0.3.1

//! `GET /api/events` — the merged event feed as Server-Sent Events.
//!
//! # Why SSE, and why its own connection
//!
//! Video has its own WebSocket per viewer, carrying fMP4 fragments and nothing
//! else. Events go here, over a **separate** stream, and the two are never
//! multiplexed. That is a hard rule, not a style choice:
//!
//! - A burst of analytics events must never be able to delay a frame.
//! - An event client that stops reading must never apply backpressure that
//!   reaches the video path.
//!
//! Sharing one socket would couple them on both counts, and the failure would
//! show up as video stuttering whenever the camera got busy — the exact
//! symptom invariant 1 exists to prevent.
//!
//! SSE rather than a second WebSocket because the traffic is one-directional
//! and low-rate: it reconnects on its own in every browser, survives proxies,
//! and needs no framing logic on the client.
//!
//! # Backpressure
//!
//! The feed reads from a bounded broadcast. A client that falls behind gets
//! `Lagged` and **skips** the events it missed rather than stalling the
//! sender. Events are advisory; a slow dashboard losing a few is correct
//! behaviour, and an unbounded queue here is how the service OOMs.

use crate::AppState;
use axum::extract::State;
use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
use axum::response::IntoResponse;
use std::convert::Infallible;
use std::time::Duration;
use tokio::sync::broadcast::error::RecvError;

/// How often to send a comment frame when nothing is happening.
///
/// Cameras are quiet for hours. Without this, an idle proxy closes the
/// connection and the client reconnects on a timer it did not choose.
const KEEPALIVE: Duration = Duration::from_secs(15);

/// `GET /api/events` — every camera's events, as they happen.
pub async fn events_sse(State(state): State<AppState>) -> impl IntoResponse {
    let mut events = state.service.subscribe_events();

    let stream = async_stream::stream! {
        loop {
            match events.recv().await {
                Ok(event) => {
                    let Ok(json) = serde_json::to_string(&event) else { continue };
                    yield Ok::<_, Infallible>(
                        SseEvent::default().event(event.kind.name()).data(json),
                    );
                }
                // Advisory data: a client that fell behind during a burst
                // skips what it missed rather than stalling the sender.
                Err(RecvError::Lagged(skipped)) => {
                    tracing::debug!(skipped, "event client lagged");
                    yield Ok(SseEvent::default().event("lagged").data(skipped.to_string()));
                }
                Err(RecvError::Closed) => break,
            }
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::new().interval(KEEPALIVE).text("keep-alive"))
}