shepherd-registry 6.7.0

/*
    Appellation: runner <module>
    Contrib: @FL03
*/
//! The canonical migration runner reads and writes `schema_versions`, never
//! `PRAGMA user_version`. The shipped schema owns that table, and every native
//! registry open observes the same version ledger.

use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Write as _;
use std::time::{SystemTime, UNIX_EPOCH};

use rusqlite::{Connection, OptionalExtension, params};
use sha2::{Digest, Sha256};

use super::embedded::{INIT_SQL, INIT_SQL_SHA256, MIGRATIONS};
use crate::error::{Error, Result};

/// The highest schema version owned by this binary.
pub const SUPPORTED_SCHEMA_VERSION: u32 = MIGRATIONS[MIGRATIONS.len() - 1].version;
/// The one historical checksum label accepted for the v5 baseline. It was
/// written by the pre-SHA migration runner and is retained as an explicit
/// compatibility fact, never silently rewritten by inspection.
pub const LEGACY_BASELINE_CHECKSUM: &str = "baseline-v5.0.0";
/// The checksum written by the historical session-bus v22 migration.
///
/// That migration created the old session-bus objects.
/// Current v22 owns the singleton-claim objects instead. The alias is accepted
/// only for v22, and inspection still requires every current v22 postcondition
/// before the explicit snapshot-backed repair may replay the canonical SQL.
pub const HISTORICAL_V22_SESSION_BUS_CHECKSUM: &str =
    "4c3a7d304ed26f2fdd694e5b28c066b0b01e28a4d302ef060ad75eedfd81884c";
pub const REPAIR_COMMAND: &str = "shepherd registry repair --confirm";
pub const ROLLBACK_COMMAND: &str =
    "shepherd registry rollback --receipt PATH --witness-sha256 SHA256 --confirm";

/// Catalog object classes are part of the typed registry-health contract.
#[derive(
    Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, serde::Serialize, serde::Deserialize,
)]
#[serde(rename_all = "snake_case")]
pub enum ObjectKind {
    Table,
    Column,
    Index,
    Trigger,
    View,
    Absence,
}

impl ObjectKind {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Table => "table",
            Self::Column => "column",
            Self::Index => "index",
            Self::Trigger => "trigger",
            Self::View => "view",
            Self::Absence => "absence",
        }
    }
}

impl FindingClass {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::MissingObject => "missing_object",
            Self::ContradictoryObject => "contradictory_object",
            Self::RequiredObjectPresent => "required_object_present",
            Self::MigrationGap => "migration_gap",
            Self::SchemaAhead => "schema_ahead",
            Self::UnknownMigration => "unknown_migration",
            Self::UnknownChecksum => "unknown_checksum",
            Self::CurrentChecksum => "current_checksum",
            Self::AcceptedLegacyChecksumAlias => "accepted_legacy_checksum_alias",
            Self::MissingLedger => "missing_ledger",
        }
    }
}

/// Exact, deterministic classification of a migration-ledger finding.
#[derive(
    Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, serde::Serialize, serde::Deserialize,
)]
#[serde(rename_all = "snake_case")]
pub enum FindingClass {
    MissingObject,
    ContradictoryObject,
    RequiredObjectPresent,
    MigrationGap,
    SchemaAhead,
    UnknownMigration,
    UnknownChecksum,
    CurrentChecksum,
    AcceptedLegacyChecksumAlias,
    MissingLedger,
}

/// One structural fact found while inspecting an applied migration.
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct StructuralFinding {
    pub version: Option<u32>,
    pub kind: ObjectKind,
    pub object: String,
    pub class: FindingClass,
    pub expected: Option<String>,
    pub found: Option<String>,
    pub repairable: bool,
}

impl StructuralFinding {
    fn sort_key(&self) -> (u32, &'static str, String, FindingClass) {
        (
            self.version.unwrap_or(0),
            self.kind.as_str(),
            self.object.clone(),
            self.class,
        )
    }
}

/// One row in the applied migration ledger and its checksum classification.
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct AppliedMigration {
    pub version: u32,
    pub checksum: String,
    pub expected_checksum: Option<String>,
    pub checksum_class: FindingClass,
}

