//! Gate execution. Each gate kind has a runner that produces a pass/fail
//! outcome plus an optional detail string (typically a stderr tail or a
//! human-readable reason). Outcomes are persisted to `gate_runs` so /state
//! and the TUI can show them.
//!
//! This module holds the context every runner reads ([`GateCtx`]), the
//! dispatcher ([`run`]) and the two gates that are pure table reads (burn-in
//! and manual confirmation). The runners themselves sit in siblings, one per
//! tool family: `cargo`, `pg`, `migration`, `code_smoke`, `probes`, with `log`
//! under all of them.
//!
//! Nothing here shares mutable state. `GateCtx` is a plain struct every runner
//! reads and none writes, which is what lets the families separate.
//!
//! # Design
//!
//! The tier/gate/deploy architecture, the typed-observability redesign, and the
//! deploy.sh-parity and account-permission notes live in the maintainer wiki.
//! <!-- wiki: sando-overview -->

use self::cargo::{cargo_test, clippy, fmt_check, hardening_test, supply_chain};
use self::code_smoke::code_smoke;
use self::migration::migration_dry_run;
use self::probes::{boot_smoke, node_health, page_smoke};
use crate::config::AppConfig;
use crate::domain::{AppId, GateKind, GateRunId, TierId, Version};
use crate::events::{self, Event, EventTx};
use crate::outcome::{GateBlocker, GateFailure, GateOutcome, LogRef, PassNote};
use crate::topology::Gate;
use anyhow::Result;
use chrono::Utc;
use sqlx::SqlitePool;
use std::collections::HashMap;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;

mod cargo;
mod code_smoke;
mod log;
mod migration;
mod pg;
mod probes;
#[cfg(test)]
mod testkit;

pub use pg::preflight_scratch_privileges;
pub(crate) use pg::{reset_scratch, run_migrator};

pub struct GateCtx {
    pub pool: SqlitePool,
    pub cfg: Arc<AppConfig>,
    pub tier: TierId,
    pub version: Version,
    /// The checkout this run's artifact was built from, when there is one.
    ///
    /// `None` for an accepted artifact: it was built elsewhere and Sando has no
    /// source tree for it. That is the boundary made visible (wiki
    /// [[sando-bento-boundary]]) — artifact-scoped gates belong to the builder,
    /// so a gate that reads source is one Sando should refuse to run here rather
    /// than resolve against a path that does not exist.
    pub worktree: Option<PathBuf>,
    /// The published, content-addressed bundle this run is about, when it has
    /// been published yet. `migration_dry_run` prefers it over the worktree, so
    /// what it proves is inside the digest rather than beside it.
    pub bundle: Option<PathBuf>,
    pub events: EventTx,
    /// Nodes the `node_health` post-deploy gate probes. Empty for build-time
    /// gate runs on the host (where `node_health` never appears); filled at
    /// promote time with each freshly-deployed node and its executor.
    pub nodes: Vec<NodeProbe>,
    /// The `build_runs.id` this gate run vouches for — the artifact identity
    /// (wiki [[release-artifact-identity]]). Recorded on every `gate_runs` row so
    /// promote can resolve the artifact through the evidence for a specific build,
    /// not through a version string that a later rebuild can silently reuse.
    /// `None` for legacy/pre-identity runs and gate unit tests.
    pub build_id: Option<i64>,
    /// Where each `[[aux_repo]]` is checked out, by topology name. Aux repos sit
    /// beside the per-sha worktree rather than under it, so a `test_target` that
    /// names one cannot be resolved against `worktree` alone. Filled from the
    /// topology at build time; empty at promote time, where the only gate that
    /// runs is `node_health` and there is no checkout at all.
    pub aux_dirs: HashMap<String, PathBuf>,
    /// The tier's public URL, for [`Gate::PageSmoke`]. `None` on every
    /// build-time run and on any tier that declares none.
    ///
    /// [`Gate::PageSmoke`]: crate::topology::Gate::PageSmoke
    pub public_url: Option<String>,
}

impl GateCtx {
    /// The `logs_root` sub-directory this run's gate logs land in.
    ///
    /// The build id, so two runs of one version keep two sets of logs. Keying on
    /// the version would have a rebuild append to the previous attempt's file,
    /// with every run pointing at the same mixed log.
    ///
    /// Falls back to the version when there is no build identity (a
    /// pre-migration-008 run, or a gate unit test), which also keeps logs written
    /// under that scheme reachable: their rows record the version path, and
    /// nothing rewrites them.
    pub fn log_scope(&self) -> String {
        self.build_id
            .map_or_else(|| self.version.to_string(), |id| id.to_string())
    }

