|
|
@@ -0,0 +1,142 @@
|
|
|
+//! Live-update plumbing for `report --serve`: a background refresher that
|
|
|
+//! re-scans the recipe `reports/` dirs on a tick, and an SSE endpoint
|
|
|
+//! (`/events`) that pushes the rows JSON to the page whenever anything
|
|
|
+//! changed (a new report file, a bench finishing, a log growing).
|
|
|
+//!
|
|
|
+//! One refresher serves all clients: it publishes the rows JSON on a `watch`
|
|
|
+//! channel only when the serialization differs from the last publish, and
|
|
|
+//! 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 std::convert::Infallible;
|
|
|
+use std::sync::Arc;
|
|
|
+use std::time::Duration;
|
|
|
+
|
|
|
+/// How often the refresher re-scans the report dirs. The scan is a local
|
|
|
+/// dir walk + small-file reads (a few dozen ms); 5 s keeps a finishing bench
|
|
|
+/// visible in the page within one poll while staying negligible I/O.
|
|
|
+pub(crate) const REFRESH_TICK: Duration = Duration::from_secs(5);
|
|
|
+
|
|
|
+/// 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<Config>,
|
|
|
+ pub(crate) data: tokio::sync::watch::Receiver<String>,
|
|
|
+}
|
|
|
+
|
|
|
+/// Background refresher: collect the rows, publish the JSON when it changed,
|
|
|
+/// sleep, repeat. The first collect runs immediately so clients see current
|
|
|
+/// data without waiting a tick. Runs on the tokio runtime; the scan is fast
|
|
|
+/// local disk I/O.
|
|
|
+pub(crate) async fn refresh_loop(
|
|
|
+ cfg: Arc<Config>,
|
|
|
+ tx: tokio::sync::watch::Sender<String>,
|
|
|
+ tick: Duration,
|
|
|
+) {
|
|
|
+ let mut last = String::new();
|
|
|
+ loop {
|
|
|
+ let json = serde_json::to_string(&collect(&cfg)).unwrap_or_default();
|
|
|
+ if json != last {
|
|
|
+ last = json;
|
|
|
+ let _ = tx.send(last.clone());
|
|
|
+ }
|
|
|
+ tokio::time::sleep(tick).await;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+/// `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))
|
|
|
+}
|
|
|
+
|
|
|
+/// `GET /r/<recipe>/<file>` — full report text in a minimal pre 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 p = dir.join(recipe).join("reports").join(file);
|
|
|
+ p.is_file()
|
|
|
+ .then(|| std::fs::read_to_string(p).ok())
|
|
|
+ .flatten()
|
|
|
+ })
|
|
|
+}
|
|
|
+
|
|
|
+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) => axum::response::Html(format!(
|
|
|
+ "<!doctype html><meta charset=\"utf-8\"><title>{recipe} — {file}</title>\
|
|
|
+ <pre style=\"font:12px/1.45 ui-monospace,monospace;white-space:pre-wrap;\
|
|
|
+ word-break:break-word;margin:16px;\">{}</pre>",
|
|
|
+ text.replace('&', "&")
|
|
|
+ .replace('<', "<")
|
|
|
+ .replace('>', ">")
|
|
|
+ ))
|
|
|
+ .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, tick: Duration) -> Router {
|
|
|
+ let (tx, rx) = tokio::sync::watch::channel(initial);
|
|
|
+ tokio::spawn(refresh_loop(config.clone(), tx, tick));
|
|
|
+ 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 })
|
|
|
+}
|