/// Read-only registry structure and checksum inventory. The runner owns the
/// inventory; CLI and tests consume this projection rather than carrying a
/// second schema map.
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RegistryInspection {
    pub supported_version: u32,
    pub current_version: Option<u32>,
    pub ledger_present: bool,
    pub applied: Vec<AppliedMigration>,
    pub accepted_legacy_aliases: Vec<AppliedMigration>,
    pub findings: Vec<StructuralFinding>,
}

impl RegistryInspection {
    pub fn ok(&self) -> bool {
        self.findings.is_empty()
    }

    pub fn canonical_repair_command(&self) -> &'static str {
        REPAIR_COMMAND
    }

    pub fn first_error(&self) -> Option<Error> {
        let finding = self.findings.first()?;
        match finding.class {
            FindingClass::SchemaAhead => Some(Error::SchemaAhead {
                found: i64::from(finding.version.unwrap_or_default()),
                supported: i64::from(self.supported_version),
            }),
            FindingClass::MigrationGap => Some(Error::MigrationGap {
                version: i64::from(finding.version.unwrap_or_default()),
            }),
            FindingClass::UnknownMigration => Some(Error::UnknownMigration {
                version: i64::from(finding.version.unwrap_or_default()),
            }),
            FindingClass::UnknownChecksum => {
                let version = i64::from(finding.version.unwrap_or_default());
                let expected = finding.expected.clone().unwrap_or_default();
                let found = finding.found.clone().unwrap_or_default();
                Some(Error::MigrationChecksum {
                    version,
                    expected,
                    found,
                })
            }
            FindingClass::MissingLedger => Some(Error::MigrationPostcondition {
                version: 1,
                object: "table:schema_versions".into(),
            }),
            FindingClass::MissingObject
            | FindingClass::ContradictoryObject
            | FindingClass::RequiredObjectPresent
            | FindingClass::AcceptedLegacyChecksumAlias => Some(Error::MigrationPostcondition {
                version: i64::from(finding.version.unwrap_or_default()),
                object: format!("{}:{}", finding.kind.as_str(), finding.object),
            }),
            FindingClass::CurrentChecksum => None,
        }
    }
}

/// Applies every pending schema migration against `conn` and returns the
/// highest applied `schema_versions.version`.
///
/// A migration is accepted only when its recorded checksum matches the exact
/// embedded SQL and every catalog postcondition for that version exists. A
/// failed DDL statement is never treated as an idempotent success: a row that
/// claims a migration ran while its schema is incomplete is corrupt state and
/// must stop the caller.
pub fn apply_all(conn: &Connection) -> Result<u32> {
    if checksum_hex(INIT_SQL.as_bytes()) != INIT_SQL_SHA256 {
        return Err(Error::Migration {
            version: 1,
            message: "0001_init.sql does not match the embedded SHA-256 contract".into(),
        });
    }
    if !schema_versions_exists(conn)? {
        conn.execute_batch(INIT_SQL)
            .map_err(|source| Error::Migration {
                version: 1,
                message: format!("0001_init.sql: {source}"),
            })?;
        record_version(conn, 1, checksum_hex(INIT_SQL.as_bytes()))?;
    }

    let supported = SUPPORTED_SCHEMA_VERSION;
    let found = read_current_version(conn)?;
    if found > supported {
        return Err(Error::SchemaAhead {
            found: i64::from(found),
            supported: i64::from(supported),
        });
    }

    let mut known = known_versions(conn)?;
    if let std::collections::btree_map::Entry::Vacant(entry) = known.entry(1) {
        // A pre-checksum database may have executed the baseline directly and
        // therefore has the schema ledger but no version row. Do not replay
        // CREATE TABLE statements into a populated database; record the exact
        // baseline bytes after verifying its mandatory root object instead.
        for &(kind, name) in postconditions(1) {
            require_postcondition(conn, 1, kind, name)?;
        }
        let checksum = checksum_hex(INIT_SQL.as_bytes());
        record_version(conn, 1, checksum.clone())?;
        entry.insert(checksum);
    }
    validate_recorded_checksums(&known)?;
    require_postconditions(conn, &known)?;

    for migration in MIGRATIONS {
        if known.contains_key(&migration.version) {
            continue;
        }

        conn.execute_batch(migration.sql)
            .map_err(|source| Error::Migration {
                version: i64::from(migration.version),
                message: format!("{}: {source}", migration.filename),
            })?;
        for &(kind, name) in migration_postconditions(migration.version) {
            require_postcondition(conn, migration.version, kind, name)?;
        }
        let checksum = checksum_hex(migration.sql.as_bytes());
        record_version(conn, migration.version, checksum.clone())?;
        known.insert(migration.version, checksum);
    }

    require_postconditions(conn, &known)?;
    read_current_version(conn)
}

