launcher/src/web/mod.rs

//! `sm121 web` — axum + SSE dashboard.
//!
//! Read-only LED board (state snapshots every 30 s + on job finish),
//! actions — serve/stop/power run as tracked jobs calling the same
//! handlers the CLI uses — and live log windows (SSH `docker logs -f`
//! attaches with ring-buffer history). Data endpoints are token-gated
//! when `--token` is set (required when binding off-loopback).
//!
//! Design: `specs/tech-architecture/web-dashboard.md`.

mod api;
pub mod events;
pub mod jobs;
pub mod logs;
pub mod probe;
pub mod ssh_stream;
pub mod topo;

use crate::config::Config;
use crate::web::events::{Event, build_snapshot};
use axum::{
    Router,
    extract::State,
    http::{HeaderMap, Request, StatusCode},
    middleware::{self, Next},
    response::{
        Html, Response,
        sse::{Event as SseEvent, KeepAlive, Sse},
    },
    routing::{get, post},
};
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;
use tokio::time::Duration;

const UI_HTML: &str = include_str!("ui.html");
const POLL_SECS: u64 = 30;
const BUS_CAP: usize = 1024;

#[derive(Clone)]
pub struct AppState {
    pub config: Arc<Config>,
    pub bus: broadcast::Sender<Event>,
    pub token: Option<String>,
    /// Latest snapshot (written by the supervisor; `/api/state` serves it).
    pub last: Arc<Mutex<Option<events::Snapshot>>>,
    /// Tracked action jobs (serve/stop/power); capped at `jobs::MAX_JOBS`.
    pub jobs: Arc<Mutex<Vec<jobs::Job>>>,
    /// Tracked log streams; capped at `logs::MAX_ATTACHES`.
    pub log_attaches: Arc<Mutex<Vec<logs::LogAttach>>>,
}

/// Probe the fleet and broadcast a fresh state snapshot (supervisor tick
/// and job-finish both route through here).
pub(crate) async fn push_state(app: &AppState) {
    let fleet = probe::probe(&app.config).await;
    let jobs = app.jobs.lock().unwrap().clone();
    let attaches: Vec<events::AttachSnap> = app
        .log_attaches
        .lock()
        .unwrap()
        .iter()
        .map(|a| events::AttachSnap {
            id: a.id.clone(),
            cluster: a.cluster.clone(),
            node: a.node.clone(),
            recipe: a.recipe.clone(),
            rank: a.rank,
            done: a.done.load(std::sync::atomic::Ordering::Relaxed),
        })
        .collect();
    let snap = build_snapshot(&app.config, &fleet, &jobs, &attaches, &fleet.ha);
    let _ = app.bus.send(Event::State {
        ts: snap.ts,
        snapshot: snap.clone(),
    });
    *lock_last(app) = Some(snap);
}

/// `GET /events` — forward every bus event as an SSE message. A fresh
/// client gets the cached snapshot first, so it paints (and flips its
/// live indicator) at once instead of waiting for the next poll tick.
async fn events_stream(
    app: State<AppState>,
) -> Sse<impl futures::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
    let head = app
        .last
        .lock()
        .unwrap()
        .clone()
        .map(|snapshot| Event::State {
            ts: snapshot.ts,
            snapshot,
        });
    let rx = app.bus.subscribe();
    Sse::new(futures::stream::unfold(
        (rx, head),
        |(mut rx, mut head)| async move {
            if let Some(ev) = head.take() {
                return Some((sse_event(&ev), (rx, None)));
            }
            loop {
                match rx.recv().await {
                    Ok(ev) => return Some((sse_event(&ev), (rx, None))),
                    Err(broadcast::error::RecvError::Lagged(n)) => {
                        tracing::debug!("SSE client lagged {n} events");
                        // Same receiver is still valid — keep reading.
                    }
                    Err(broadcast::error::RecvError::Closed) => return None,
                }
            }
        },
    ))
    // comment lines between events: NAT/VPN/proxy idle timers (and frozen
    // tabs on reconnect) see traffic, so a dead stream is noticed in seconds,
    // not minutes
    .keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
}

fn sse_event(ev: &Event) -> Result<SseEvent, std::convert::Infallible> {
    Ok(SseEvent::default().data(serde_json::to_string(ev).unwrap_or_default()))
}

