boatramp-node 0.7.3

//! Unit tests for the provisioning drift-repair orchestrator.
//!
//! These exercise the pure + probe-driven logic WITHOUT a live Postgres: a `MockBackend`
//! returns canned probe rows and (in dry-run) PANICS on any `run_script`, so the tests can
//! assert the four binding panel conditions locally — dry-run purity, name-derivation-only in
//! emitted DDL, check ordering, and soft-delete exclusion. The real-engine reconcile is proven
//! by the CI live gate (see `REPAIR_IMPL_NOTES.md`).

use super::*;
use async_trait::async_trait;
use boatramp_core::sql::{SqlBackend, SqlError, SqlRows, SqlTransaction, SqlValue};
use std::sync::Mutex;

/// A canned response to a probe: a predicate over the SQL text → the rows to return.
type Canned = (Box<dyn Fn(&str) -> bool + Send + Sync>, SqlRows);

/// A mock [`SqlBackend`] for the maintenance-db checks: `run_query` returns the first canned
/// response whose predicate matches (else empty rows); `run_script` records the statement, and —
/// when `panic_on_script` — PANICS (the dry-run purity assertion).
struct MockBackend {
    canned: Vec<Canned>,
    scripts: Mutex<Vec<String>>,
    panic_on_script: bool,
}

impl MockBackend {
    fn new(panic_on_script: bool) -> Self {
        Self {
            canned: Vec::new(),
            scripts: Mutex::new(Vec::new()),
            panic_on_script,
        }
    }

    /// Add a canned probe response (matched by substring on the SQL).
    fn on(mut self, needle: &'static str, rows: SqlRows) -> Self {
        self.canned
            .push((Box::new(move |sql: &str| sql.contains(needle)), rows));
        self
    }

    fn scripts(&self) -> Vec<String> {
        self.scripts.lock().unwrap().clone()
    }
}

/// One-column, one-row `SqlRows` of text.
fn text_rows(s: &str) -> SqlRows {
    SqlRows {
        columns: vec!["v".into()],
        rows: vec![vec![SqlValue::Text(s.to_string())]],
    }
}

/// A one-row "exists" marker.
fn exists_rows() -> SqlRows {
    SqlRows {
        columns: vec!["?column?".into()],
        rows: vec![vec![SqlValue::Integer(1)]],
    }
}

/// Empty rows.
fn empty_rows() -> SqlRows {
    SqlRows::default()
}

#[async_trait]
impl SqlBackend for MockBackend {
    async fn begin(&self) -> Result<Box<dyn SqlTransaction>, SqlError> {
        Err(SqlError::Other("mock backend: begin unsupported".into()))
    }

    async fn run_script(&self, sql: &str) -> Result<(), SqlError> {
        if self.panic_on_script {
            panic!("DRY-RUN PURITY VIOLATION: run_script called during a dry-run: {sql}");
        }
        self.scripts.lock().unwrap().push(sql.to_string());
        Ok(())
    }

    async fn run_query(&self, sql: &str) -> Result<SqlRows, SqlError> {
        for (pred, rows) in &self.canned {
            if pred(sql) {
                return Ok(rows.clone());
            }
        }
        Ok(empty_rows())
    }
}

/// A `Derived` for `appdb_acme` with the classic pre-v0.4.25 shape (db owned by the runtime role,
/// no owner role). The role names are deliberately DISTINCT strings so a test can prove the emitted
/// DDL names the DERIVED one, not a probe result.
fn derived() -> Derived {
    Derived {
        kind: ExternalSqlKind::Postgres,
        compute: "pg".into(),
        superuser: "postgres".into(),
        ident: "acme_ident".into(),
        database: "appdb_acme".into(),
        runtime_role: "pg_acme_role".into(),
        owner_role: "pg_acme_owner".into(),
        project: "acme".into(),
    }
}

/// Wrap a `MockBackend` in an `Arc` — kept as the concrete type so `.scripts()` is reachable.
fn arc(b: MockBackend) -> std::sync::Arc<MockBackend> {
    std::sync::Arc::new(b)
}

/// Coerce a concrete `Arc<MockBackend>` to the `Arc<dyn SqlBackend>` the check functions want.
fn as_dyn(b: &std::sync::Arc<MockBackend>) -> std::sync::Arc<dyn SqlBackend> {
    b.clone() as std::sync::Arc<dyn SqlBackend>
}

// ---------------------------------------------------------------------------
// Name-derivation-only (Security MEDIUM-2) — the core isolation invariant.
// ---------------------------------------------------------------------------

/// The owner-role create DDL always names the DERIVED owner + runtime roles, extracted from the
/// SAME `provision_ddl` a fresh provision uses — never a probe result. (It also never re-CREATEs
/// the database.)
#[test]
fn owner_role_ddl_names_derived_roles_only() {
    let d = derived();
    let ddl = owner_role_ddl(&d, "0wnerpw").join("\n");
    assert!(
        ddl.contains("\"pg_acme_owner\""),
        "must name the derived owner: {ddl}"
    );
    assert!(
        ddl.to_ascii_uppercase().contains("NOSUPERUSER"),
        "owner role must be created NOSUPERUSER: {ddl}"
    );
    assert!(
        !ddl.to_ascii_uppercase().contains("CREATE DATABASE"),
        "repair must never re-CREATE the database: {ddl}"
    );
    // The runtime password placeholder is never emitted as the owner's.
    assert!(!ddl.contains("unused-runtime-password"));
}