/// Inspect one already-open connection without changing any SQLite state.
/// Callers opening a path for this purpose must use read-only + no-follow and
/// query-only posture; [`Registry`](crate::registry::Registry) does so at its
/// public boundary.
pub fn inspect(conn: &Connection) -> Result<RegistryInspection> {
    let ledger_present = schema_versions_exists(conn)?;
    if !ledger_present {
        return Ok(RegistryInspection {
            supported_version: SUPPORTED_SCHEMA_VERSION,
            current_version: None,
            ledger_present: false,
            applied: Vec::new(),
            accepted_legacy_aliases: Vec::new(),
            findings: vec![StructuralFinding {
                version: Some(1),
                kind: ObjectKind::Table,
                object: "schema_versions".into(),
                class: FindingClass::MissingLedger,
                expected: Some("table".into()),
                found: None,
                repairable: false,
            }],
        });
    }

    let mut findings = Vec::new();
    let mut applied = Vec::new();
    let mut aliases = Vec::new();
    let known = known_versions(conn)?;
    let current_version = known.keys().next_back().copied();
    if known.is_empty() {
        findings.push(StructuralFinding {
            version: Some(1),
            kind: ObjectKind::Absence,
            object: "schema_versions:version=1".into(),
            class: FindingClass::MissingLedger,
            expected: Some("baseline ledger row".into()),
            found: None,
            repairable: false,
        });
    }

    for (&version, checksum) in &known {
        let expected = expected_checksum(version);
        let checksum_class = match (&expected, version, checksum.as_str()) {
            (Some(_expected), 1, LEGACY_BASELINE_CHECKSUM)
            | (Some(_expected), 22, HISTORICAL_V22_SESSION_BUS_CHECKSUM) => {
                FindingClass::AcceptedLegacyChecksumAlias
            }
            (Some(expected), _, found) if expected == found => FindingClass::CurrentChecksum,
            (Some(_), _, _) => FindingClass::UnknownChecksum,
            (None, _, _) => FindingClass::UnknownMigration,
        };
        let row = AppliedMigration {
            version,
            checksum: checksum.clone(),
            expected_checksum: expected.clone(),
            checksum_class,
        };
        if checksum_class == FindingClass::AcceptedLegacyChecksumAlias {
            aliases.push(row.clone());
        }
        applied.push(row);

        match checksum_class {
            FindingClass::UnknownChecksum => findings.push(StructuralFinding {
                version: Some(version),
                kind: ObjectKind::Absence,
                object: format!("migration:{version}:checksum"),
                class: FindingClass::UnknownChecksum,
                expected,
                found: Some(checksum.clone()),
                repairable: false,
            }),
            FindingClass::UnknownMigration => findings.push(StructuralFinding {
                version: Some(version),
                kind: ObjectKind::Absence,
                object: format!("migration:{version}"),
                class: FindingClass::UnknownMigration,
                expected: None,
                found: Some(checksum.clone()),
                repairable: false,
            }),
            _ => {}
        }

        if checksum_class == FindingClass::AcceptedLegacyChecksumAlias && version == 22 {
            for &(kind, name) in historical_v22_postconditions() {
                match catalog_state(conn, kind, name)? {
                    CatalogState::Present { kind: actual } if actual == kind => {}
                    CatalogState::Absent => findings.push(StructuralFinding {
                        version: Some(version),
                        kind: object_kind(kind),
                        object: name.into(),
                        class: FindingClass::MissingObject,
                        expected: Some(kind.into()),
                        found: None,
                        repairable: false,
                    }),
                    CatalogState::Present { kind: actual } => findings.push(StructuralFinding {
                        version: Some(version),
                        kind: object_kind(kind),
                        object: name.into(),
                        class: FindingClass::ContradictoryObject,
                        expected: Some(kind.into()),
                        found: Some(actual),
                        repairable: false,
                    }),
                }
            }
        }
    }

    if let Some(found) = current_version {
        if found > SUPPORTED_SCHEMA_VERSION {
            findings.push(StructuralFinding {
                version: Some(found),
                kind: ObjectKind::Absence,
                object: "schema_versions".into(),
                class: FindingClass::SchemaAhead,
                expected: Some(SUPPORTED_SCHEMA_VERSION.to_string()),
                found: Some(found.to_string()),
                repairable: false,
            });
        }
        for version in 1..=found.min(SUPPORTED_SCHEMA_VERSION) {
            if !known.contains_key(&version) {
                findings.push(StructuralFinding {
                    version: Some(version),
                    kind: ObjectKind::Absence,
                    object: format!("migration:{version}"),
                    class: FindingClass::MigrationGap,
                    expected: Some("applied ledger row".into()),
                    found: None,
                    repairable: false,
                });
            }
        }
    }

    let known_for_postconditions = known.clone();
    for (&version, checksum) in &known {
        if expected_checksum(version).is_none() || !checksum_is_accepted(version, checksum) {
            continue;
        }
        for &(kind, name) in recorded_postconditions(version, &known_for_postconditions) {
            let object_kind = object_kind(kind);
            match catalog_state(conn, kind, name)? {
                CatalogState::Absent if kind == "absent" => {}
                CatalogState::Present { kind: actual } if kind == "absent" => {
                    findings.push(StructuralFinding {
                        version: Some(version),
                        kind: ObjectKind::Absence,
                        object: name.into(),
                        class: FindingClass::RequiredObjectPresent,
                        expected: Some("absent".into()),
                        found: Some(actual),
                        repairable: false,
                    });
                }
                CatalogState::Absent => findings.push(StructuralFinding {
                    version: Some(version),
                    kind: object_kind,
                    object: name.into(),
                    class: FindingClass::MissingObject,
                    expected: Some(kind.into()),
                    found: None,
                    repairable: true,
                }),
                CatalogState::Present { kind: actual } if actual == kind => {}
                CatalogState::Present { kind: actual } => findings.push(StructuralFinding {
                    version: Some(version),
                    kind: object_kind,
                    object: name.into(),
                    class: FindingClass::ContradictoryObject,
                    expected: Some(kind.into()),
                    found: Some(actual),
                    repairable: false,
                }),
            }
        }
    }

    applied.sort_by_key(|row| row.version);
    aliases.sort_by_key(|row| row.version);
    findings.sort_by_key(StructuralFinding::sort_key);
    Ok(RegistryInspection {
        supported_version: SUPPORTED_SCHEMA_VERSION,
        current_version,
        ledger_present,
        applied,
        accepted_legacy_aliases: aliases,
        findings,
    })
}

