//! Applying what the ledger says is missing.
//!
//! Lifted from goingson's runner, which was written when that app left
//! `sqlx-sqlite` and needed to keep reading a ledger `sqlx` had written. Two
//! things carried over unchanged and neither is a free choice there: the table
//! shape and the checksum function (sha384 over the raw file bytes, verified
//! against `sqlx` 0.9). They are kept here so that an app with `sqlx` history
//! can adopt this runner by naming its old ledger and nothing else — see
//! [`Migrator::ledger`].
//!
//! What did not carry over is the *default* name. A generated app has no
//! installs in the field and no `sqlx` past, and calling its ledger
//! `_sqlx_migrations` would be a new app inheriting a compatibility note about
//! a library it never linked.

use std::time::Instant;

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

/// One migration, as [`crate::embed::from_dir`] emits it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Migration {
    /// The integer prefix of the filename. Applied in this order.
    pub version: i64,
    /// The rest of the filename, underscores as spaces.
    pub description: &'static str,
    /// The file, verbatim. Hashed as-is.
    pub sql: &'static str,
}

/// A migration that could not be applied.
///
/// Every member needs a person. There is no variant meaning "retry": a run that
/// stops has either found a database it does not recognise or a migration that
/// does not apply, and running it again produces the same answer.
#[derive(Debug, thiserror::Error)]
pub enum MigrateError {
    #[error("database error running migrations: {0}")]
    Db(#[from] rusqlite::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")]
    Dirty(i64),

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

/// sha384 of a migration's bytes.
///
/// `sqlx` 0.9's function, so a ledger it wrote verifies against this one. Not a
/// free choice for an adopting app, and not worth changing for a new one.
#[must_use]
pub fn checksum(sql: &str) -> Vec<u8> {
    Sha384::digest(sql.as_bytes()).to_vec()
}

/// The ledger a fresh app writes.
const DEFAULT_LEDGER: &str = "_quasi_migrations";

/// A set of migrations and the ledger recording which of them ran.
pub struct Migrator {
    migrations: &'static [Migration],
    ledger: &'static str,
}

impl Migrator {
    /// A migrator over this table, recording to the default ledger.
    #[must_use]
    pub fn new(migrations: &'static [Migration]) -> Self {
        Self {
            migrations,
            ledger: DEFAULT_LEDGER,
        }
    }

    /// Record to a ledger of this name instead.
    ///
    /// For an app that has `sqlx` history: `.ledger("_sqlx_migrations")` makes
    /// this runner read the rows already there, find nothing pending, and do
    /// nothing. Writing a fresh ledger beside an existing one would instead
    /// make an upgraded install believe it had applied nothing and re-run every
    /// migration against a populated database.
    ///
    /// # Panics
    ///
    /// If the name is not a bare identifier. It is interpolated into SQL, since
    /// a table name cannot be bound as a parameter, and a startup literal is the
    /// right place to be strict about that.
    #[must_use]
    pub fn ledger(mut self, name: &'static str) -> Self {
        assert!(
            !name.is_empty()
                && name.chars().all(|c| c.is_ascii_alphanumeric() || c == '_')
                && !name.starts_with(|c: char| c.is_ascii_digit()),
            "ledger name `{name}` is not a bare identifier"
        );
        self.ledger = name;
        self
    }

    /// Create the ledger if absent.
    ///
    /// The DDL is `sqlx` 0.9's verbatim, which is what makes adopting an
    /// existing ledger a no-op rather than a conflicting `CREATE TABLE`.
    fn ensure_ledger(&self, conn: &Connection) -> Result<(), rusqlite::Error> {
        conn.execute_batch(&format!(
            "CREATE TABLE IF NOT EXISTS {} (
                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
            );",
            self.ledger
        ))
    }

    /// 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.
    ///
    /// # Errors
    ///
    /// If a migration fails, if an applied migration's file has changed since,
    /// or if the ledger carries a failed row from a previous run.
    #[tracing::instrument(skip_all)]
    pub fn run(&self, conn: &mut Connection) -> Result<(), MigrateError> {
        self.ensure_ledger(conn)?;

        // A `success = false` row means a previous run died between applying a
        // migration and committing its ledger row. This runner cannot write
        // one, since it commits both together, but an install upgraded from
        // `sqlx` may carry one and it still needs a person.
        let dirty: Option<i64> = conn
            .query_row(
                &format!(
                    "SELECT version FROM {} WHERE success = false ORDER BY version LIMIT 1",
                    self.ledger
                ),
                [],
                |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 {} ORDER BY version",
                self.ledger
            ))?;
            let rows = stmt.query_map([], |row| {
                Ok((row.get::<_, i64>(0)?, row.get::<_, Vec<u8>>(1)?))
            })?;
            rows.collect::<Result<_, _>>()?
        };

