//! HTTP + WS contract (mirrors Sando's shape).
//!
//! - `GET  /state` — latest build + its target x step matrix (loose JSON).
//! - `POST /build` — `{app, version?, targets?[]}` kick a build.
//! - `POST /retry` — `{app, target, version?}` re-run one target.
//! - `GET  /logs/{app}/{version}/{target}/{step}` — post-mortem step log.
//! - `GET  /events` — WS; broadcasts `EventEnvelope` JSON frames.
//! - `GET  /metrics` — Prometheus.

use crate::domain::{AppId, Target};
use crate::error::{Error, Result};
use crate::runner;
use crate::state::AppState;
use axum::extract::{Path, State, WebSocketUpgrade};
use axum::response::IntoResponse;
use axum::routing::{get, post};
use axum::{Json, Router};
use serde::{Deserialize, Serialize};
use sqlx::Row;

pub fn router(state: AppState) -> Router {
    let prom = state.prom.clone();
    let token = state.api_token.clone();

    // Build triggers require a bearer token (when configured); reads stay open
    // so the TUI's state/event/log polling needs no credential (CF2).
    let mutating = Router::new()
        .route("/build", post(build))
        .route("/retry", post(retry))
        .route_layer(axum::middleware::from_fn(move |req, next| {
            require_bearer(token.clone(), req, next)
        }));

    let open = Router::new()
        .route("/state", get(get_state))
        .route("/status.json", get(get_status_json))
        .route("/release/{app}/{version}", get(get_release))
        .route("/logs/{app}/{version}/{target}/{step}", get(get_step_log))
        .route("/events", get(events_ws));

    Router::new()
        .merge(mutating)
        .merge(open)
        .with_state(state)
        // `/metrics` is intentionally open (Prometheus scrape convention): it
        // exposes only build/host telemetry — counts, durations, app/host names —
        // and never a secret or token (no secret reaches a metric label). The
        // daemon binds the tailnet only, so the scrape surface is tailnet-internal.
        .route("/metrics", get(crate::metrics::render).with_state(prom))
        // Build/retry bodies are small JSON; cap request bodies well below
        // axum's 2 MB default so a flood of oversized POSTs can't buffer freely.
        .layer(axum::extract::DefaultBodyLimit::max(64 * 1024))
}

/// Bearer-token gate for the build triggers. No token configured -> pass
/// (main() only allows that on a loopback bind). Constant-time comparison.
async fn require_bearer(
    token: Option<std::sync::Arc<str>>,
    req: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    let Some(expected) = token.as_deref() else {
        return next.run(req).await;
    };
    let path = req.uri().path().to_string();
    let ok = req
        .headers()
        .get(axum::http::header::AUTHORIZATION)
        .and_then(|h| h.to_str().ok())
        .and_then(|h| h.strip_prefix("Bearer "))
        .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
    if ok {
        next.run(req).await
    } else {
        tracing::warn!(path = %path, "rejected unauthenticated request to a build trigger");
        (
            axum::http::StatusCode::UNAUTHORIZED,
            "missing or invalid bearer token\n",
        )
            .into_response()
    }
}

#[derive(Serialize)]
struct StateView {
    build: Option<BuildView>,
}

/// One app's latest build, plus what the topology says it ships.
///
/// `/state` answers with the newest build row across every app, which means a
/// release that half-failed on one app becomes invisible the moment any other
/// app builds. `/status.json` reads per app instead, so nothing can be hidden
/// by a later, unrelated build.
pub(crate) struct AppStatusView {
    pub(crate) app: String,
    pub(crate) kind: crate::topology::Kind,
    /// Every target this app ships, from its in-repo manifest -- including ones
    /// the latest build never ran.
    pub(crate) declared_targets: Vec<String>,
    /// `None` until this app has ever been built.
    pub(crate) build: Option<BuildView>,
    /// Targets with a `releases` row at the build's version. Empty when
    /// publishing is not in play for this app.
    pub(crate) published_targets: Vec<String>,
}