    /// This run's log pointer for `gate`. Always paired with
    /// [`Self::log_path`], which resolves the same ref to an absolute path.
    pub fn log_ref(&self, gate: GateKind) -> LogRef {
        LogRef::new(&self.log_scope(), gate)
    }

    /// Where `gate`'s log is written on this host: `logs_root` joined to
    /// [`Self::log_ref`]. The two are derived from one scope so a row's
    /// `log_ref` can never name a file the gate did not write.
    pub fn log_path(&self, gate: GateKind) -> PathBuf {
        self.cfg
            .logs_root
            .join(self.log_scope())
            .join(format!("{}.log", gate.as_str()))
    }

    /// Absolute directory a `test_target` runs in: under the worktree, or under
    /// the named aux repo's checkout.
    ///
    /// An `aux_repo` naming nothing this run knows about resolves to `None`
    /// rather than to a wrong path. Callers treat that as "not present in this
    /// run" and skip, the same as a target missing from an older sha —
    /// `--check-config` is what stops a genuine typo from reaching here
    /// (`Topology::ensure_test_target_aux_repos_exist`).
    pub fn target_dir(&self, target: &crate::config::TestTarget) -> Option<PathBuf> {
        match target.aux_repo.as_deref() {
            None => Some(self.worktree.as_ref()?.join(&target.dir)),
            Some(name) => Some(self.aux_dirs.get(name)?.join(&target.dir)),
        }
    }

    /// The checkout, or a typed refusal for a gate that cannot work without one.
    ///
    /// Every caller of this is a gate whose evidence is about the *artifact*
    /// rather than about the artifact in an environment, which the boundary
    /// assigns to the builder. Reaching this arm means a tier asked Sando to
    /// re-run a builder's gate against a bundle it was handed, and the honest
    /// answer is to say so rather than to pass on having run nothing.
    pub fn worktree_for(&self, gate: GateKind) -> std::result::Result<&Path, GateOutcome> {
        self.worktree.as_deref().ok_or_else(|| {
            GateOutcome::failed(GateFailure::NeedsSource {
                gate,
                artifact: self.bundle.as_ref().map_or_else(
                    || "an artifact built elsewhere".into(),
                    |b| b.display().to_string(),
                ),
            })
        })
    }

    /// Where a `migration_check` finds its migrations.
    ///
    /// The bundle wins when it carries them. That is the point of shipping
    /// migrations as a `release_contents` entry: it puts them inside the digest,
    /// so the dry run proves something about the bytes that ship rather than
    /// about a checkout that happens to sit next to them. The worktree is the
    /// fallback for a build whose config has not opted in yet, and for an
    /// accepted artifact there is no fallback at all — if the builder did not
    /// bundle its migrations, Sando cannot dry-run them and says so.
    pub fn migrations_dir(&self, dir: &Path) -> Option<PathBuf> {
        if let Some(bundle) = &self.bundle {
            let in_bundle = bundle.join(dir);
            if in_bundle.is_dir() {
                return Some(in_bundle);
            }
        }
        let in_worktree = self.worktree.as_ref()?.join(dir);
        in_worktree.is_dir().then_some(in_worktree)
    }
}

/// One node the `node_health` gate verifies: its id, the systemd unit to
/// confirm active after the restart, an optional HTTP readiness URL, and the
/// executor that reaches it (the same transport the deploy used).
pub struct NodeProbe {
    pub node: crate::domain::NodeId,
    pub service: String,
    pub health_url: Option<String>,
    pub executor: Arc<dyn ops_exec::Executor>,
}