/// Bearer-token gate for data endpoints (`/` stays public — it is static
/// HTML with no secrets). Accepts `Authorization: Bearer <t>` or `?token=<t>`
/// (SSE `EventSource` cannot set headers, so the UI uses the query form).
async fn token_guard(
    State(app): State<AppState>,
    headers: HeaderMap,
    req: Request<axum::body::Body>,
    next: Next,
) -> Result<Response, (StatusCode, String)> {
    if let Some(expected) = &app.token {
        // ponytail: raw query param, no percent-decoding — tokens are
        // operator-generated; decode if exotic tokens are ever used.
        let from_query = req.uri().query().and_then(|q| {
            q.split('&').find_map(|kv| {
                kv.split_once('=')
                    .and_then(|(k, v)| (k == "token").then_some(v))
            })
        });
        let got = headers
            .get("authorization")
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "))
            .or(from_query);
        if got != Some(expected.as_str()) {
            return Err((StatusCode::UNAUTHORIZED, "unauthorized".into()));
        }
    }
    Ok(next.run(req).await)
}

/// Supervisor: poll the fleet and broadcast a state snapshot every tick.
async fn supervisor(app: AppState) {
    loop {
        // isolate each tick: a probe bug or a poisoned lock must not kill
        // the poll loop for good (a dead supervisor means a dead live feed)
        let tick = app.clone();
        if let Err(e) = tokio::spawn(async move { push_state(&tick).await }).await {
            tracing::warn!("sm121 web: supervisor tick panicked: {e}");
        }
        tokio::time::sleep(Duration::from_secs(POLL_SECS)).await;
    }
}

/// `AppState.last` guard that survives a poisoned mutex (a panic while
/// holding the lock would otherwise take the whole live feed down).
fn lock_last(app: &AppState) -> std::sync::MutexGuard<'_, Option<events::Snapshot>> {
    app.last
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

fn is_loopback(addr: SocketAddr) -> bool {
    addr.ip().is_loopback()
}

/// Build the router (separate from `run` so tests can drive it without
/// binding a port).
pub fn router(app: AppState) -> Router {
    let protected = Router::new()
        .route("/api/state", get(api::state))
        .route("/api/serve", post(api::serve))
        .route("/api/stop", post(api::stop))
        .route("/api/power", post(api::power))
        .route("/api/recipes", get(api::recipes))
        .route(
            "/api/logs",
            get(api::logs_history)
                .post(api::logs_attach)
                .delete(api::logs_detach),
        )
        .route("/api/raw/config", get(api::raw_config))
        .route("/api/raw/recipe/{name}", get(api::raw_recipe))
        .route("/api/raw/effective", get(api::raw_effective))
        .route("/events", get(events_stream))
        .layer(middleware::from_fn_with_state(app.clone(), token_guard));
    // Recipes panel: the very page `sm121 report --serve` serves, nested at
    // `/recipes` — its own SSE stream and `/r/<recipe>/<file>` viewer move
    // under the prefix (see `report::html_page_at`), live refresh included.
    let rows = crate::report::collect(&app.config, false);
    let recipes = crate::report::serve::router(
        app.config.clone(),
        crate::report::html_page_at(&rows, "/recipes"),
        serde_json::to_string(&rows).unwrap_or_default(),
        false,
    )
    .layer(middleware::from_fn_with_state(app.clone(), token_guard));
    Router::new()
        .route("/", get(|| async { Html(UI_HTML) }))
        .merge(protected)
        .with_state(app)
        .nest("/recipes", recipes)
}

/// Run the dashboard until the process exits.
pub async fn run(
    config: Config,
    port: u16,
    bind: String,
    token: Option<String>,
    open: bool,
) -> anyhow::Result<()> {
    let addr: SocketAddr = format!("{bind}:{port}").parse()?;
    if !is_loopback(addr) && token.is_none() {
        anyhow::bail!(
            "refusing to bind {bind} (non-loopback) without --token — the dashboard can act on the cluster"
        );
    }
    let (tx, _) = broadcast::channel(BUS_CAP);
    let app = AppState {
        config: Arc::new(config),
        bus: tx,
        token,
        last: Arc::new(Mutex::new(None)),
        jobs: Arc::new(Mutex::new(Vec::new())),
        log_attaches: Arc::new(Mutex::new(Vec::new())),
    };
    tokio::spawn(supervisor(app.clone()));
    let listener = tokio::net::TcpListener::bind(addr).await?;
    if open {
        let url = format!("http://{addr}/");
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(400)).await;
            let prog = if cfg!(target_os = "macos") {
                "open"
            } else {
                "xdg-open"
            };
            let _ = std::process::Command::new(prog).arg(&url).spawn();
        });
    }
    tracing::info!("sm121 web: http://{addr}/ (poll every {POLL_SECS}s)");
    axum::serve(listener, router(app)).await?;
    Ok(())
}

#[cfg(test)]
mod api_tests;
#[cfg(test)]
mod tests;
#[cfg(test)]
mod ui_js_tests;