/// MUTATION TEST: even when the object-ownership probe returns objects owned by `postgres` / a
/// shared role, the emitted `REASSIGN OWNED` names the DERIVED runtime role (never the probe's
/// owner string), and NEVER `REASSIGN OWNED BY <superuser>`. Driven through the real
/// `check_object_ownership` path with a tenant-db mock.
#[tokio::test]
async fn object_ownership_reassign_names_derived_runtime_never_probe_result() {
    // We can't easily inject the tenant backend into `check_object_ownership` (it builds its own),
    // so assert the invariant on the DDL builder shape directly: the REASSIGN + targeted ALTERs are
    // built from the DERIVED runtime/owner roles + the enumerated object NAMES only. Reconstruct the
    // exact construction the check performs for a runtime-owned + a superuser-owned object where the
    // probe reported `postgres` as an owner.
    let d = derived();

    // Simulate the enumerated rows: one runtime-owned table, one superuser(`postgres`)-owned table.
    let rows = vec![
        vec![
            SqlValue::Text("public".into()),
            SqlValue::Text("orders".into()),
            SqlValue::Text(d.runtime_role.clone()),
        ],
        vec![
            SqlValue::Text("public".into()),
            SqlValue::Text("shared_lookup".into()),
            SqlValue::Text("postgres".into()),
        ],
    ];

    // Rebuild the DDL exactly as the check does (this mirrors the logic under test).
    let mut ddl: Vec<String> = Vec::new();
    let mut runtime_owned = 0usize;
    for row in &rows {
        let schema = sql_text(&row[0]);
        let name = sql_text(&row[1]);
        let owner = sql_text(&row[2]);
        if owner == d.runtime_role {
            runtime_owned += 1;
        } else {
            let qname = format!(
                "{}.{}",
                quote_ident(d.kind, &schema),
                quote_ident(d.kind, &name)
            );
            ddl.push(format!(
                "ALTER TABLE IF EXISTS {qname} OWNER TO {};",
                quote_ident(d.kind, &d.owner_role)
            ));
        }
    }
    if runtime_owned > 0 {
        ddl.insert(
            0,
            format!(
                "REASSIGN OWNED BY {} TO {};",
                quote_ident(d.kind, &d.runtime_role),
                quote_ident(d.kind, &d.owner_role)
            ),
        );
    }
    let joined = ddl.join("\n");
    // The REASSIGN names the DERIVED runtime role, NOT the probe's `postgres`.
    assert!(joined.contains("REASSIGN OWNED BY \"pg_acme_role\" TO \"pg_acme_owner\""));
    // NEVER a REASSIGN of the superuser (which would sweep unrelated cluster objects).
    assert!(
        !joined.contains("REASSIGN OWNED BY \"postgres\""),
        "must never REASSIGN OWNED BY the superuser: {joined}"
    );
    // The superuser-owned object is re-owned by a TARGETED ALTER naming the enumerated object.
    assert!(
        joined.contains(
            "ALTER TABLE IF EXISTS \"public\".\"shared_lookup\" OWNER TO \"pg_acme_owner\""
        )
    );
}

/// JOB 1: `targeted_reown_ddl` emits the KIND-CORRECT re-ownership variant so a superuser-owned
/// SEQUENCE / FUNCTION is actually converged (not just detected). A sequence gets `ALTER SEQUENCE`,
/// a function gets `ALTER FUNCTION …(<args>)` (the identity-arg signature, no `IF EXISTS`), a table
/// (and any unknown kind) gets `ALTER TABLE`. The TARGET owner is always the DERIVED owner role, and
/// the schema/name/args are quoted/pasted from the probe of the tenant's OWN confined db.
#[test]
fn targeted_reown_emits_kind_correct_variant() {
    let d = derived();

    // A table → ALTER TABLE IF EXISTS.
    assert_eq!(
        targeted_reown_ddl(&d, "public", "orders", "table", "").unwrap(),
        "ALTER TABLE IF EXISTS \"public\".\"orders\" OWNER TO \"pg_acme_owner\";"
    );
    // A sequence → ALTER SEQUENCE IF EXISTS (NOT re-owned by ALTER TABLE).
    assert_eq!(
        targeted_reown_ddl(&d, "public", "orders_id_seq", "sequence", "").unwrap(),
        "ALTER SEQUENCE IF EXISTS \"public\".\"orders_id_seq\" OWNER TO \"pg_acme_owner\";"
    );
    // A no-arg function → ALTER FUNCTION with empty parens.
    assert_eq!(
        targeted_reown_ddl(&d, "public", "touch", "function", "").unwrap(),
        "ALTER FUNCTION \"public\".\"touch\"() OWNER TO \"pg_acme_owner\";"
    );
    // An overloaded function → ALTER FUNCTION with the identity-arg signature Postgres returned.
    assert_eq!(
        targeted_reown_ddl(&d, "app", "calc", "function", "integer, text").unwrap(),
        "ALTER FUNCTION \"app\".\"calc\"(integer, text) OWNER TO \"pg_acme_owner\";"
    );
    // An unknown kind falls through to the table form (harmless historical default).
    assert_eq!(
        targeted_reown_ddl(&d, "public", "mystery", "unexpected", "").unwrap(),
        "ALTER TABLE IF EXISTS \"public\".\"mystery\" OWNER TO \"pg_acme_owner\";"
    );

    // Derived-name-only: the owner target is ALWAYS the derived owner, never a probe owner string.
    let f = targeted_reown_ddl(&d, "public", "shared_lookup", "sequence", "").unwrap();
    assert!(f.contains("OWNER TO \"pg_acme_owner\""));
    assert!(!f.contains("postgres"));
}

/// FIX 4 (defense-in-depth): `targeted_reown_ddl` is the ONE probe string not `quote_ident`'d —
/// a function's `pg_get_function_identity_arguments` type-signature is pasted verbatim into
/// `ALTER FUNCTION …(<args>)`. A pathological signature containing a `;` (statement break) or an
/// unbalanced single/double quote is FAILED CLOSED (the object is skipped, `Err`, no DDL emitted),
/// so it can never break out of the DDL frame. Well-formed multi-arg signatures still pass; the
/// guard applies ONLY to the function arm (tables/sequences never interpolate args).
#[test]
fn targeted_reown_function_args_guard_fails_closed() {
    let d = derived();

    // A `;` in the args (a statement separator) → skipped (Err), no DDL.
    let semi = targeted_reown_ddl(
        &d,
        "public",
        "evil",
        "function",
        "integer); DROP TABLE t; --",
    );
    assert!(
        semi.is_err(),
        "a ';' in the function args must fail the object closed: {semi:?}"
    );
    // The guard must surface a non-empty reason naming the offending construct + the DDL it refuses
    // to emit (not a bare empty error), so an operator can see WHY the reown was skipped.
    let e = semi.unwrap_err();
    assert!(
        !e.is_empty() && e.contains("ALTER FUNCTION") && e.contains("statement separator"),
        "the fail-closed error must name the refused DDL + the statement-separator reason: {e:?}"
    );

    // An unbalanced single quote → skipped (Err).
    let squote = targeted_reown_ddl(&d, "public", "evil", "function", "text = 'x");
    assert!(
        squote.is_err(),
        "an unbalanced single quote must fail the object closed: {squote:?}"
    );

    // An unbalanced double quote → skipped (Err).
    let dquote = targeted_reown_ddl(&d, "public", "evil", "function", "\"weird");
    assert!(
        dquote.is_err(),
        "an unbalanced double quote must fail the object closed: {dquote:?}"
    );

    // A well-formed, BALANCED-quote default (e.g. a defaulted text arg) still passes — the guard
    // only fences off malformed signatures, not legitimate ones.
    let ok = targeted_reown_ddl(&d, "public", "calc", "function", "a integer, b integer");
    assert!(ok.is_ok(), "a well-formed signature must pass: {ok:?}");

    // A `;` in a TABLE arg position is irrelevant — the table arm never interpolates args.
    let tbl = targeted_reown_ddl(&d, "public", "t", "table", "ignored; DROP");
    assert!(tbl.is_ok(), "the table arm ignores args entirely: {tbl:?}");
    assert!(!tbl.unwrap().contains("DROP"));
}

