shepherd-registry 6.5.9

/*
    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;
use std::fmt::Write as _;
use std::time::{SystemTime, UNIX_EPOCH};

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

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

/// 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 = MIGRATIONS.last().map_or(1, |migration| migration.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)
}

/// `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 expected = if version == 1 {
            checksum_hex(INIT_SQL.as_bytes())
        } else if let Some(migration) = MIGRATIONS
            .iter()
            .find(|migration| migration.version == version)
        {
            checksum_hex(migration.sql.as_bytes())
        } 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::MigrationChecksum {
                version: i64::from(version),
                expected: "embedded migration is missing".into(),
                found: recorded.clone(),
            });
        };
        if recorded != &expected {
            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
}