/// Read the canonical migration bytes for one version. This is intentionally
/// private to the runner's repair implementation so no caller can author a
/// second migration map.
pub(crate) fn migration_sql(version: u32) -> Option<&'static str> {
    if version == 1 {
        Some(INIT_SQL)
    } else {
        MIGRATIONS
            .iter()
            .find(|migration| migration.version == version)
            .map(|migration| migration.sql)
    }
}

/// Apply one already-proven missing migration inside a caller-owned
/// `BEGIN IMMEDIATE` boundary. Migration source files historically vary in
/// whether they include BEGIN/COMMIT; those wrappers are removed while their
/// statements remain byte-for-byte the canonical source authority.
pub(crate) fn apply_missing_in_transaction(conn: &Connection, version: u32) -> Result<()> {
    let Some(sql) = migration_sql(version) else {
        return Err(Error::UnknownMigration {
            version: i64::from(version),
        });
    };
    let body = strip_transaction_wrappers(sql);
    conn.execute_batch(&body)
        .map_err(|source| Error::Migration {
            version: i64::from(version),
            message: format!("repair migration {version}: {source}"),
        })?;
    Ok(())
}

/// Advance from an already repaired ledger row through every ordinary pending
/// migration while the caller holds the same explicit repair transaction.
pub(crate) fn apply_pending_in_transaction(conn: &Connection) -> Result<Vec<u32>> {
    let mut known = known_versions(conn)?;
    validate_recorded_checksums(&known)?;
    require_postconditions(conn, &known)?;
    let mut advanced = Vec::new();
    for migration in MIGRATIONS {
        if known.contains_key(&migration.version) {
            continue;
        }
        let body = strip_transaction_wrappers(migration.sql);
        conn.execute_batch(&body)
            .map_err(|source| Error::Migration {
                version: i64::from(migration.version),
                message: format!("{}: {source}", migration.filename),
            })?;
        for &(kind, name) in migration_postconditions(migration.version) {
            require_postcondition(conn, migration.version, kind, name)?;
        }
        let checksum = checksum_hex(migration.sql.as_bytes());
        record_version(conn, migration.version, checksum.clone())?;
        known.insert(migration.version, checksum);
        advanced.push(migration.version);
    }
    require_postconditions(conn, &known)?;
    Ok(advanced)
}

