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: watch the recipe dirs, re-collect after a debounced
/// filesystem event (or the keepalive safety net), publish the rows JSON on
/// the watch channel only when it changed. The first collect runs
/// immediately so clients see current data without waiting. If the watcher
/// cannot be created (unsupported FS), the keepalive alone keeps the data
/// fresh.
pub(crate) async fn refresh_loop(
    cfg: Arc<Config>,
    tx: tokio::sync::watch::Sender<String>,
    keepalive: Duration,
    archived: bool,
) {
    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::recipe_dirs(&cfg) {
            if dir.is_dir() {
                let _ = w.watch(&dir, notify::RecursiveMode::Recursive);
            }
        }
    }
    let mut last = String::new();
    loop {
        tokio::select! {
            _ = ev_rx.recv() => {}
            _ = tokio::time::sleep(keepalive) => {}
        }
        // debounce: let the burst settle, then drain the remaining events
        tokio::time::sleep(DEBOUNCE).await;
        while ev_rx.try_recv().is_ok() {}
        let mut rows = collect(&cfg, archived);
        // Live cross-check: an open ledger run with no running container
        // anywhere (all its nodes reachable) is marked stale — otherwise a
        // killed serve or a node that died before `stop` could verify would
        // show as "up" forever. No open run with a cluster → no probe.
        if rows
            .iter()
            .any(|r| r.up.is_some() && r.up_cluster.is_some())
        {
            let fleet = crate::web::probe::probe(&cfg).await;
            let mut cluster_nodes: HashMap<String, Vec<String>> = HashMap::new();
            for c in &cfg.clusters {
                cluster_nodes.insert(
                    c.name.clone(),
                    c.nodes.iter().map(|n| n.name.clone()).collect(),
                );
            }
            let mut reachable: HashMap<String, bool> = HashMap::new();
            let mut live: std::collections::HashSet<String> = std::collections::HashSet::new();
            for (node, st) in &fleet.nodes {
                reachable.insert(node.clone(), st.reachable);
                for c in &st.containers {
                    live.insert(c.recipe.clone());
                }
            }
            crate::report::stale::mark_stale(&mut rows, &cluster_nodes, &reachable, &live);
        }
        let json = serde_json::to_string(&rows).unwrap_or_default();
        if json != last {
            last = json;
            let _ = tx.send(last.clone());
        }
    }
}

/// `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>` — 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(
    axum::extract::State(state): axum::extract::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 });
            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,
) -> Router {
    let (tx, rx) = tokio::sync::watch::channel(initial);
    tokio::spawn(refresh_loop(config.clone(), tx, KEEPALIVE, archived));
    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))
        .route("/repo/{*path}", get(super::repo::repo_route))
        .with_state(AppState {
            config,
            data: rx,
            base: base.to_string(),
        })
}