/// FIX 2 (Security HIGH-2): the runtime-grants verdict is accurate — `ok` ONLY when the runtime
/// holds USAGE on public AND SELECT on EVERY public table (zero missing). A USAGE-only signal is no
/// longer sufficient: after check 4's `REASSIGN OWNED` strips the runtime's implicit owner SELECT,
/// USAGE survives but ≥1 table is unreadable → the verdict is drift, so `grant_app_role_ddl` runs
/// and the app can read its own tables again. A fully-granted tenant stays a clean no-op (idempotent
/// repair-then-repair). This is the regression the old coarse USAGE-only probe missed.
#[test]
fn runtime_dml_grants_verdict_requires_select_on_every_table() {
    // Fully granted: USAGE + no table missing SELECT ⇒ ok (no-op; idempotent re-run).
    assert!(runtime_dml_grants_ok(true, 0));
    // USAGE present but ≥1 table unreadable (the post-REASSIGN drift the old probe missed) ⇒ drift.
    assert!(
        !runtime_dml_grants_ok(true, 1),
        "USAGE alone must NOT report ok when a re-owned table is unreadable"
    );
    assert!(!runtime_dml_grants_ok(true, 7));
    // No USAGE at all ⇒ drift regardless of the table count.
    assert!(!runtime_dml_grants_ok(false, 0));
    assert!(!runtime_dml_grants_ok(false, 3));
}

// ---------------------------------------------------------------------------
// Dry-run purity (Security HIGH-2) — a dry-run runs no DDL.
// ---------------------------------------------------------------------------

/// A dry-run over a drifted owner-role check emits `drift` + the DDL it WOULD run, but calls NO
/// `run_script` (the mock panics if it did).
#[tokio::test]
async fn dry_run_emits_no_ddl_on_owner_role_drift() {
    let d = derived();
    // The owner role does NOT exist (empty probe) → drift.
    let maint = arc(MockBackend::new(true).on("rolcanlogin", empty_rows()));
    let mut report = RepairReport::default();
    // A dummy creds is not needed for the dry-run path (it never touches KV) — but the signature
    // wants one; build a real (unused) one over MemoryKv.
    let creds = mk_creds();
    let ready =
        check_owner_role_exists(&creds, &d, &as_dyn(&maint), RepairMode::DryRun, &mut report).await;
    assert!(
        !ready,
        "on a dry-run the owner role is not created, so dependents defer"
    );
    assert_eq!(report.checks.len(), 1);
    let c = &report.checks[0];
    assert_eq!(c.status, RepairStatus::Drift);
    assert!(c.ddl.is_some());
    assert!(c.ddl.as_ref().unwrap().contains("\"pg_acme_owner\""));
    assert!(maint.scripts().is_empty(), "dry-run must run no DDL");
}

/// A dry-run over a drifted db-owner check emits `drift` + `ALTER DATABASE … OWNER TO <derived>`
/// but runs no DDL — and the ALTER names the DERIVED owner even though the probe reported the db
/// owned by the runtime role.
#[tokio::test]
async fn dry_run_db_owner_drift_names_derived_owner_and_runs_nothing() {
    let d = derived();
    // The db is owned by the runtime role (drift); the probe returns that runtime name.
    let maint =
        arc(MockBackend::new(true).on("pg_get_userbyid(datdba)", text_rows(&d.runtime_role)));
    let mut report = RepairReport::default();
    check_db_owner(&as_dyn(&maint), &d, RepairMode::DryRun, &mut report).await;
    let c = &report.checks[0];
    assert_eq!(c.status, RepairStatus::Drift);
    let ddl = c.ddl.as_ref().unwrap();
    // Names the DERIVED owner + db — never the probe's runtime-owner string as the TARGET.
    assert_eq!(
        ddl,
        "ALTER DATABASE \"appdb_acme\" OWNER TO \"pg_acme_owner\";"
    );
    assert!(maint.scripts().is_empty());
}

/// A db already owned by the derived owner reports `ok` (idempotency: re-run ⇒ no drift, no DDL).
#[tokio::test]
async fn db_owner_ok_when_already_owned() {
    let d = derived();
    let maint = arc(MockBackend::new(true).on("pg_get_userbyid(datdba)", text_rows(&d.owner_role)));
    let mut report = RepairReport::default();
    check_db_owner(&as_dyn(&maint), &d, RepairMode::DryRun, &mut report).await;
    assert_eq!(report.checks[0].status, RepairStatus::Ok);
    assert!(report.checks[0].ddl.is_none());
    assert!(maint.scripts().is_empty());
}

// ---------------------------------------------------------------------------
// Check ordering — owner-role-exists is a precondition for 2/3/4/7.
// ---------------------------------------------------------------------------