/// Derive the only migration that can be replayed safely from one inventory.
/// A replay is permitted only when the current highest row is the sole
/// defective version and every positive postcondition it owns is absent.
pub(crate) fn exact_repair_version(inspection: &RegistryInspection) -> Result<u32> {
    if !inspection.ledger_present {
        return Err(Error::RepairRefused("schema ledger is absent".into()));
    }
    if inspection.findings.is_empty() {
        return Err(Error::RepairRefused(
            "registry has no repairable structural defect".into(),
        ));
    }
    let mut versions = BTreeSet::new();
    for finding in &inspection.findings {
        if finding.class != FindingClass::MissingObject
            || !finding.repairable
            || finding.kind == ObjectKind::Absence
        {
            return Err(Error::RepairRefused(format!(
                "finding {}:{} is not a fully absent migration object",
                finding.kind.as_str(),
                finding.object
            )));
        }
        versions.insert(finding.version.ok_or_else(|| {
            Error::RepairRefused("repairable finding has no migration version".into())
        })?);
    }
    if versions.len() != 1 {
        return Err(Error::RepairRefused(
            "more than one migration has a structural defect".into(),
        ));
    }
    let version = *versions.iter().next().expect("one version was checked");
    if inspection.current_version != Some(version) {
        return Err(Error::RepairRefused(
            "a later migration is recorded, so replay would be ambiguous".into(),
        ));
    }
    let expected = postconditions(version)
        .iter()
        .filter(|(kind, _)| *kind != "absent")
        .count();
    if expected == 0 || inspection.findings.len() != expected {
        return Err(Error::RepairRefused(
            "the migration is only partially absent or has no replayable objects".into(),
        ));
    }
    Ok(version)
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum CatalogState {
    Absent,
    Present { kind: String },
}

fn object_kind(kind: &str) -> ObjectKind {
    match kind {
        "table" => ObjectKind::Table,
        "column" => ObjectKind::Column,
        "index" => ObjectKind::Index,
        "trigger" => ObjectKind::Trigger,
        "view" => ObjectKind::View,
        "absent" => ObjectKind::Absence,
        _ => ObjectKind::Absence,
    }
}

fn catalog_state(conn: &Connection, kind: &str, name: &str) -> Result<CatalogState> {
    if kind == "column" {
        let Some((table, column)) = name.split_once('.') else {
            return Ok(CatalogState::Absent);
        };
        let table_state = catalog_state(conn, "table", table)?;
        match table_state {
            CatalogState::Absent => return Ok(CatalogState::Absent),
            CatalogState::Present { kind } if kind != "table" => {
                return Ok(CatalogState::Present { kind });
            }
            CatalogState::Present { .. } => {}
        }
        // SQLite does not bind PRAGMA identifiers on every supported version;
        // use the catalog's quote-safe identifier form instead.
        let escaped = table.replace('"', "\"\"");
        let mut statement = conn.prepare(&format!("PRAGMA table_info(\"{escaped}\")"))?;
        let found = statement
            .query_map([], |row| row.get::<_, String>(1))?
            .collect::<rusqlite::Result<Vec<_>>>()?
            .into_iter()
            .any(|value| value == column);
        return Ok(if found {
            CatalogState::Present {
                kind: "column".into(),
            }
        } else {
            CatalogState::Absent
        });
    }
    if kind == "absent" {
        return conn
            .query_row(
                "SELECT type FROM sqlite_master WHERE name = ?1 LIMIT 1",
                [name],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map(|kind| match kind {
                Some(kind) => CatalogState::Present { kind },
                None => CatalogState::Absent,
            })
            .map_err(Error::from);
    }
    conn.query_row(
        "SELECT type FROM sqlite_master WHERE name = ?1 LIMIT 1",
        [name],
        |row| row.get::<_, String>(0),
    )
    .optional()
    .map(|actual| match actual {
        Some(actual) if actual == kind => CatalogState::Present { kind: actual },
        Some(actual) => CatalogState::Present { kind: actual },
        None => CatalogState::Absent,
    })
    .map_err(Error::from)
}

fn expected_checksum(version: u32) -> Option<String> {
    if version == 1 {
        Some(checksum_hex(INIT_SQL.as_bytes()))
    } else {
        MIGRATIONS
            .iter()
            .find(|migration| migration.version == version)
            .map(|migration| checksum_hex(migration.sql.as_bytes()))
    }
}

fn checksum_is_accepted(version: u32, checksum: &str) -> bool {
    expected_checksum(version).is_some_and(|expected| {
        checksum == expected
            || (version == 1 && checksum == LEGACY_BASELINE_CHECKSUM)
            || (version == 22 && checksum == HISTORICAL_V22_SESSION_BUS_CHECKSUM)
    })
}

fn historical_v22_postconditions() -> &'static [(&'static str, &'static str)] {
    &[
        ("column", "sessions.harness"),
        ("column", "sessions.status"),
        ("column", "sessions.last_seen_at"),
        ("table", "session_heartbeats"),
        ("index", "idx_session_heartbeats_session_ts"),
        ("table", "notifications"),
        ("index", "idx_notifications_pending"),
    ]
}

fn strip_transaction_wrappers(sql: &str) -> String {
    sql.lines()
        .filter(|line| {
            let trimmed = line.trim();
            !trimmed.eq_ignore_ascii_case("BEGIN;") && !trimmed.eq_ignore_ascii_case("COMMIT;")
        })
        .collect::<Vec<_>>()
        .join("\n")
}

/// `SELECT COALESCE(MAX(version), 0) FROM schema_versions` -- the highest
/// applied version, or 0 on an empty (or absent) table.
fn read_current_version(conn: &Connection) -> Result<u32> {
    let max: i64 = conn.query_row(
        "SELECT COALESCE(MAX(version), 0) FROM schema_versions",
        [],
        |row| row.get(0),
    )?;
    u32::try_from(max)
        .map_err(|_| Error::unknown(format!("schema_versions.version out of range: {max}")))
}

/// An order-normalized `sqlite_master` dump -- the registry parity surface.
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn dump_sqlite_master(conn: &Connection) -> Result<String> {
    let mut stmt =
        conn.prepare("SELECT type, name, tbl_name, sql FROM sqlite_master ORDER BY type, name")?;
    let rows = stmt.query_map([], |row| {
        let kind: String = row.get(0)?;
        let name: String = row.get(1)?;
        let tbl_name: String = row.get(2)?;
        let sql: Option<String> = row.get(3)?;
        Ok((kind, name, tbl_name, sql))
    })?;

    let mut out = String::new();
    for row in rows {
        let (kind, name, tbl_name, sql) = row?;
        let sql = sql.unwrap_or_default().replace('\n', "\\n");
        writeln!(out, "{kind}\t{name}\t{tbl_name}\t{sql}")
            .expect("writing to a String cannot fail");
    }
    Ok(out)
}

fn schema_versions_exists(conn: &Connection) -> Result<bool> {
    let exists: bool = conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'schema_versions')",
        [],
        |row| row.get(0),
    )?;
    Ok(exists)
}

