launcher/src/web/api.rs

//! REST endpoints. P1: `GET /api/state` (latest snapshot; probes once if
//! the supervisor hasn't ticked yet).

use axum::{extract::State, http::StatusCode, response::Json};
use serde::Deserialize;

use super::AppState;
use super::events::Snapshot;

pub(crate) async fn state(app: State<AppState>) -> Json<Snapshot> {
    if app.last.lock().unwrap().is_none() {
        // First request before the first supervisor tick: probe now so the
        // UI gets data immediately (probe is read-only `docker ps` per node
        // and degrades to `reachable: false` on failure). Two racing first
        // requests may both probe — harmless (idempotent, last write wins).
        super::push_state(&app).await;
    }
    app.last
        .lock()
        .unwrap()
        .as_ref()
        .map(|s| Json(s.clone()))
        .unwrap()
}

// ---------------------------------------------------------------------------
// P2: actions (serve/stop/power run as tracked jobs via the same handlers
// the CLI uses) + recipe list.

type ApiError = (StatusCode, String);

#[derive(Deserialize)]
pub(crate) struct ServeReq {
    pub cluster: String,
    pub recipe: String,
    #[serde(default)]
    pub node: Option<String>,
    #[serde(default)]
    pub solo: Option<bool>,
    #[serde(default)]
    pub force: bool,
    #[serde(default)]
    pub fit: Option<String>,
    #[serde(default)]
    pub clear: bool,
    #[serde(default)]
    pub mirror: Option<String>,
    #[serde(default)]
    pub keep: bool,
    #[serde(default)]
    pub dry_run: bool,
}

/// `POST /api/serve` — same dispatch as the CLI (docker vs tmux by
/// `recipe.deploy`), run as a tracked job.
pub(crate) async fn serve(
    app: State<AppState>,
    Json(req): Json<ServeReq>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let id = super::jobs::new_job(&app, "serve", &req.cluster, Some(&req.recipe))?;
    let config = app.config.clone();
    let app2 = (*app).clone();
    let opts = crate::docker::ServeOpts {
        keep: req.keep,
        skip_before_serve: false,
        no_wait_ready: false,
        dry_run: req.dry_run,
        auto_power_on: false,
        sync_image: true,
        sync_image_source: None,
        no_push: false,
        image_source: crate::docker::ImageSource::Repo,
        build_node: None,
        force_build: false,
        force: req.force,
        solo: req.solo,
        mirror: req.mirror,
        hf_cache: None,
        stage: None,
        fit: req.fit,
        clear: req.clear,
    };
    let recipe = req.recipe.clone();
    let cluster = req.cluster.clone();
    let node = req.node.clone();
    super::jobs::spawn(app2, id.clone(), async move {
        crate::handle_serve(&config, &recipe, &cluster, node, opts, false, false).await
    });
    Ok(Json(serde_json::json!({ "job": id })))
}

#[derive(Deserialize)]
pub(crate) struct StopReq {
    pub cluster: String,
    pub recipe: String,
    #[serde(default)]
    pub node: Option<String>,
}

/// `POST /api/stop` — stops the recipe's containers (auto-discovers the
/// cluster when the recipe is running exactly once), as a tracked job.
pub(crate) async fn stop(
    app: State<AppState>,
    Json(req): Json<StopReq>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let id = super::jobs::new_job(&app, "stop", &req.cluster, Some(&req.recipe))?;
    let config = app.config.clone();
    let app2 = (*app).clone();
    let recipe = req.recipe.clone();
    let cluster = req.cluster.clone();
    let node = req.node.clone();
    super::jobs::spawn(app2, id.clone(), async move {
        let nodes = vec![node.unwrap_or_default()];
        let nodes = if nodes[0].is_empty() { vec![] } else { nodes };
        crate::docker::stop::handle(&config, &recipe, &cluster, &nodes, true, false, false).await
    });
    Ok(Json(serde_json::json!({ "job": id })))
}

