//! Build-run tracking: one `build_runs` row per `/rebuild`, updated as the
//! pipeline moves through its phases, terminating in passed/failed/aborted.
//!
//! This is the resource that makes Sando driveable headlessly. `/state` only
//! ever reflects the last *successful* deploy, so on a red pipeline a poller
//! of `/state` sees stale-green for the whole build. A
//! `RunId` returned by `/rebuild` + `GET /runs/{id}` gives a non-TUI caller
//! one pollable resource tied to the build it triggered, carrying the phase,
//! the per-gate status, and a `failure_summary`
//! (first compile error / first failed gate) so the cause is in the API, not
//! behind `sudo journalctl`.
//!
//! Terminal writes (`mark_passed`/`mark_failed`/`mark_aborted`) are guarded on
//! `result = 'building'`, so whichever site settles the run first wins: a
//! build-step compile error, the first red gate, or the task-level catch for
//! pre-build bails. Later writes are silent no-ops.

use crate::domain::{AppId, RunId, Version};
use anyhow::Result;
use chrono::Utc;
use serde::Serialize;
use sqlx::{Row, SqlitePool};

/// In-flight sub-state. Plain strings in the DB; this enum names the values so
/// call sites can't typo them.
#[derive(Debug, Clone, Copy)]
pub enum Phase {
    Fetching,
    Compiling,
    Staging,
    Gating,
}

impl Phase {
    pub fn as_str(self) -> &'static str {
        match self {
            Phase::Fetching => "fetching",
            Phase::Compiling => "compiling",
            Phase::Staging => "staging",
            Phase::Gating => "gating",
        }
    }
}

/// Insert a fresh `building` run for `app` at `sha` and return its id.
pub async fn create(pool: &SqlitePool, app: &AppId, sha: &str) -> Result<RunId> {
    let id: i64 = sqlx::query_scalar(
        "INSERT INTO build_runs (app, sha, phase, result, started_at)
         VALUES (?, ?, 'queued', 'building', ?) RETURNING id",
    )
    .bind(app)
    .bind(sha)
    .bind(Utc::now().to_rfc3339())
    .fetch_one(pool)
    .await?;
    Ok(RunId(id))
}

/// Advance the in-flight phase. No-op once the run is terminal so a late
/// phase write can't resurrect a finished row.
pub async fn set_phase(pool: &SqlitePool, run_id: RunId, phase: Phase) -> Result<()> {
    sqlx::query("UPDATE build_runs SET phase = ? WHERE id = ? AND result = 'building'")
        .bind(phase.as_str())
        .bind(run_id.0)
        .execute(pool)
        .await?;
    Ok(())
}

/// Forward-advance a tier to `version` in a single atomic UPDATE. `previous_version`
/// is set from the row's *old* `current_version` (SQLite evaluates every RHS against
/// the original row), so there is no read-modify-write to lose under concurrency
/// (CF3) — no separate SELECT exists to race. `burn_in_started_at = now` starts the
/// tier's burn-in clock.
///
/// This is the *only* forward-advance writer of `tier_state`. The host build path
/// and `/promote` both go through it; keeping the fetch-then-write shape out of the
/// codebase is the point (ultra-fuzz Run 2, S1). Callers MUST hold `deploy_lock` so
/// the logical advance is serialized against `/rollback`.
///
/// Returns the raw `sqlx::Error` so the route layer can map it to its typed
/// `Error::Db`; anyhow callers (the build pipeline) get `?`-conversion for free.
/// `build_id` is the `build_runs.id` of the build landing on the tier, the
/// artifact identity (wiki [[release-artifact-identity]]). It advances in
/// lockstep with the version label: `previous_build_id` takes the old
/// `current_build_id` in the same self-referential UPDATE, so the build the
/// tier just stepped off is the rollback target. `None` for a caller with no
/// build identity writes NULL, treated as "no recorded build" downstream, with
/// the version-string path as the fallback.
///
/// `burn_in_started_at = now` starts the tier's burn-in clock. Because the
/// clock resets on every advance and the clock is what burn-in reads, the clock
/// always belongs to the build now current on the tier — this is what stops a
/// promote from crediting one build with another build's elapsed burn-in.
///
/// `advanced_at = now` records when the tier's deployed identity last changed.
/// It tracks the burn-in clock on an advance but, unlike it, is never nulled by
/// rollback or reset_burn_in — so the startup reconcile can trust it as the
/// instant to compare a `deploys` row against (wiki [[sando-overview]], migration
/// 009 / [`crate::reconcile`]).
pub async fn advance_tier(
    pool: &SqlitePool,
    app: &AppId,
    tier: &str,
    version: &Version,
    build_id: Option<i64>,
) -> Result<(), sqlx::Error> {
    let now = Utc::now().to_rfc3339();
    sqlx::query(
        "UPDATE tier_state
         SET previous_version  = current_version,
             current_version   = ?,
             previous_build_id = current_build_id,
             current_build_id  = ?,
             burn_in_started_at = ?,
             advanced_at        = ?
         WHERE app = ? AND tier = ?",
    )
    .bind(version)
    .bind(build_id)
    .bind(&now)
    .bind(&now)
    .bind(app)
    .bind(tier)
    .execute(pool)
    .await?;
    Ok(())
}

