wsb 0.155.49

use axum::{extract::State, response::sse::{Event, KeepAlive, Sse}};
use futures::stream::Stream;
use serde::Serialize;
use std::convert::Infallible;
use crate::web::state::WebState;

/// Sent as the first SSE event to give the newly-connected client recent history.
#[derive(Serialize)]
struct LogHistory<'a> {
    lines: &'a [crate::logging::LogLine],
}

/// GET /api/v1/logs
///
/// SSE stream of log lines.
///
/// Protocol:
/// - On connect, a single `history` event is sent containing the last N log
///   lines (as a JSON object `{ lines: [...] }`) so the client immediately
///   sees context rather than a blank panel.
/// - After that, each new log line is sent as a `log` event with the
///   `LogLine` JSON as the data payload.
/// - If the client falls behind in the broadcast buffer (lagged), the
///   stream continues from the newest available message (some lines are lost).
pub async fn stream_logs(
    State(state): State<WebState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    // Subscribe before taking the snapshot so we don't race-lose events
    // that arrive between snapshot and subscribe.
    let mut rx = state.log_tx.subscribe();
    let history = state.log_ring.snapshot().await;

    let stream = async_stream::stream! {
        // Emit recent history as a single batch event so the client
        // can render all past lines before the live stream begins.
        let hist = LogHistory { lines: &history };
        match serde_json::to_string(&hist) {
            Ok(json) => yield Ok(Event::default().event("history").data(json)),
            Err(e) => {
                // Use eprintln — same reason: must not feed back into the broadcast channel.
                eprintln!("[wsb] log SSE history serialize error: {e}");
                return;
            }
        }

        loop {
            match rx.recv().await {
                Ok(line) => {
                    match serde_json::to_string(&line) {
                        Ok(json) => yield Ok(Event::default().event("log").data(json)),
                        Err(e) => {
                            // Serialization failure is a bug. Use eprintln rather than
                            // tracing to avoid feeding back into the broadcast channel.
                            eprintln!("[wsb] log SSE serialize error: {e}");
                        }
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                    // Fell behind — continue from the newest message.
                    // MUST NOT use tracing! here: any log emitted inside the log
                    // SSE handler feeds back into the broadcast channel, creating a
                    // self-reinforcing loop of lag warnings.
                    eprintln!("[wsb] log SSE client lagged by {n} events");
                    continue;
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}