launcher/src/report/serve.rs

//! Live-update plumbing for `report --serve`: a background refresher that
//! re-scans the recipe `reports/` dirs when the file system says something
//! changed, and an SSE endpoint (`/events`) that pushes the rows JSON to the
//! page whenever the scan result differs (a new report file, a bench
//! finishing, a log growing).
//!
//! The refresher is event-driven, not a fixed poll: a `notify` watcher
//! (FSEvents on macOS, inotify on Linux) over the recipe dirs feeds an mpsc
//! channel; after a short debounce burst the rows are re-collected and
//! published on the `watch` channel only when the serialization changed.
//! Filesystem events are filtered to the report-relevant paths (anything
//! under a `reports/` dir, or a `recipe.toml`), so build churn elsewhere in
//! the repo is ignored. A slow keepalive scan stays as a safety net for
//! missed notifications (legacy log-only reports, dirs created late).
//!
//! One refresher serves all clients: each SSE connection forwards its own
//! `watch::Receiver`.

use crate::config::Config;
use crate::report::collect;
use axum::{
    Router,
    extract::Path,
    response::{Response, Sse, sse::Event as SseEvent},
    routing::get,
};
use futures::{Stream, StreamExt};
use notify::Watcher;
use std::collections::HashMap;
use std::convert::Infallible;
use std::sync::Arc;
use std::time::Duration;

/// Safety-net rescan interval: covers missed watcher notifications (a
/// legacy `.log`-only report whose score lands in a Modify the filter
/// ignores, a source dir created after startup). The normal path is the
/// event-driven one, so this can stay slow.
pub(crate) const KEEPALIVE: Duration = Duration::from_secs(30);

/// Wait after a filesystem event before re-scanning, so a write burst
/// (the launcher copying a report triplet) collapses into one collect.
const DEBOUNCE: Duration = Duration::from_millis(500);

/// Shared server state: the config (for `/r/...` and `/repo/...` file
/// loads) plus the latest rows JSON (each SSE connection clones the
/// receiver) plus the URL base this server is mounted under ("" for
/// `report --serve`, "/recipes" when the dashboard nests it) — the viewer
/// link targets carry it.
#[derive(Clone)]
pub(crate) struct AppState {
    pub(crate) config: Arc<Config>,
    pub(crate) data: tokio::sync::watch::Receiver<String>,
    pub(crate) base: String,
}

/// True when the event touches a report-relevant path: anything under a
/// `reports/` directory, or a `recipe.toml` (row fields come from it).
/// Everything else (build churn, git, the rest of the repo) is ignored so
/// the watcher can watch a wide tree cheaply.
fn is_relevant(ev: &notify::Event) -> bool {
    ev.paths.iter().any(|p| {
        p.ancestors()
            .any(|a| a.file_name().is_some_and(|n| n == "reports"))
            || p.file_name().is_some_and(|n| n == "recipe.toml")
    })
}