/// Record the version once it's been read from the worktree's Cargo.toml.
pub async fn set_version(pool: &SqlitePool, run_id: RunId, version: &Version) -> Result<()> {
    sqlx::query("UPDATE build_runs SET version = ? WHERE id = ? AND result = 'building'")
        .bind(version.to_string())
        .bind(run_id.0)
        .execute(pool)
        .await?;
    Ok(())
}

/// Record the build's content identity once its bundle has been staged and
/// hashed: the full 64-hex `bundle_digest` and the `staged_path` it lives at.
/// This is what promote/burn-in/retention key on once build id becomes the
/// identity (wiki [[release-artifact-identity]]). Guarded on `building` so a
/// settled run is never mutated.
pub async fn set_identity(
    pool: &SqlitePool,
    run_id: RunId,
    bundle_digest: &str,
    staged_path: &str,
) -> Result<()> {
    sqlx::query(
        "UPDATE build_runs SET bundle_digest = ?, staged_path = ?
         WHERE id = ? AND result = 'building'",
    )
    .bind(bundle_digest)
    .bind(staged_path)
    .bind(run_id.0)
    .execute(pool)
    .await?;
    Ok(())
}

/// Record what the bundle runs on.
///
/// Separate from [`set_identity`] and not best-effort: the digest identifies the
/// bytes, the platform is what makes two bundles of one version distinguishable,
/// and a row that lost it holds an artifact no node declaring a platform will
/// accept. A dropped write here is a bundle that can be placed nowhere, so the
/// caller is told.
pub async fn set_platform(
    pool: &SqlitePool,
    run_id: RunId,
    platform: &crate::domain::Platform,
) -> Result<()> {
    sqlx::query("UPDATE build_runs SET platform = ? WHERE id = ? AND result = 'building'")
        .bind(platform.to_string())
        .bind(run_id.0)
        .execute(pool)
        .await?;
    Ok(())
}

/// Settle the run green. First terminal write wins (guarded on `building`).
pub async fn mark_passed(pool: &SqlitePool, run_id: RunId) -> Result<()> {
    sqlx::query(
        "UPDATE build_runs SET result = 'passed', phase = 'done', finished_at = ?
         WHERE id = ? AND result = 'building'",
    )
    .bind(Utc::now().to_rfc3339())
    .bind(run_id.0)
    .execute(pool)
    .await?;
    Ok(())
}

/// Settle the run red with a human-readable cause. First terminal write wins,
/// so the most specific failure (build compile error, first red gate) recorded
/// before the task-level catch is the one that sticks.
pub async fn mark_failed(pool: &SqlitePool, run_id: RunId, summary: &str) -> Result<()> {
    // Bound the stored summary — it's a headline, not the log. The full output
    // is at the gate's log_ref / journald.
    let summary: String = summary.chars().take(600).collect();
    sqlx::query(
        "UPDATE build_runs SET result = 'failed', phase = 'done', failure_summary = ?, finished_at = ?
         WHERE id = ? AND result = 'building'",
    )
    .bind(&summary)
    .bind(Utc::now().to_rfc3339())
    .bind(run_id.0)
    .execute(pool)
    .await?;
    Ok(())
}