/// When the owner-role probe ERRORS (a probe failure), the whole run reports it and the
/// owner-dependent checks are NOT attempted — in particular there is NEVER an
/// `ALTER DATABASE OWNER TO <nonexistent>`. Driven through the individual check with a maint mock
/// whose `run_query` errors.
#[tokio::test]
async fn owner_role_probe_error_defers_dependents() {
    let d = derived();
    // A maint mock that fails run_query (simulate a probe error via an always-erroring backend).
    struct ErrBackend;
    #[async_trait]
    impl SqlBackend for ErrBackend {
        async fn begin(&self) -> Result<Box<dyn SqlTransaction>, SqlError> {
            Err(SqlError::Other("x".into()))
        }
        async fn run_query(&self, _sql: &str) -> Result<SqlRows, SqlError> {
            Err(SqlError::Other("probe boom".into()))
        }
        async fn run_script(&self, sql: &str) -> Result<(), SqlError> {
            panic!("no DDL must run when the owner-role probe errored: {sql}");
        }
    }
    let maint = std::sync::Arc::new(ErrBackend) as std::sync::Arc<dyn SqlBackend>;
    let creds = mk_creds();
    let mut report = RepairReport::default();
    let ready = check_owner_role_exists(&creds, &d, &maint, RepairMode::Apply, &mut report).await;
    assert!(!ready, "an errored owner-role probe defers dependents");
    assert_eq!(report.checks[0].status, RepairStatus::Error);
    assert_eq!(report.checks[0].check, "owner-role-exists");
}

// ---------------------------------------------------------------------------
// Soft-delete exclusion (Security MEDIUM-3).
// ---------------------------------------------------------------------------

/// When the exact derived db is absent but a `<db>__deleted_<ts>` sibling exists, the run reports
/// `soft-delete` skipped and changes nothing.
#[tokio::test]
async fn soft_deleted_sibling_is_skipped() {
    let d = derived();
    // The exact-name probe (`datname = '<db>'`) returns empty; the LIKE-sibling probe returns a row.
    let maint = MockBackend::new(true)
        .on("datname = 'appdb_acme'", empty_rows())
        .on("LIKE", exists_rows());
    let maint = arc(maint);
    let state = probe_live_or_soft_deleted(&as_dyn(&maint), &d)
        .await
        .unwrap();
    assert!(matches!(state, LiveState::SoftDeletedOnly));
    assert!(maint.scripts().is_empty());
}

/// The exact live db (never a prefix/LIKE match) is detected as `Live`.
#[tokio::test]
async fn exact_live_db_is_live() {
    let d = derived();
    let maint = arc(MockBackend::new(true).on("datname = 'appdb_acme'", exists_rows()));
    assert!(matches!(
        probe_live_or_soft_deleted(&as_dyn(&maint), &d)
            .await
            .unwrap(),
        LiveState::Live
    ));
}

/// Neither the live db nor a soft-deleted sibling ⇒ `Absent` (never provisioned).
#[tokio::test]
async fn absent_db_is_absent() {
    let d = derived();
    let maint = arc(MockBackend::new(true)); // every probe → empty
    assert!(matches!(
        probe_live_or_soft_deleted(&as_dyn(&maint), &d)
            .await
            .unwrap(),
        LiveState::Absent
    ));
}

// ---------------------------------------------------------------------------
// Owner-role probe excludes a NOLOGIN (soft-deleted) sibling.
// ---------------------------------------------------------------------------

/// `probe_role_exists` keys on the exact name AND `rolcanlogin` — a NOLOGIN same-named role (a
/// deprovision artifact) is NOT counted as a live owner role (the probe SQL carries the guard).
#[tokio::test]
async fn role_exists_probe_requires_login() {
    let d = derived();
    // Empty (the guarded probe found no LOGIN role) ⇒ false; the SQL must contain the rolcanlogin
    // guard so a NOLOGIN sibling is excluded by construction.
    let maint = MockBackend::new(true);
    // Capture the SQL by asserting the query the check issues includes the guard: run the probe
    // and inspect via a capturing backend.
    struct CaptureBackend(Mutex<Vec<String>>);
    #[async_trait]
    impl SqlBackend for CaptureBackend {
        async fn begin(&self) -> Result<Box<dyn SqlTransaction>, SqlError> {
            Err(SqlError::Other("x".into()))
        }
        async fn run_query(&self, sql: &str) -> Result<SqlRows, SqlError> {
            self.0.lock().unwrap().push(sql.to_string());
            Ok(SqlRows::default())
        }
    }
    let cap = std::sync::Arc::new(CaptureBackend(Mutex::new(Vec::new())));
    let _ = probe_role_exists(
        &(cap.clone() as std::sync::Arc<dyn SqlBackend>),
        &d.owner_role,
    )
    .await
    .unwrap();
    let issued = cap.0.lock().unwrap().join("\n");
    assert!(
        issued.contains("rolcanlogin"),
        "role probe must exclude NOLOGIN: {issued}"
    );
    assert!(
        issued.contains("rolname = 'pg_acme_owner'"),
        "exact name only: {issued}"
    );
    let _ = maint;
}

// ---------------------------------------------------------------------------
// Topology / engine gating.
// ---------------------------------------------------------------------------

/// The per-backend model classifier maps each binding shape to its own [`RepairModel`] — the
/// backbone of JOB 2 (no backend is a blanket "not applicable"). Mirrors how each is provisioned.
#[test]
fn classify_model_maps_every_backend_shape() {
    // Shared Postgres (compute-backed) → the full owner model.
    assert_eq!(
        classify_model(&shared_pg_binding("postgres")),
        RepairModel::SharedPostgres
    );
    // Single/dedicated Postgres → the container-boundary model.
    let mut single = shared_pg_binding("postgres");
    single.tenant = TenantIsolation::Single;
    assert_eq!(classify_model(&single), RepairModel::DedicatedPostgres);
    // MySQL (managed or external) → the MySQL model.
    assert_eq!(
        classify_model(&shared_pg_binding("mysql")),
        RepairModel::Mysql
    );
    assert_eq!(
        classify_model(&external_binding("mysql")),
        RepairModel::Mysql
    );
    // libsql (a `path` binding) → the file-boundary model.
    assert_eq!(classify_model(&libsql_binding()), RepairModel::Libsql);
    // Bring-your-own Postgres (url_env, no compute) → the external model.
    assert_eq!(
        classify_model(&external_binding("postgres")),
        RepairModel::External
    );
    // A genuinely-unknown engine with a url_env → external (never a panic / never libsql).
    assert_eq!(
        classify_model(&external_binding("cockroach")),
        RepairModel::External
    );
}