/// Background refresher — lazy by design (the web process must stay small;
/// a full collect scans every recipe + report file): a full re-collect runs
/// ONLY on a debounced filesystem event (or once at startup); the keepalive
/// tick just re-runs the live stale cross-check (the only row field that
/// changes without a file change) and only when a run is open — otherwise it
/// does zero work. The cross-check rides on the main dashboard's fleet
/// probe (`fleet_rx` bus State events — no extra SSH) and falls back to a
/// direct probe when no bus is wired (CLI `report --serve`).
pub(crate) async fn refresh_loop(
    cfg: Arc<Config>,
    tx: tokio::sync::watch::Sender<String>,
    keepalive: Duration,
    archived: bool,
    fleet_rx: Option<tokio::sync::broadcast::Receiver<crate::web::events::Event>>,
) {
    let mut fleet_rx = fleet_rx;
    let (ev_tx, mut ev_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
    let watcher = notify::recommended_watcher(move |res: Result<notify::Event, notify::Error>| {
        if let Ok(ev) = res
            && is_relevant(&ev)
        {
            let _ = ev_tx.send(());
        }
    });
    let mut watcher = watcher.ok();
    if let Some(w) = &mut watcher {
        for dir in crate::config::watchable_recipe_dirs(&cfg) {
            let _ = w.watch(&dir, notify::RecursiveMode::Recursive);
        }
    }
    let mut last = String::new();
    let mut rows: Vec<crate::report::ReportRow> = Vec::new();
    let mut fleet: Option<crate::web::events::Snapshot> = None;
    let cluster_nodes: HashMap<String, Vec<String>> = cfg
        .clusters
        .iter()
        .map(|c| {
            (
                c.name.clone(),
                c.nodes.iter().map(|n| n.name.clone()).collect(),
            )
        })
        .collect();
    let has_open_run = |rows: &[crate::report::ReportRow]| {
        rows.iter()
            .any(|r| r.up.is_some() && r.up_cluster.is_some())
    };
    loop {
        tokio::select! {
            // Filesystem event (debounced) → the only full-rescan trigger.
            _ = ev_rx.recv() => {
                // debounce: let the burst settle, then drain the remaining events
                tokio::time::sleep(DEBOUNCE).await;
                while ev_rx.try_recv().is_ok() {}
                rows = collect(&cfg, archived);
                cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
                let json = serde_json::to_string(&rows).unwrap_or_default();
                if json != last {
                    last = json;
                    let _ = tx.send(last.clone());
                }
            }
            // Bus State event (the main dashboard's 30 s fleet probe) →
            // just track the fleet; the keepalive tick consumes it.
            ev = async {
                match fleet_rx.as_mut() {
                    Some(rx) => rx.recv().await,
                    None => std::future::pending::<Result<
                        crate::web::events::Event,
                        tokio::sync::broadcast::error::RecvError,
                    > >().await,
                }
            } => {
                if let Ok(crate::web::events::Event::State { snapshot, .. }) = ev {
                    fleet = Some(snapshot);
                }
            }
            // Keepalive safety net (missed watcher events): full collect only
            // while uncollected (startup); afterwards only the cheap
            // cross-check — and only when a run is open.
            _ = tokio::time::sleep(keepalive) => {
                if rows.is_empty() {
                    rows = collect(&cfg, archived);
                    cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
                } else if has_open_run(&rows) {
                    cross_check(&cfg, &mut rows, &cluster_nodes, &mut fleet).await;
                } else {
                    continue; // nothing to do — zero work, zero allocation
                }
                let json = serde_json::to_string(&rows).unwrap_or_default();
                if json != last {
                    last = json;
                    let _ = tx.send(last.clone());
                }
            }
        }
    }
}

/// Reconcile the open-ledger rows against the live fleet: prefer the main
/// dashboard's last fleet snapshot (no extra SSH); probe directly only when
/// no snapshot has arrived yet (CLI `report --serve`, first 30 s of web).
async fn cross_check(
    cfg: &Config,
    rows: &mut [crate::report::ReportRow],
    cluster_nodes: &HashMap<String, Vec<String>>,
    fleet: &mut Option<crate::web::events::Snapshot>,
) {
    let (reachable, live) = match fleet {
        Some(snap) => {
            let mut reachable: HashMap<String, bool> = HashMap::new();
            let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
            for c in &snap.clusters {
                for n in &c.nodes {
                    reachable.insert(n.name.clone(), n.reachable);
                    for ct in &n.containers {
                        live.insert(ct.recipe.clone());
                    }
                }
            }
            (reachable, live)
        }
        None => {
            let f = crate::web::probe::probe(cfg).await;
            let mut reachable: HashMap<String, bool> = HashMap::new();
            let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
            for (node, st) in &f.nodes {
                reachable.insert(node.clone(), st.reachable);
                for c in &st.containers {
                    live.insert(c.recipe.clone());
                }
            }
            (reachable, live)
        }
    };
    crate::report::stale::mark_stale(rows, cluster_nodes, &reachable, &live);
    // The inverse direction: a container in the fleet with NO local ledger
    // run (the add-on machine) still renders as running.
    crate::report::stale::mark_fleet_up(rows, &live);
}

/// `GET /events` — the latest rows JSON on connect, then one message per
/// refresher update. The stream ends when the refresher (the sender) drops.
async fn events_stream(
    axum::extract::State(s): axum::extract::State<AppState>,
) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
    let mut rx = s.data;
    let first = rx.borrow_and_update().clone();
    let first = futures::stream::once(async move { Ok(SseEvent::default().data(first)) });
    let rest = futures::stream::unfold(rx, |mut rx| async move {
        if rx.changed().await.is_err() {
            return None;
        }
        let json = rx.borrow_and_update().clone();
        Some((Ok(SseEvent::default().data(json)), rx))
    });
    Sse::new(first.chain(rest))
        // the rows only change when a bench report changes — a silent
        // stream would be indistinguishable from a dead one for anything
        // between the server and the browser
        .keep_alive(axum::response::sse::KeepAlive::new().interval(Duration::from_secs(15)))
}