/// Run a single gate end-to-end: insert the in-flight row, execute the gate,
/// update the row with the outcome. Returns the outcome for the caller.
pub async fn run(ctx: &GateCtx, gate: &Gate) -> Result<GateOutcome> {
    let kind = gate.kind();
    let started_at = Utc::now().to_rfc3339();

    let id: i64 = sqlx::query_scalar(
        "INSERT INTO gate_runs (app, version, tier, gate_kind, started_at, build_id)
         VALUES (?, ?, ?, ?, ?, ?)
         RETURNING id",
    )
    .bind(&ctx.cfg.id)
    .bind(&ctx.version)
    .bind(&ctx.tier)
    .bind(kind)
    .bind(&started_at)
    .bind(ctx.build_id)
    .fetch_one(&ctx.pool)
    .await?;
    let run_id = GateRunId(id);

    tracing::info!(
        run_id = %run_id, tier = %ctx.tier, version = %ctx.version, gate = %kind,
        "gate start",
    );
    events::emit(
        &ctx.events,
        Event::GateStart {
            run_id,
            tier: ctx.tier.clone(),
            version: ctx.version.clone(),
            gate: kind,
        },
    );

    let outcome = match gate {
        // cargo_test bounds its own run internally (it kills the specific child).
        Gate::CargoTest => cargo_test(ctx, run_id).await,
        // hardening_test bounds its own run internally, same as cargo_test.
        Gate::HardeningTest => hardening_test(ctx, run_id).await,
        // Each bounds itself the same way cargo_test does: one deadline across
        // every target, so N crates cannot multiply the ceiling by N.
        Gate::Clippy => clippy(ctx, run_id).await,
        Gate::Fmt => fmt_check(ctx, run_id).await,
        Gate::CargoAudit => supply_chain(ctx, run_id, GateKind::CargoAudit).await,
        Gate::CargoDeny => supply_chain(ctx, run_id, GateKind::CargoDeny).await,
        // migration_dry_run's psql restore + sqlx migrate could wedge; bound the
        // whole gate here. Its bash restore sets kill_on_drop, so a timeout-drop
        // doesn't orphan it.
        Gate::MigrationDryRun => {
            let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
            match tokio::time::timeout(ceiling, migration_dry_run(ctx, run_id)).await {
                Ok(res) => res,
                Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout {
                    gate: GateKind::MigrationDryRun,
                    after_s: ctx.cfg.gate_timeout_secs as u32,
                })
                .with_log_ref(ctx.log_ref(GateKind::MigrationDryRun))),
            }
        }
        // code_smoke boots the real binary (migrate-from-scratch + seed + serve),
        // any step of which could wedge; bound the whole gate here. Both child
        // processes set kill_on_drop, so a timeout-drop can't orphan them. A
        // timeout leaves the throwaway DB behind; the next run's createdb drops
        // it first (DROP IF EXISTS), same as migration_dry_run's scratch reset.
        Gate::CodeSmoke => {
            let ceiling = std::time::Duration::from_secs(ctx.cfg.gate_timeout_secs);
            match tokio::time::timeout(ceiling, code_smoke(ctx, run_id)).await {
                Ok(res) => res,
                Err(_elapsed) => Ok(GateOutcome::failed(GateFailure::Timeout {
                    gate: GateKind::CodeSmoke,
                    after_s: ctx.cfg.gate_timeout_secs as u32,
                })
                .with_log_ref(ctx.log_ref(GateKind::CodeSmoke))),
            }
        }
        Gate::BootSmoke => boot_smoke(ctx, run_id).await,
        Gate::NodeHealth => node_health(ctx).await,
        Gate::PageSmoke => page_smoke(ctx).await,
        Gate::BurnIn { hours } => burn_in(ctx, *hours).await,
        Gate::ManualConfirm => manual_confirm(ctx).await,
    };

    let outcome = outcome.unwrap_or_else(|e| {
        GateOutcome::failed(GateFailure::Unclassified {
            legacy_detail: Some(format!("gate runner errored: {e}")),
        })
    });

    let outcome_json = serde_json::to_string(&outcome)
        .unwrap_or_else(|e| format!("{{\"_serialize_error\":{e:?}}}"));
    sqlx::query(
        "UPDATE gate_runs
         SET finished_at = ?, status = ?, outcome_json = ?, log_ref = ?
         WHERE id = ?",
    )
    .bind(Utc::now().to_rfc3339())
    .bind(outcome.status_str())
    .bind(&outcome_json)
    .bind(outcome.log_ref.as_ref().map(super::outcome::LogRef::as_str))
    .bind(id)
    .execute(&ctx.pool)
    .await?;

    tracing::info!(
        tier = %ctx.tier, version = %ctx.version, gate = %kind,
        status = outcome.status_str(), "gate done",
    );
    events::emit(
        &ctx.events,
        Event::GateDone {
            run_id,
            tier: ctx.tier.clone(),
            version: ctx.version.clone(),
            gate: kind,
            outcome: outcome.clone(),
        },
    );

    Ok(outcome)
}