        for migration in self.migrations {
            let digest = checksum(migration.sql);

            if let Some(recorded) = applied.get(&migration.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: migration.version,
                        description: migration.description.to_owned(),
                    });
                }
                continue;
            }

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

        Ok(())
    }
}

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

    const FIRST: Migration = Migration {
        version: 1,
        description: "initial schema",
        sql: "CREATE TABLE note (id INTEGER PRIMARY KEY, body TEXT NOT NULL);",
    };

    const SECOND: Migration = Migration {
        version: 2,
        description: "archived flag",
        sql: "ALTER TABLE note ADD COLUMN archived INTEGER NOT NULL DEFAULT 0;",
    };

    static ONE: &[Migration] = &[FIRST];
    static BOTH: &[Migration] = &[FIRST, SECOND];

    fn columns(conn: &Connection) -> Vec<String> {
        let mut stmt = conn
            .prepare("SELECT name FROM pragma_table_info('note')")
            .unwrap();
        let rows = stmt.query_map([], |row| row.get::<_, String>(0)).unwrap();
        rows.collect::<Result<_, _>>().unwrap()
    }

    fn applied(conn: &Connection, ledger: &str) -> Vec<i64> {
        let mut stmt = conn
            .prepare(&format!("SELECT version FROM {ledger} ORDER BY version"))
            .unwrap();
        let rows = stmt.query_map([], |row| row.get::<_, i64>(0)).unwrap();
        rows.collect::<Result<_, _>>().unwrap()
    }

    #[test]
    fn an_empty_database_gets_every_migration() {
        let mut conn = Connection::open_in_memory().unwrap();
        Migrator::new(BOTH).run(&mut conn).unwrap();
        assert_eq!(columns(&conn), ["id", "body", "archived"]);
        assert_eq!(applied(&conn, DEFAULT_LEDGER), [1, 2]);
    }

    #[test]
    fn a_second_run_applies_nothing() {
        let mut conn = Connection::open_in_memory().unwrap();
        Migrator::new(BOTH).run(&mut conn).unwrap();
        // The second run is the one that would fail loudly if it re-applied:
        // `CREATE TABLE` on a table that exists.
        Migrator::new(BOTH).run(&mut conn).unwrap();
        assert_eq!(applied(&conn, DEFAULT_LEDGER), [1, 2]);
    }

    #[test]
    fn a_new_migration_applies_over_an_existing_database() {
        let mut conn = Connection::open_in_memory().unwrap();
        Migrator::new(ONE).run(&mut conn).unwrap();
        assert_eq!(columns(&conn), ["id", "body"]);

        Migrator::new(BOTH).run(&mut conn).unwrap();
        assert_eq!(columns(&conn), ["id", "body", "archived"]);
    }

    #[test]
    fn editing_a_shipped_migration_is_refused() {
        let mut conn = Connection::open_in_memory().unwrap();
        Migrator::new(ONE).run(&mut conn).unwrap();

        // Same version and description, one character of SQL different.
        static EDITED: &[Migration] = &[Migration {
            version: 1,
            description: "initial schema",
            sql: "CREATE TABLE note (id INTEGER PRIMARY KEY, body TEXT);",
        }];
        let refused = Migrator::new(EDITED).run(&mut conn).unwrap_err();
        assert!(matches!(
            refused,
            MigrateError::ChecksumMismatch { version: 1, .. }
        ));
    }

    #[test]
    fn a_failed_row_from_a_previous_run_stops_everything() {
        let mut conn = Connection::open_in_memory().unwrap();
        Migrator::new(ONE).run(&mut conn).unwrap();
        conn.execute(
            &format!(
                "INSERT INTO {DEFAULT_LEDGER} (version, description, success, checksum, execution_time)
                 VALUES (9, 'half done', FALSE, X'00', 0)"
            ),
            [],
        )
        .unwrap();

        let stopped = Migrator::new(BOTH).run(&mut conn).unwrap_err();
        assert!(matches!(stopped, MigrateError::Dirty(9)));
    }

    #[test]
    fn a_broken_migration_names_itself() {
        let mut conn = Connection::open_in_memory().unwrap();
        static BROKEN: &[Migration] = &[Migration {
            version: 1,
            description: "not sql",
            sql: "CREATE TABL note (id INTEGER);",
        }];

        let failed = Migrator::new(BROKEN).run(&mut conn).unwrap_err();
        let MigrateError::Apply {
            version,
            description,
            ..
        } = failed
        else {
            panic!("expected an apply failure");
        };
        assert_eq!(version, 1);
        assert_eq!(description, "not sql");

        // And nothing was recorded, so fixing the file and re-running works.
        assert_eq!(applied(&conn, DEFAULT_LEDGER), Vec::<i64>::new());
    }

    #[test]
    fn an_sqlx_ledger_is_adopted_rather_than_duplicated() {
        let mut conn = Connection::open_in_memory().unwrap();
        // What an install upgraded from sqlx looks like: the schema is there
        // and so are the rows, written by a library this app no longer links.
        conn.execute_batch(FIRST.sql).unwrap();
        conn.execute_batch(
            "CREATE TABLE _sqlx_migrations (
                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
            );",
        )
        .unwrap();
        conn.execute(
            "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time)
             VALUES (1, 'initial schema', TRUE, ?1, 0)",
            rusqlite::params![checksum(FIRST.sql)],
        )
        .unwrap();

        Migrator::new(BOTH)
            .ledger("_sqlx_migrations")
            .run(&mut conn)
            .unwrap();

        // Migration 1 was recognised as done and only 2 ran.
        assert_eq!(applied(&conn, "_sqlx_migrations"), [1, 2]);
        assert_eq!(columns(&conn), ["id", "body", "archived"]);
    }

    #[test]
    #[should_panic(expected = "not a bare identifier")]
    fn a_ledger_name_that_is_not_an_identifier_is_a_bug() {
        let _ = Migrator::new(ONE).ledger("ledger; DROP TABLE note");
    }
}