#[derive(Deserialize)]
pub(crate) struct PowerReq {
    /// "off" | "on" | "switch"
    pub action: String,
    pub cluster: String,
    /// switch target (required only for `switch`)
    #[serde(default)]
    pub to: Option<String>,
}

/// `POST /api/power` — HA power actions as a tracked job. The UI enforces
/// the type-the-name confirm; this endpoint trusts the caller (it is
/// token-gated / loopback-bound).
pub(crate) async fn power(
    app: State<AppState>,
    Json(req): Json<PowerReq>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let id = super::jobs::new_job(&app, "power", &req.cluster, None)?;
    let config = app.config.clone();
    let app2 = (*app).clone();
    let action = req.action.clone();
    let cluster = req.cluster.clone();
    let to = req.to.clone();
    let stale = stale_from_snapshot(&app, &req.cluster);
    super::jobs::spawn_report(app2, id.clone(), async move {
        match action.as_str() {
            "off" => {
                let c = crate::config::resolve_cluster_name(&config.clusters, &cluster)?;
                crate::power::power_off(&config, c, &[], false).await?;
                Ok(Vec::new())
            }
            "on" => {
                let c = crate::config::resolve_cluster_name(&config.clusters, &cluster)?;
                // nodes the last probe saw down: their plugs may be stale-on
                // (a graceful shutdown leaves the plug on), so power on
                // cycles those instead of issuing a no-op turn_on
                crate::power::power_on(&config, c, &[], false, &stale).await
            }
            "switch" => {
                let to = to
                    .as_deref()
                    .ok_or_else(|| anyhow::anyhow!("`to` cluster is required for power switch"))?;
                let from = crate::config::resolve_cluster_name(&config.clusters, &cluster)?;
                let to = crate::config::resolve_cluster_name(&config.clusters, to)?;
                crate::power::power_switch(&config, from, to, false).await?;
                Ok(Vec::new())
            }
            other => Err(anyhow::anyhow!(
                "unknown power action '{other}' (off|on|switch)"
            )),
        }
    });
    Ok(Json(serde_json::json!({ "job": id })))
}

/// Nodes of `cluster` that the last fleet probe saw unreachable — the
/// `power on` cycle list (a plug left on by a graceful shutdown needs a
/// cycle, not a no-op `turn_on`).
pub(crate) fn stale_from_snapshot(app: &AppState, cluster: &str) -> Vec<String> {
    let snap = app
        .last
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
        .clone();
    let Some(snap) = snap else { return Vec::new() };
    let mut out = Vec::new();
    for c in &snap.clusters {
        if c.name != cluster {
            continue;
        }
        for n in &c.nodes {
            if !n.reachable {
                out.push(n.name.clone());
            }
        }
    }
    out
}

/// `GET /api/recipes` — all discoverable recipes (cwd + `recipe_sources` +
/// the global config dir), with deploy target + node requirements.
pub(crate) async fn recipes(app: State<AppState>) -> Json<serde_json::Value> {
    let dirs = crate::config::recipe_dirs(&app.config);
    let mut out = Vec::new();
    for dir in dirs {
        let mut entries = match tokio::fs::read_dir(&dir).await {
            Ok(e) => e,
            Err(_) => continue,
        };
        loop {
            let Ok(Some(entry)) = entries.next_entry().await else {
                break;
            };
            let path = entry.path();
            if !path.is_dir() {
                continue;
            }
            let Some(name) = entry.file_name().to_str().map(String::from) else {
                continue;
            };
            let file = path.join("recipe.toml");
            let Ok(toml_str) = tokio::fs::read_to_string(&file).await else {
                continue;
            };
            match toml::from_str::<crate::config::Recipe>(&toml_str) {
                Ok(r) => out.push(serde_json::json!({
                    "name": name,
                    "engine": r.engine,
                    "deploy": format!("{:?}", r.deploy).to_lowercase(),
                    "required_nodes": r.required_nodes,
                })),
                Err(e) => tracing::debug!("skipping recipe {}: {e}", name),
            }
        }
    }
    Json(serde_json::json!({ "recipes": out }))
}