fn known_versions(conn: &Connection) -> Result<BTreeMap<u32, String>> {
    let mut stmt = conn.prepare("SELECT version, checksum FROM schema_versions")?;
    let rows = stmt.query_map([], |row| {
        Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?))
    })?;
    let mut versions = BTreeMap::new();
    for row in rows {
        let (version, checksum) = row?;
        let version = u32::try_from(version)
            .map_err(|_| Error::unknown(format!("negative schema_versions.version: {version}")))?;
        versions.insert(version, checksum);
    }
    Ok(versions)
}

fn validate_recorded_checksums(known: &BTreeMap<u32, String>) -> Result<()> {
    for (&version, recorded) in known {
        let Some(expected) = expected_checksum(version) else {
            // `apply_all` already rejects an ahead schema. A gap below the
            // supported maximum with no embedded file is a corrupt ledger.
            return Err(Error::UnknownMigration {
                version: i64::from(version),
            });
        };
        if !checksum_is_accepted(version, recorded) {
            return Err(Error::MigrationChecksum {
                version: i64::from(version),
                expected,
                found: recorded.clone(),
            });
        }
    }
    Ok(())
}

fn require_postconditions(conn: &Connection, known: &BTreeMap<u32, String>) -> Result<()> {
    for &version in known.keys() {
        for &(kind, name) in recorded_postconditions(version, known) {
            require_postcondition(conn, version, kind, name)?;
        }
    }
    Ok(())
}