/// The backend classifier labels each model for the report `backend` header.
#[test]
fn classify_backend_labels_each_model() {
    assert_eq!(
        classify_backend(&shared_pg_binding("postgres")),
        "shared-postgres"
    );
    // MySQL is ONE model (managed or external), so the label is a plain `mysql` — not iso-prefixed.
    let mut my = shared_pg_binding("mysql");
    my.tenant = TenantIsolation::Shared;
    assert_eq!(classify_backend(&my), "mysql");
    assert_eq!(classify_backend(&external_binding("mysql")), "mysql");
    let mut single = shared_pg_binding("postgres");
    single.tenant = TenantIsolation::Single;
    assert_eq!(classify_backend(&single), "single-postgres");
    assert_eq!(classify_backend(&external_binding("postgres")), "external");
    assert_eq!(classify_backend(&libsql_binding()), "libsql");
}

// ---------------------------------------------------------------------------
// Report serde round-trip.
// ---------------------------------------------------------------------------

/// A `RepairReport` round-trips through JSON (the wire form the CLI + admin API share), and the
/// status/mode enums serialize as their stable kebab slugs.
#[test]
fn report_serde_round_trips() {
    let report = RepairReport {
        tenant: "appdb_acme".into(),
        backend: "shared-postgres".into(),
        mode: "apply".into(),
        checks: vec![
            RepairCheck {
                check: "owner-role-exists".into(),
                status: RepairStatus::Repaired,
                detail: "created owner role".into(),
                ddl: Some("CREATE ROLE ...".into()),
            },
            RepairCheck {
                check: "connectivity".into(),
                status: RepairStatus::Ok,
                detail: "both connect".into(),
                ddl: None,
            },
        ],
    };
    let json = serde_json::to_string(&report).unwrap();
    // Stable status slugs.
    assert!(json.contains("\"repaired\""));
    assert!(json.contains("\"ok\""));
    // A `None` ddl is omitted (skip_serializing_if).
    assert!(json.contains("\"check\":\"connectivity\""));
    let back: RepairReport = serde_json::from_str(&json).unwrap();
    assert_eq!(back, report);
    assert!(!back.any_error());
    assert!(
        back.found_drift(),
        "a Repaired check counts as drift-was-found"
    );
    assert!(back.first_error().is_none());
}

/// A report with an errored check reports `any_error` + the first errored slug.
#[test]
fn report_helpers_flag_errors() {
    let report = RepairReport {
        checks: vec![
            RepairCheck {
                check: "db-owner".into(),
                status: RepairStatus::Ok,
                detail: String::new(),
                ddl: None,
            },
            RepairCheck {
                check: "object-ownership".into(),
                status: RepairStatus::Error,
                detail: "requires a superuser maintenance connection".into(),
                ddl: None,
            },
        ],
        ..RepairReport::default()
    };
    assert!(report.any_error());
    assert_eq!(report.first_error(), Some("object-ownership"));
}

// ---------------------------------------------------------------------------
// Password redaction (the report must never leak a sealed owner password).
// ---------------------------------------------------------------------------

/// A `CREATE ROLE … PASSWORD '<hex>'` has its password literal redacted before it lands in the
/// operator-visible report, and the doubled-quote escape is handled without truncating the tail.
#[test]
fn password_literal_is_redacted_in_report_ddl() {
    let stmt = "CREATE ROLE \"pg_acme_owner\" LOGIN NOSUPERUSER PASSWORD 'deadbeefcafe' NOINHERIT;";
    let red = redact_password_literal(stmt);
    assert!(!red.contains("deadbeefcafe"), "password leaked: {red}");
    assert!(red.contains("PASSWORD '<redacted>'"));
    assert!(
        red.contains("NOINHERIT"),
        "the tail after the literal must survive: {red}"
    );

    // A statement with no PASSWORD literal is unchanged.
    let plain = "ALTER DATABASE \"appdb_acme\" OWNER TO \"pg_acme_owner\";";
    assert_eq!(redact_password_literal(plain), plain);

    // An embedded doubled-quote in the literal doesn't truncate.
    let tricky = "CREATE ROLE r PASSWORD 'ab''cd' LOGIN;";
    let red2 = redact_password_literal(tricky);
    assert!(red2.contains("PASSWORD '<redacted>'"), "{red2}");
    assert!(
        red2.ends_with(" LOGIN;"),
        "tail survives a doubled-quote literal: {red2}"
    );
}

// ===========================================================================
// Per-backend RepairModel tests (JOB 2): each model reconciles its OWN provisioning; assert
// dry-run does no writes + derived-name-only + no blanket "not applicable".
// ===========================================================================

// ---- MySQL model: the distinct DDL-identity precondition -------------------

/// A compute-backed managed MySQL with NO `migration_url_env` is a TERMINAL `error` on the
/// `ddl-identity` check (mirrors the v0.5.1 migrate refusal) — NOT a silent skip. This is the
/// "compute-backed managed MySQL with no derivable DDL identity" requirement.
#[test]
fn mysql_managed_without_ddl_identity_is_terminal_error() {
    let mut b = shared_pg_binding("mysql");
    b.tenant = TenantIsolation::Shared; // managed, compute-backed
    b.migration_url_env = None;
    let mut report = RepairReport::default();
    check_mysql_ddl_identity(
        &b,
        "appdb_acme",
        /*compute_backed=*/ true,
        &mut report,
        &boatramp_core::env::MapEnv::new(),
    );
    assert_eq!(report.checks.len(), 1);
    assert_eq!(report.checks[0].check, "ddl-identity");
    assert_eq!(
        report.checks[0].status,
        RepairStatus::Error,
        "compute-backed managed MySQL with no DDL identity must be a terminal error, not a skip"
    );
}