/// Settle the run as superseded by a newer `/rebuild`.
pub async fn mark_aborted(pool: &SqlitePool, run_id: RunId) -> Result<()> {
    sqlx::query(
        "UPDATE build_runs SET result = 'aborted', phase = 'done',
                failure_summary = 'superseded by a newer /rebuild', finished_at = ?
         WHERE id = ? AND result = 'building'",
    )
    .bind(Utc::now().to_rfc3339())
    .bind(run_id.0)
    .execute(pool)
    .await?;
    Ok(())
}

/// Settle any `build_runs` left `result = 'building'` by a daemon that died
/// mid-build (crash, OOM, `systemctl restart`, a self-update that SIGKILLed an
/// in-flight build). Without this they stay `'building'` forever and `/state` +
/// `GET /runs/{id}/wait` report an ever-growing elapsed for a build that will
/// never settle (the Run-2 SERIOUS-2 gap). Run once at startup before serving.
/// Returns the number of orphaned runs reconciled.
pub async fn recover_orphaned_running(pool: &SqlitePool) -> Result<u64> {
    let res = sqlx::query(
        "UPDATE build_runs SET result = 'aborted', phase = 'done',
                failure_summary = 'daemon restarted mid-build', finished_at = ?
         WHERE result = 'building'",
    )
    .bind(Utc::now().to_rfc3339())
    .execute(pool)
    .await?;
    Ok(res.rows_affected())
}

/// One gate's status within a run view.
#[derive(Debug, Serialize)]
pub struct RunGateView {
    pub kind: String,
    /// `'passed' | 'failed' | 'blocked'` or NULL while in-flight.
    pub status: Option<String>,
    /// Relative path under `cfg.logs_root` for the full byte stream.
    pub log_ref: Option<String>,
}

/// The `GET /runs/{id}` payload.
#[derive(Debug, Serialize)]
pub struct RunView {
    pub run_id: i64,
    pub sha: String,
    pub version: Option<String>,
    pub phase: String,
    /// `'building' | 'passed' | 'failed' | 'aborted'`.
    pub result: String,
    pub started_at: String,
    pub finished_at: Option<String>,
    /// Headline cause when `result = 'failed'`: first compile error or first
    /// red gate. NULL otherwise.
    pub failure_summary: Option<String>,
    /// Gates run on the host tier for this run's version, latest row per kind.
    /// Empty until the run reaches a version + the gating phase.
    pub gates: Vec<RunGateView>,
}

/// Load a run plus its host-tier gate statuses. `None` if the id is unknown.
pub async fn get(pool: &SqlitePool, run_id: RunId) -> Result<Option<RunView>> {
    let Some(row) = sqlx::query(
        "SELECT id, app, sha, version, phase, result, started_at, finished_at, failure_summary
         FROM build_runs WHERE id = ?",
    )
    .bind(run_id.0)
    .fetch_optional(pool)
    .await?
    else {
        return Ok(None);
    };

    let version: Option<String> = row.get("version");
    // The run's own app, read from its row rather than passed in: a run id is
    // unique across products, and the row is the authority on which product it
    // belongs to. Asking the caller would let a lookup for one product's run
    // return another's gates.
    let app: String = row.get("app");
    // **This run's gates, keyed on this run.** A build run's id IS the build
    // identity every gate it ran recorded (`gate_runs.build_id`), so asking for
    // them is an exact question with an exact answer.
    //
    // It was keyed on (tier, version) until 2026-08-20, and a rebuild at an
    // unchanged version is the normal way to retry a red build: runs 60, 61 and
    // 62 of mnw-server 0.11.20 wrote over each other's rows, and `/runs/62`
    // answered with whichever run had touched each gate last. Two reads a minute
    // apart, same run, no promote between them, disagreed about whether
    // `hardening_test` had passed or not run at all.
    //
    // A pre-migration-008 run left rows with a NULL `build_id` and so reports no
    // gates here. That is the honest answer for a row that never recorded which
    // build it vouched for, and the same call the migration made in refusing to
    // backfill one.
    let gates: Vec<RunGateView> = sqlx::query(
        "SELECT gate_kind, status, log_ref
         FROM gate_runs g
         WHERE app = ?1 AND build_id = ?2
           AND id = (SELECT MAX(id) FROM gate_runs
                     WHERE app = ?1 AND build_id = ?2
                       AND gate_kind = g.gate_kind)
         ORDER BY gate_kind",
    )
    .bind(&app)
    .bind(run_id.0)
    .fetch_all(pool)
    .await?
    .into_iter()
    .map(|gr| RunGateView {
        kind: gr.get("gate_kind"),
        status: gr.get("status"),
        log_ref: gr.get("log_ref"),
    })
    .collect();

    Ok(Some(RunView {
        run_id: row.get("id"),
        sha: row.get("sha"),
        version,
        phase: row.get("phase"),
        result: row.get("result"),
        started_at: row.get("started_at"),
        finished_at: row.get("finished_at"),
        failure_summary: row.get("failure_summary"),
        gates,
    }))
}

