//! The migration dry run: prove every pending migration applies to a restored
//! copy of what production is actually holding, before a deploy makes that
//! irreversible.

use super::GateCtx;
use super::log::GateLog;
use super::pg::{pg_create_db, pg_url_with_dbname, reset_scratch, restore_dump, run_migrator};
use crate::classify;
use crate::domain::{GateKind, GateRunId};
use crate::outcome::{GateBlocker, GateFailure, GateOutcome, PassNote};
use anyhow::Result;
use chrono::Utc;

pub(super) async fn migration_dry_run(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
    let log = GateLog::open(ctx, run_id, GateKind::MigrationDryRun).await;
    let outcome = migration_dry_run_inner(ctx, &log).await;
    log.close().await;
    outcome.map(|o| o.with_log_ref(ctx.log_ref(GateKind::MigrationDryRun)))
}

/// The staged interior of [`migration_dry_run`], writing every step through the
/// gate's live log. The caller owns the sink so it can flush it on every exit
/// path, and attaches the `log_ref` once instead of at each return.
/// Runs one configured check per database, in config order, and stops at the
/// first that does not pass — a red gate is a red gate, and continuing would
/// bury it under a second restore's output.
///
/// The server's check runs against `scratch_db_url` itself and is deliberately
/// last-writer for it: `cargo_test` reuses that database in migrated state, so
/// every other check must name its own `scratch_db` (enforced at config load).
async fn migration_dry_run_inner(ctx: &GateCtx, log: &GateLog) -> Result<GateOutcome> {
    let Some(scratch_url) = ctx.cfg.scratch_db_url.as_deref() else {
        log.line("scratch_db_url unset in daemon config\n").await;
        return Ok(GateOutcome::blocked(GateBlocker::ScratchDbUrlUnset));
    };

    let mut checked = Vec::new();
    let mut primary_backup_path = String::new();
    for check in &ctx.cfg.migration_checks {
        let label = check.dir.display().to_string();
        log.line(&format!("==== migration_check: {label} ====\n"))
            .await;
        match run_migration_check(ctx, log, scratch_url, check).await? {
            CheckResult::Passed { backup_path } => {
                if primary_backup_path.is_empty() {
                    primary_backup_path = backup_path;
                }
                checked.push(label);
            }
            CheckResult::Stopped(outcome) => return Ok(outcome),
        }
    }

    log.line(&format!(
        "all {} migration check(s) passed: {}",
        checked.len(),
        checked.join(", ")
    ))
    .await;
    Ok(GateOutcome::passed(PassNote::Migrated {
        backup_path: primary_backup_path,
        checks: checked,
    }))
}

/// One check's verdict: it passed (against `backup_path`), or it produced the
/// outcome the whole gate reports.
enum CheckResult {
    Passed { backup_path: String },
    Stopped(GateOutcome),
}