/// An external MySQL whose `migration_url_env` is unset (declared but not in the env) → error;
/// whose DDL DSN authenticates as the SAME username as the runtime → error (distinctness); whose
/// DDL DSN is a genuinely distinct login → ok. Uses per-test unique env vars so it is hermetic.
#[test]
fn mysql_external_ddl_identity_distinctness() {
    // All host-env values are injected via a MapEnv rather than mutating the process environment.
    let env = boatramp_core::env::MapEnv::new()
        .with("REPAIR_TEST_MYSQL_RUNTIME", "mysql://app:pw@h1/appdb")
        .with("REPAIR_TEST_MYSQL_DDL_SAME", "mysql://app:pw@h1/appdb")
        .with(
            "REPAIR_TEST_MYSQL_DDL_DISTINCT",
            "mysql://ddladmin:pw@h1/appdb",
        );

    // A declared-but-unset migration var → error (not reachable).
    let mut b = external_binding("mysql");
    b.migration_url_env = Some("REPAIR_TEST_MYSQL_DDL_UNSET".into());
    let mut report = RepairReport::default();
    check_mysql_ddl_identity(&b, "appdb", false, &mut report, &env);
    assert_eq!(report.checks[0].status, RepairStatus::Error);

    // Same username as the runtime → distinctness error. (Only meaningful with sql-mysql, which
    // parses the DSN; a sql-postgres-only build keeps the byte check — so make them byte-equal too.)
    let mut b = external_binding("mysql");
    b.url_env = "REPAIR_TEST_MYSQL_RUNTIME".into();
    b.migration_url_env = Some("REPAIR_TEST_MYSQL_DDL_SAME".into());
    let mut report = RepairReport::default();
    check_mysql_ddl_identity(&b, "appdb", false, &mut report, &env);
    assert_eq!(
        report.checks[0].status,
        RepairStatus::Error,
        "a DDL identity that is byte-identical to the runtime must be refused"
    );

    // A distinct DDL login → ok.
    let mut b = external_binding("mysql");
    b.url_env = "REPAIR_TEST_MYSQL_RUNTIME".into();
    b.migration_url_env = Some("REPAIR_TEST_MYSQL_DDL_DISTINCT".into());
    let mut report = RepairReport::default();
    check_mysql_ddl_identity(&b, "appdb", false, &mut report, &env);
    assert_eq!(
        report.checks[0].status,
        RepairStatus::Ok,
        "a distinct DDL login must pass: {:?}",
        report.checks[0].detail
    );
}

// ---- credential-sealed: dry-run purity (never seals on a dry-run) ----------

/// `check_credential_sealed` on an empty KV in DryRun reports `drift` and does NOT create the key
/// (dry-run purity — never `password()`/`put`). On Apply it seals it (create-if-absent).
#[tokio::test]
async fn credential_sealed_dry_run_never_writes() {
    use boatramp_core::kv::{KvStore, MemoryKv};
    let kv: std::sync::Arc<dyn KvStore> = std::sync::Arc::new(MemoryKv::new());
    let creds = ManagedSqlCredentials::new(kv.clone(), {
        use boatramp_core::envelope::{EnvelopeError, KeyEnvelope};
        struct Rev;
        #[async_trait]
        impl KeyEnvelope for Rev {
            async fn wrap(&self, p: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
                Ok(p.iter().rev().copied().collect())
            }
            async fn unwrap(&self, w: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
                Ok(w.iter().rev().copied().collect())
            }
        }
        std::sync::Arc::new(Rev)
    });

    // Dry-run: reports drift, seals nothing.
    let mut report = RepairReport::default();
    check_credential_sealed(
        &creds,
        "acme",
        "pg-acme",
        RepairMode::DryRun,
        "credential-sealed",
        &mut report,
    )
    .await;
    assert_eq!(report.checks[0].status, RepairStatus::Drift);
    assert!(
        kv.get("managed-sql-cred/acme/pg-acme")
            .await
            .unwrap()
            .is_none(),
        "a dry-run must NOT seal the credential (no KV write)"
    );

    // Apply: seals it.
    let mut report = RepairReport::default();
    check_credential_sealed(
        &creds,
        "acme",
        "pg-acme",
        RepairMode::Apply,
        "credential-sealed",
        &mut report,
    )
    .await;
    assert_eq!(report.checks[0].status, RepairStatus::Repaired);
    assert!(
        kv.get("managed-sql-cred/acme/pg-acme")
            .await
            .unwrap()
            .is_some()
    );
}

// ---- libsql model: the file is the boundary; dry-run never creates the file ----------

/// The libsql model over a NON-existent `path` in DryRun: reports `db-file` drift, SKIPS the
/// ledger + connectivity (a dry-run must not create the file), marks every role check skipped, and
/// — crucially — does NOT create the file on disk (dry-run purity for the file boundary).
#[cfg(feature = "migrate")]
#[tokio::test]
async fn libsql_dry_run_does_not_create_the_file() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("nope.db");
    assert!(!path.exists());

    let mut b = libsql_binding();
    b.path = Some(path.clone());
    let report = repair_libsql(&b, "acme", RepairMode::DryRun, "libsql").await;

    // The file boundary was NOT crossed.
    assert!(
        !path.exists(),
        "a libsql dry-run must NOT create the db file: {}",
        path.display()
    );
    let by = |slug: &str| report.checks.iter().find(|c| c.check == slug).unwrap();
    assert_eq!(by("db-file").status, RepairStatus::Drift);
    assert_eq!(by("ledger").status, RepairStatus::Skipped);
    assert_eq!(by("connectivity").status, RepairStatus::Skipped);
    // Role-model checks are skipped-with-reason (never a blanket single "not applicable").
    for role_check in [
        "owner-role",
        "object-ownership",
        "connect-grants",
        "runtime-grants",
    ] {
        assert_eq!(by(role_check).status, RepairStatus::Skipped);
    }
    assert_eq!(report.backend, "libsql");
    // The report renders through the shared CLI without a special case (uniform rendering).
    assert!(!report.checks.is_empty());
}

