//! 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::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/...` file loads) plus the /// latest rows JSON (each SSE connection clones the receiver). #[derive(Clone)] pub(crate) struct AppState { pub(crate) config: Arc, pub(crate) data: tokio::sync::watch::Receiver, } /// 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: ¬ify::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, tx: tokio::sync::watch::Sender, keepalive: Duration, ) { let (ev_tx, mut ev_rx) = tokio::sync::mpsc::unbounded_channel::<()>(); let watcher = notify::recommended_watcher(move |res: Result| { 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 json = serde_json::to_string(&collect(&cfg)).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, ) -> Sse>> { 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)) } /// `GET /r//` — 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 { 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() }) } /// `GET /r//` — file viewer page: rendered markdown (with a /// raw toggle), syntax-highlighted TOML/JSON, plain logs. const VIEW_TEMPLATE: &str = include_str!("view.html"); async fn report_file( axum::extract::State(state): axum::extract::State, 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"); let meta = serde_json::json!({ "recipe": recipe, "file": file, "content": clean }); 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( "404report not found" .to_string(), ), ) .into_response(), } } /// Build the report server router: static page at `/`, live SSE at /// `/events`, full report text at `/r//`. `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, page: String, initial: String) -> Router { let (tx, rx) = tokio::sync::watch::channel(initial); tokio::spawn(refresh_loop(config.clone(), tx, KEEPALIVE)); 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)) .with_state(AppState { config, data: rx }) }