/// `GET /r/<recipe>/<file>` — full file text (report, log, json, or the
/// recipe's own `recipe.toml`) served through the viewer page. Both path
/// parts must be plain names (no traversal); the resolve walks every
/// recipe dir, so stay strict.
pub(crate) fn load_report(config: &Config, recipe: &str, file: &str) -> Option<String> {
    let plain = |s: &str| {
        !s.contains("..")
            && s.chars()
                .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
    };
    if !plain(recipe) || !plain(file) {
        return None;
    }
    crate::config::recipe_dirs(config)
        .into_iter()
        .find_map(|dir| {
            let base = dir.join(recipe);
            // "recipe.toml" resolves from the recipe dir root, everything
            // else from its reports/ subdir
            let p = if file == "recipe.toml" {
                base.join("recipe.toml")
            } else {
                base.join("reports").join(file)
            };
            p.is_file()
                .then(|| std::fs::read_to_string(p).ok())
                .flatten()
        })
}

/// Braille spinner frame or progress bar — llama-benchy's rich in-place
/// progress redraws.
fn is_spinner_frame(line: &str) -> bool {
    let t = line.trim_start();
    t.starts_with(['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']) || t.contains('━')
}

/// Progress-redraw identity: the spinner frames differ only by the glyph,
/// the elapsed clock and the run/progress counters — same key = same phase
/// redrawn in place.
fn progress_key(line: &str) -> String {
    line.split_whitespace()
        .filter(|t| {
            // keep only wordish tokens (the test name): bars — the tip glyph
            // varies as the bar advances (━╸╺…) — spinner glyphs, @ signs and
            // other graphics carry no alphanumerics; counters (0/18), elapsed
            // clocks (0:01:48) and the "run N" pair are progress churn.
            t.chars().any(|c| c.is_alphanumeric())
                && !t.chars().all(|c| c.is_ascii_digit() || c == '/')
                && !t.contains(':')
                && *t != "run"
        })
        .collect::<Vec<_>>()
        .join(" ")
}

/// Make old pty-captured bench logs read like the new ones: drop pty
/// control-char artifacts (`^D`/backspace echoes), collapse each spinner
/// phase's redraw run down to its last frame, drop consecutive duplicate
/// lines. Idempotent on logs that are already clean.
pub(crate) fn clean_log(content: &str) -> String {
    let lines: Vec<String> = content
        .lines()
        .map(|l| {
            let l: String = l
                .chars()
                .filter(|c| !matches!(c, '\u{4}' | '\u{8}'))
                .collect();
            // the pty echoes control chars as caret notation (^D, ^H) —
            // printable '^' + @.._ pairs at the line start
            let mut l = l.as_str();
            while l.starts_with('^')
                && l[1..]
                    .chars()
                    .next()
                    .is_some_and(|c| ('@'..='_').contains(&c))
            {
                l = &l[2..];
            }
            l.to_string()
        })
        .collect();
    let mut keep = vec![true; lines.len()];
    let mut i = 0;
    while i < lines.len() {
        let prog = is_spinner_frame(&lines[i]);
        let key = if prog {
            progress_key(&lines[i])
        } else {
            String::new()
        };
        let mut j = i + 1;
        while j < lines.len() {
            let same = if prog {
                is_spinner_frame(&lines[j]) && progress_key(&lines[j]) == key
            } else {
                !lines[i].is_empty() && lines[j] == lines[i]
            };
            if !same {
                break;
            }
            keep[j] = false;
            j += 1;
        }
        if j > i + 1 && prog {
            // a redraw run keeps only its final frame (exact dups keep the first)
            keep[i] = false;
            keep[j - 1] = true;
        }
        i = j;
    }
    lines
        .iter()
        .enumerate()
        .filter(|(k, _)| keep[*k])
        .map(|(_, l)| l.as_str())
        .collect::<Vec<_>>()
        .join("\n")
}