/// Compact view of the latest build run for `/state`'s liveness line.
#[derive(Debug, Serialize)]
pub struct BuildSummary {
    pub run_id: i64,
    pub sha: String,
    pub version: Option<String>,
    pub phase: String,
    pub result: String,
    pub failure_summary: Option<String>,
    /// Seconds from start to finish (or to now while building). Lets a
    /// `/state` poller show "building <ver>, phase=<x>, elapsed Ns" instead of
    /// a version frozen at the last success for the whole ~10-min build.
    pub elapsed_s: i64,
}

/// The most recent build run for `app`, for `/state`. `None` until that
/// product's first `/rebuild`.
pub async fn latest_summary(pool: &SqlitePool, app: &AppId) -> Result<Option<BuildSummary>> {
    let Some(row) = sqlx::query(
        "SELECT id, sha, version, phase, result, failure_summary, started_at, finished_at
         FROM build_runs WHERE app = ? ORDER BY id DESC LIMIT 1",
    )
    .bind(app)
    .fetch_optional(pool)
    .await?
    else {
        return Ok(None);
    };
    let started_at: String = row.get("started_at");
    let finished_at: Option<String> = row.get("finished_at");
    Ok(Some(BuildSummary {
        run_id: row.get("id"),
        sha: row.get("sha"),
        version: row.get("version"),
        phase: row.get("phase"),
        result: row.get("result"),
        failure_summary: row.get("failure_summary"),
        elapsed_s: elapsed_seconds(&started_at, finished_at.as_deref()),
    }))
}

/// Seconds between an rfc3339 `started_at` and (`finished_at` or now), clamped
/// at 0. A parse failure yields 0 rather than erroring the whole `/state` call.
fn elapsed_seconds(started_at: &str, finished_at: Option<&str>) -> i64 {
    let Ok(start) = chrono::DateTime::parse_from_rfc3339(started_at) else {
        return 0;
    };
    let end = match finished_at {
        Some(f) => chrono::DateTime::parse_from_rfc3339(f)
            .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc)),
        None => Utc::now(),
    };
    (end - start.with_timezone(&Utc)).num_seconds().max(0)
}