#[derive(Serialize)]
pub(crate) struct BuildView {
    pub(crate) id: i64,
    pub(crate) app: String,
    pub(crate) version: String,
    pub(crate) status: String,
    pub(crate) created_at: String,
    pub(crate) targets: Vec<TargetView>,
}

#[derive(Serialize)]
pub(crate) struct TargetView {
    pub(crate) target: String,
    pub(crate) status: String,
    pub(crate) current_step: Option<String>,
    pub(crate) error: Option<String>,
    pub(crate) steps: Vec<StepView>,
}

#[derive(Serialize)]
pub(crate) struct StepView {
    pub(crate) run_id: i64,
    pub(crate) step: String,
    pub(crate) status: String,
    pub(crate) log_ref: Option<String>,
}

async fn get_state(State(s): State<AppState>) -> Result<Json<StateView>> {
    let latest = sqlx::query(
        "SELECT id, app, version, status, created_at FROM builds ORDER BY id DESC LIMIT 1",
    )
    .fetch_optional(&s.pool)
    .await?;

    let Some(b) = latest else {
        return Ok(Json(StateView { build: None }));
    };
    Ok(Json(StateView {
        build: Some(build_view(&s, &b).await?),
    }))
}

/// Hydrate one `builds` row into its target x step matrix.
///
/// Shared by `/state` and `/status.json` so the two cannot disagree about what
/// a build looks like.
///
/// The matrix is the per-`(app, version)` RELEASE view, not one `builds` row's
/// targets: for each target, the latest `target_runs` row for this app+version
/// across EVERY build of that version. A `/retry` inserts a fresh single-target
/// build, so keying on `build_id` would collapse a 5-target release to the one
/// retried row and lose the answer to "did every target of this version ship".
/// Aggregating by `(app, version)` keeps the full matrix and folds the retry's
/// newer result into just that target's cell.
async fn build_view(s: &AppState, b: &sqlx::sqlite::SqliteRow) -> Result<BuildView> {
    let build_id: i64 = b.get("id");
    let app: String = b.get("app");
    let version: String = b.get("version");

    let target_rows = sqlx::query(
        "SELECT id, target, status, current_step, error FROM target_runs tr
         WHERE app = ?1 AND version = ?2
           AND id = (SELECT MAX(id) FROM target_runs
                     WHERE app = ?1 AND version = ?2 AND target = tr.target)
         ORDER BY target",
    )
    .bind(&app)
    .bind(&version)
    .fetch_all(&s.pool)
    .await?;

    let mut targets = Vec::with_capacity(target_rows.len());
    for tr in target_rows {
        let target_run_id: i64 = tr.get("id");
        // Latest row per step for this target run.
        let steps: Vec<StepView> = sqlx::query(
            "SELECT id, step, status, log_ref FROM step_runs sr
             WHERE target_run_id = ?1
               AND id = (SELECT MAX(id) FROM step_runs
                         WHERE target_run_id = ?1 AND step = sr.step)
             ORDER BY id",
        )
        .bind(target_run_id)
        .fetch_all(&s.pool)
        .await?
        .into_iter()
        .map(|r| StepView {
            run_id: r.get("id"),
            step: r.get("step"),
            status: r.get("status"),
            log_ref: r.get("log_ref"),
        })
        .collect();

        targets.push(TargetView {
            target: tr.get("target"),
            status: tr.get("status"),
            current_step: tr.get("current_step"),
            error: tr.get("error"),
            steps,
        });
    }

    Ok(BuildView {
        id: build_id,
        app: b.get("app"),
        version: b.get("version"),
        status: b.get("status"),
        created_at: b.get("created_at"),
        targets,
    })
}

