crates/nornir-saga/src/ui/web.rs
//! Web frontend (feature `web`): an `axum` server + Server-Sent Events that make
//! a **replay actually move in the browser**. Mirrors the knut-pipelines `web`
//! viewer pattern — a frontend over the *same* [`Viewer`](crate::ui::Viewer)
//! model + [`Player`](crate::ui::Player) as the CLI, never a replacement.
//!
//! Routes:
//! * `GET /` — a **self-contained** HTML page (inline SVG + vanilla JS,
//! no external CDN) that draws the cursor, its motion trail, the current
//! route, counters, and the event log, and re-animates them live;
//! * `GET /saga.json` — the [`Saga`] being replayed;
//! * `GET /events` — an SSE stream of [`Viewer::state_json`] snapshots, one
//! per replayed movement, **paced by the player's time-scaled delay** so the
//! page moves the way the session was recorded (speed-adjustable).
//!
//! **Demoable with no live Njord:** `nornir saga replay --web` (or the built-in
//! [`demo_saga`](crate::ui::demo_saga)) serves a synthetic saga and streams it
//! over SSE so the page animates.
use std::convert::Infallible;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use axum::extract::State;
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{Html, IntoResponse};
use axum::routing::get;
use axum::Router;
use futures_util::Stream;
use futures_util::StreamExt;
use tokio::sync::mpsc;
use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::ui::msg::Saga;
use crate::ui::player::Player;
/// Shared server state: the saga to replay + how to pace it.
#[derive(Clone)]
struct AppState {
saga: Arc<Saga>,
/// Playback speed multiplier (1.0 = real time).
speed: f64,
/// Whether to pace frames by the saga's (scaled) inter-event timing. `false`
/// (tests) streams every frame instantly.
paced: bool,
}
/// The longest a single inter-frame wait may be, so a big recorded gap never
/// stalls the demo for minutes.
const MAX_STEP_DELAY: Duration = Duration::from_millis(1500);
/// Build the axum router for `saga`, paced at `speed` (or instant when `!paced`).
fn router(saga: Saga, speed: f64, paced: bool) -> Router {
let state = AppState { saga: Arc::new(saga), speed, paced };
Router::new()
.route("/", get(index))
.route("/saga.json", get(saga_json))
.route("/events", get(events))
.with_state(state)
}
/// Serve the web replay viewer on `addr` until the process is stopped.
pub async fn serve(saga: Saga, speed: f64, addr: SocketAddr) -> anyhow::Result<()> {
let app = router(saga, speed, true);
let listener = tokio::net::TcpListener::bind(addr)
.await
.map_err(|e| anyhow::anyhow!("bind {addr}: {e}"))?;
let bound = listener.local_addr().map_err(|e| anyhow::anyhow!(e.to_string()))?;
eprintln!("# nornir saga web replay on http://{bound}");
axum::serve(listener, app)
.await
.map_err(|e| anyhow::anyhow!("web serve: {e}"))
}
/// `GET /` — the self-contained HTML page.
async fn index() -> Html<&'static str> {
Html(INDEX_HTML)
}
/// `GET /saga.json` — the saga being replayed.
async fn saga_json(State(st): State<AppState>) -> impl IntoResponse {
(
[(axum::http::header::CONTENT_TYPE, "application/json")],
st.saga.to_json(),
)
}
/// `GET /events` — SSE stream of viewer-state snapshots. Each connection gets
/// its own replay through a fresh [`Player`].
async fn events(State(st): State<AppState>) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
let (tx, rx) = mpsc::unbounded_channel::<String>();
tokio::spawn(run_replay(st, tx));
let stream = UnboundedReceiverStream::new(rx).map(|data| Ok(Event::default().data(data)));
Sse::new(stream).keep_alive(KeepAlive::default())
}
/// Drive one connection's replay into the SSE channel, paced by the player.
async fn run_replay(st: AppState, tx: mpsc::UnboundedSender<String>) {
let mut player = Player::new((*st.saga).clone()).with_speed(st.speed);
// Initial (empty-model) frame so the page starts clean.
let _ = tx.send(player.state_json());
while !player.finished() {
if st.paced {
let d = player.next_delay_ms().min(MAX_STEP_DELAY.as_millis() as u64);
if d > 0 {
tokio::time::sleep(Duration::from_millis(d)).await;
}
}
match player.step() {
Some(frame) => {
if tx.send(frame.state_json).is_err() {
return; // client hung up
}
}
None => break,
}
}
let _ = tx.send(r#"{"done":true}"#.to_string());
}
/// The self-contained web page: inline SVG + vanilla-JS SSE client. No external
/// requests (the SVG namespace URI is a constant, not a fetch).
const INDEX_HTML: &str = r##"<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>nornir saga — UI-movement replay</title>
<style>
:root { color-scheme: dark; }
body { margin:0; font:14px/1.4 system-ui,sans-serif; background:#0f1419; color:#e6e6e6; }
header { padding:10px 16px; background:#161b22; border-bottom:1px solid #30363d; }
header b { color:#58a6ff; }
#meta { color:#8b949e; margin-top:4px; font-size:13px; }
#stage { position:relative; }
svg { display:block; background:#0b0f14; }
.trail { stroke:#1f6feb; fill:none; stroke-width:2; opacity:.7; }
.cursor { fill:#f0883e; stroke:#ffd7a8; stroke-width:2; }
.ping { fill:none; stroke:#f85149; stroke-width:2; }
#log { padding:6px 16px; color:#8b949e; font-family:ui-monospace,monospace;
font-size:12px; white-space:pre; max-height:180px; overflow:auto;
border-top:1px solid #30363d; }
</style></head>
<body>
<header>
<b>nornir saga</b> — UI-movement replay
<div id="meta">connecting…</div>
</header>
<div id="stage"><svg id="svg" width="820" height="520" viewBox="0 0 820 520">
<path id="trail" class="trail"></path>
<circle id="ping" class="ping" r="0"></circle>
<circle id="cursor" class="cursor" r="8" cx="-20" cy="-20"></circle>
</svg></div>
<div id="log"></div>
<script>
const svg = document.getElementById("svg");
const cursor = document.getElementById("cursor");
const trailEl = document.getElementById("trail");
const pingEl = document.getElementById("ping");
const metaEl = document.getElementById("meta");
const logEl = document.getElementById("log");
let lastClicks = 0;
function apply(s) {
const c = s.cursor || {x:-20,y:-20};
cursor.setAttribute("cx", c.x); cursor.setAttribute("cy", c.y);
const t = s.trail || [];
if (t.length) trailEl.setAttribute("d", "M" + t.map(p => p.x + "," + p.y).join(" L"));
// pulse a ping ring on a new click.
const clicks = (s.counts && s.counts.clicks) || 0;
if (clicks > lastClicks) {
lastClicks = clicks;
pingEl.setAttribute("cx", c.x); pingEl.setAttribute("cy", c.y);
let r = 4; pingEl.setAttribute("r", r);
const iv = setInterval(() => { r += 4; pingEl.setAttribute("r", r);
pingEl.style.opacity = Math.max(0, 1 - r/40);
if (r > 40) { clearInterval(iv); pingEl.setAttribute("r", 0); pingEl.style.opacity = 1; } }, 30);
}
const cc = s.counts || {};
metaEl.textContent =
`route ${s.route||"—"} · focus ${s.focus||"—"} · scroll ${(s.scroll&&s.scroll.y)||0} — `
+ `${cc.moves||0} moves · ${cc.clicks||0} clicks · ${cc.navs||0} navs · `
+ `${cc.edits||0} edits · applied ${s.applied||0} @ ${s.t_ms||0}ms`;
if (s.log) logEl.textContent = s.log.join("\n");
}
const es = new EventSource("events");
es.onmessage = (e) => {
const s = JSON.parse(e.data);
if (s.done) { metaEl.textContent += " · replay complete"; es.close(); return; }
apply(s);
};
es.onerror = () => { /* EventSource auto-reconnects; leave the last frame up */ };
</script>
</body></html>"##;
#[cfg(test)]
mod tests {
use super::*;
use crate::ui::demo_saga;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use http_body_util::BodyExt;
use tower::ServiceExt; // oneshot
async fn body_string(resp: axum::response::Response) -> String {
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
String::from_utf8(bytes.to_vec()).unwrap()
}
#[tokio::test]
async fn index_serves_a_self_contained_page() {
let app = router(demo_saga(), 1.0, false);
let resp = app
.oneshot(Request::builder().uri("/").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = body_string(resp).await;
assert!(body.contains("EventSource(\"events\")"), "wires the SSE client");
assert!(!body.contains("http://"), "no external CDN/host fetch");
assert!(!body.to_lowercase().contains("https://cdn"), "no CDN");
}
#[tokio::test]
async fn saga_json_endpoint_serves_the_saga() {
let app = router(demo_saga(), 1.0, false);
let resp = app
.oneshot(Request::builder().uri("/saga.json").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = body_string(resp).await;
let v: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(v["version"], Saga::VERSION);
assert!(v["events"].as_array().unwrap().len() > 40);
}
#[tokio::test]
async fn events_endpoint_streams_snapshots_to_completion() {
// Instant (unpaced) stream so the test doesn't wait on saga timing.
let app = router(demo_saga(), 1.0, false);
let resp = app
.oneshot(Request::builder().uri("/events").body(Body::empty()).unwrap())
.await
.unwrap();
assert_eq!(resp.status(), StatusCode::OK);
let body = body_string(resp).await;
// SSE frames are `data: <json>\n\n`. The last data frame is the done marker.
let datas: Vec<&str> = body
.lines()
.filter_map(|l| l.strip_prefix("data: "))
.collect();
assert!(datas.len() > 40, "one frame per movement + initial + done");
assert_eq!(*datas.last().unwrap(), r#"{"done":true}"#, "terminal marker closes the stream");
// A mid-stream frame is a real viewer snapshot.
let mid: serde_json::Value = serde_json::from_str(datas[datas.len() / 2]).unwrap();
assert!(mid.get("counts").is_some(), "frames carry the viewer model");
}
}