/// `GET /r/<recipe>/<file>` with `?follow=1` — SSE stream of a `.log`
/// report file, updated as the file grows (the bench-log follow). The first
/// message carries the full cleaned content (the client's baseline); each
/// subsequent message carries only the appended tail (the delta since the
/// last poll). The client appends deltas to its baseline. The stream polls
/// the file every 2 s and ends when the client disconnects. Non-`.log`
/// files and a missing `?follow=1` fall through to the static `report_file`.
async fn report_file_or_stream(
    axum::extract::State(state): axum::extract::State<AppState>,
    axum::extract::Path((recipe, file)): Path<(String, String)>,
    axum::extract::Query(params): axum::extract::Query<std::collections::HashMap<String, String>>,
) -> axum::response::Response {
    use axum::response::IntoResponse;
    let follow = params.get("follow").is_some_and(|s| s == "1");
    if !follow || !file.ends_with(".log") {
        return report_file(state, Path((recipe, file))).await;
    }
    let config = state.config.clone();
    type SseRes = Result<SseEvent, std::convert::Infallible>;
    let stream = futures::stream::unfold(
        (config, recipe, file, 0usize),
        |mut st: (Arc<Config>, String, String, usize)| async move {
            let (cfg, recipe, file, last_len) = &mut st;
            if let Some(content) = load_report(cfg, recipe, file) {
                let clean = crate::ansi::strip_ansi(content.as_bytes());
                let clean = String::from_utf8_lossy(&clean)
                    .replace("\r\n", "\n")
                    .replace('\r', "\n");
                let clean = clean_log(&clean);
                if clean.len() > *last_len {
                    let delta = clean.split_at(*last_len).1.to_string();
                    *last_len = clean.len();
                    return Some((
                        Ok(SseEvent::default().data(delta)) as SseRes,
                        (cfg.clone(), recipe.clone(), file.clone(), *last_len),
                    ));
                }
            }
            tokio::time::sleep(Duration::from_secs(2)).await;
            Some((
                Ok(SseEvent::default().comment("keepalive")) as SseRes,
                (cfg.clone(), recipe.clone(), file.clone(), *last_len),
            ))
        },
    );
    Sse::new(stream)
        .keep_alive(axum::response::sse::KeepAlive::new().interval(Duration::from_secs(15)))
        .into_response()
}

/// `GET /r/<recipe>/<file>` — file viewer page: rendered markdown (with a
/// raw toggle), syntax-highlighted TOML/JSON, plain logs.
pub(crate) const VIEW_TEMPLATE: &str = include_str!("view.html");

async fn report_file(
    state: AppState,
    axum::extract::Path((recipe, file)): Path<(String, String)>,
) -> Response {
    use axum::http::StatusCode;
    use axum::response::IntoResponse;
    match load_report(&state.config, &recipe, &file) {
        Some(text) => {
            // Strip ANSI escapes (old logs were saved raw; new ones are
            // already clean) and normalize line ends: the old pty-saved
            // logs carry CRLF (ONLCR) and the Live-soup ones bare \r —
            // CRLF first, then lone \r, so nothing doubles.
            let clean = crate::ansi::strip_ansi(text.as_bytes());
            let clean = String::from_utf8_lossy(&clean)
                .replace("\r\n", "\n")
                .replace('\r', "\n");
            // old pty-captured logs carry spinner-redraw soup and pty
            // control-char artifacts — render them like the clean ones
            let clean = if file.ends_with(".log") {
                clean_log(&clean)
            } else {
                clean
            };
            let links = if file == "recipe.toml" {
                super::repo::toml_links(&clean, &recipe, &state.base)
            } else {
                Vec::new()
            };
            let meta = serde_json::json!({ "recipe": recipe, "file": file, "content": clean, "links": links, "base": state.base });
            let meta = serde_json::to_string(&meta)
                .unwrap_or_default()
                .replace('<', "\\u003c");
            axum::response::Html(VIEW_TEMPLATE.replace("{{META}}", &meta)).into_response()
        }
        None => (
            StatusCode::NOT_FOUND,
            axum::response::Html(
                "<!doctype html><meta charset=\"utf-8\"><title>404</title>report not found"
                    .to_string(),
            ),
        )
            .into_response(),
    }
}

/// Build the report server router: static page at `/`, live SSE at
/// `/events`, full report text at `/r/<recipe>/<file>`. `initial` is the
/// rows JSON snapshot used to seed the watch channel (from the same
/// `collect` that rendered the page, so the first paint and the first SSE
/// message agree).
pub(crate) fn router(
    config: Arc<Config>,
    page: String,
    initial: String,
    archived: bool,
    base: &str,
    fleet_rx: Option<tokio::sync::broadcast::Receiver<crate::web::events::Event>>,
) -> Router {
    let (tx, rx) = tokio::sync::watch::channel(initial);
    tokio::spawn(refresh_loop(
        config.clone(),
        tx,
        KEEPALIVE,
        archived,
        fleet_rx,
    ));
    Router::new()
        .route(
            "/",
            get(move || {
                let page = page.clone();
                async move { axum::response::Html(page) }
            }),
        )
        .route("/events", get(events_stream))
        .route("/r/{recipe}/{file}", get(report_file_or_stream))
        .route("/repo/{*path}", get(super::repo::repo_route))
        .with_state(AppState {
            config,
            data: rx,
            base: base.to_string(),
        })
}