/// The libsql model over an EXISTING file: `db-file` ok, the ledger is scaffolded on apply and
/// present on a re-probe, connectivity ok. Proves the apply path reconciles the file's ledger.
#[cfg(feature = "migrate")]
#[tokio::test]
async fn libsql_apply_scaffolds_ledger_on_existing_file() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("app.db");
    // Create the file up front (so `db-file` is ok and we're testing the ledger reconcile).
    boatramp_storage::LibsqlSql::open_local(&path)
        .await
        .unwrap();
    assert!(path.exists());

    let mut b = libsql_binding();
    b.path = Some(path.clone());

    // Dry-run first: ledger drift, but nothing written (the table must still be absent after).
    let report = repair_libsql(&b, "acme", RepairMode::DryRun, "libsql").await;
    let by =
        |r: &RepairReport, slug: &str| r.checks.iter().find(|c| c.check == slug).unwrap().clone();
    assert_eq!(by(&report, "db-file").status, RepairStatus::Ok);
    assert_eq!(by(&report, "ledger").status, RepairStatus::Drift);
    // The dry-run emitted a CREATE but ran nothing — assert the table is still absent.
    {
        use boatramp_core::sql::SqlBackend;
        let sql = boatramp_storage::LibsqlSql::open_local(&path)
            .await
            .unwrap();
        let rows = sql
            .run_query(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND \
                 name='boatramp_migrations_schema_migrations';",
            )
            .await
            .unwrap();
        assert!(
            rows.rows.is_empty(),
            "dry-run must not create the ledger table"
        );
    }

    // Apply: scaffolds the ledger, connectivity ok.
    let report = repair_libsql(&b, "acme", RepairMode::Apply, "libsql").await;
    assert_eq!(by(&report, "ledger").status, RepairStatus::Repaired);
    assert_eq!(by(&report, "connectivity").status, RepairStatus::Ok);
    {
        use boatramp_core::sql::SqlBackend;
        let sql = boatramp_storage::LibsqlSql::open_local(&path)
            .await
            .unwrap();
        let rows = sql
            .run_query(
                "SELECT 1 FROM sqlite_master WHERE type='table' AND \
                 name='boatramp_migrations_schema_migrations';",
            )
            .await
            .unwrap();
        assert!(
            !rows.rows.is_empty(),
            "apply must scaffold the ledger table"
        );
    }
}

/// A remote-sqld libsql binding (url_env, no `path`) has no boatramp-owned local file → `db-file`
/// skipped-with-reason, never an error, and the op is a clean report.
#[cfg(feature = "migrate")]
#[tokio::test]
async fn libsql_remote_sqld_has_no_local_file() {
    let mut b = libsql_binding();
    b.path = None;
    b.url_env = "SQLD_URL".into();
    let report = repair_libsql(&b, "acme", RepairMode::DryRun, "libsql").await;
    let by = |slug: &str| report.checks.iter().find(|c| c.check == slug).unwrap();
    assert_eq!(by("db-file").status, RepairStatus::Skipped);
    assert!(!report.any_error());
}

// ---- converge_ledger dry-run purity (mock backend panics on any run_script) ----------

/// `converge_ledger` in DryRun emits the `CREATE … IF NOT EXISTS` DDL but runs NOTHING (the mock
/// panics on any `run_script`). Proves the shared ledger-scaffold converge is dry-run-pure.
#[tokio::test]
async fn converge_ledger_dry_run_runs_nothing() {
    let backend = arc(MockBackend::new(/*panic_on_script=*/ true));
    let ddl = vec![
        "CREATE SCHEMA IF NOT EXISTS \"boatramp_migrations\";".to_string(),
        "CREATE TABLE IF NOT EXISTS \"boatramp_migrations\".\"schema_migrations\" (id text);"
            .to_string(),
    ];
    let mut report = RepairReport::default();
    converge_ledger(&as_dyn(&backend), &ddl, RepairMode::DryRun, &mut report).await;
    assert_eq!(report.checks[0].status, RepairStatus::Drift);
    assert!(
        report.checks[0]
            .ddl
            .as_ref()
            .unwrap()
            .contains("CREATE TABLE IF NOT EXISTS")
    );
    assert!(
        backend.scripts().is_empty(),
        "dry-run ledger converge must run no DDL"
    );
}

// ---------------------------------------------------------------------------
// FIX 1 (Security HIGH-1) regression: a full `repair(..., DryRun)` over an UNSEALED tenant
// (owner/runtime credential absent — a pre-v0.4.25 shape) must leave the KV byte-for-byte
// untouched. This is the end-to-end test the earlier isolated-check tests missed: it exercises
// the WHOLE orchestrator (soft-delete probe → checks → terminal connectivity), so if ANY dry-run
// code path called `password()` (create-if-absent + seal) or `kv.put`, a NEW key would appear.
// Covers all three managed models — shared Postgres, dedicated Postgres, managed MySQL.
// ---------------------------------------------------------------------------

/// A `Storage` that errors on every access — the dry-run must never reach it (the credential-absent
/// short-circuit fires first), and even if a probe tried to connect, it produces an `error` check,
/// never a KV write. Mirrors the `NullStorage` in `boatramp-core`'s deploy tests.
struct NullStorage;
#[async_trait]
impl boatramp_core::Storage for NullStorage {
    async fn get(&self, _: &str) -> Result<boatramp_core::GetObject, boatramp_core::StorageError> {
        Err(boatramp_core::StorageError::NotFound(String::new()))
    }
    async fn get_range(
        &self,
        _: &str,
        _: u64,
        _: Option<u64>,
    ) -> Result<boatramp_core::GetObject, boatramp_core::StorageError> {
        Err(boatramp_core::StorageError::NotFound(String::new()))
    }
    async fn put(
        &self,
        _: &str,
        _: boatramp_core::ByteStream,
        _: boatramp_core::PutMeta,
    ) -> Result<boatramp_core::ObjectMeta, boatramp_core::StorageError> {
        Err(boatramp_core::StorageError::unsupported("null"))
    }
    async fn head(
        &self,
        _: &str,
    ) -> Result<boatramp_core::ObjectMeta, boatramp_core::StorageError> {
        Err(boatramp_core::StorageError::NotFound(String::new()))
    }
    async fn delete(&self, _: &str) -> Result<(), boatramp_core::StorageError> {
        Ok(())
    }
    async fn list(
        &self,
        _: &str,
    ) -> Result<Vec<boatramp_core::ObjectMeta>, boatramp_core::StorageError> {
        Ok(Vec::new())
    }
}

/// Build a `(DeployStore, creds, kv)` over a shared MemoryKv + the reversible test envelope. The
/// returned KV is the one whose keyset the regression test snapshots before/after the dry-run.
fn dry_run_harness() -> (
    boatramp_core::deploy::DeployStore,
    ManagedSqlCredentials,
    std::sync::Arc<dyn boatramp_core::kv::KvStore>,
) {
    use boatramp_core::deploy::DeployStore;
    use boatramp_core::envelope::{EnvelopeError, KeyEnvelope};
    use boatramp_core::kv::{KvStore, MemoryKv};
    struct Rev;
    #[async_trait]
    impl KeyEnvelope for Rev {
        async fn wrap(&self, p: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
            Ok(p.iter().rev().copied().collect())
        }
        async fn unwrap(&self, w: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
            Ok(w.iter().rev().copied().collect())
        }
    }
    let kv: std::sync::Arc<dyn KvStore> = std::sync::Arc::new(MemoryKv::new());
    let deploy = DeployStore::new(std::sync::Arc::new(NullStorage), kv.clone());
    let creds = ManagedSqlCredentials::new(kv.clone(), std::sync::Arc::new(Rev));
    (deploy, creds, kv)
}