/// `GET /status.json` -- every app's latest build, in the shared cross-service
/// payload every operator surface renders. See `crate::status`.
async fn get_status_json(State(s): State<AppState>) -> Result<Json<ops_status::Payload>> {
    let view = status_view(&s).await?;
    Ok(Json(crate::status::payload(&view, chrono::Utc::now())))
}

/// One release, answering "did every target of `{version}` ship?"
#[derive(Serialize)]
struct ReleaseView {
    app: String,
    version: String,
    /// Every target the app's manifest declares for this release.
    declared_targets: Vec<String>,
    /// The per-`(app, version)` matrix: latest run per target across every build
    /// (including retries) of this version.
    build: BuildView,
    /// Targets with a `releases` row at this version.
    published_targets: Vec<String>,
    /// Every declared target has a latest run of `ok` — the release built clean.
    all_targets_green: bool,
    /// `all_targets_green` AND every declared target is published.
    complete: bool,
}

/// `GET /release/{app}/{version}` -- the release view for a SPECIFIC version, not
/// just the latest. Unlike `/state`, a retry can't hide it and an unrelated
/// newer build can't mask it; this is the durable "did 0.5.0 fully ship" query.
async fn get_release(
    State(s): State<AppState>,
    Path((app, version)): Path<(String, String)>,
) -> Result<Json<ReleaseView>> {
    let cfg = s
        .topo
        .app(&AppId::new(app.clone()))
        .ok_or_else(|| Error::BadRequest(format!("unknown app `{app}`")))?;
    let declared_targets: Vec<String> = cfg.targets.iter().map(ToString::to_string).collect();

    // A representative build row for this exact version (newest wins for the
    // build-level fields; the target matrix is aggregated across all of them).
    let row = sqlx::query(
        "SELECT id, app, version, status, created_at FROM builds
         WHERE app = ? AND version = ? ORDER BY id DESC LIMIT 1",
    )
    .bind(&app)
    .bind(&version)
    .fetch_optional(&s.pool)
    .await?
    .ok_or(Error::NotFound)?;
    let build = build_view(&s, &row).await?;

    let published_targets = sqlx::query_scalar::<_, String>(
        "SELECT DISTINCT target FROM releases WHERE app = ? AND version = ? ORDER BY target",
    )
    .bind(&app)
    .bind(&version)
    .fetch_all(&s.pool)
    .await?;

    // Green = every declared target has a latest run that is `ok`.
    let all_targets_green = declared_targets.iter().all(|d| {
        build
            .targets
            .iter()
            .any(|t| &t.target == d && t.status == "ok")
    });
    let complete = all_targets_green
        && declared_targets
            .iter()
            .all(|d| published_targets.contains(d));

    Ok(Json(ReleaseView {
        app,
        version,
        declared_targets,
        build,
        published_targets,
        all_targets_green,
        complete,
    }))
}

/// One [`AppStatusView`] per app in the topology, name-ordered.
///
/// Every declared app appears whether or not it has ever been built: an app
/// missing from the surface is indistinguishable from an app that is fine, and
/// the whole point of the viewer is that absence of evidence must be visible.
pub(crate) async fn status_view(s: &AppState) -> Result<Vec<AppStatusView>> {
    let mut names: Vec<&String> = s.topo.app.keys().collect();
    names.sort();

    let mut apps = Vec::with_capacity(names.len());
    for name in names {
        let cfg = &s.topo.app[name];

        let latest = sqlx::query(
            "SELECT id, app, version, status, created_at FROM builds
             WHERE app = ? ORDER BY id DESC LIMIT 1",
        )
        .bind(name)
        .fetch_optional(&s.pool)
        .await?;

        let build = match latest {
            Some(row) => Some(build_view(s, &row).await?),
            None => None,
        };

        let published_targets = match &build {
            Some(b) => {
                sqlx::query_scalar::<_, String>(
                    "SELECT DISTINCT target FROM releases
                     WHERE app = ? AND version = ? ORDER BY target",
                )
                .bind(name)
                .bind(&b.version)
                .fetch_all(&s.pool)
                .await?
            }
            None => Vec::new(),
        };

        apps.push(AppStatusView {
            app: name.clone(),
            kind: cfg.kind,
            declared_targets: cfg.targets.iter().map(ToString::to_string).collect(),
            build,
            published_targets,
        });
    }
    Ok(apps)
}

