//! The schema migration runner.
//!
//! # Why the ledger table is still called `_sqlx_migrations`
//!
//! It is not a leftover. Every GoingsOn install in the field has that table with
//! one row per applied migration, written by `sqlx::migrate!` before the move to
//! rusqlite. Renaming it, or writing a fresh ledger beside it, would make an
//! upgraded install believe it had applied nothing and re-run all 64 migrations
//! against a populated database -- which fails at the first `CREATE TABLE`, and
//! would be worse if it didn't.
//!
//! So this runner adopts the existing ledger verbatim: same table name, same
//! columns, same semantics, and the same checksum function (sha384 over the raw
//! file bytes, verified against sqlx 0.9 in `migration_checksum_tests.rs`). An
//! upgraded install reads its own 64 rows, finds nothing pending, and does
//! nothing. That is the entire compatibility requirement, and it is why 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(())
}