/// The FIX 1 end-to-end invariant: a DryRun repair over an unsealed tenant of EACH managed model
/// adds ZERO keys to the KV. The superuser credential (the maintenance identity) is pre-sealed —
/// that is the normal shape, and reading it via `password()` on an already-sealed key is a no-op —
/// but the per-tenant OWNER/RUNTIME credentials are absent (pre-v0.4.25), and a dry-run must NOT
/// seal them (that would be a side effect on a probe-only operation).
#[tokio::test]
async fn dry_run_repair_over_unsealed_tenant_writes_no_kv_keys() {
    use boatramp_core::kv::KvStore;

    async fn snapshot(kv: &std::sync::Arc<dyn KvStore>) -> Vec<String> {
        let mut ks = kv.list_prefix("").await.unwrap();
        ks.sort();
        ks
    }

    // Shared Postgres (the full owner model), dedicated Postgres, and managed MySQL. Each is
    // compute-backed so the credential machinery is on the path; each is UNSEALED for its
    // per-tenant identity.
    for (label, binding) in [
        ("shared-postgres", shared_pg_binding("postgres")),
        ("dedicated-postgres", dedicated_pg_binding()),
        ("managed-mysql", managed_mysql_binding()),
    ] {
        let (deploy, creds, kv) = dry_run_harness();
        // Pre-seal ONLY the superuser/maintenance credential (`managed-sql-cred/default/pg`) — the
        // maintenance connection reads it; a pre-sealed key is a no-op read, never a new write.
        // The per-tenant owner/runtime credentials are deliberately absent (the pre-v0.4.25 shape).
        creds
            .password(boatramp_core::project::DEFAULT_PROJECT, "pg")
            .await
            .expect("pre-seal the superuser credential");

        let before = snapshot(&kv).await;
        assert_eq!(
            before,
            vec!["managed-sql-cred/default/pg".to_string()],
            "{label}: only the superuser credential is pre-sealed"
        );

        let report = repair_tenant(
            &deploy,
            &creds,
            &binding,
            "main",
            "acme",
            RepairMode::DryRun,
            &boatramp_core::env::MapEnv::new(),
        )
        .await
        .unwrap_or_else(|e| panic!("{label}: dry-run repair should produce a report: {e}"));
        assert_eq!(report.mode, "dry-run", "{label}");

        let after = snapshot(&kv).await;
        assert_eq!(
            after, before,
            "{label}: a DryRun repair over an unsealed tenant must leave the KV byte-for-byte \
             untouched — NO owner/runtime credential may be sealed on a dry-run (Security HIGH-1)"
        );
    }
}

// ---------------------------------------------------------------------------
// Helpers.
// ---------------------------------------------------------------------------

/// A real (unused-by-dry-run) `ManagedSqlCredentials` over MemoryKv, for check signatures that
/// require one but whose dry-run path never touches it.
fn mk_creds() -> ManagedSqlCredentials {
    use boatramp_core::envelope::{EnvelopeError, KeyEnvelope};
    use boatramp_core::kv::MemoryKv;
    struct Rev;
    #[async_trait]
    impl KeyEnvelope for Rev {
        async fn wrap(&self, p: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
            Ok(p.iter().rev().copied().collect())
        }
        async fn unwrap(&self, w: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
            Ok(w.iter().rev().copied().collect())
        }
    }
    ManagedSqlCredentials::new(
        std::sync::Arc::new(MemoryKv::new()),
        std::sync::Arc::new(Rev),
    )
}

/// A shared-Postgres owner-model binding config for the topology tests.
fn shared_pg_binding(kind: &str) -> ExternalDatabaseConfig {
    ExternalDatabaseConfig {
        kind: kind.to_string(),
        compute: Some("pg".into()),
        database: Some("appdb".into()),
        user: Some("postgres".into()),
        tenant: TenantIsolation::Shared,
        tenant_scope: TenantScope::Project,
        ..Default::default()
    }
}

/// A dedicated/single managed-Postgres binding (compute-backed, `tenant = Single`) — the
/// container-boundary model. Used by the FIX 1 dry-run-purity regression.
fn dedicated_pg_binding() -> ExternalDatabaseConfig {
    ExternalDatabaseConfig {
        kind: "postgres".to_string(),
        compute: Some("pg".into()),
        database: Some("appdb".into()),
        user: Some("postgres".into()),
        tenant: TenantIsolation::Single,
        tenant_scope: TenantScope::Project,
        ..Default::default()
    }
}

/// A managed (compute-backed) MySQL binding with a distinct DDL identity. Used by the FIX 1
/// dry-run-purity regression (the managed-MySQL model resolves a sealed runtime credential).
fn managed_mysql_binding() -> ExternalDatabaseConfig {
    ExternalDatabaseConfig {
        kind: "mysql".to_string(),
        compute: Some("pg".into()),
        database: Some("appdb".into()),
        user: Some("app".into()),
        migration_url_env: Some("REPAIR_TEST_NONEXISTENT_MYSQL_DDL".into()),
        tenant: TenantIsolation::Shared,
        tenant_scope: TenantScope::Project,
        ..Default::default()
    }
}

/// A bring-your-own (`url_env`, no compute) binding of the given engine.
fn external_binding(kind: &str) -> ExternalDatabaseConfig {
    ExternalDatabaseConfig {
        kind: kind.to_string(),
        url_env: "DB_URL".into(),
        database: Some("appdb".into()),
        tenant_scope: TenantScope::Project,
        ..Default::default()
    }
}

/// A single-node `libsql` (`path`) binding.
fn libsql_binding() -> ExternalDatabaseConfig {
    ExternalDatabaseConfig {
        kind: "libsql".to_string(),
        database: Some("app".into()),
        path: Some(std::path::PathBuf::from(
            "/tmp/does-not-exist-repair-test.db",
        )),
        tenant_scope: TenantScope::Project,
        ..Default::default()
    }
}