//! Startup crash-recovery reconcile of `deploys` against `tier_state`.
//!
//! A promote lands a build on a tier's nodes and *then* advances tier_state:
//! [`crate::routes`] inserts the per-node `deploys` row (outcome `ok`) inside the
//! canary loop, and only after every node succeeds does [`crate::runs::advance_tier`]
//! record the new version. If sandod dies between those two writes — OOM, a
//! `systemctl restart`, a self-update SIGKILL — the node is serving the new
//! version but tier_state still names the old one, and `partial_reason` is NULL
//! because no failure path executed. `/state` then shows a clean tier on the
//! wrong version, and a later `/rollback` walks back to `previous_version`, which
//! is now *two* versions behind what the node actually runs (audit 2026-07-21,
//! sando-h5-crash-mid-promote).
//!
//! [`recover_unrecorded_deploys`] runs once at startup, next to
//! [`crate::runs::recover_orphaned_running`], and flags any tier whose latest
//! successful deploy postdates the last recorded state change yet names a
//! different version/build than tier_state holds. The flag is a `partial_reason`,
//! the same channel the promote/rollback failure paths use, so `/state` and the
//! TUI go loud and force the operator to reconcile by hand rather than trusting a
//! stale-green tier.
//!
//! The comparison instant is `tier_state.advanced_at` (migration 009), set on
//! every advance and rollback and never nulled — unlike `burn_in_started_at`,
//! which rollback and `reset_burn_in` clear while the deployed identity stays
//! real. A clean promote sets `advanced_at` *after* its `deploys` row, so its own
//! landing never postdates it; a rollback sets it after the roll-off row, so a
//! cleanly rolled-back tier is not mistaken for an unrecorded deploy.
//!
//! The `deploys` row is written `in_progress` *before* the node is touched and
//! finalized to ok/failed right after, so the only sliver still uncovered is a
//! crash between the pre-write and the swap — where the node was never changed,
//! so there is nothing to reconcile. Any row still `in_progress` at startup is an
//! orphaned in-flight deploy: it is flagged (the swap may or may not have landed)
//! and settled to a terminal `failed` so the audit trail carries no eternal
//! in-flight row.

use anyhow::Result;
use chrono::{DateTime, Utc};
use sqlx::{Row, SqlitePool};

/// One tier's most recent deploy row, whatever its outcome.
struct LastDeploy {
    version: String,
    build_id: Option<i64>,
    /// `'ok' | 'failed' | 'in_progress'`.
    outcome: String,
    /// `finished_at` when the row is terminal, else `started_at` — the instant to
    /// weigh against the tier's last state change. `started_at` is NOT NULL.
    instant: String,
}