#[derive(Deserialize, Default)]
struct BuildBody {
    app: String,
    #[serde(default)]
    version: Option<String>,
    #[serde(default)]
    targets: Vec<String>,
}

async fn build(
    State(s): State<AppState>,
    Json(body): Json<BuildBody>,
) -> Result<Json<serde_json::Value>> {
    let app = AppId::new(body.app);
    let targets = parse_targets(body.targets)?;
    let version = runner::resolve_version(&s, &app, body.version)?;
    let targets = runner::resolve_targets(&s, &app, targets)?;
    let build_id = runner::start_build(s, app, version.clone(), targets)
        .await
        .map_err(Error::Other)?;
    Ok(Json(
        serde_json::json!({ "accepted": true, "build_id": build_id, "version": version.to_string() }),
    ))
}

#[derive(Deserialize)]
struct RetryBody {
    app: String,
    target: String,
    #[serde(default)]
    version: Option<String>,
}

async fn retry(
    State(s): State<AppState>,
    Json(body): Json<RetryBody>,
) -> Result<Json<serde_json::Value>> {
    let app = AppId::new(body.app);
    let target: Target = body.target.parse().map_err(Error::BadRequest)?;
    let version = runner::resolve_version(&s, &app, body.version)?;
    let targets = runner::resolve_targets(&s, &app, vec![target])?;
    let build_id = runner::start_build(s, app, version.clone(), targets)
        .await
        .map_err(Error::Other)?;
    Ok(Json(
        serde_json::json!({ "accepted": true, "build_id": build_id, "target": target.to_string() }),
    ))
}

fn parse_targets(raw: Vec<String>) -> Result<Vec<Target>> {
    raw.into_iter()
        .map(|t| t.parse::<Target>().map_err(Error::BadRequest))
        .collect()
}

async fn get_step_log(
    State(s): State<AppState>,
    Path((app, version, target, step)): Path<(String, String, String, String)>,
) -> Result<axum::response::Response> {
    fn safe(seg: &str) -> bool {
        !seg.is_empty() && !seg.contains('/') && !seg.contains('\\') && seg != "." && seg != ".."
    }
    if ![&app, &version, &target, &step]
        .into_iter()
        .all(|s| safe(s))
    {
        return Err(Error::NotFound);
    }
    let path = s
        .cfg
        .logs_root
        .join(&app)
        .join(&version)
        .join(&target)
        .join(format!("{step}.log"));
    // Stream the log in chunks rather than reading the whole (potentially large,
    // verbose-build) file into memory.
    let file = match tokio::fs::File::open(&path).await {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Err(Error::NotFound),
        Err(e) => return Err(Error::Other(e.into())),
    };
    let (tx, rx) =
        tokio::sync::mpsc::channel::<std::result::Result<axum::body::Bytes, std::io::Error>>(8);
    tokio::spawn(async move {
        use tokio::io::AsyncReadExt;
        let mut file = file;
        let mut buf = vec![0u8; 64 * 1024];
        loop {
            match file.read(&mut buf).await {
                Ok(0) => break,
                Ok(n) => {
                    if tx
                        .send(Ok(axum::body::Bytes::copy_from_slice(&buf[..n])))
                        .await
                        .is_err()
                    {
                        break;
                    }
                }
                Err(e) => {
                    let _ = tx.send(Err(e)).await;
                    break;
                }
            }
        }
    });
    let body = axum::body::Body::from_stream(tokio_stream::wrappers::ReceiverStream::new(rx));
    Ok((
        [(
            axum::http::header::CONTENT_TYPE,
            "text/plain; charset=utf-8",
        )],
        body,
    )
        .into_response())
}