fn recorded_postconditions(
    version: u32,
    known: &BTreeMap<u32, String>,
) -> &'static [(&'static str, &'static str)] {
    // Migration 0020 retires the mailbox objects created by 0007 and rebuilt
    // by 0016. Until 0020 is recorded, version 16's live postcondition is the
    // mailbox table/view; after 0020, its durable postcondition is absence.
    if version == 16 && !known.contains_key(&20) {
        migration_postconditions(16)
    } else {
        postconditions(version)
    }
}

fn require_postcondition(conn: &Connection, version: u32, kind: &str, name: &str) -> Result<()> {
    let exists = match kind {
        "column" => {
            let (table, column) =
                name.split_once('.')
                    .ok_or_else(|| Error::MigrationPostcondition {
                        version: i64::from(version),
                        object: format!("{kind}:{name}"),
                    })?;
            let mut statement = conn.prepare(&format!("PRAGMA table_info({table})"))?;
            statement
                .query_map([], |row| row.get::<_, String>(1))?
                .collect::<rusqlite::Result<Vec<_>>>()?
                .iter()
                .any(|value| value == column)
        }
        "table" | "view" | "index" | "trigger" => conn.query_row(
            "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = ?1 AND name = ?2)",
            (kind, name),
            |row| row.get::<_, bool>(0),
        )?,
        "absent" => !conn.query_row(
            "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name = ?1)",
            [name],
            |row| row.get::<_, bool>(0),
        )?,
        _ => {
            return Err(Error::MigrationPostcondition {
                version: i64::from(version),
                object: format!("{kind}:{name}"),
            });
        }
    };
    if exists {
        Ok(())
    } else {
        Err(Error::MigrationPostcondition {
            version: i64::from(version),
            object: format!("{kind}:{name}"),
        })
    }
}

