//! The schema migration runner.
//!
//! # Why the ledger table is called `_sqlx_migrations`
//!
//! Every Balanced Breakfast install in the field carries that table, one row per
//! applied migration, written by `sqlx::migrate!`. Renaming it, or writing a
//! fresh ledger beside it, makes an upgraded install believe it has applied
//! nothing and re-run every migration against a populated database, which fails
//! at the first `CREATE TABLE`.
//!
//! So this runner adopts that ledger verbatim: same table name, same columns,
//! same semantics, and the same checksum function (sha384 over the raw file
//! bytes, as sqlx 0.9 computes it). An upgraded install reads its own rows,
//! finds nothing pending, and does nothing. The checksum algorithm is not a free
//! choice.

use std::time::Instant;

use rusqlite::{Connection, OptionalExtension};
use sha2::{Digest, Sha384};

include!(concat!(env!("OUT_DIR"), "/migrations.rs"));

/// The ledger table. Named by sqlx; see the module docs before changing it.
const LEDGER: &str = "_sqlx_migrations";

/// A migration that failed to record itself needs a human, not a retry.
#[derive(Debug, thiserror::Error)]
pub enum MigrateError {
    #[error("database error running migrations: {0}")]
    Db(#[from] rusqlite::Error),

    #[error("could not check out a connection to run migrations: {0}")]
    Pool(#[from] r2d2::Error),

    #[error(
        "migration {version} ({description}) was already applied, but its file has changed since. \
         Applied migrations are immutable -- add a new migration instead of editing a shipped one."
    )]
    ChecksumMismatch { version: i64, description: String },

    #[error(
        "migration {0} is partially applied; fix it and remove its row from the `{LEDGER}` table"
    )]
    Dirty(i64),

    #[error("migration {version} ({description}) failed: {source}")]
    Apply {
        version: i64,
        description: String,
        #[source]
        source: rusqlite::Error,
    },
}

/// sha384 of a migration's bytes, matching what sqlx recorded.
pub(crate) fn checksum(sql: &str) -> Vec<u8> {
    Sha384::digest(sql.as_bytes()).to_vec()
}

/// Every migration compiled into this binary, in version order.
pub(crate) fn all() -> impl Iterator<Item = (i64, &'static str, &'static str)> {
    MIGRATIONS.iter().copied()
}

/// Create the ledger if absent. The DDL is sqlx 0.9's verbatim, so this is a
/// no-op on an install that sqlx already set up.
fn ensure_ledger(conn: &Connection) -> Result<(), rusqlite::Error> {
    conn.execute_batch(&format!(
        "CREATE TABLE IF NOT EXISTS {LEDGER} (
            version BIGINT PRIMARY KEY,
            description TEXT NOT NULL,
            installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
            success BOOLEAN NOT NULL,
            checksum BLOB NOT NULL,
            execution_time BIGINT NOT NULL
        );"
    ))
}