async fn events_ws(ws: WebSocketUpgrade, State(s): State<AppState>) -> impl IntoResponse {
    use axum::extract::ws::Message;
    use tokio::sync::broadcast::error::RecvError;

    ws.on_upgrade(move |mut socket| async move {
        // Subscribe to both channels and merge them: a lag on the high-rate log
        // stream emits its own `lagged` frame without dropping anything on the
        // status stream (and vice versa), so a busy build's chunk firehose can't
        // evict a TargetFailed/PublishOk the operator needs to see.
        let mut status_rx = s.events.subscribe_status();
        let mut logs_rx = s.events.subscribe_logs();
        loop {
            let recv = tokio::select! {
                r = status_rx.recv() => r,
                r = logs_rx.recv() => r,
            };
            match recv {
                Ok(env) => {
                    let json = match serde_json::to_string(&env) {
                        Ok(s) => s,
                        Err(e) => {
                            tracing::warn!(error = %e, "events ws: serialize failed");
                            continue;
                        }
                    };
                    if socket.send(Message::Text(json.into())).await.is_err() {
                        break;
                    }
                }
                Err(RecvError::Lagged(n)) => {
                    let _ = socket
                        .send(Message::Text(
                            format!(r#"{{"kind":"lagged","skipped":{n}}}"#).into(),
                        ))
                        .await;
                }
                Err(RecvError::Closed) => break,
            }
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::Config;
    use crate::ota::OtaRegistry;
    use crate::topology::Topology;
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use http_body_util::BodyExt;
    use std::collections::HashMap;
    use std::sync::Arc;
    use tokio::sync::Mutex;
    use tower::ServiceExt;

    async fn test_state(root: &std::path::Path) -> AppState {
        let cfg = Config::for_tests(root);
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let repo = root.join("goingson");
        std::fs::create_dir_all(&repo).unwrap();
        std::fs::write(repo.join("bento.toml"), "targets = [\"linux/x86_64\"]\n").unwrap();
        let topo = Topology::from_str_for_tests(&format!(
            r#"
[[host]]
name = "fw13"
ssh = "local"
targets = ["linux/x86_64"]

[app.goingson]
repo = "{}"
"#,
            repo.display()
        ))
        .unwrap();
        let executors = Arc::new(crate::state::build_executors(&topo));
        let syncs = Arc::new(crate::state::build_syncs(&topo));
        let host_locks = crate::state::build_host_locks(&topo);
        AppState {
            pool,
            topo: Arc::new(topo),
            cfg: Arc::new(cfg),
            prom: crate::metrics::test_handle(),
            events: crate::events::channel(),
            ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
            executors,
            syncs,
            active: Arc::new(Mutex::new(HashMap::new())),
            api_token: None,
            host_locks,
        }
    }

    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 state_is_empty_initially() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/state")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(body_string(resp).await, r#"{"build":null}"#);
    }

    #[tokio::test]
    async fn status_json_lists_every_declared_app_before_any_build() {
        // The mapping is tested in `crate::status`. This asserts the route is
        // wired and that an app with no build history still reaches the
        // surface -- an app missing from the viewer is indistinguishable from
        // an app that is fine.
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/status.json")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);

        let payload: ops_status::Payload = serde_json::from_str(&body_string(resp).await).unwrap();
        assert_eq!(payload.source, crate::status::SOURCE);
        assert_eq!(payload.schema, ops_status::SCHEMA_VERSION);
        assert_eq!(payload.validate(), Ok(()));

        let goingson = payload.node("app:goingson").expect("declared app appears");
        assert_eq!(goingson.status, ops_status::Status::Pending);
        assert!(payload.actions.contains_key("build-goingson"));
        assert!(payload.node("target:goingson:linux/x86_64").is_some());
    }

    // ---- CF2: bearer-token auth on build triggers ----

    #[tokio::test]
    async fn build_route_requires_bearer_when_token_set() {
        let tmp = tempfile::tempdir().unwrap();
        let mut state = test_state(tmp.path()).await;
        state.api_token = Some(std::sync::Arc::from("s3cr3t"));
        let app = router(state);

        let build_req = || {
            Request::builder()
                .method("POST")
                .uri("/build")
                .header("content-type", "application/json")
                .body(Body::from(r#"{"app":"goingson"}"#))
                .unwrap()
        };

        // No token -> 401.
        let resp = app.clone().oneshot(build_req()).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        // Wrong token -> 401.
        let mut bad = build_req();
        bad.headers_mut()
            .insert("authorization", "Bearer nope".parse().unwrap());
        let resp = app.clone().oneshot(bad).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);

        // Correct token -> passes auth (NOT 401; the build itself is accepted).
        let mut good = build_req();
        good.headers_mut()
            .insert("authorization", "Bearer s3cr3t".parse().unwrap());
        let resp = app.clone().oneshot(good).await.unwrap();
        assert_ne!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn read_route_open_when_token_set() {
        let tmp = tempfile::tempdir().unwrap();
        let mut state = test_state(tmp.path()).await;
        state.api_token = Some(std::sync::Arc::from("s3cr3t"));
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/state")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn step_log_rejects_traversal() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/logs/goingson/0.4.1/..%2f..%2fetc/passwd")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn step_log_streams_existing_file() {
        let tmp = tempfile::tempdir().unwrap();
        let state = test_state(tmp.path()).await;
        // Write a log at the path the route resolves.
        let dir = state
            .cfg
            .logs_root
            .join("goingson")
            .join("0.4.1")
            .join("linux-x86_64");
        std::fs::create_dir_all(&dir).unwrap();
        std::fs::write(dir.join("build.log"), b"compiling\nlinked\n").unwrap();
        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/logs/goingson/0.4.1/linux-x86_64/build")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        assert_eq!(body_string(resp).await, "compiling\nlinked\n");
    }

    #[tokio::test]
    async fn bad_request_body_is_a_json_error_envelope() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/build")
                    .header("content-type", "application/json")
                    .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        // JSON envelope, message preserved.
        let body = body_string(resp).await;
        let v: serde_json::Value = serde_json::from_str(&body).unwrap();
        assert!(v["error"].as_str().unwrap().contains("unknown app"));
    }

    #[tokio::test]
    async fn build_rejects_unshipped_target_as_bad_request() {
        // windows/x86_64 is a valid target but the test app doesn't ship it.
        // That's a client error (400), not a server error (500).
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/build")
                    .header("content-type", "application/json")
                    // Explicit version so resolve_version doesn't read a
                    // (nonexistent) tauri.conf first — we're testing the target
                    // validation specifically.
                    .body(Body::from(
                        r#"{"app":"goingson","version":"0.4.1","targets":["windows/x86_64"]}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert!(
            body_string(resp).await.contains("does not ship target"),
            "names the problem"
        );
    }

    #[tokio::test]
    async fn build_rejects_unknown_app_as_bad_request() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/build")
                    .header("content-type", "application/json")
                    .body(Body::from(r#"{"app":"nope","targets":["linux/x86_64"]}"#))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert!(
            body_string(resp).await.contains("unknown app"),
            "names the problem"
        );
    }

    #[tokio::test]
    async fn build_rejects_malformed_target_as_bad_request() {
        // A non-parseable target string is also a 400 (via parse_targets).
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri("/build")
                    .header("content-type", "application/json")
                    .body(Body::from(
                        r#"{"app":"goingson","targets":["not-a-target"]}"#,
                    ))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    /// A two-target app + a two-host topology, for the release-view aggregation.
    async fn two_target_state(root: &std::path::Path) -> AppState {
        let cfg = Config::for_tests(root);
        let pool = crate::db::open(&cfg.db_path).await.unwrap();
        let repo = root.join("demo");
        std::fs::create_dir_all(&repo).unwrap();
        std::fs::write(
            repo.join("bento.toml"),
            "targets = [\"linux/x86_64\", \"macos/aarch64\"]\n",
        )
        .unwrap();
        let topo = Topology::from_str_for_tests(&format!(
            "[[host]]\nname = \"fw13\"\nssh = \"local\"\ntargets = [\"linux/x86_64\"]\n\
             [[host]]\nname = \"mbp\"\nssh = \"mbp\"\ntargets = [\"macos/aarch64\"]\n\
             [app.demo]\nrepo = \"{}\"\n",
            repo.display()
        ))
        .unwrap();
        let executors = Arc::new(crate::state::build_executors(&topo));
        let syncs = Arc::new(crate::state::build_syncs(&topo));
        let host_locks = crate::state::build_host_locks(&topo);
        AppState {
            pool,
            topo: Arc::new(topo),
            cfg: Arc::new(cfg),
            prom: crate::metrics::test_handle(),
            events: crate::events::channel(),
            ota: Arc::new(OtaRegistry::standard("https://makenot.work")),
            executors,
            syncs,
            active: Arc::new(Mutex::new(HashMap::new())),
            api_token: None,
            host_locks,
        }
    }

    async fn insert_target_run(pool: &sqlx::SqlitePool, ver: &str, target: &str, status: &str) {
        let bid: i64 = sqlx::query_scalar(
            "INSERT INTO builds (app, version, status, created_at) VALUES ('demo', ?, ?, '2026-07-23T00:00:00Z') RETURNING id",
        )
        .bind(ver)
        .bind(status)
        .fetch_one(pool)
        .await
        .unwrap();
        sqlx::query(
            "INSERT INTO target_runs (build_id, app, version, target, status, started_at)
             VALUES (?, 'demo', ?, ?, ?, '2026-07-23T00:00:00Z')",
        )
        .bind(bid)
        .bind(ver)
        .bind(target)
        .bind(status)
        .execute(pool)
        .await
        .unwrap();
    }

    /// The audit's H2: `/retry` inserts a whole new single-target build, so a
    /// build-id-keyed view collapses a multi-target release to the retried row.
    /// The release view aggregates by (app, version), so a green retry of one
    /// target folds into that cell while the other target stays visible.
    #[tokio::test]
    async fn release_view_survives_a_single_target_retry() {
        let tmp = tempfile::tempdir().unwrap();
        let state = two_target_state(tmp.path()).await;
        let pool = state.pool.clone();

        // Original release build: linux ok, macos failed.
        insert_target_run(&pool, "0.5.0", "linux/x86_64", "ok").await;
        insert_target_run(&pool, "0.5.0", "macos/aarch64", "failed").await;
        // Retry macos only — a fresh single-target build that succeeds.
        insert_target_run(&pool, "0.5.0", "macos/aarch64", "ok").await;

        let app = router(state);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/release/demo/0.5.0")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();

        // Both targets are present (the matrix did not collapse to the retry),
        // and macos shows the newer `ok`, not the earlier `failed`.
        let targets = v["build"]["targets"].as_array().unwrap();
        assert_eq!(targets.len(), 2, "full matrix preserved across the retry");
        let macos = targets
            .iter()
            .find(|t| t["target"] == "macos/aarch64")
            .unwrap();
        assert_eq!(macos["status"], "ok", "retry result wins the cell");
        assert_eq!(v["all_targets_green"], true);
    }

    #[tokio::test]
    async fn release_view_404s_for_an_unbuilt_version() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(two_target_state(tmp.path()).await);
        let resp = app
            .oneshot(
                Request::builder()
                    .uri("/release/demo/9.9.9")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
    }

    // ---- /retry handler (the route itself, distinct from the release-view
    //      aggregation a retry produces above) ----

    fn retry_req(json: &str) -> Request<Body> {
        Request::builder()
            .method("POST")
            .uri("/retry")
            .header("content-type", "application/json")
            .body(Body::from(json.to_string()))
            .unwrap()
    }

    #[tokio::test]
    async fn retry_accepts_a_single_shipped_target() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        // Explicit version so the handler doesn't read a (nonexistent) version
        // file; the point here is the accept path, not version resolution.
        let resp = app
            .oneshot(retry_req(
                r#"{"app":"goingson","target":"linux/x86_64","version":"0.4.1"}"#,
            ))
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let v: serde_json::Value = serde_json::from_str(&body_string(resp).await).unwrap();
        assert_eq!(v["accepted"], true);
        assert_eq!(v["target"], "linux/x86_64", "echoes the retried target");
        assert!(
            v["build_id"].as_i64().is_some(),
            "a fresh single-target build id is returned"
        );
    }

    #[tokio::test]
    async fn retry_rejects_an_unshipped_target_as_bad_request() {
        // goingson does not ship windows; a retry of it is a 400, not a 500.
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(retry_req(
                r#"{"app":"goingson","target":"windows/x86_64","version":"0.4.1"}"#,
            ))
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert!(body_string(resp).await.contains("does not ship target"));
    }

    #[tokio::test]
    async fn retry_rejects_a_malformed_target_as_bad_request() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(retry_req(
                r#"{"app":"goingson","target":"not-a-target","version":"0.4.1"}"#,
            ))
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn retry_rejects_an_unknown_app_as_bad_request() {
        let tmp = tempfile::tempdir().unwrap();
        let app = router(test_state(tmp.path()).await);
        let resp = app
            .oneshot(retry_req(r#"{"app":"nope","target":"linux/x86_64"}"#))
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        assert!(body_string(resp).await.contains("unknown app"));
    }

    // ---- /events websocket: the HTTP->WS wiring (upgrade + serialize + send).
    //      The channel merge and lagged framing themselves are covered in
    //      `crate::events`; this asserts a real client on `/events` receives an
    //      emitted event as the flat-`kind` JSON the TUI parses. ----

    #[tokio::test]
    async fn events_ws_streams_emitted_events_as_flat_json() {
        use futures_util::StreamExt;
        use tokio_tungstenite::tungstenite::Message as WsMessage;

        let tmp = tempfile::tempdir().unwrap();
        let state = test_state(tmp.path()).await;
        let events = state.events.clone();
        let app = router(state);

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });

        let (mut ws, _resp) = tokio_tungstenite::connect_async(format!("ws://{addr}/events"))
            .await
            .expect("client connects to /events");

        // The handler subscribes to the bus inside `on_upgrade`, which can land
        // just after the handshake returns; a broadcast has no backlog for a
        // late subscriber, so re-emit until the first frame arrives rather than
        // racing that window with a single send.
        let ev = crate::events::Event::PublishFailed {
            app: AppId::new("goingson"),
            target: "macos/aarch64".parse().unwrap(),
            channel: "stable".into(),
            error: "boom".into(),
        };
        let mut frame = None;
        for _ in 0..100 {
            crate::events::emit(&events, ev.clone());
            if let Ok(Some(Ok(WsMessage::Text(t)))) =
                tokio::time::timeout(std::time::Duration::from_millis(50), ws.next()).await
            {
                frame = Some(t);
                break;
            }
        }
        let frame = frame.expect("received an event frame within the retry budget");
        let v: serde_json::Value = serde_json::from_str(&frame).unwrap();
        assert_eq!(v["kind"], "publish_failed", "flat kind-tagged wire shape");
        assert_eq!(v["channel"], "stable");
        assert!(v.get("event").is_none(), "not nested under `event`");
    }
}