/// Reconcile `deploys` against `tier_state` after a crash. Flags any tier whose
/// most recent deploy postdates the last recorded state change but disagrees with
/// what tier_state holds (a promote that died before advancing) or is still
/// in-flight (a promote interrupted mid-node). Sets `partial_reason` — only when
/// it is currently NULL, so a more specific promote/rollback failure reason
/// stands — and settles orphaned `in_progress` rows. Returns how many tiers were
/// flagged. Run once at startup, before serving.
pub async fn recover_unrecorded_deploys(pool: &SqlitePool) -> Result<u64> {
    let tiers = sqlx::query(
        "SELECT tier, current_version, current_build_id, advanced_at, partial_reason
         FROM tier_state",
    )
    .fetch_all(pool)
    .await?;

    let mut flagged = 0u64;
    for row in tiers {
        let tier: String = row.get("tier");
        // A tier already flagged partial carries a more specific reason from the
        // path that flagged it; don't overwrite it.
        if row.get::<Option<String>, _>("partial_reason").is_some() {
            continue;
        }
        let Some(landing) = latest_deploy(pool, &tier).await? else {
            continue;
        };
        let current_version: Option<String> = row.get("current_version");
        let current_build_id: Option<i64> = row.get("current_build_id");
        let advanced_at: Option<String> = row.get("advanced_at");

        // Only a deploy that postdates the last recorded state change is suspect:
        // a clean promote/rollback stamps `advanced_at` after its own rows.
        if !postdates(&landing.instant, advanced_at.as_deref()) {
            continue;
        }
        let reason = match landing.outcome.as_str() {
            // Interrupted mid-node: the swap may or may not have landed, and no
            // outcome was ever recorded. Force a look regardless of identity.
            "in_progress" => Some(format!(
                "crash-recovery: a deploy of {deployed} to this tier was in flight when the \
                 daemon last stopped and never recorded an outcome. The node's binary may or \
                 may not have been swapped. Verify what the tier is actually running, then \
                 re-promote or roll back deliberately.",
                deployed = describe(&landing.version, landing.build_id),
            )),
            // Landed successfully but tier_state names something else — the promote
            // died after the deploy but before `advance_tier`.
            "ok" if !identity_matches(&landing, current_version.as_deref(), current_build_id) => {
                Some(format!(
                    "crash-recovery: node(s) on this tier were deployed {deployed} at {when} but \
                     tier_state still records {recorded}. A promote most likely died after the \
                     deploy landed but before tier_state advanced. Verify what the tier is \
                     actually running, then re-promote or roll back deliberately — a blind \
                     /rollback would target the wrong version.",
                    deployed = describe(&landing.version, landing.build_id),
                    when = landing.instant,
                    recorded = current_version.as_deref().unwrap_or("no version"),
                ))
            }
            // A terminal 'failed' row was handled by the deploy's own failure path
            // (which sets its own partial_reason); nothing to add here.
            _ => None,
        };
        let Some(reason) = reason else { continue };

        // `AND partial_reason IS NULL` keeps the write a no-op if anything set a
        // reason since the read above (startup is single-threaded, but the guard
        // costs nothing and states the intent).
        let res = sqlx::query(
            "UPDATE tier_state SET partial_reason = ? WHERE tier = ? AND partial_reason IS NULL",
        )
        .bind(&reason)
        .bind(&tier)
        .execute(pool)
        .await?;
        if res.rows_affected() > 0 {
            tracing::error!(tier = %tier, %reason, "startup reconcile: interrupted/unrecorded deploy on tier");
            flagged += 1;
        }
    }

    // Settle any orphaned in-flight deploy rows so the trail carries no eternal
    // 'in_progress'. Done after the flag decisions above, which read that state.
    // At startup (single-threaded, pre-serving) any 'in_progress' row is an orphan.
    settle_orphaned_in_progress(pool).await?;
    Ok(flagged)
}

/// The most recent deploy on `tier`, any outcome. Rows are ordered by `id`
/// (monotonic autoincrement), so this is the last one regardless of clock skew.
async fn latest_deploy(pool: &SqlitePool, tier: &str) -> Result<Option<LastDeploy>> {
    let row = sqlx::query(
        "SELECT version, build_id, outcome, started_at, finished_at FROM deploys
         WHERE tier = ? ORDER BY id DESC LIMIT 1",
    )
    .bind(tier)
    .fetch_optional(pool)
    .await?;
    Ok(row.map(|r| {
        let started_at: String = r.get("started_at");
        let finished_at: Option<String> = r.get("finished_at");
        LastDeploy {
            version: r.get("version"),
            build_id: r.get("build_id"),
            outcome: r.get("outcome"),
            instant: finished_at.unwrap_or(started_at),
        }
    }))
}

/// Mark every `in_progress` deploy row terminal-failed with an interruption note,
/// mirroring [`crate::runs::recover_orphaned_running`] for build runs. Returns the
/// number of rows settled.
async fn settle_orphaned_in_progress(pool: &SqlitePool) -> Result<u64> {
    let outcome =
        crate::outcome::DeployOutcome::failed(crate::outcome::DeployFailureKind::Unclassified {
            detail: "daemon restarted mid-deploy".into(),
        });
    let outcome_json = serde_json::to_string(&outcome)?;
    let res = sqlx::query(
        "UPDATE deploys SET outcome = 'failed', finished_at = ?, outcome_json = ?
         WHERE outcome = 'in_progress'",
    )
    .bind(Utc::now().to_rfc3339())
    .bind(&outcome_json)
    .execute(pool)
    .await?;
    let n = res.rows_affected();
    if n > 0 {
        tracing::warn!(
            settled = n,
            "startup reconcile: settled orphaned in-flight deploy row(s)"
        );
    }
    Ok(n)
}