/// Apply every migration not yet in the ledger.
///
/// Each migration and its ledger row commit together, so a crash mid-run leaves
/// the database at a migration boundary rather than half-applied.
#[tracing::instrument(skip_all)]
pub fn run(conn: &mut Connection) -> Result<(), MigrateError> {
    ensure_ledger(conn)?;

    // A `success = false` row means a previous run died between applying a
    // migration and committing its ledger row. sqlx wrote these; this runner
    // cannot (it commits both together), but an install upgraded from sqlx may
    // carry one, and it still needs a human.
    let dirty: Option<i64> = conn
        .query_row(
            &format!("SELECT version FROM {LEDGER} WHERE success = false ORDER BY version LIMIT 1"),
            [],
            |row| row.get(0),
        )
        .optional()?;
    if let Some(version) = dirty {
        return Err(MigrateError::Dirty(version));
    }

    let applied: std::collections::BTreeMap<i64, Vec<u8>> = {
        let mut stmt = conn.prepare(&format!(
            "SELECT version, checksum FROM {LEDGER} ORDER BY version"
        ))?;
        let rows = stmt.query_map([], |row| {
            Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
        })?;
        rows.collect::<Result<_, _>>()?
    };

    for (version, description, sql) in all() {
        let digest = checksum(sql);

        if let Some(recorded) = applied.get(&version) {
            // Already applied. The only question left is whether the file still
            // hashes to what was recorded; if not, a shipped migration was
            // edited and every install that ran it now disagrees with this one.
            if *recorded != digest {
                return Err(MigrateError::ChecksumMismatch {
                    version,
                    description: description.to_owned(),
                });
            }
            continue;
        }

        tracing::info!(version, description, "applying migration");
        let started = Instant::now();
        let tx = conn.transaction()?;
        tx.execute_batch(sql)
            .map_err(|source| MigrateError::Apply {
                version,
                description: description.to_owned(),
                source,
            })?;
        tx.execute(
            &format!(
                "INSERT INTO {LEDGER} (version, description, success, checksum, execution_time)
                 VALUES (?1, ?2, TRUE, ?3, ?4)"
            ),
            rusqlite::params![
                version,
                description,
                digest,
                i64::try_from(started.elapsed().as_nanos()).unwrap_or(i64::MAX),
            ],
        )?;
        tx.commit()?;
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn every_migration_is_embedded() {
        let versions: Vec<i64> = all().map(|(v, _, _)| v).collect();
        assert!(!versions.is_empty(), "no migrations were embedded");
        assert!(
            versions.windows(2).all(|w| w[0] < w[1]),
            "migrations are not in strictly increasing version order: {versions:?}"
        );
    }

    #[test]
    fn description_matches_sqlx_filename_parsing() {
        // sqlx turns `004_feed_tags.sql` into the description "feed tags"; the
        // ledger rows in the field carry that spelling.
        let (_, description, _) = all().find(|(v, _, _)| *v == 4).expect("migration 004");
        assert_eq!(description, "feed tags");
    }

    #[test]
    fn running_twice_is_a_no_op() {
        let mut conn = Connection::open_in_memory().unwrap();
        run(&mut conn).unwrap();
        let first: i64 = conn
            .query_row(&format!("SELECT COUNT(*) FROM {LEDGER}"), [], |r| r.get(0))
            .unwrap();
        run(&mut conn).unwrap();
        let second: i64 = conn
            .query_row(&format!("SELECT COUNT(*) FROM {LEDGER}"), [], |r| r.get(0))
            .unwrap();
        assert_eq!(first, second);
        assert_eq!(first, all().count() as i64);
    }

    #[test]
    fn a_ledger_written_by_sqlx_is_adopted_untouched() {
        // Simulates an upgraded install: sqlx applied everything, so the new
        // runner must find nothing pending and leave the rows alone.
        let mut conn = Connection::open_in_memory().unwrap();
        run(&mut conn).unwrap();
        conn.execute(&format!("UPDATE {LEDGER} SET execution_time = 12345"), [])
            .unwrap();

        run(&mut conn).unwrap();

        let untouched: i64 = conn
            .query_row(
                &format!("SELECT COUNT(*) FROM {LEDGER} WHERE execution_time = 12345"),
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(untouched, all().count() as i64);
    }

    #[test]
    fn an_edited_migration_is_refused() {
        let mut conn = Connection::open_in_memory().unwrap();
        run(&mut conn).unwrap();
        conn.execute(
            &format!("UPDATE {LEDGER} SET checksum = X'00' WHERE version = 1"),
            [],
        )
        .unwrap();

        let err = run(&mut conn).unwrap_err();
        assert!(matches!(
            err,
            MigrateError::ChecksumMismatch { version: 1, .. }
        ));
    }

    /// What sqlx 0.9 records, read out of a Balanced Breakfast install that sqlx
    /// migrated. Not recomputed here: these are the bytes in the field, and
    /// reproducing them is what lets an upgraded install find nothing pending.
    const SQLX_CHECKSUMS: &[(i64, &str)] = &[
        (
            1,
            "f130b2307cda80086a379ae69723033ece65f0bfba06203354ab6e0d1613613677e2a98a5b7683b94c88e2479ccd4294",
        ),
        (
            2,
            "601dcfdb67d8fa2d2aa0a34dbde1dbea1bde71477d3c34d8327ea90594ab9db6d04669f0103f9de15aa7591d23b877fc",
        ),
        (
            3,
            "57b560ae0cea9bf6d2767feb8d113dbc065fde4ffc163ae966da945b8c0f33150e2e5a4de7c6a814e1faf915971b48fe",
        ),
        (
            4,
            "ab80308289562efbf9c808c37ac0b89f424c67bf5d054f61379f5e955c1aa75effe06d64311e6ec02f1ee3322b9116f8",
        ),
        (
            5,
            "bbed9b6af1942d85147f9558ab6c71b7018f194f70d3090abbd3a2a23c12073234e599df7aa0007f8c108462f3a45ca2",
        ),
        (
            6,
            "ca68cbf17892742631940c807f87fb9302c9e14cdf65432ce1c4877aa7d8f621cc551705478aae102640041eaaffe95a",
        ),
        (
            7,
            "7a8e9e6f930a4b87951cc062f8b9dc2f16e4f958a2772e8694a129d5dbc53d8573f013859e5e3cd31b4ed6b18fc49a63",
        ),
        (
            8,
            "efd01690547a1e12b964bada2f0eb163ee7946ed76538f70f685e5e17b89c5bfbab5931565aafd80d5b2d536d664f85e",
        ),
        (
            9,
            "8f5e8a3ac01a1e9251a7960f02d62dbc507ff5bd1978f555e6058a9977ed85ef1a11e699eace98b372626211adb8ebf8",
        ),
        (
            10,
            "c365b1e57c7722e83130f6d3fd7563abf6e68bd38bcba58ff94c19f64f7fae265b734aab0846e494eff5020ead77343b",
        ),
        (
            11,
            "5fce7042c81c0e58376b18cd6973708a25a0d4a473e2f89738653ccc390c20add47fa43751667da4278bb6ffb414b76c",
        ),
        (
            12,
            "44547197a05d3febe98770858d33d207eadfc1e365cba9075932347fff28bc0609b14a3f6adfa3e901dd36c9b46affb5",
        ),
        (
            13,
            "361c6ac7e31576ee9840a4af66afc7108f6414d6d1ecc8e2926886eb974f381f01706d0f3555d9a05f329b614af28bbd",
        ),
    ];

    /// The compatibility test that matters. If this fails, every install in the
    /// field sees a checksum mismatch on its first launch after upgrading and
    /// refuses to start -- so a failure here is never "update the constant", it
    /// is "the migration file was edited, put it back".
    #[test]
    fn checksums_match_what_sqlx_recorded() {
        let embedded: std::collections::BTreeMap<i64, &str> =
            all().map(|(v, _, sql)| (v, sql)).collect();

        for (version, expected_hex) in SQLX_CHECKSUMS {
            let sql = embedded
                .get(version)
                .unwrap_or_else(|| panic!("migration {version} is no longer embedded"));
            let actual = checksum(sql).iter().fold(String::new(), |mut out, b| {
                use std::fmt::Write as _;
                let _ = write!(out, "{b:02x}");
                out
            });
            assert_eq!(
                &actual, expected_hex,
                "migration {version} no longer hashes to what sqlx recorded in the field"
            );
        }
    }

    #[test]
    fn a_dirty_ledger_row_stops_the_run() {
        let mut conn = Connection::open_in_memory().unwrap();
        run(&mut conn).unwrap();
        conn.execute(
            &format!("UPDATE {LEDGER} SET success = false WHERE version = 1"),
            [],
        )
        .unwrap();

        let err = run(&mut conn).unwrap_err();
        assert!(matches!(err, MigrateError::Dirty(1)));
    }
}