wsb-daemon 0.248.6265

//! The one stream the page listens on.
//!
//! A browser allows about six connections to one origin, and an event source
//! holds its connection open for as long as it lives. Every long-lived stream
//! is therefore a connection the page can never use for a request — and this
//! app had three of them running at all times (workspace changes, background
//! operations, log output), plus the ones a session, an operation view or the
//! issue list open on top. With six held, ordinary requests do not fail: they
//! *queue*, for ever, until something else lets go. That is what a tree that
//! spins and never opens looks like from the inside.
//!
//! So everything that is pushed to every client, regardless of what the user
//! is looking at, is carried here on one connection and told apart by event
//! name. What remains separate is what cannot be shared: a stream scoped to
//! one session, one operation's output, or a filtered issue query.

use wsb::prelude::*;
use crate::web::ops::OpsEvent;
use crate::web::state::WebState;
use axum::{
    extract::State,
    response::sse::{Event, KeepAlive, Sse},
};
use futures::stream::Stream;
use std::convert::Infallible;
use tokio::sync::broadcast::error::RecvError;

use super::workspace::WorkspaceChangedEvent;

/// The first log event a client receives: the recent lines it missed.
#[derive(Serialize)]
struct LogHistory<'a> {
    lines: &'a [wsb::logging::LogLine],
}

/// GET /api/v1/events — every workspace-wide push, on one connection.
///
/// Event names:
/// - `connected` once, when the stream is live. A client that has state from
///   before this moment cannot know what it missed while it was disconnected,
///   so this is its cue to re-read what it holds.
/// - `changed` / `file_changed` — the workspace tree and files.
/// - `github` — the GitHub mirror moved: which scopes, so a page re-reads only
///   what it is showing.
/// - `ops` — the background operation registry, snapshot first.
/// - `history` / `log` — recent log lines, then each new one.
///
/// Every channel is subscribed *before* its snapshot is taken, so nothing that
/// happens between the two is lost.
pub async fn events_stream(
    State(state): State<WebState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let mut git_rx = state.git_status_tx.subscribe();
    let mut file_rx = state.file_changed_tx.subscribe();
    let mut ws_changed_rx = state.workspace_changed_tx.subscribe();
    // The daemon's log is the process's, the same for every worktree.
    let (log_tx, log_ring) = wsb::logging::log_stream();
    let mut log_rx = log_tx.subscribe();
    // A workspace with no GitHub mirror has nothing to announce, and waits on
    // nothing rather than on a channel that would close and end the stream.
    let mut github_rx = state.github.as_ref().ok().map(|github| github.changed_tx.subscribe());
    let registry = state.ops.clone();
    let mut ops_rx = registry.subscribe();

    // Reconciled, like the REST snapshot and by the same function: a client
    // that opened the stream must be told what a client that asked would be
    // told. Two answers to one question is one answer nobody is looking at,
    // and it would be the stale one.
    let ops_snapshot = state.ops.snapshot().await;
    let log_history = log_ring.snapshot().await;

    let stream = async_stream::stream! {
        yield Ok(Event::default().event("connected").data("{}"));

        let initial_ops = OpsEvent::Snapshot { ops: ops_snapshot };
        if let Ok(json) = serde_json::to_string(&initial_ops) {
            yield Ok(Event::default().event("ops").data(json));
        }

        let history = LogHistory { lines: &log_history };
        if let Ok(json) = serde_json::to_string(&history) {
            yield Ok(Event::default().event("history").data(json));
        }

        loop {
            tokio::select! {
                git_result = git_rx.recv() => {
                    match git_result {
                        Ok(repo) => {
                            let changed = WorkspaceChangedEvent::repo_status(&repo.path);
                            if let Ok(json) = serde_json::to_string(&changed) {
                                yield Ok(Event::default().event("changed").data(json));
                            }
                        }
                        Err(RecvError::Lagged(_)) => {
                            // We cannot know which listings moved; say so.
                            let changed = WorkspaceChangedEvent::everything();
                            if let Ok(json) = serde_json::to_string(&changed) {
                                yield Ok(Event::default().event("changed").data(json));
                            }
                        }
                        Err(RecvError::Closed) => break,
                    }
                }
                ws_result = ws_changed_rx.recv() => {
                    match ws_result {
                        Ok(changed) => {
                            if let Ok(json) = serde_json::to_string(&changed) {
                                yield Ok(Event::default().event("changed").data(json));
                            }
                        }
                        Err(RecvError::Lagged(_)) => {
                            let changed = WorkspaceChangedEvent::everything();
                            if let Ok(json) = serde_json::to_string(&changed) {
                                yield Ok(Event::default().event("changed").data(json));
                            }
                        }
                        Err(RecvError::Closed) => break,
                    }
                }
                github_result = async {
                    match github_rx.as_mut() {
                        Some(rx) => rx.recv().await,
                        None => std::future::pending().await,
                    }
                } => {
                    match github_result {
                        Ok(changed) => {
                            if let Ok(json) = serde_json::to_string(&changed) {
                                yield Ok(Event::default().event("github").data(json));
                            }
                        }
                        // Fell behind: which scopes moved is unknown, so every
                        // one of them might have. An empty list means "re-read
                        // whatever you are showing".
                        Err(RecvError::Lagged(_)) => {
                            let changed = crate::web::github::GitHubChanged { scopes: Vec::new() };
                            if let Ok(json) = serde_json::to_string(&changed) {
                                yield Ok(Event::default().event("github").data(json));
                            }
                        }
                        Err(RecvError::Closed) => break,
                    }
                }
                file_result = file_rx.recv() => {
                    match file_result {
                        Ok(event) => {
                            if let Ok(json) = serde_json::to_string(&event) {
                                yield Ok(Event::default().event("file_changed").data(json));
                            }
                        }
                        Err(RecvError::Lagged(_)) => {
                            // Fell behind — an empty path means "every open file".
                            let event = crate::web::state::FileChangedEvent {
                                path: String::new(),
                                mtime_secs: None,
                            };
                            if let Ok(json) = serde_json::to_string(&event) {
                                yield Ok(Event::default().event("file_changed").data(json));
                            }
                        }
                        Err(RecvError::Closed) => break,
                    }
                }
                ops_result = ops_rx.recv() => {
                    match ops_result {
                        Ok(event) => {
                            if let Ok(json) = serde_json::to_string(&event) {
                                yield Ok(Event::default().event("ops").data(json));
                            }
                        }
                        Err(RecvError::Lagged(n)) => {
                            // Fell behind — the full snapshot makes the client
                            // consistent again.
                            tracing::warn!("events SSE client lagged by {n} ops events, re-sending snapshot");
                            let recovery = OpsEvent::Snapshot { ops: registry.snapshot().await };
                            if let Ok(json) = serde_json::to_string(&recovery) {
                                yield Ok(Event::default().event("ops").data(json));
                            }
                        }
                        Err(RecvError::Closed) => break,
                    }
                }
                log_result = log_rx.recv() => {
                    match log_result {
                        Ok(line) => {
                            // eprintln, never tracing, on failure here: a log
                            // line about logging feeds straight back into this
                            // channel.
                            match serde_json::to_string(&line) {
                                Ok(json) => yield Ok(Event::default().event("log").data(json)),
                                Err(e) => wsb::errln!("[wsb] log SSE serialize error: {e}"),
                            }
                        }
                        // Lagged log lines are lost; the stream continues from
                        // the newest available line.
                        Err(RecvError::Lagged(_)) => {}
                        Err(RecvError::Closed) => break,
                    }
                }
            }
        }
    };

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