/// Whether `deploy_finished_at` is later than the tier's last recorded state
/// change. A NULL `advanced_at` (state never advanced, yet a deploy landed) and
/// any unparseable timestamp both count as postdating — fail toward flagging, so
/// a corrupt or missing instant forces operator attention rather than hiding.
fn postdates(deploy_finished_at: &str, advanced_at: Option<&str>) -> bool {
    let Some(advanced_at) = advanced_at else {
        return true;
    };
    match (
        DateTime::parse_from_rfc3339(deploy_finished_at),
        DateTime::parse_from_rfc3339(advanced_at),
    ) {
        (Ok(d), Ok(a)) => d > a,
        _ => true,
    }
}

/// Whether the landed artifact is the one tier_state records. Build identity wins
/// when both sides carry it (a re-deploy of the same version from a different
/// build is still a mismatch); otherwise fall back to the version string. A NULL
/// `current_version` with a real landing is a mismatch.
fn identity_matches(
    landing: &LastDeploy,
    current_version: Option<&str>,
    current_build_id: Option<i64>,
) -> bool {
    match (landing.build_id, current_build_id) {
        (Some(landed), Some(current)) => landed == current,
        _ => current_version == Some(landing.version.as_str()),
    }
}

/// Human label for a landed artifact: the version, plus the build id when known.
fn describe(version: &str, build_id: Option<i64>) -> String {
    match build_id {
        Some(id) => format!("{version} (build {id})"),
        None => version.to_string(),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{Duration, Utc};
    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
    }

    async fn seed_tier(pool: &SqlitePool, tier: &str) {
        sqlx::query(
            "INSERT INTO tiers (name, ord, provisioned, canary) VALUES (?, 0, 1, 'sequential')",
        )
        .bind(tier)
        .execute(pool)
        .await
        .unwrap();
        sqlx::query("INSERT INTO tier_state (tier) VALUES (?)")
            .bind(tier)
            .execute(pool)
            .await
            .unwrap();
    }

    /// Set the tier's recorded state the way an advance/rollback would.
    async fn set_state(
        pool: &SqlitePool,
        tier: &str,
        current_version: Option<&str>,
        advanced_at: Option<&str>,
    ) {
        if let Some(v) = current_version {
            sqlx::query(
                "INSERT OR IGNORE INTO versions (version, git_sha, built_at, artifact_path)
                 VALUES (?, 'sha', '2026-07-21T00:00:00Z', '/r/x')",
            )
            .bind(v)
            .execute(pool)
            .await
            .unwrap();
        }
        sqlx::query("UPDATE tier_state SET current_version = ?, advanced_at = ? WHERE tier = ?")
            .bind(current_version)
            .bind(advanced_at)
            .bind(tier)
            .execute(pool)
            .await
            .unwrap();
    }

    /// FK parents `deploys` needs: a `versions` row and a `nodes` row. sqlx
    /// enables `foreign_keys` by default, so these must exist first.
    async fn seed_deploy_fks(pool: &SqlitePool, tier: &str, version: &str) {
        sqlx::query(
            "INSERT OR IGNORE INTO versions (version, git_sha, built_at, artifact_path)
             VALUES (?, 'sha', '2026-07-21T00:00:00Z', '/r/x')",
        )
        .bind(version)
        .execute(pool)
        .await
        .unwrap();
        sqlx::query(
            "INSERT OR IGNORE INTO nodes (name, tier, ssh_target, release_root)
             VALUES ('n1', ?, 'local', '/opt/mnw')",
        )
        .bind(tier)
        .execute(pool)
        .await
        .unwrap();
    }

    /// Insert a `deploys` row, seeding its FK parents first.
    async fn insert_deploy(
        pool: &SqlitePool,
        tier: &str,
        version: &str,
        outcome: &str,
        finished_at: &str,
    ) {
        seed_deploy_fks(pool, tier, version).await;
        sqlx::query(
            "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome)
             VALUES (?, ?, 'n1', ?, ?, ?)",
        )
        .bind(version)
        .bind(tier)
        .bind(finished_at)
        .bind(finished_at)
        .bind(outcome)
        .execute(pool)
        .await
        .unwrap();
    }

    async fn partial_reason(pool: &SqlitePool, tier: &str) -> Option<String> {
        sqlx::query_scalar("SELECT partial_reason FROM tier_state WHERE tier = ?")
            .bind(tier)
            .fetch_one(pool)
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn clean_promote_is_not_flagged() {
        // deploy lands, THEN tier_state advances: advanced_at postdates the row.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let deployed = Utc::now();
        insert_deploy(&pool, "b", "0.10.15", "ok", &deployed.to_rfc3339()).await;
        set_state(
            &pool,
            "b",
            Some("0.10.15"),
            Some(&(deployed + Duration::seconds(1)).to_rfc3339()),
        )
        .await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
        assert!(partial_reason(&pool, "b").await.is_none());
    }

    #[tokio::test]
    async fn crash_mid_promote_is_flagged() {
        // tier_state still on the OLD version; an 'ok' deploy of the NEW version
        // postdates the last advance — the exact SIGKILL-before-advance window.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let advanced = Utc::now() - Duration::hours(1);
        set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
        insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
        let reason = partial_reason(&pool, "b").await.expect("flagged");
        assert!(reason.contains("0.10.15"), "names the deployed version");
        assert!(reason.contains("0.10.14"), "names the recorded version");
        assert!(reason.contains("crash-recovery"));
    }

    #[tokio::test]
    async fn rolled_back_tier_is_not_flagged() {
        // Promote to 0.10.15 landed (ok deploy), then a rollback moved tier_state
        // to 0.10.14 and stamped advanced_at AFTER the roll-off deploy. The stale
        // deploy row must not read as unrecorded.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let deployed = Utc::now() - Duration::minutes(5);
        insert_deploy(&pool, "b", "0.10.15", "ok", &deployed.to_rfc3339()).await;
        set_state(&pool, "b", Some("0.10.14"), Some(&Utc::now().to_rfc3339())).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
        assert!(partial_reason(&pool, "b").await.is_none());
    }

    #[tokio::test]
    async fn an_already_partial_tier_keeps_its_reason() {
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let advanced = Utc::now() - Duration::hours(1);
        set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
        insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;
        sqlx::query(
            "UPDATE tier_state SET partial_reason = 'canary rollback incomplete' WHERE tier = 'b'",
        )
        .execute(&pool)
        .await
        .unwrap();

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
        assert_eq!(
            partial_reason(&pool, "b").await.as_deref(),
            Some("canary rollback incomplete"),
            "the specific reason is not clobbered",
        );
    }

    #[tokio::test]
    async fn a_failed_deploy_does_not_flag() {
        // The latest deploy is a failure (its own path handles partial state);
        // there is no successful landing that tier_state failed to record.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let advanced = Utc::now() - Duration::hours(1);
        set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
        insert_deploy(&pool, "b", "0.10.15", "failed", &Utc::now().to_rfc3339()).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
        assert!(partial_reason(&pool, "b").await.is_none());
    }

    #[tokio::test]
    async fn a_tier_with_no_deploys_is_skipped() {
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        set_state(&pool, "b", None, None).await;
        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
    }

    #[tokio::test]
    async fn crash_on_first_ever_promote_is_flagged() {
        // Never advanced (advanced_at NULL, current_version NULL) but a node was
        // deployed: prod is running something tier_state has no record of.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
        assert!(
            partial_reason(&pool, "b")
                .await
                .unwrap()
                .contains("0.10.15")
        );
    }

    #[tokio::test]
    async fn build_id_mismatch_is_flagged_even_when_version_matches() {
        // Same version string, different build — a re-deploy landed a build the
        // tier never recorded. Build identity is the tie-breaker.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        seed_deploy_fks(&pool, "b", "0.10.15").await;
        // build_runs 7 and 9, referenced by tier_state.current_build_id and
        // deploys.build_id respectively.
        for id in [7, 9] {
            sqlx::query(
                "INSERT INTO build_runs (id, sha, phase, result, started_at)
                 VALUES (?, 'sha', 'done', 'passed', '2026-07-21T00:00:00Z')",
            )
            .bind(id)
            .execute(&pool)
            .await
            .unwrap();
        }
        let advanced = Utc::now() - Duration::hours(1);
        // tier_state: version 0.10.15, build 7.
        sqlx::query(
            "UPDATE tier_state SET current_version = '0.10.15', current_build_id = 7, advanced_at = ? WHERE tier = 'b'",
        )
        .bind(advanced.to_rfc3339())
        .execute(&pool)
        .await
        .unwrap();
        // deploy: same version 0.10.15 but build 9, postdating the advance.
        sqlx::query(
            "INSERT INTO deploys (version, tier, node, started_at, finished_at, outcome, build_id)
             VALUES ('0.10.15', 'b', 'n1', ?, ?, 'ok', 9)",
        )
        .bind(Utc::now().to_rfc3339())
        .bind(Utc::now().to_rfc3339())
        .execute(&pool)
        .await
        .unwrap();

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
    }

    /// Insert an in-flight deploy row: `outcome = 'in_progress'`, `finished_at`
    /// NULL, `started_at` set — the shape the promote loop writes before touching
    /// a node.
    async fn insert_in_progress(pool: &SqlitePool, tier: &str, version: &str, started_at: &str) {
        seed_deploy_fks(pool, tier, version).await;
        sqlx::query(
            "INSERT INTO deploys (version, tier, node, started_at, outcome)
             VALUES (?, ?, 'n1', ?, 'in_progress')",
        )
        .bind(version)
        .bind(tier)
        .bind(started_at)
        .execute(pool)
        .await
        .unwrap();
    }

    async fn outcome_of_latest(pool: &SqlitePool, tier: &str) -> String {
        sqlx::query_scalar("SELECT outcome FROM deploys WHERE tier = ? ORDER BY id DESC LIMIT 1")
            .bind(tier)
            .fetch_one(pool)
            .await
            .unwrap()
    }

    #[tokio::test]
    async fn orphaned_in_progress_deploy_is_flagged_and_settled() {
        // A deploy was in flight (row written before the swap) when the daemon
        // died: flag the tier and settle the row so it isn't in_progress forever.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let advanced = Utc::now() - Duration::hours(1);
        set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
        insert_in_progress(&pool, "b", "0.10.15", &Utc::now().to_rfc3339()).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
        let reason = partial_reason(&pool, "b").await.expect("flagged");
        assert!(
            reason.contains("in flight"),
            "names the interruption: {reason}"
        );
        assert!(reason.contains("0.10.15"));
        // The orphan row is settled to a terminal outcome.
        assert_eq!(outcome_of_latest(&pool, "b").await, "failed");
    }

    #[tokio::test]
    async fn in_progress_row_predating_last_advance_is_only_settled_not_flagged() {
        // An old in_progress row from before the tier's last clean advance is
        // stale bookkeeping, not an unrecorded deploy: settle it, don't flag.
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let old = Utc::now() - Duration::hours(2);
        insert_in_progress(&pool, "b", "0.10.13", &old.to_rfc3339()).await;
        // A later clean promote advanced the tier past that row.
        let deployed = Utc::now() - Duration::minutes(10);
        insert_deploy(&pool, "b", "0.10.14", "ok", &deployed.to_rfc3339()).await;
        set_state(&pool, "b", Some("0.10.14"), Some(&Utc::now().to_rfc3339())).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
        assert!(partial_reason(&pool, "b").await.is_none());
        // But the stale in_progress row is still settled.
        let leftover: i64 =
            sqlx::query_scalar("SELECT COUNT(*) FROM deploys WHERE outcome = 'in_progress'")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(leftover, 0, "no in_progress rows survive");
    }

    #[tokio::test]
    async fn is_idempotent_after_flagging() {
        let pool = pool().await;
        seed_tier(&pool, "b").await;
        let advanced = Utc::now() - Duration::hours(1);
        set_state(&pool, "b", Some("0.10.14"), Some(&advanced.to_rfc3339())).await;
        insert_deploy(&pool, "b", "0.10.15", "ok", &Utc::now().to_rfc3339()).await;

        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 1);
        // Second pass: the reason is already set, so nothing new is flagged.
        assert_eq!(recover_unrecorded_deploys(&pool).await.unwrap(), 0);
    }
}