/// The summary of the first gate `run_id` failed, if any. The build pipeline
/// uses it to populate `failure_summary` when `run_all` reports a red pipeline.
/// Reads the typed `outcome_json` so the stored headline matches what the TUI
/// renders.
///
/// Keyed on the run, not on (tier, version). Version-keyed, it would read the
/// first failure any run of that version had ever recorded, so a rebuild would
/// inherit its predecessor's headline and report a failure beside the same
/// gate listed as passed.
pub async fn first_failed_gate_summary(
    pool: &SqlitePool,
    app: &AppId,
    run_id: RunId,
) -> Option<String> {
    let row = sqlx::query(
        "SELECT gate_kind, outcome_json FROM gate_runs
         WHERE app = ? AND build_id = ? AND status = 'failed'
         ORDER BY id ASC LIMIT 1",
    )
    .bind(app)
    .bind(run_id.0)
    .fetch_optional(pool)
    .await
    .ok()
    .flatten()?;
    let kind: String = row.get("gate_kind");
    let outcome_json: Option<String> = row.get("outcome_json");
    let summary = outcome_json
        .and_then(|s| serde_json::from_str::<crate::outcome::GateOutcome>(&s).ok())
        .map_or_else(
            || "gate failed".to_string(),
            |o| match o.status {
                crate::outcome::GateStatus::Failed { failure } => failure.summary(),
                other => format!("{other:?}"),
            },
        );
    Some(format!("{kind}: {summary}"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use sqlx::sqlite::SqlitePoolOptions;

    async fn pool() -> SqlitePool {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        crate::db::migrate(&pool).await.unwrap();
        pool
    }

    /// Two products do not see each other's builds or tier state.
    ///
    /// The whole point of the app dimension, in one test. Without it these reads
    /// answer from a single global pile: pom's `/state` reports MNW's latest
    /// build, and advancing pom's `host` tier moves MNW's, silently, because a
    /// shared row looks exactly like a correct one.
    #[tokio::test]
    async fn one_apps_state_is_invisible_to_another() {
        let pool = pool().await;
        let mnw = AppId::new("mnw");
        let pom = AppId::new("pom");

        // Each product needs its own tier and version rows to advance against.
        for app in [&mnw, &pom] {
            sqlx::query("INSERT INTO tiers (app, name, ord) VALUES (?, 'host', 0)")
                .bind(app)
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query("INSERT INTO tier_state (app, tier) VALUES (?, 'host')")
                .bind(app)
                .execute(&pool)
                .await
                .unwrap();
            sqlx::query(
                "INSERT INTO versions (app, version, git_sha, built_at, artifact_path)
                 VALUES (?, '1.0.0', 'sha', '2026-08-06T00:00:00Z', '/r')",
            )
            .bind(app)
            .execute(&pool)
            .await
            .unwrap();
        }

        let mnw_run = create(&pool, &mnw, "aaaaaaa").await.unwrap();
        let pom_run = create(&pool, &pom, "bbbbbbb").await.unwrap();

        // Latest build is per product, not "whichever ran last".
        assert_eq!(
            latest_summary(&pool, &mnw).await.unwrap().unwrap().sha,
            "aaaaaaa"
        );
        assert_eq!(
            latest_summary(&pool, &pom).await.unwrap().unwrap().sha,
            "bbbbbbb",
            "pom's newest build is pom's, even though mnw's is older"
        );

        // Advancing one product's `host` tier leaves the other's alone.
        let v = Version::parse("1.0.0").unwrap();
        advance_tier(&pool, &pom, "host", &v, Some(pom_run.0))
            .await
            .unwrap();
        let mnw_current: Option<String> =
            sqlx::query_scalar("SELECT current_version FROM tier_state WHERE app = 'mnw'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(
            mnw_current, None,
            "advancing pom's host tier must not advance mnw's"
        );

        // And a run resolves its gates through its own product.
        assert_eq!(get(&pool, mnw_run).await.unwrap().unwrap().sha, "aaaaaaa");
    }

    #[tokio::test]
    async fn create_then_get_roundtrips_building() {
        let pool = pool().await;
        let id = create(&pool, &AppId::default(), "abc1234").await.unwrap();
        let v = get(&pool, id).await.unwrap().expect("run exists");
        assert_eq!(v.sha, "abc1234");
        assert_eq!(v.result, "building");
        assert_eq!(v.phase, "queued");
        assert!(v.version.is_none());
        assert!(v.gates.is_empty());
        assert!(v.failure_summary.is_none());
    }

    #[tokio::test]
    async fn recover_orphaned_running_settles_building_runs() {
        let pool = pool().await;
        // Two in-flight runs (as if the daemon died mid-build) + one already
        // settled, which must be left untouched.
        let run_a = create(&pool, &AppId::default(), "aaaaaaa").await.unwrap();
        let run_b = create(&pool, &AppId::default(), "bbbbbbb").await.unwrap();
        let run_c = create(&pool, &AppId::default(), "ccccccc").await.unwrap();
        mark_passed(&pool, run_c).await.unwrap();

        let reconciled = recover_orphaned_running(&pool).await.unwrap();
        assert_eq!(reconciled, 2, "both 'building' runs reconciled");

        for id in [run_a, run_b] {
            let rec = get(&pool, id).await.unwrap().unwrap();
            assert_eq!(rec.result, "aborted");
            assert_eq!(rec.phase, "done");
            assert!(rec.finished_at.is_some());
            assert_eq!(
                rec.failure_summary.as_deref(),
                Some("daemon restarted mid-build")
            );
        }
        // The already-settled run is unchanged.
        assert_eq!(get(&pool, run_c).await.unwrap().unwrap().result, "passed");

        // Idempotent: a second pass finds nothing to do.
        assert_eq!(recover_orphaned_running(&pool).await.unwrap(), 0);
    }

    #[tokio::test]
    async fn phase_and_version_advance_then_pass() {
        let pool = pool().await;
        let id = create(&pool, &AppId::default(), "abc1234").await.unwrap();
        set_phase(&pool, id, Phase::Compiling).await.unwrap();
        let ver: Version = "0.10.2".parse().unwrap();
        set_version(&pool, id, &ver).await.unwrap();
        mark_passed(&pool, id).await.unwrap();

        let v = get(&pool, id).await.unwrap().unwrap();
        assert_eq!(v.result, "passed");
        assert_eq!(v.phase, "done");
        assert_eq!(v.version.as_deref(), Some("0.10.2"));
        assert!(v.finished_at.is_some());
    }

    #[tokio::test]
    async fn first_terminal_write_wins() {
        let pool = pool().await;
        let id = create(&pool, &AppId::default(), "abc1234").await.unwrap();
        mark_failed(&pool, id, "error[E0063]: missing field user_pages_host")
            .await
            .unwrap();
        // A later pass attempt (e.g. the task catch racing a build-step error)
        // must not overwrite the recorded failure.
        mark_passed(&pool, id).await.unwrap();
        // And a second failure summary doesn't clobber the first.
        mark_failed(&pool, id, "something else").await.unwrap();

        let v = get(&pool, id).await.unwrap().unwrap();
        assert_eq!(v.result, "failed");
        assert_eq!(
            v.failure_summary.as_deref(),
            Some("error[E0063]: missing field user_pages_host")
        );
    }

    #[tokio::test]
    async fn phase_write_after_terminal_is_noop() {
        let pool = pool().await;
        let id = create(&pool, &AppId::default(), "abc1234").await.unwrap();
        mark_passed(&pool, id).await.unwrap();
        set_phase(&pool, id, Phase::Gating).await.unwrap();
        let v = get(&pool, id).await.unwrap().unwrap();
        assert_eq!(
            v.phase, "done",
            "a late phase write must not move a finished run"
        );
    }

    #[test]
    fn elapsed_seconds_uses_finished_when_present() {
        // Both timestamps present → exact span, no wall-clock dependency.
        let s = elapsed_seconds("2026-06-13T00:00:00Z", Some("2026-06-13T00:02:05Z"));
        assert_eq!(s, 125);
        // Unparseable start → 0, never a panic / negative.
        assert_eq!(elapsed_seconds("not-a-date", None), 0);
    }

    #[tokio::test]
    async fn latest_summary_reports_most_recent_run() {
        let pool = pool().await;
        assert!(
            latest_summary(&pool, &AppId::default())
                .await
                .unwrap()
                .is_none()
        );
        let _old = create(&pool, &AppId::default(), "old1234").await.unwrap();
        let new = create(&pool, &AppId::default(), "new5678").await.unwrap();
        set_phase(&pool, new, Phase::Compiling).await.unwrap();
        let sum = latest_summary(&pool, &AppId::default())
            .await
            .unwrap()
            .expect("a run exists");
        assert_eq!(sum.run_id, new.0);
        assert_eq!(sum.sha, "new5678");
        assert_eq!(sum.phase, "compiling");
        assert_eq!(sum.result, "building");
    }

    #[tokio::test]
    async fn get_unknown_id_is_none() {
        let pool = pool().await;
        assert!(get(&pool, RunId(999)).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn failure_summary_is_bounded() {
        let pool = pool().await;
        let id = create(&pool, &AppId::default(), "abc1234").await.unwrap();
        mark_failed(&pool, id, &"x".repeat(5_000)).await.unwrap();
        let v = get(&pool, id).await.unwrap().unwrap();
        assert!(v.failure_summary.unwrap().len() <= 600);
    }
}