/// `POST /api/repos/sync` — clone/pull every configured recipe repo right
/// now (the same best-effort batch the command start runs; unreachable
/// repos are skipped, never fatal). A single in-flight guard: a second
/// sync while one is running gets a 409, not a second git storm. The
/// recipes panel's ⟳ button calls it; the report refresher's file watcher
/// then picks up the new/changed recipes on its own.
pub(crate) async fn repos_sync(app: State<AppState>) -> Result<Json<serde_json::Value>, ApiError> {
    static SYNCING: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
    if SYNCING.swap(true, std::sync::atomic::Ordering::SeqCst) {
        return Err((
            StatusCode::CONFLICT,
            "a repo sync is already running".into(),
        ));
    }
    let config = app.config.clone();
    let roots = crate::recipe_repos::sync_recipe_repos(&config).await;
    SYNCING.store(false, std::sync::atomic::Ordering::SeqCst);
    Ok(Json(
        serde_json::json!({ "ok": true, "repos": roots.len() }),
    ))
}

// ---------------------------------------------------------------------------
// P3: log windows — attach a streaming `docker logs -f` per node+recipe+rank
// (idempotent), detach, and fetch ring history for a fresh window.

#[derive(Deserialize)]
pub(crate) struct LogsReq {
    pub cluster: String,
    pub node: String,
    pub recipe: String,
    #[serde(default)]
    pub rank: u32,
    #[serde(default)]
    pub lines: usize,
}

/// `POST /api/logs` — attach (reuses a live stream for the same key).
pub(crate) async fn logs_attach(
    app: State<AppState>,
    Json(req): Json<LogsReq>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let (id, history) = super::logs::attach(
        &app,
        &app.log_attaches,
        &req.cluster,
        &req.node,
        &req.recipe,
        req.rank,
    )
    .map_err(|e| {
        (
            if e.starts_with("max") {
                StatusCode::CONFLICT
            } else {
                StatusCode::BAD_REQUEST
            },
            e,
        )
    })?;
    Ok(Json(
        serde_json::json!({ "attach_id": id, "history": history }),
    ))
}

/// `DELETE /api/logs` — stop the stream and drop the attach.
#[derive(Deserialize)]
pub(crate) struct DetachReq {
    pub attach_id: String,
}

pub(crate) async fn logs_detach(
    app: State<AppState>,
    Json(req): Json<DetachReq>,
) -> Result<StatusCode, ApiError> {
    let id = req.attach_id;
    if super::logs::detach(&app.log_attaches, &id) {
        Ok(StatusCode::NO_CONTENT)
    } else {
        Err((StatusCode::NOT_FOUND, format!("no log attach '{id}'")))
    }
}

/// `GET /api/logs?cluster=&node=&recipe=&rank=&lines=` — ring history for a
/// fresh window (0 lines = `logs::TAIL_LINES`).
pub(crate) async fn logs_history(
    app: State<AppState>,
    axum::extract::Query(q): axum::extract::Query<LogsReq>,
) -> Result<Json<serde_json::Value>, ApiError> {
    let lines = if q.lines == 0 {
        super::logs::TAIL_LINES
    } else {
        q.lines
    };
    let (id, history) = super::logs::by_key(
        &app.log_attaches,
        &q.cluster,
        &q.node,
        &q.recipe,
        q.rank,
        lines,
    )
    .ok_or_else(|| {
        (
            StatusCode::NOT_FOUND,
            "no log attach for that node/recipe".to_string(),
        )
    })?;
    Ok(Json(
        serde_json::json!({ "attach_id": id, "history": history }),
    ))
}

// ---------------------------------------------------------------------------
// P4: raw content — the active config file, a recipe verbatim, and the
// merged effective config (post cluster-source merge) as JSON.