/// Exact catalog postconditions for every migration. The kind is part of the
/// contract: a table with a view's name, or an index with a trigger's name, is
/// tampering, not a successful migration. SQL text remains the migration
/// authority; this list only states the catalog facts its execution must leave.
fn postconditions(version: u32) -> &'static [(&'static str, &'static str)] {
    match version {
        1 => &[
            ("table", "schema_versions"),
            ("table", "projects"),
            ("table", "sessions"),
            ("table", "profiles_defs"),
            ("table", "mem_entries"),
            ("table", "index_symbols"),
            ("table", "index_concepts"),
            ("table", "index_issues"),
            ("table", "index_prs"),
            ("table", "index_releases"),
            ("table", "index_milestones"),
            ("table", "logs_events"),
            ("table", "artifacts"),
            ("table", "locks_history"),
            ("view", "v_open_issues"),
            ("view", "v_canonical_types"),
            ("view", "v_drift_risk"),
            ("view", "v_mem_recent_7d"),
            ("view", "v_active_locks"),
        ],
        2 => &[("table", "styles")],
        3 => &[
            ("view", "v_canonical_types"),
            ("view", "v_canonical_symbols"),
            ("table", "lane_closures"),
            ("index", "idx_lane_closures_project_sprint"),
        ],
        4 => &[
            ("table", "index_fts_artifacts"),
            ("table", "index_fts_symbols"),
            ("trigger", "artifacts_ai"),
            ("trigger", "artifacts_ad"),
            ("trigger", "artifacts_au"),
            ("trigger", "index_symbols_ai"),
            ("trigger", "index_symbols_ad"),
            ("trigger", "index_symbols_au"),
        ],
        5 => &[
            ("table", "watch_paths"),
            ("index", "idx_watch_paths_label"),
            ("index", "idx_watch_paths_source"),
            ("trigger", "trg_watch_paths_updated_at"),
        ],
        6 => &[
            ("table", "index_cache_usage"),
            ("index", "idx_cache_usage_sprint"),
            ("index", "idx_cache_usage_role_ts"),
            ("view", "v_cache_usage"),
        ],
        7 => &[
            ("table", "teammates"),
            ("table", "heartbeats"),
            ("table", "escalations"),
            ("table", "deliverables"),
            ("table", "discovery_findings"),
            ("table", "audit_findings"),
            ("view", "v_teammates_live"),
        ],
        8 => &[("table", "worktrees")],
        9 => &[("table", "locks_history"), ("view", "v_active_locks")],
        10 => &[
            ("table", "sprint_metrics"),
            ("view", "v_sprint_metrics_avg"),
        ],
        11 => &[("table", "mem_entries"), ("view", "v_mem_recent_7d")],
        12 => &[
            ("table", "loops"),
            ("table", "loop_iterations"),
            ("view", "v_loops_active"),
        ],
        13 => &[("table", "focus"), ("view", "v_focus_current")],
        14 => &[("table", "compile_runs"), ("view", "v_compile_runs_sprint")],
        15 => &[("table", "index_struct_shapes")],
        16 => &[
            ("absent", "mailbox"),
            ("absent", "v_mailbox_unread_per_recipient"),
        ],
        17 => &[("table", "focus"), ("view", "v_focus_current")],
        18 => &[
            ("table", "eval_runs"),
            ("index", "idx_eval_runs_project"),
            ("view", "v_eval_latest"),
        ],
        19 => &[("column", "teammates.declared_state")],
        20 => &[
            ("table", "session_signals"),
            ("index", "idx_session_signals_pending"),
            ("absent", "mailbox"),
            ("absent", "v_mailbox_unread_per_recipient"),
        ],
        21 => &[("table", "spawn_leads")],
        22 => &[
            ("table", "dispatch_singleton_claims"),
            ("column", "dispatch_singleton_claims.write_scope"),
        ],
        23 => &[
            ("table", "dispatch_singleton_publications"),
            ("index", "idx_singleton_publications_preparing"),
            ("index", "idx_singleton_publications_key"),
            ("index", "idx_singleton_claims_publication_nonce"),
            ("column", "dispatch_singleton_claims.publication_nonce"),
        ],
        _ => &[],
    }
}

fn migration_postconditions(version: u32) -> &'static [(&'static str, &'static str)] {
    if version == 16 {
        &[
            ("table", "mailbox"),
            ("index", "idx_mailbox_recipient_unread"),
            ("index", "idx_mailbox_ack_pending"),
            ("view", "v_mailbox_unread_per_recipient"),
        ]
    } else {
        postconditions(version)
    }
}

fn record_version(conn: &Connection, version: u32, checksum: String) -> Result<()> {
    let applied_at = i64::try_from(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0),
    )
    .unwrap_or(i64::MAX);
    conn.execute(
        "INSERT INTO schema_versions (version, applied_at, checksum) VALUES (?1, ?2, ?3)",
        params![i64::from(version), applied_at, checksum],
    )?;
    Ok(())
}

/// A lowercase SHA-256 digest of the exact migration bytes stored in the
/// `schema_versions.checksum` column.
pub(crate) fn checksum_hex(bytes: &[u8]) -> String {
    let digest = Sha256::digest(bytes);
    let mut out = String::with_capacity(digest.len() * 2);
    for byte in digest {
        write!(out, "{byte:02x}").expect("writing to a String cannot fail");
    }
    out
}