/// Run every gate in order and return the kinds that did not pass (empty means
/// green). We deliberately do NOT short-circuit on first failure — every gate's
/// outcome is recorded in `gate_runs`, which is the operator's only visibility
/// into pipeline health. Hiding later gates because an earlier one failed makes
/// diagnosis worse.
///
/// Returning the failing kinds rather than a bare bool is what lets the promote
/// path name them in the tier's `partial_reason` and in the error it returns to
/// the operator, instead of a generic "something was red".
pub async fn run_all(ctx: &GateCtx, gates: &[Gate]) -> Result<Vec<GateKind>> {
    let mut failed = Vec::new();
    for g in gates {
        let o = run(ctx, g).await?;
        if !o.is_passed() {
            failed.push(g.kind());
        }
    }
    Ok(failed)
}

/// Live check: has `tier`'s burn-in window of `hours` elapsed since its clock
/// (`tier_state.burn_in_started_at`, started by a promote onto the tier)? Used
/// by the promote-time gate check (`unsatisfied_gates`) so a stale `blocked`
/// row never masks an elapsed — or not-yet-elapsed — window. The `burn_in` gate
/// runner below wraps the same state with a richer outcome for `/state`.
pub async fn burn_in_satisfied(
    pool: &SqlitePool,
    app: &AppId,
    tier: &TierId,
    hours: u32,
) -> Result<bool> {
    let started: Option<String> =
        sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?")
            .bind(app)
            .bind(tier)
            .fetch_optional(pool)
            .await?
            .flatten();
    let Some(started) = started else {
        return Ok(false);
    };
    let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
    Ok(Utc::now() - started >= chrono::Duration::hours(hours as i64))
}

async fn burn_in(ctx: &GateCtx, hours: u32) -> Result<GateOutcome> {
    // Check tier_state.burn_in_started_at on this tier; pass if enough time
    // has elapsed. The clock is started by /promote when a version lands on
    // the burn-in tier.
    let started: Option<String> =
        sqlx::query_scalar("SELECT burn_in_started_at FROM tier_state WHERE app = ? AND tier = ?")
            .bind(&ctx.cfg.id)
            .bind(&ctx.tier)
            .fetch_optional(&ctx.pool)
            .await?
            .flatten();
    let Some(started) = started else {
        return Ok(GateOutcome::blocked(GateBlocker::BurnInClockNotStarted));
    };
    let started = chrono::DateTime::parse_from_rfc3339(&started)?.with_timezone(&Utc);
    let elapsed = Utc::now() - started;
    let needed = chrono::Duration::hours(hours as i64);
    if elapsed >= needed {
        Ok(GateOutcome::passed(PassNote::BurnInElapsed {
            hours: elapsed.num_hours() as u32,
        }))
    } else {
        let remaining = (needed - elapsed).num_hours().max(0) as u32;
        Ok(GateOutcome::blocked(GateBlocker::BurnInRemaining {
            hours_remaining: remaining,
            hours_total: hours,
        }))
    }
}

