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;

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, 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);
    let _ = app.bus.send(Event::State {
        ts: snap.ts,
        snapshot: snap.clone(),
    });
    *app.last.lock().unwrap() = Some(snap);
}

/// `GET /events` — forward every bus event as an SSE message.
async fn events_stream(
    app: State<AppState>,
) -> Sse<impl futures::Stream<Item = Result<SseEvent, std::convert::Infallible>>> {
    let rx = app.bus.subscribe();
    Sse::new(futures::stream::unfold(rx, |mut rx| async move {
        loop {
            match rx.recv().await {
                Ok(ev) => {
                    let json = serde_json::to_string(&ev).unwrap_or_default();
                    return Some((Ok(SseEvent::default().data(json)), rx));
                }
                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,
            }
        }
    }))
}

/// 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 {
        push_state(&app).await;
        tokio::time::sleep(Duration::from_secs(POLL_SECS)).await;
    }
}

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));
    Router::new()
        .route("/", get(|| async { Html(UI_HTML) }))
        .merge(protected)
        .with_state(app)
}

/// 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;