/// Restore one database's dump into its scratch DB and run its migrations on top.
async fn run_migration_check(
    ctx: &GateCtx,
    log: &GateLog,
    scratch_url: &str,
    check: &crate::config::MigrationCheck,
) -> Result<CheckResult> {
    let label = check.dir.display().to_string();

    let backup: Option<(String, String)> = sqlx::query_as(
        "SELECT local_path, fetched_at FROM backups
         WHERE app = ? AND name = ? ORDER BY id DESC LIMIT 1",
    )
    .bind(&ctx.cfg.id)
    .bind(&check.backup)
    .fetch_optional(&ctx.pool)
    .await?;
    let Some((backup_path, fetched_at)) = backup else {
        log.line(&format!(
            "no {} backup fetched; call /backup/fetch first\n",
            check.backup
        ))
        .await;
        return Ok(CheckResult::Stopped(GateOutcome::blocked(
            GateBlocker::NoBackupAvailable {
                check: label,
                backup: check.backup.clone(),
            },
        )));
    };

    // Presence is not freshness. A fetch that quietly stopped working leaves this
    // row in place, and restoring it dry-runs the migrations against a schema prod
    // has moved past — green, and worthless. Block on age instead. An unparsable
    // timestamp is treated as stale: this row is daemon-written RFC 3339, so a
    // value that will not parse means something is wrong, and failing closed on a
    // freshness check is the whole point.
    let age_hours = chrono::DateTime::parse_from_rfc3339(&fetched_at).map_or(i64::MAX, |t| {
        (Utc::now() - t.with_timezone(&Utc)).num_hours()
    });
    let max_age_hours = ctx.cfg.backup_max_age_hours;
    if age_hours > i64::from(max_age_hours) {
        let msg = format!(
            "backup {backup_path} was fetched {fetched_at} ({age_hours}h ago, max \
             {max_age_hours}h); re-run /backup/fetch\n"
        );
        log.line(&msg).await;
        return Ok(CheckResult::Stopped(GateOutcome::blocked(
            GateBlocker::BackupStale {
                age_hours,
                max_age_hours,
                check: label,
            },
        )));
    }

    // A check with its own `scratch_db` gets that database created here rather
    // than by a host bootstrap step: adding a `[[migration_check]]` should not
    // silently depend on someone having remembered to `createdb` on the Sando
    // host, which is exactly the class of footgun this gate exists to remove.
    // DROP + CREATE also makes the database sando-owned, so the PG15+ public
    // schema grants `reset_scratch` applies next are the owner's to give.
    let db_url = match check.scratch_db.as_deref() {
        None => scratch_url.to_string(),
        Some(dbname) => {
            let maintenance_url = pg_url_with_dbname(scratch_url, "postgres");
            log.line(&format!("---- create scratch db {dbname} ----\n"))
                .await;
            if let Err(e) = pg_create_db(&maintenance_url, dbname).await {
                let msg = format!("{label}: creating scratch db {dbname}: {e}");
                log.line(&msg).await;
                return Ok(CheckResult::Stopped(GateOutcome::failed(
                    GateFailure::RestoreFailed { reason: msg },
                )));
            }
            pg_url_with_dbname(scratch_url, dbname)
        }
    };

    let owner_role = check
        .owner_role
        .as_deref()
        .unwrap_or(&ctx.cfg.scratch_owner_role);
    log.line("---- reset_scratch ----\n").await;
    if let Err(e) = reset_scratch(&db_url, owner_role).await {
        let msg = format!("{label}: scratch reset: {e}");
        log.line(&msg).await;
        return Ok(CheckResult::Stopped(GateOutcome::failed(
            GateFailure::RestoreFailed { reason: msg },
        )));
    }
    log.line(&format!("---- restore_dump ({backup_path}) ----\n"))
        .await;
    if let Err(e) = restore_dump(&db_url, &backup_path, log).await {
        let msg = format!("{label}: restore: {e}");
        log.line(&msg).await;
        return Ok(CheckResult::Stopped(GateOutcome::failed(
            GateFailure::RestoreFailed { reason: msg },
        )));
    }

    let Some(migrations_dir) = ctx.migrations_dir(&check.dir) else {
        // Neither the bundle nor a checkout holds them. For an accepted
        // artifact that means the builder did not ship its migrations, and a
        // dry run over nothing would report green having proved nothing.
        let msg = format!(
            "{label}: no migrations at {} in the bundle or a checkout",
            check.dir.display()
        );
        log.line(&msg).await;
        return Ok(CheckResult::Stopped(GateOutcome::failed(
            GateFailure::RestoreFailed { reason: msg },
        )));
    };
    log.line("---- run_migrator ----\n").await;
    match run_migrator(&db_url, &migrations_dir).await {
        Ok(()) => {
            log.line(&format!("{label}: restored {backup_path} + migrated\n"))
                .await;
            Ok(CheckResult::Passed { backup_path })
        }
        Err(e) => {
            let err_s = format!("{label}: {e}");
            log.line(&err_s).await;
            Ok(CheckResult::Stopped(GateOutcome::failed(
                classify::classify_migration_error(&err_s, None),
            )))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::gates::testkit::{
        dry_run_ctx, mt_check, seed_backup, seed_named_backup, with_check,
    };

    #[tokio::test]
    async fn migration_dry_run_blocks_when_a_checks_own_dump_was_never_fetched() {
        // The hazard the check list exists for: multithreaded applies its own
        // migrations at boot against its own database, so the server's dump must
        // never stand in for it. A fetched `server` row with no `multithreaded`
        // row is exactly that substitution, and it has to block.
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = dry_run_ctx(tmp.path(), 48).await;
        with_check(&mut ctx, mt_check());
        seed_named_backup(&ctx, "server", 1).await;
        let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;

        let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
        log.close().await;

        let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
            panic!("a missing multithreaded dump must block");
        };
        let GateBlocker::NoBackupAvailable { check, backup } = blocker else {
            panic!("expected NoBackupAvailable, got {blocker:?}");
        };
        assert_eq!(backup, "multithreaded", "names the dump that is missing");
        assert!(
            check.contains("multithreaded/migrations"),
            "names the check that wanted it, got {check}"
        );
    }

    #[tokio::test]
    async fn migration_dry_run_freshness_is_per_dump() {
        // A fresh server dump must not make a 45-day-old multithreaded dump look
        // current: the clock is per-database, or the second check inherits the
        // first's freshness and the gate is theatre.
        let tmp = tempfile::tempdir().unwrap();
        let mut ctx = dry_run_ctx(tmp.path(), 48).await;
        with_check(&mut ctx, mt_check());
        seed_named_backup(&ctx, "server", 1).await;
        seed_named_backup(&ctx, "multithreaded", 24 * 45).await;
        let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;

        let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
        log.close().await;

        let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
            panic!("a 45-day-old multithreaded dump must block");
        };
        let GateBlocker::BackupStale { check, .. } = blocker else {
            panic!("expected BackupStale, got {blocker:?}");
        };
        assert!(
            check.contains("multithreaded/migrations"),
            "names the check whose dump is stale, got {check}"
        );
    }

    #[tokio::test]
    async fn migration_dry_run_blocks_on_a_stale_backup() {
        // The failure this closes: the gate used to check only that a backups row
        // existed, so a fetch that silently stopped working left it green against
        // an ever-older schema. Sando ran 45 days that way.
        let tmp = tempfile::tempdir().unwrap();
        let ctx = dry_run_ctx(tmp.path(), 48).await;
        seed_backup(&ctx, 24 * 45).await;
        let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;

        let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
        log.close().await;

        let crate::outcome::GateStatus::Blocked { blocker } = outcome.status else {
            panic!("a 45-day-old backup must block");
        };
        let GateBlocker::BackupStale {
            age_hours,
            max_age_hours,
            ..
        } = blocker
        else {
            panic!("expected BackupStale, got {blocker:?}");
        };
        assert_eq!(max_age_hours, 48);
        assert!(
            age_hours >= 24 * 45,
            "reports the real age, got {age_hours}"
        );
    }

    #[tokio::test]
    async fn migration_dry_run_accepts_a_fresh_backup() {
        // The other side of the boundary: a backup inside the window must not be
        // blocked on freshness. It fails later (there is no such dump on disk),
        // which is exactly the proof the age check let it through.
        let tmp = tempfile::tempdir().unwrap();
        let ctx = dry_run_ctx(tmp.path(), 48).await;
        seed_backup(&ctx, 6).await;
        let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;

        let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
        log.close().await;

        assert!(
            !matches!(
                outcome.status,
                crate::outcome::GateStatus::Blocked {
                    blocker: GateBlocker::BackupStale { .. }
                }
            ),
            "a 6h-old backup is fresh, got {:?}",
            outcome.status,
        );
    }

    #[tokio::test]
    async fn migration_dry_run_treats_an_unparsable_fetched_at_as_stale() {
        // Fail closed: `fetched_at` is daemon-written RFC 3339, so a value that
        // will not parse means the row is untrustworthy — and a freshness check
        // that shrugs at a timestamp it cannot read is not a freshness check.
        let tmp = tempfile::tempdir().unwrap();
        let ctx = dry_run_ctx(tmp.path(), 48).await;
        sqlx::query(
            "INSERT INTO backups (fetched_at, source, local_path, byte_size)
             VALUES ('not-a-timestamp', 'file:///x.sql.gz', '/tmp/x.sql.gz', 1000000)",
        )
        .execute(&ctx.pool)
        .await
        .unwrap();
        let log = GateLog::open(&ctx, GateRunId(0), GateKind::MigrationDryRun).await;

        let outcome = migration_dry_run_inner(&ctx, &log).await.unwrap();
        log.close().await;

        assert!(
            matches!(
                outcome.status,
                crate::outcome::GateStatus::Blocked {
                    blocker: GateBlocker::BackupStale { .. }
                }
            ),
            "an unreadable fetched_at must block, got {:?}",
            outcome.status,
        );
    }
}