async fn manual_confirm(ctx: &GateCtx) -> Result<GateOutcome> {
    // Pass iff a row in gate_runs exists with status='passed' for this
    // (tier, version, manual_confirm) that was inserted out-of-band by an
    // operator action. Since the harness inserts the in-flight row itself,
    // look for a prior confirmation row.
    let prior_at: Option<String> = sqlx::query_scalar(
        "SELECT finished_at FROM gate_runs
         WHERE app = ? AND tier = ? AND version = ? AND gate_kind = 'manual_confirm'
           AND status = 'passed'
         ORDER BY id DESC LIMIT 1",
    )
    .bind(&ctx.cfg.id)
    .bind(&ctx.tier)
    .bind(&ctx.version)
    .fetch_optional(&ctx.pool)
    .await?;
    match prior_at {
        Some(at_str) => {
            let at = chrono::DateTime::parse_from_rfc3339(&at_str)
                .map_or_else(|_| Utc::now(), |d| d.with_timezone(&Utc));
            Ok(GateOutcome::passed(PassNote::OperatorConfirmed { at }))
        }
        None => Ok(GateOutcome::blocked(
            GateBlocker::AwaitingOperatorConfirmation,
        )),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::gates::testkit::{aux_target, resolving_ctx, target};
    use sqlx::sqlite::SqlitePoolOptions;

    #[tokio::test]
    async fn a_plain_target_resolves_under_the_worktree() {
        let ctx = resolving_ctx("/w/abc123", &[]);
        assert_eq!(
            ctx.target_dir(&target("shared/tagtree")),
            Some(PathBuf::from("/w/abc123/shared/tagtree")),
        );
    }

    #[tokio::test]
    async fn an_aux_target_resolves_beside_the_worktree_not_under_it() {
        // The whole point: `Libraries/docengine` is a sibling of the per-sha
        // worktree, so a worktree-relative path can never reach it.
        let ctx = resolving_ctx("/w/abc123", &[("docengine", "/w/Libraries/docengine")]);
        assert_eq!(
            ctx.target_dir(&aux_target("", "docengine")),
            Some(PathBuf::from("/w/Libraries/docengine")),
        );
        // A subdirectory of an aux repo resolves under its checkout.
        assert_eq!(
            ctx.target_dir(&aux_target("crates/inner", "docengine")),
            Some(PathBuf::from("/w/Libraries/docengine/crates/inner")),
        );
    }

    #[tokio::test]
    async fn an_aux_target_with_no_checkout_this_run_resolves_to_nothing() {
        // Promote-time gates carry no aux dirs. Resolving to a wrong path (say,
        // the worktree) would run the gate against whatever happened to sit
        // there; `None` makes the caller skip, and --check-config is what
        // catches a real typo.
        let ctx = resolving_ctx("/w/abc123", &[]);
        assert_eq!(ctx.target_dir(&aux_target("", "docengine")), None);
    }

    #[test]
    fn labels_name_the_repo_an_aux_target_lives_in() {
        assert_eq!(target("server").label(), "server");
        assert_eq!(aux_target("", "docengine").label(), "docengine (aux)");
        assert_eq!(
            aux_target("crates/inner", "docengine").label(),
            "docengine/crates/inner (aux)",
        );
    }

    #[test]
    fn every_gate_kind_round_trips_through_its_wire_string() {
        // gate_kind is a TEXT column and a WS event field; as_str and FromStr
        // disagreeing would make a gate's evidence unreadable by
        // unsatisfied_gates, which fails the promote closed with no explanation.
        for k in [
            GateKind::CargoTest,
            GateKind::HardeningTest,
            GateKind::Clippy,
            GateKind::Fmt,
            GateKind::CargoAudit,
            GateKind::CargoDeny,
            GateKind::MigrationDryRun,
            GateKind::CodeSmoke,
            GateKind::BootSmoke,
            GateKind::NodeHealth,
            GateKind::BurnIn,
            GateKind::ManualConfirm,
        ] {
            assert_eq!(
                k.as_str().parse::<GateKind>().unwrap(),
                k,
                "round trip for {k:?}"
            );
        }
    }

    /// burn_in returns a typed Blocked when the clock isn't started; the
    /// runner persists status='blocked' + outcome_json (the json carries
    /// blocker.kind = 'burn_in_clock_not_started').
    #[tokio::test]
    async fn burn_in_blocked_persists_typed_outcome() {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        crate::db::migrate(&pool).await.unwrap();
        // Topology sync expects a tier row before gate_runs can reference it.
        sqlx::query("INSERT INTO tiers (name, ord, provisioned, canary) VALUES ('host', 0, 0, 'sequential')")
            .execute(&pool).await.unwrap();
        sqlx::query("INSERT INTO tier_state (tier) VALUES ('host')")
            .execute(&pool)
            .await
            .unwrap();
        // versions FK target.
        sqlx::query("INSERT INTO versions (version, git_sha, built_at, artifact_path) VALUES ('0.1.0', 'abc1234', '2026-01-01T00:00:00Z', '/tmp/x')")
            .execute(&pool).await.unwrap();

        let cfg = std::sync::Arc::new(crate::config::AppConfig::for_tests());
        let ctx = GateCtx {
            public_url: None,
            pool: pool.clone(),
            cfg,
            tier: TierId::new("host"),
            version: "0.1.0".parse().unwrap(),
            worktree: Some(std::path::PathBuf::from("/tmp/unused")),
            bundle: None,
            events: events::channel(),
            nodes: Vec::new(),
            build_id: None,
            aux_dirs: HashMap::new(),
        };
        let out = run(&ctx, &Gate::BurnIn { hours: 24 }).await.unwrap();
        assert_eq!(out.status_str(), "blocked");
        assert!(!out.is_passed());

        // Read the persisted row.
        let row: (Option<String>, Option<String>) =
            sqlx::query_as("SELECT status, outcome_json FROM gate_runs ORDER BY id DESC LIMIT 1")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(row.0.as_deref(), Some("blocked"), "typed status");
        let json: serde_json::Value = serde_json::from_str(row.1.as_deref().unwrap()).unwrap();
        assert_eq!(json["status"]["kind"], "blocked");
        assert_eq!(
            json["status"]["blocker"]["kind"],
            "burn_in_clock_not_started"
        );
    }
}