/// `GET /api/raw/config` — the active config file, verbatim.
pub(crate) async fn raw_config() -> Result<axum::response::Response, ApiError> {
    let path = crate::gconfig::resolve_config_path("config.toml");
    let content = tokio::fs::read_to_string(&path).await.map_err(|e| {
        (
            StatusCode::NOT_FOUND,
            format!("no config file at {}: {e}", path.display()),
        )
    })?;
    Ok(axum::response::Response::builder()
        .status(200)
        .header("content-type", "text/plain; charset=utf-8")
        .body(axum::body::Body::from(content))
        .unwrap())
}

/// `GET /api/raw/recipe/<name>` — recipe.toml verbatim (search order as
/// `load_recipe`: cwd, recipe_sources, global dir).
pub(crate) async fn raw_recipe(
    app: State<AppState>,
    axum::extract::Path(name): axum::extract::Path<String>,
) -> Result<axum::response::Response, ApiError> {
    if name.is_empty() || name.contains('/') || name.contains("..") {
        return Err((StatusCode::BAD_REQUEST, "invalid recipe name".to_string()));
    }
    let dirs = crate::config::recipe_dirs(&app.config);
    for dir in &dirs {
        let p = dir.join(&name).join("recipe.toml");
        if let Ok(content) = tokio::fs::read_to_string(&p).await {
            return Ok(axum::response::Response::builder()
                .status(200)
                .header("content-type", "text/plain; charset=utf-8")
                .body(axum::body::Body::from(content))
                .unwrap());
        }
    }
    Err((StatusCode::NOT_FOUND, format!("recipe '{name}' not found")))
}

/// `GET /api/raw/effective` — the merged `Config` as JSON (post
/// cluster-source merge), HA token redacted.
pub(crate) async fn raw_effective(app: State<AppState>) -> Json<serde_json::Value> {
    use crate::config::NcclConfig;
    let ha = app.config.home_assistant.as_ref().map(|h| {
        let masked: String = h.token.chars().take(12).collect();
        let token = if h.token.len() > 12 {
            format!("{masked}…<redacted>")
        } else {
            "<redacted>".to_string()
        };
        serde_json::json!({
            "url": h.url,
            "token": token,
            "shutdown_delay_seconds": h.shutdown_delay_seconds,
        })
    });
    Json(serde_json::json!({
        "config_path": crate::gconfig::resolve_config_path("config.toml").display().to_string(),
        "ssh_user": app.config.ssh_user,
        "ssh_key_path": app.config.ssh_key_path.display().to_string(),
        "clusters": app.config.clusters.iter().map(|c| serde_json::json!({
            "name": c.name,
            "nodes": c.nodes.iter().map(|n| serde_json::json!({
                "name": n.name,
                "ssh_ip": n.ssh_ip,
                "rank": n.rank,
                "entity": n.entity,
                "host_key_set": n.host_key.is_some(),
            })).collect::<Vec<_>>(),
            "extra_entities": c.extra_entities.iter().map(|e| serde_json::json!({
                "entity": e.entity(),
                "blocks": e.blocks(),
            })).collect::<Vec<_>>(),
            "nccl": app.config.nccl.get(&c.name).map(|n: &NcclConfig| serde_json::json!({
                "ib_hca": n.ib_hca,
                "socket_ifname": n.socket_ifname,
                "ib_gid_index": n.ib_gid_index,
                "ib_disable": n.ib_disable,
            })),
            "before_serve": app.config.before_serve.get(&c.name).map(|b| b.commands.len()),
        })).collect::<Vec<_>>(),
        "home_assistant": ha,
        "init_port": app.config.init_port,
        "mount_aliases": app.config.mount_aliases,
        "recipe_sources": app.config.recipe_sources,
        "cluster_sources": app.config.cluster_sources,
        "drop_caches": app.config.drop_caches,
    }))
}