//! Preparing a profile database before the webview opens it (#390, ADR 0021).
//!
//! tauri-plugin-sql migrates only the database registered with
//! `add_migrations` — the default profile, `simpl_resultat.db` (lib.rs) — and
//! drops that list after the first load. A secondary profile
//! (`profile_<id>.db`) is created from the consolidated schema of the version
//! that creates it, and until #390 never received a migration afterwards.
//!
//! This module catches such a database up with the very same `Migration` list,
//! keeping the history in the `_sqlx_migrations` table sqlx uses for the
//! default profile, and repairs a value an old bug left in every profile.
//! It works on a plain `rusqlite::Connection`, so every rule is tested against
//! real SQLite and the databases each released version created.

use rusqlite::{params, Connection, OpenFlags, TransactionBehavior};
use serde::Serialize;
use sha2::{Digest, Sha384};
use std::collections::BTreeSet;
use std::fs;
use std::path::Path;
use std::time::{Duration, Instant};
use tauri_plugin_sql::{Migration, MigrationKind};

use super::DEFAULT_PROFILE_DB;

/// Subfolder, next to the profile databases, that holds the copy taken before
/// a catch-up. Removed with the profile (`delete_profile_db`).
pub const MIGRATION_BACKUP_DIR: &str = "migration-backups";

/// The DDL sqlx 0.8 creates (sqlx-sqlite `migrate.rs`), so the default profile
/// and the secondary ones keep their history in one format.
const SQLX_MIGRATIONS_DDL: &str = "CREATE TABLE IF NOT EXISTS _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
);";

/// Until #390 the Transactions page bound the JS boolean `true`, which
/// tauri-plugin-sql stores as the TEXT 'true' (an INTEGER column keeps a text
/// that is not a number). Only `'true'` was ever written; any other text is
/// read as "not set", like `toSqlFlag` on the TypeScript side.
const REPAIR_MANUAL_FLAGS_SQL: &str = "UPDATE transactions \
     SET is_manually_categorized = CASE WHEN is_manually_categorized = 'true' THEN 1 ELSE 0 END \
     WHERE typeof(is_manually_categorized) = 'text'";

/// What a catch-up did, for the Journal.
#[derive(Debug, Default, Clone, PartialEq, Serialize)]
pub struct CatchUp {
    /// Highest version whose schema change the database already carried.
    pub floor: Option<i64>,
    /// Versions written to the history without being run: the database already
    /// had their effect (first history of a secondary profile, or a gap left by
    /// an interrupted run).
    pub recorded: Vec<i64>,
    /// Versions this run applied, in order.
    pub applied: Vec<i64>,
    /// Rows a version could only apply with foreign keys off, set back to NULL
    /// as their key declares (ON DELETE SET NULL), as (table, rows).
    pub fk_repaired: Vec<(String, i64)>,
    /// Foreign key violations the database already had, found by the final
    /// check, as (table, rows). Reported, never touched.
    pub fk_violations: Vec<(String, i64)>,
    /// File name of the copy taken before the first applied version.
    pub backup_file: Option<String>,
}

/// What `prepare_at` did to one profile database. Paths never leave this
/// module: the report names files only, since error-level Journal lines can be
/// attached to a feedback report.
#[derive(Debug, Default, Clone, PartialEq, Serialize)]
pub struct ProfilePreparation {
    pub catch_up: Option<CatchUp>,
    /// Why the catch-up stopped, when it did. The profile still opens.
    pub catch_up_error: Option<String>,
    /// Rows whose manual-categorization flag was repaired.
    pub repaired_flags: usize,
    pub elapsed_ms: u64,
}

fn read_err(e: rusqlite::Error) -> String {
    format!("schema_read_failed: {}", e)
}

fn table_exists(conn: &Connection, table: &str) -> Result<bool, String> {
    conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)",
        [table],
        |row| row.get(0),
    )
    .map_err(read_err)
}

fn column_exists(conn: &Connection, table: &str, column: &str) -> Result<bool, String> {
    conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name = ?2)",
        [table, column],
        |row| row.get(0),
    )
    .map_err(read_err)
}

/// Highest migration version whose schema change this database carries,
/// checked newest first. Each marker is a column or table that version
/// introduced and that no earlier state had.
///
/// Only databases created before #390 need it: since then a new profile gets
/// its history with its schema (`history_sql`). Those older databases form a
/// closed set, one state per consolidated schema a release shipped (fixtures
/// in `database/fixtures/`): v0.3.0-v0.8.3 → 6, v0.8.4 → 8, v0.9.0 and v0.9.1
/// → 10, v0.10.0-v0.14.0 → 16, v0.15.0-v0.17.x → 17. The 14, 12 and 9 markers
/// cover databases a development build created between two releases. Never
/// add a marker for a later migration: no database without a history can
/// carry it.
pub fn infer_floor(conn: &Connection) -> Result<i64, String> {
    // v17 — import format on the sources. `amount_mode` alone is no marker:
    // import_config_templates has carried it since v5.
    if column_exists(conn, "import_sources", "header_signature")? {
        return Ok(17);
    }
    // v15 added `kind`. The only released state carrying it (v0.10.0-v0.14.0)
    // was created with v16's data conversion already folded in.
    if column_exists(conn, "balance_accounts", "kind")? {
        return Ok(16);
    }
    if table_exists(conn, "balance_securities")? {
        return Ok(14);
    }
    // v12; v13 is an idempotent reclass and is simply re-run.
    if column_exists(conn, "balance_accounts", "vehicle_type")? {
        return Ok(12);
    }
    // v10; v0.9.1 (v11, an idempotent orphan cleanup) has the same schema.
    if column_exists(conn, "balance_categories", "asset_type")? {
        return Ok(10);
    }
    if table_exists(conn, "balance_categories")? {
        return Ok(9);
    }
    if column_exists(conn, "categories", "i18n_key")? {
        return Ok(8);
    }
    // v3 to v6, all in the consolidated schema multi-profile shipped with.
    if table_exists(conn, "import_config_templates")?
        && column_exists(conn, "categories", "is_inputable")?
        && column_exists(conn, "import_sources", "has_header")?
    {
        return Ok(6);
    }
    Err("unknown_layout".to_string())
}

fn checksum(sql: &str) -> Vec<u8> {
    Sha384::digest(sql.as_bytes()).to_vec()
}

fn recorded_versions(conn: &Connection) -> Result<BTreeSet<i64>, String> {
    if !table_exists(conn, "_sqlx_migrations")? {
        return Ok(BTreeSet::new());
    }
    let dirty: Option<i64> = conn
        .query_row(
            "SELECT MIN(version) FROM _sqlx_migrations WHERE success = 0",
            [],
            |row| row.get(0),
        )
        .map_err(read_err)?;
    if let Some(version) = dirty {
        // Same refusal as sqlx: a half-applied version needs a human.
        return Err(format!("dirty_history:v{}", version));
    }
    let mut stmt = conn
        .prepare("SELECT version FROM _sqlx_migrations")
        .map_err(read_err)?;
    let versions = stmt
        .query_map([], |row| row.get::<_, i64>(0))
        .map_err(read_err)?
        .collect::<Result<BTreeSet<i64>, _>>()
        .map_err(read_err)?;
    Ok(versions)
}

fn record(conn: &Connection, migration: &Migration, execution_time: i64) -> rusqlite::Result<usize> {
    conn.execute(
        "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) \
         VALUES (?1, ?2, 1, ?3, ?4)",
        params![
            migration.version,
            migration.description,
            checksum(migration.sql),
            execution_time
        ],
    )
}

/// One row of `PRAGMA foreign_key_check`: (table, rowid, foreign key id).
type Violation = (String, Option<i64>, i64);

fn violations(conn: &Connection) -> rusqlite::Result<Vec<Violation>> {
    let mut stmt = conn.prepare("PRAGMA foreign_key_check")?;
    let rows = stmt.query_map([], |row| Ok((row.get(0)?, row.get(1)?, row.get(3)?)))?;
    rows.collect()
}

fn count_by_table<'a>(tables: impl Iterator<Item = &'a str>) -> Vec<(String, i64)> {
    let mut counts: Vec<(String, i64)> = Vec::new();
    for table in tables {
        match counts.iter_mut().find(|(name, _)| name == table) {
            Some((_, n)) => *n += 1,
            None => counts.push((table.to_string(), 1)),
        }
    }
    counts.sort();
    counts
}

fn quote_ident(name: &str) -> String {
    format!("\"{}\"", name.replace('"', "\"\""))
}

fn is_foreign_key_failure(error: &rusqlite::Error) -> bool {
    matches!(
        error,
        rusqlite::Error::SqliteFailure(failure, _)
            if failure.extended_code == rusqlite::ffi::SQLITE_CONSTRAINT_FOREIGNKEY
    )
}

/// Undo the violations a version introduced, as each key declares: only
/// `ON DELETE SET NULL` can be repaired (the column goes back to NULL). Any
/// other new violation is an error, so the caller rolls the version back.
fn repair_new_violations(conn: &Connection, before: &BTreeSet<Violation>) -> Result<Vec<(String, i64)>, String> {
    let new: Vec<Violation> = violations(conn)
        .map_err(|e| e.to_string())?
        .into_iter()
        .filter(|v| !before.contains(v))
        .collect();
    for (table, rowid, fkid) in &new {
        let unrepairable = || format!("unrepairable foreign key in {}", table);
        let rowid = rowid.ok_or_else(unrepairable)?;
        let columns: Vec<(String, String)> = conn
            .prepare("SELECT \"from\", on_delete FROM pragma_foreign_key_list(?1) WHERE id = ?2")
            .and_then(|mut stmt| {
                stmt.query_map(params![table, fkid], |row| Ok((row.get(0)?, row.get(1)?)))?
                    .collect()
            })
            .map_err(|e| e.to_string())?;
        if columns.is_empty() || columns.iter().any(|(_, action)| action != "SET NULL") {
            return Err(unrepairable());
        }
        let assignments = columns
            .iter()
            .map(|(column, _)| format!("{} = NULL", quote_ident(column)))
            .collect::<Vec<_>>()
            .join(", ");
        conn.execute(
            &format!("UPDATE {} SET {} WHERE rowid = ?1", quote_ident(table), assignments),
            [rowid],
        )
        .map_err(|e| e.to_string())?;
    }
    let left = violations(conn)
        .map_err(|e| e.to_string())?
        .into_iter()
        .any(|v| !before.contains(&v));
    if left {
        return Err("foreign key violations left after the repair".to_string());
    }
    Ok(count_by_table(new.iter().map(|(table, _, _)| table.as_str())))
}

fn run_in_transaction(conn: &mut Connection, migration: &Migration, started: Instant) -> rusqlite::Result<()> {
    let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
    tx.execute_batch(migration.sql)?;
    record(&tx, migration, started.elapsed().as_nanos() as i64)?;
    tx.commit()
}

/// Second attempt of a version that failed on a foreign key: once, with
/// foreign keys off, then repair what it left. Foreign keys are back on
/// whatever happens.
fn retry_without_foreign_keys(
    conn: &mut Connection,
    migration: &Migration,
    started: Instant,
) -> Result<Vec<(String, i64)>, String> {
    conn.execute_batch("PRAGMA foreign_keys = OFF").map_err(|e| e.to_string())?;
    let attempt = (|| -> Result<Vec<(String, i64)>, String> {
        let before: BTreeSet<Violation> = violations(conn).map_err(|e| e.to_string())?.into_iter().collect();
        let tx = conn
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(|e| e.to_string())?;
        tx.execute_batch(migration.sql).map_err(|e| e.to_string())?;
        let repaired = repair_new_violations(&tx, &before)?;
        record(&tx, migration, started.elapsed().as_nanos() as i64).map_err(|e| e.to_string())?;
        tx.commit().map_err(|e| e.to_string())?;
        Ok(repaired)
    })();
    let restored = conn.execute_batch("PRAGMA foreign_keys = ON").map_err(|e| e.to_string());
    let repaired = attempt?;
    restored?;
    Ok(repaired)
}

fn apply_pending(conn: &mut Connection, pending: &[&Migration], report: &mut CatchUp) -> Result<(), String> {
    for migration in pending {
        let started = Instant::now();
        // Dropping the transaction on an error rolls this version back; the
        // earlier ones stay committed and recorded.
        match run_in_transaction(conn, migration, started) {
            Ok(()) => {}
            Err(e) if is_foreign_key_failure(&e) => {
                let repaired = retry_without_foreign_keys(conn, migration, started)
                    .map_err(|e| format!("migration_failed:v{}: {}", migration.version, e))?;
                for (table, rows) in repaired {
                    match report.fk_repaired.iter_mut().find(|(name, _)| *name == table) {
                        Some((_, n)) => *n += rows,
                        None => report.fk_repaired.push((table, rows)),
                    }
                }
            }
            Err(e) => return Err(format!("migration_failed:v{}: {}", migration.version, e)),
        }
        report.applied.push(migration.version);
    }
    Ok(())
}

/// Bring a secondary profile database to the last migration.
///
/// - `floor` comes from the schema itself, not from the history, so an
///   interrupted first run can never make a version below it look pending.
///   That is what keeps the destructive v2 seed (DELETE FROM categories) from
///   ever being replayed onto a populated database.
/// - `backup(conn, first_pending)` runs before the first write, and only when
///   something is pending. A failed backup stops everything.
/// - The history (table and every version up to the floor) is written in one
///   transaction; then each pending version runs in its own transaction and
///   the first failure stops the run, earlier versions kept.
/// - Foreign keys stay on, as sqlx keeps them for the default profile. A
///   version that fails on a foreign key is retried once without them, then
///   the rows it left pointing nowhere are set back to NULL when their key
///   declares ON DELETE SET NULL; anything else rolls it back. The case it
///   exists for: v7 adds subcategories under category 31, which a profile
///   whose categories were restored from a newer export no longer has.
pub fn catch_up<F>(conn: &mut Connection, migrations: &[Migration], backup: F) -> Result<CatchUp, String>
where
    F: FnOnce(&Connection, i64) -> Result<String, String>,
{
    let floor = infer_floor(conn)?;
    let recorded = recorded_versions(conn)?;
    let has_history = table_exists(conn, "_sqlx_migrations")?;

    let mut ups: Vec<&Migration> = migrations
        .iter()
        .filter(|m| matches!(m.kind, MigrationKind::Up))
        .collect();
    ups.sort_by_key(|m| m.version);
    let to_record: Vec<&Migration> = ups
        .iter()
        .copied()
        .filter(|m| m.version <= floor && !recorded.contains(&m.version))
        .collect();
    let pending: Vec<&Migration> = ups
        .iter()
        .copied()
        .filter(|m| m.version > floor && !recorded.contains(&m.version))
        .collect();

    // Second line of defence, unreachable while `infer_floor` never returns
    // below 6: the initial schema and the v2 seed must never run on a
    // database that already holds transactions.
    if pending.iter().any(|m| m.version <= 2) && table_exists(conn, "transactions")? {
        return Err("refused_initial_migrations".to_string());
    }

    let mut report = CatchUp {
        floor: Some(floor),
        ..CatchUp::default()
    };

    if let Some(first) = pending.first() {
        report.backup_file = Some(backup(conn, first.version)?);
    }

    if !has_history || !to_record.is_empty() {
        let tx = conn
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(|e| format!("history_write_failed: {}", e))?;
        tx.execute_batch(SQLX_MIGRATIONS_DDL)
            .map_err(|e| format!("history_write_failed: {}", e))?;
        for migration in &to_record {
            record(&tx, migration, 0).map_err(|e| format!("history_write_failed: {}", e))?;
        }
        tx.commit().map_err(|e| format!("history_write_failed: {}", e))?;
        report.recorded = to_record.iter().map(|m| m.version).collect();
    }

    if pending.is_empty() {
        return Ok(report);
    }

    // Foreign keys on, as sqlx runs the default profile's migrations.
    conn.execute_batch("PRAGMA foreign_keys = ON")
        .map_err(|e| format!("pragma_failed: {}", e))?;
    if let Err(e) = apply_pending(conn, &pending, &mut report) {
        return Err(if report.applied.is_empty() {
            e
        } else {
            format!("{} (applied before stopping: {:?})", e, report.applied)
        });
    }
    let left = violations(conn).map_err(read_err)?;
    report.fk_violations = count_by_table(left.iter().map(|(table, _, _)| table.as_str()));
    Ok(report)
}

/// Normalize the manual-categorization flag of every transaction stored as
/// text. Returns the number of rows changed; 0 on a database without the table.
pub fn repair_manual_flags(conn: &Connection) -> Result<usize, String> {
    if !table_exists(conn, "transactions")? {
        return Ok(0);
    }
    conn.execute(REPAIR_MANUAL_FLAGS_SQL, [])
        .map_err(|e| format!("flag_repair_failed: {}", e))
}

/// Copy the database to `<dir>/<stem>-before-v<first_pending>.db` and return
/// the file name. The copy is written as `.tmp` and renamed only once
/// complete, so an existing final file is trusted as is and kept; a `.tmp`
/// left by an interrupted run is replaced. The path is bound, never formatted
/// into the SQL, so a folder such as `C:\Users\L'Heureux` works.
pub fn backup_to(conn: &Connection, dir: &Path, stem: &str, first_pending: i64) -> Result<String, String> {
    let name = format!("{}-before-v{}.db", stem, first_pending);
    let final_path = dir.join(&name);
    if final_path.exists() {
        return Ok(name);
    }
    fs::create_dir_all(dir).map_err(|e| format!("backup_failed: {}", e))?;
    let tmp_path = dir.join(format!("{}.tmp", name));
    if tmp_path.exists() {
        fs::remove_file(&tmp_path).map_err(|e| format!("backup_failed: {}", e))?;
    }
    let tmp = tmp_path
        .to_str()
        .ok_or_else(|| "backup_failed: path is not valid UTF-8".to_string())?;
    conn.execute("VACUUM INTO ?1", [tmp])
        .map_err(|e| format!("backup_failed: {}", e))?;
    // VACUUM INTO does not sync its output; the rename must not publish a
    // copy a power cut could still truncate. Write access: Windows refuses
    // to flush a file opened read-only.
    fs::OpenOptions::new()
        .write(true)
        .open(&tmp_path)
        .and_then(|file| file.sync_all())
        .map_err(|e| format!("backup_failed: {}", e))?;
    fs::rename(&tmp_path, &final_path).map_err(|e| format!("backup_failed: {}", e))?;
    #[cfg(unix)]
    if let Ok(folder) = fs::File::open(dir) {
        // Best effort: makes the rename itself durable.
        let _ = folder.sync_all();
    }
    Ok(name)
}

/// SQL that writes the complete history of a database created from the
/// current consolidated schema. `get_new_profile_init_sql` returns it right
/// after the schema, so a new profile is born with its history and never goes
/// through `infer_floor`. `INSERT OR IGNORE`: a version already recorded keeps
/// its row.
pub fn history_sql(migrations: &[Migration]) -> String {
    let mut sql = String::from(SQLX_MIGRATIONS_DDL);
    for migration in migrations.iter().filter(|m| matches!(m.kind, MigrationKind::Up)) {
        let hex: String = checksum(migration.sql).iter().map(|b| format!("{:02X}", b)).collect();
        sql.push_str(&format!(
            "\nINSERT OR IGNORE INTO _sqlx_migrations (version, description, success, checksum, execution_time) \
             VALUES ({}, '{}', 1, X'{}', 0);",
            migration.version,
            migration.description.replace('\'', "''"),
            hex
        ));
    }
    sql
}

/// Replace the profile folder in a message with a placeholder. The folder
/// holds the OS user name, and error-level Journal lines can be attached to a
/// feedback report (Loi 25).
pub fn scrub(message: String, dir: &Path) -> String {
    let full = dir.to_string_lossy();
    if full.is_empty() {
        return message;
    }
    message.replace(full.as_ref(), "<profile folder>")
}

/// Remove the catch-up copies of one profile (and any `.tmp` left behind).
/// Returns how many files were removed.
pub fn remove_backups(dir: &Path, stem: &str) -> Result<usize, String> {
    let entries = match fs::read_dir(dir) {
        Ok(entries) => entries,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(0),
        Err(e) => return Err(format!("backup_cleanup_failed: {}", e)),
    };
    let prefix = format!("{}-before-v", stem);
    let mut removed = 0;
    for entry in entries {
        let entry = entry.map_err(|e| format!("backup_cleanup_failed: {}", e))?;
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        if name.starts_with(&prefix) && (name.ends_with(".db") || name.ends_with(".db.tmp")) {
            fs::remove_file(entry.path()).map_err(|e| format!("backup_cleanup_failed: {}", e))?;
            removed += 1;
        }
    }
    Ok(removed)
}

/// Prepare one profile database before tauri-plugin-sql opens it.
///
/// A missing file is left alone: never create a database here, the plugin
/// does that on first load. The default profile is never caught up here (the
/// plugin migrates it); every profile gets the flag repair. A catch-up that
/// stops is reported in `catch_up_error` and the profile still opens, like it
/// did before #390; only a database that cannot be opened is an error.
pub fn prepare_at(db_path: &Path, db_filename: &str, migrations: &[Migration]) -> Result<ProfilePreparation, String> {
    let started = Instant::now();
    let mut report = ProfilePreparation::default();
    if !db_path.exists() {
        return Ok(report);
    }
    let dir = db_path
        .parent()
        .ok_or_else(|| "open_failed: database has no parent folder".to_string())?;
    let mut conn = Connection::open_with_flags(
        db_path,
        OpenFlags::SQLITE_OPEN_READ_WRITE | OpenFlags::SQLITE_OPEN_URI | OpenFlags::SQLITE_OPEN_NO_MUTEX,
    )
    .map_err(|e| scrub(format!("open_failed: {}", e), dir))?;
    conn.busy_timeout(Duration::from_secs(5))
        .map_err(|e| scrub(format!("open_failed: {}", e), dir))?;

    if db_filename != DEFAULT_PROFILE_DB {
        let backup_dir = dir.join(MIGRATION_BACKUP_DIR);
        let stem = db_filename.trim_end_matches(".db");
        match catch_up(&mut conn, migrations, |conn, first| backup_to(conn, &backup_dir, stem, first)) {
            Ok(done) => report.catch_up = Some(done),
            Err(e) => report.catch_up_error = Some(scrub(e, dir)),
        }
    }

    report.repaired_flags = repair_manual_flags(&conn).map_err(|e| scrub(e, dir))?;
    report.elapsed_ms = started.elapsed().as_millis() as u64;
    Ok(report)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeMap;
    use std::path::PathBuf;
    use std::sync::atomic::{AtomicUsize, Ordering};

    // The SQL each release ran to create a secondary profile
    // (`get_new_profile_init_sql` of that tag), copied verbatim from git.
    const CONSOLIDATED_V0_3_0: &str = include_str!("fixtures/consolidated_v0.3.0.sql");
    const SEED_V0_3_0: &str = include_str!("fixtures/seed_v0.3.0.sql");
    const SEED_V0_4_0: &str = include_str!("fixtures/seed_v0.4.0.sql");
    const CONSOLIDATED_V0_8_4: &str = include_str!("fixtures/consolidated_v0.8.4.sql");
    const CONSOLIDATED_V0_9_0: &str = include_str!("fixtures/consolidated_v0.9.0.sql");
    const CONSOLIDATED_V0_9_1: &str = include_str!("fixtures/consolidated_v0.9.1.sql");
    const CONSOLIDATED_V0_10_0: &str = include_str!("fixtures/consolidated_v0.10.0.sql");

    const SECONDARY: &str = "profile_0a1b2c3d.db";

    struct State {
        name: &'static str,
        init: Vec<&'static str>,
        floor: i64,
    }

    fn released_states() -> Vec<State> {
        vec![
            State { name: "A1 v0.3.0-v0.8.3", init: vec![CONSOLIDATED_V0_3_0, SEED_V0_3_0], floor: 6 },
            State { name: "A2 v0.4.0", init: vec![CONSOLIDATED_V0_3_0, SEED_V0_4_0], floor: 6 },
            State { name: "B v0.8.4", init: vec![CONSOLIDATED_V0_8_4], floor: 8 },
            State { name: "C v0.9.0", init: vec![CONSOLIDATED_V0_9_0], floor: 10 },
            State { name: "D v0.9.1", init: vec![CONSOLIDATED_V0_9_1], floor: 10 },
            State { name: "E v0.10.0-v0.14.0", init: vec![CONSOLIDATED_V0_10_0], floor: 16 },
        ]
    }

    fn migrations() -> Vec<Migration> {
        crate::migrations()
    }

    fn max_version() -> i64 {
        migrations().iter().map(|m| m.version).max().unwrap()
    }

    fn versions(from: i64, to: i64) -> Vec<i64> {
        (from..=to).collect()
    }

    /// Cargo runs tests in parallel: every test gets its own folder.
    fn unique_dir(tag: &str) -> PathBuf {
        static COUNTER: AtomicUsize = AtomicUsize::new(0);
        let dir = std::env::temp_dir().join(format!(
            "simpl-resultat-profile-db-{}-{}-{}",
            std::process::id(),
            tag,
            COUNTER.fetch_add(1, Ordering::SeqCst)
        ));
        let _ = fs::remove_dir_all(&dir);
        fs::create_dir_all(&dir).unwrap();
        dir
    }

    fn create_db(dir: &Path, name: &str, init: &[&str]) -> PathBuf {
        let path = dir.join(name);
        let conn = Connection::open(&path).unwrap();
        for sql in init {
            conn.execute_batch(sql).unwrap();
        }
        path
    }

    fn open(path: &Path) -> Connection {
        Connection::open(path).unwrap()
    }

    fn count(conn: &Connection, sql: &str) -> i64 {
        conn.query_row(sql, [], |row| row.get(0)).unwrap()
    }

    fn has_table(conn: &Connection, table: &str) -> bool {
        table_exists(conn, table).unwrap()
    }

    fn no_backup(_: &Connection, _: i64) -> Result<String, String> {
        Ok("not-written.db".to_string())
    }

    /// One source whose mapping reads debit/credit columns, one file, and one
    /// transaction classified by hand while the bug stood (TEXT 'true').
    /// Returns the transaction's category.
    fn seed_user_rows(conn: &Connection) -> i64 {
        conn.execute(
            "INSERT INTO import_sources (name, date_format, delimiter, encoding, column_mapping, skip_lines, has_header) \
             VALUES ('Visa', 'YYYY-MM-DD', ',', 'utf-8', \
                     '{\"date\":0,\"description\":1,\"debitAmount\":2,\"creditAmount\":3}', 0, 1)",
            [],
        )
        .unwrap();
        let source = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO imported_files (source_id, filename, file_hash, row_count, status) \
             VALUES (?1, 'visa.csv', 'hash', 1, 'completed')",
            [source],
        )
        .unwrap();
        let file = conn.last_insert_rowid();
        let category = count(conn, "SELECT MIN(id) FROM categories");
        conn.execute(
            "INSERT INTO transactions (date, description, amount, category_id, source_id, file_id, is_manually_categorized) \
             VALUES ('2026-03-02', 'IGA', -42.5, ?1, ?2, ?3, 'true')",
            params![category, source, file],
        )
        .unwrap();
        category
    }

    /// An envelope account in the old `tfsa` category, a stock account with a
    /// priced line, and an orphan snapshot: what v11, v12, v13, v15 and v16
    /// exist to transform.
    fn seed_balance_rows(conn: &Connection) {
        conn.execute(
            "INSERT INTO balance_accounts (balance_category_id, name, currency) \
             VALUES ((SELECT id FROM balance_categories WHERE key = 'tfsa'), 'CELI', 'CAD')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO balance_accounts (balance_category_id, name, symbol, currency) \
             VALUES ((SELECT id FROM balance_categories WHERE key = 'stock'), 'Apple', ' aapl ', 'CAD')",
            [],
        )
        .unwrap();
        let apple = conn.last_insert_rowid();
        conn.execute("INSERT INTO balance_snapshots (snapshot_date) VALUES ('2026-04-30')", [])
            .unwrap();
        let snapshot = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO balance_snapshot_lines (snapshot_id, account_id, quantity, unit_price, value) \
             VALUES (?1, ?2, 10, 150, 1500)",
            params![snapshot, apple],
        )
        .unwrap();
        conn.execute("INSERT INTO balance_snapshots (snapshot_date) VALUES ('2026-03-31')", [])
            .unwrap();
    }

    // -- schema comparison --------------------------------------------------

    type ColumnShape = (String, String, i64, Option<String>, i64);
    type ForeignKeyShape = (String, String, Option<String>, String, String);
    type IndexShape = (String, i64, Vec<String>);

    #[derive(Debug, PartialEq)]
    struct Shape {
        columns: BTreeMap<String, BTreeSet<ColumnShape>>,
        foreign_keys: BTreeMap<String, BTreeSet<ForeignKeyShape>>,
        indexes: BTreeMap<String, BTreeSet<IndexShape>>,
    }

    /// Tables, columns (as a set: ALTER appends where the consolidated schema
    /// declares mid-table), foreign keys and indexes. Automatic index names
    /// depend on table history, so they compare by origin and columns.
    fn shape(conn: &Connection) -> Shape {
        let tables: Vec<String> = conn
            .prepare(
                "SELECT name FROM sqlite_master WHERE type = 'table' \
                 AND name NOT LIKE 'sqlite_%' AND name <> '_sqlx_migrations' ORDER BY name",
            )
            .unwrap()
            .query_map([], |row| row.get(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        let mut result = Shape {
            columns: BTreeMap::new(),
            foreign_keys: BTreeMap::new(),
            indexes: BTreeMap::new(),
        };
        for table in tables {
            let columns: BTreeSet<ColumnShape> = conn
                .prepare("SELECT name, type, \"notnull\", dflt_value, pk FROM pragma_table_info(?1)")
                .unwrap()
                .query_map([&table], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)))
                .unwrap()
                .collect::<Result<_, _>>()
                .unwrap();
            let foreign_keys: BTreeSet<ForeignKeyShape> = conn
                .prepare(
                    "SELECT \"table\", \"from\", \"to\", on_update, on_delete FROM pragma_foreign_key_list(?1)",
                )
                .unwrap()
                .query_map([&table], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)))
                .unwrap()
                .collect::<Result<_, _>>()
                .unwrap();
            let listed: Vec<(String, i64, String)> = conn
                .prepare("SELECT name, \"unique\", origin FROM pragma_index_list(?1)")
                .unwrap()
                .query_map([&table], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
                .unwrap()
                .collect::<Result<_, _>>()
                .unwrap();
            let mut indexes = BTreeSet::new();
            for (name, unique, origin) in listed {
                let columns: Vec<String> = conn
                    .prepare("SELECT name FROM pragma_index_info(?1) ORDER BY seqno")
                    .unwrap()
                    .query_map([&name], |r| r.get::<_, Option<String>>(0))
                    .unwrap()
                    .map(|c| c.unwrap().unwrap_or_else(|| "<expr>".to_string()))
                    .collect();
                let key = if origin == "c" { name } else { origin };
                indexes.insert((key, unique, columns));
            }
            result.columns.insert(table.clone(), columns);
            result.foreign_keys.insert(table.clone(), foreign_keys);
            result.indexes.insert(table, indexes);
        }
        result
    }

    fn fresh_profile_shape() -> Shape {
        let dir = unique_dir("reference");
        let path = create_db(&dir, SECONDARY, &[crate::database::CONSOLIDATED_SCHEMA]);
        shape(&open(&path))
    }

    /// Every version recorded once, applied, with the checksum sqlx computes.
    fn assert_history_complete(conn: &Connection, context: &str) {
        let rows: Vec<(i64, i64, Vec<u8>)> = conn
            .prepare("SELECT version, success, checksum FROM _sqlx_migrations ORDER BY version")
            .unwrap()
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        let expected: Vec<(i64, i64, Vec<u8>)> = migrations()
            .iter()
            .map(|m| (m.version, 1, checksum(m.sql)))
            .collect();
        assert_eq!(rows, expected, "{}", context);
    }

    // -- the released states ------------------------------------------------

    #[test]
    fn every_released_state_infers_its_floor() {
        for state in released_states() {
            let dir = unique_dir("floor");
            let path = create_db(&dir, SECONDARY, &state.init);
            assert_eq!(infer_floor(&open(&path)), Ok(state.floor), "{}", state.name);
        }
    }

    #[test]
    fn every_released_state_catches_up_to_the_consolidated_schema() {
        let reference = fresh_profile_shape();
        let max = max_version();
        for state in released_states() {
            let dir = unique_dir("released");
            let path = create_db(&dir, SECONDARY, &state.init);
            let category = seed_user_rows(&open(&path));

            let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

            assert_eq!(report.catch_up_error, None, "{}", state.name);
            let done = report.catch_up.clone().unwrap();
            assert_eq!(done.floor, Some(state.floor), "{}", state.name);
            assert_eq!(done.recorded, versions(1, state.floor), "{}", state.name);
            assert_eq!(done.applied, versions(state.floor + 1, max), "{}", state.name);
            assert_eq!(done.fk_repaired, Vec::<(String, i64)>::new(), "{}", state.name);
            assert_eq!(done.fk_violations, Vec::<(String, i64)>::new(), "{}", state.name);
            let expected_backup = format!("profile_0a1b2c3d-before-v{}.db", state.floor + 1);
            assert_eq!(done.backup_file.as_deref(), Some(expected_backup.as_str()), "{}", state.name);
            assert_eq!(report.repaired_flags, 1, "{}", state.name);

            let conn = open(&path);
            assert_eq!(shape(&conn), reference, "{}", state.name);
            assert_history_complete(&conn, state.name);
            let flag: (i64, String, i64) = conn
                .query_row(
                    "SELECT is_manually_categorized, typeof(is_manually_categorized), category_id FROM transactions",
                    [],
                    |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
                )
                .unwrap();
            assert_eq!(flag, (1, "integer".to_string(), category), "{}", state.name);
            let mode: String = conn
                .query_row("SELECT amount_mode FROM import_sources WHERE name = 'Visa'", [], |r| r.get(0))
                .unwrap();
            assert_eq!(mode, "debit_credit", "{}: v17 backfill", state.name);
            assert!(dir.join(MIGRATION_BACKUP_DIR).join(&expected_backup).exists(), "{}", state.name);
        }
    }

    #[test]
    fn a_profile_from_v0_15_to_v0_17_gets_its_history_and_nothing_else() {
        // State F: created with the current consolidated schema, before #390
        // wrote a history. Nothing to apply, so no copy either; the flag is
        // still repaired.
        let dir = unique_dir("state-f");
        let path = create_db(&dir, SECONDARY, &[crate::database::CONSOLIDATED_SCHEMA]);
        let category = seed_user_rows(&open(&path));

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report.catch_up_error, None);
        let done = report.catch_up.unwrap();
        assert_eq!(done.floor, Some(17));
        assert_eq!(done.recorded, versions(1, max_version()));
        assert!(done.applied.is_empty());
        assert_eq!(done.backup_file, None);
        assert_eq!(report.repaired_flags, 1);
        let conn = open(&path);
        assert_history_complete(&conn, "state F");
        assert_eq!(count(&conn, "SELECT category_id FROM transactions"), category);
        assert!(!dir.join(MIGRATION_BACKUP_DIR).exists());
    }

    #[test]
    fn the_bilan_data_of_a_v0_9_profile_is_converted() {
        for state in released_states().into_iter().filter(|s| s.floor == 10) {
            let dir = unique_dir("bilan");
            let path = create_db(&dir, SECONDARY, &state.init);
            seed_balance_rows(&open(&path));

            let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();
            assert_eq!(report.catch_up_error, None, "{}", state.name);

            let conn = open(&path);
            // v11: the snapshot without lines is gone.
            assert_eq!(count(&conn, "SELECT COUNT(*) FROM balance_snapshots"), 1, "{}", state.name);
            // v12 then v13: the envelope moves onto the account, which leaves the
            // deprecated `tfsa` category for `other`.
            let celi: (String, String) = conn
                .query_row(
                    "SELECT a.vehicle_type, c.key FROM balance_accounts a \
                     JOIN balance_categories c ON c.id = a.balance_category_id WHERE a.name = 'CELI'",
                    [],
                    |r| Ok((r.get(0)?, r.get(1)?)),
                )
                .unwrap();
            assert_eq!(celi, ("tfsa".to_string(), "other".to_string()), "{}", state.name);
            assert_eq!(
                count(&conn, "SELECT COUNT(*) FROM balance_categories WHERE key IN ('tfsa','rrsp') AND is_active = 1"),
                0,
                "{}",
                state.name
            );
            // v15 then v16: the priced account becomes detailed, holding one position.
            let kind: String = conn
                .query_row("SELECT kind FROM balance_accounts WHERE name = 'Apple'", [], |r| r.get(0))
                .unwrap();
            assert_eq!(kind, "detailed", "{}", state.name);
            let holding: (String, String, f64, f64, f64) = conn
                .query_row(
                    "SELECT s.symbol, s.asset_type, h.quantity, h.unit_price, h.value \
                     FROM balance_snapshot_holdings h JOIN balance_securities s ON s.id = h.security_id",
                    [],
                    |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?)),
                )
                .unwrap();
            assert_eq!(holding, ("AAPL".to_string(), "stock".to_string(), 10.0, 150.0, 1500.0), "{}", state.name);
            let line: (Option<f64>, Option<f64>, f64) = conn
                .query_row("SELECT quantity, unit_price, value FROM balance_snapshot_lines", [], |r| {
                    Ok((r.get(0)?, r.get(1)?, r.get(2)?))
                })
                .unwrap();
            assert_eq!(line, (None, None, 1500.0), "{}", state.name);
        }
    }

    #[test]
    fn development_states_catch_up_too() {
        let all = migrations();
        let run = |conn: &Connection, wanted: &[i64]| {
            for version in wanted {
                let migration = all.iter().find(|m| m.version == *version).unwrap();
                conn.execute_batch(migration.sql).unwrap();
            }
        };
        let reference = fresh_profile_shape();
        let max = max_version();
        let cases: Vec<(&str, &str, Vec<i64>, i64)> = vec![
            ("v0.8.4 + v9", CONSOLIDATED_V0_8_4, vec![9], 9),
            ("v0.9.0 + v12", CONSOLIDATED_V0_9_0, vec![12], 12),
            ("v0.9.0 + v12..v14", CONSOLIDATED_V0_9_0, vec![12, 13, 14], 14),
        ];
        for (name, init, extra, floor) in cases {
            let dir = unique_dir("dev");
            let path = create_db(&dir, SECONDARY, &[init]);
            run(&open(&path), &extra);

            let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

            assert_eq!(report.catch_up_error, None, "{}", name);
            let done = report.catch_up.unwrap();
            assert_eq!(done.floor, Some(floor), "{}", name);
            assert_eq!(done.applied, versions(floor + 1, max), "{}", name);
            let conn = open(&path);
            assert_eq!(shape(&conn), reference, "{}", name);
            assert_history_complete(&conn, name);
        }
    }

    #[test]
    fn a_new_profile_is_born_with_its_history() {
        // What createProfile runs, statement by statement, through the plugin.
        let dir = unique_dir("new");
        let init = crate::commands::get_new_profile_init_sql().unwrap();
        let statements: Vec<&str> = init.iter().map(String::as_str).collect();
        let path = create_db(&dir, SECONDARY, &statements);
        let mut conn = open(&path);
        assert_history_complete(&conn, "new profile");

        let report = catch_up(&mut conn, &migrations(), |_: &Connection, _: i64| -> Result<String, String> {
            panic!("a new profile needs no backup")
        })
        .unwrap();

        assert!(report.recorded.is_empty());
        assert!(report.applied.is_empty());
        assert_eq!(report.backup_file, None);
        assert!(!dir.join(MIGRATION_BACKUP_DIR).exists());
    }

    #[test]
    fn the_legacy_floor_stops_at_v17() {
        // `infer_floor` serves only databases created before #390. The current
        // consolidated schema must still carry the newest legacy marker; a
        // later migration needs no marker of its own.
        let dir = unique_dir("legacy-top");
        let path = create_db(&dir, SECONDARY, &[crate::database::CONSOLIDATED_SCHEMA]);
        assert_eq!(infer_floor(&open(&path)), Ok(17));
    }

    #[test]
    fn history_sql_escapes_descriptions() {
        let list = vec![Migration {
            version: 1,
            description: "the user's file",
            sql: "SELECT 1;",
            kind: MigrationKind::Up,
        }];
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(&history_sql(&list)).unwrap();
        let description: String = conn
            .query_row("SELECT description FROM _sqlx_migrations", [], |r| r.get(0))
            .unwrap();
        assert_eq!(description, "the user's file");
    }

    #[test]
    fn scrub_removes_the_profile_folder() {
        let dir = Path::new("/home/someone/.config/com.simpl.resultat");
        let message = "backup_failed: unable to open database: /home/someone/.config/com.simpl.resultat/migration-backups/x.tmp".to_string();
        assert_eq!(
            scrub(message, dir),
            "backup_failed: unable to open database: <profile folder>/migration-backups/x.tmp"
        );
    }

    #[test]
    fn a_second_open_changes_nothing() {
        for state in released_states() {
            let dir = unique_dir("twice");
            let path = create_db(&dir, SECONDARY, &state.init);
            seed_user_rows(&open(&path));
            prepare_at(&path, SECONDARY, &migrations()).unwrap();

            let again = prepare_at(&path, SECONDARY, &migrations()).unwrap();

            assert_eq!(again.catch_up_error, None, "{}", state.name);
            let done = again.catch_up.unwrap();
            assert!(done.recorded.is_empty(), "{}", state.name);
            assert!(done.applied.is_empty(), "{}", state.name);
            assert_eq!(done.backup_file, None, "{}", state.name);
            assert_eq!(again.repaired_flags, 0, "{}", state.name);
            assert_eq!(fs::read_dir(dir.join(MIGRATION_BACKUP_DIR)).unwrap().count(), 1, "{}", state.name);
        }
    }

    // -- an interrupted first run -------------------------------------------

    #[test]
    fn an_interrupted_first_run_never_replays_the_seed() {
        // A history table without its rows is what a crash between the CREATE
        // and the INSERTs of a non-atomic first write would leave. Replaying v2
        // would empty the categories and every transaction's category.
        for partial in [vec![], vec![1_i64, 2, 3]] {
            let dir = unique_dir("partial");
            let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_3_0, SEED_V0_3_0]);
            let category;
            {
                let conn = open(&path);
                category = seed_user_rows(&conn);
                conn.execute(
                    "INSERT INTO categories (id, name, type, sort_order) VALUES (9001, 'Mine', 'expense', 99)",
                    [],
                )
                .unwrap();
                conn.execute_batch(SQLX_MIGRATIONS_DDL).unwrap();
                for migration in migrations().iter().filter(|m| partial.contains(&m.version)) {
                    record(&conn, migration, 0).unwrap();
                }
            }
            let categories_before = count(&open(&path), "SELECT COUNT(*) FROM categories");

            let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

            assert_eq!(report.catch_up_error, None, "{:?}", partial);
            let done = report.catch_up.unwrap();
            let expected_recorded: Vec<i64> = versions(1, 6).into_iter().filter(|v| !partial.contains(v)).collect();
            assert_eq!(done.recorded, expected_recorded, "{:?}", partial);
            assert_eq!(done.applied, versions(7, max_version()), "{:?}", partial);
            let conn = open(&path);
            // v7 adds the three insurance subcategories, nothing is deleted.
            assert_eq!(count(&conn, "SELECT COUNT(*) FROM categories"), categories_before + 3, "{:?}", partial);
            assert_eq!(count(&conn, "SELECT COUNT(*) FROM categories WHERE id = 9001"), 1, "{:?}", partial);
            assert_eq!(count(&conn, "SELECT category_id FROM transactions"), category, "{:?}", partial);
            assert_history_complete(&conn, "partial");
        }
    }

    #[test]
    fn a_profile_without_category_31_still_catches_up() {
        // A restore of a newer export replaces the v2 taxonomy, so v7's
        // subcategories have no parent to point at. v7 is retried without
        // foreign keys, and its three rows become root categories, as the key's
        // ON DELETE SET NULL declares.
        let dir = unique_dir("no31");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_3_0, SEED_V0_3_0]);
        open(&path)
            .execute_batch(
                "DELETE FROM keywords; DELETE FROM categories; \
                 INSERT INTO categories (id, name, type, sort_order) VALUES (1000, 'Logement', 'expense', 1);",
            )
            .unwrap();

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report.catch_up_error, None);
        let done = report.catch_up.unwrap();
        assert_eq!(done.applied, versions(7, max_version()));
        assert_eq!(done.fk_repaired, vec![("categories".to_string(), 3)]);
        assert_eq!(done.fk_violations, Vec::<(String, i64)>::new());
        let conn = open(&path);
        assert_history_complete(&conn, "no31");
        assert!(violations(&conn).unwrap().is_empty());
        assert_eq!(
            count(&conn, "SELECT COUNT(*) FROM categories WHERE id IN (310, 311, 312) AND parent_id IS NULL"),
            3
        );
        // What a restore of this profile's own export does: every category
        // inserted in id order, foreign keys on.
        let create_sql: String = conn
            .query_row("SELECT sql FROM sqlite_master WHERE name = 'categories'", [], |r| r.get(0))
            .unwrap();
        let copy = Connection::open_in_memory().unwrap();
        copy.execute_batch(&format!("PRAGMA foreign_keys = ON; {};", create_sql)).unwrap();
        let rows: Vec<(i64, String, Option<i64>, String)> = conn
            .prepare("SELECT id, name, parent_id, type FROM categories ORDER BY id")
            .unwrap()
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        for (id, name, parent, kind) in rows {
            copy.execute(
                "INSERT INTO categories (id, name, parent_id, type) VALUES (?1, ?2, ?3, ?4)",
                params![id, name, parent, kind],
            )
            .unwrap();
        }
    }

    #[test]
    fn a_foreign_key_that_cannot_be_set_to_null_rolls_the_version_back() {
        let dir = unique_dir("unrepairable");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        let mut list: Vec<Migration> = migrations().into_iter().filter(|m| m.version <= 16).collect();
        // Both keys of a snapshot line are CASCADE / RESTRICT, never SET NULL.
        list.push(Migration {
            version: 17,
            description: "points nowhere",
            sql: "INSERT INTO balance_snapshot_lines (snapshot_id, account_id, value) VALUES (999, 999, 1);",
            kind: MigrationKind::Up,
        });
        let mut conn = open(&path);

        let error = catch_up(&mut conn, &list, no_backup).unwrap_err();

        assert!(error.starts_with("migration_failed:v17: unrepairable foreign key"), "{}", error);
        assert_eq!(count(&conn, "SELECT COUNT(*) FROM balance_snapshot_lines"), 0);
        assert_eq!(count(&conn, "SELECT MAX(version) FROM _sqlx_migrations"), 16);
        let foreign_keys: i64 = conn.query_row("PRAGMA foreign_keys", [], |r| r.get(0)).unwrap();
        assert_eq!(foreign_keys, 1);
    }

    #[test]
    fn an_unknown_layout_is_left_untouched() {
        let dir = unique_dir("unknown");
        let path = create_db(&dir, SECONDARY, &["CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT);"]);

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report.catch_up_error.as_deref(), Some("unknown_layout"));
        assert_eq!(report.repaired_flags, 0);
        let conn = open(&path);
        assert!(!has_table(&conn, "_sqlx_migrations"));
        assert!(!dir.join(MIGRATION_BACKUP_DIR).exists());
    }

    #[test]
    fn a_dirty_history_is_refused() {
        let dir = unique_dir("dirty");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        open(&path)
            .execute_batch(&format!(
                "{} INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) \
                 VALUES (17, 'half', 0, x'00', -1);",
                SQLX_MIGRATIONS_DDL
            ))
            .unwrap();

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report.catch_up_error.as_deref(), Some("dirty_history:v17"));
        assert!(!column_exists(&open(&path), "import_sources", "header_signature").unwrap());
    }

    #[test]
    fn a_failing_version_stops_the_run_and_keeps_the_earlier_ones() {
        let dir = unique_dir("failing");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        let mut list: Vec<Migration> = migrations().into_iter().filter(|m| m.version <= 16).collect();
        list.push(Migration {
            version: 17,
            description: "works",
            sql: "CREATE TABLE ok_one (x INTEGER);",
            kind: MigrationKind::Up,
        });
        list.push(Migration {
            version: 18,
            description: "breaks halfway",
            sql: "CREATE TABLE half_done (x INTEGER); INSERT INTO no_such_table VALUES (1);",
            kind: MigrationKind::Up,
        });
        let mut conn = open(&path);

        let error = catch_up(&mut conn, &list, no_backup).unwrap_err();

        assert!(error.starts_with("migration_failed:v18"), "{}", error);
        assert!(error.ends_with("(applied before stopping: [17])"), "{}", error);
        assert!(has_table(&conn, "ok_one"));
        assert!(!has_table(&conn, "half_done"), "the failing version is rolled back");
        assert_eq!(count(&conn, "SELECT MAX(version) FROM _sqlx_migrations"), 17);
        let foreign_keys: i64 = conn.query_row("PRAGMA foreign_keys", [], |r| r.get(0)).unwrap();
        assert_eq!(foreign_keys, 1, "foreign keys are back on after a failure");
    }

    // -- the backup ---------------------------------------------------------

    #[test]
    fn the_backup_holds_the_database_as_it_was() {
        let dir = unique_dir("backup");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        seed_user_rows(&open(&path));

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        let name = report.catch_up.unwrap().backup_file.unwrap();
        assert_eq!(name, "profile_0a1b2c3d-before-v17.db");
        let copy = open(&dir.join(MIGRATION_BACKUP_DIR).join(&name));
        assert!(!column_exists(&copy, "import_sources", "header_signature").unwrap());
        assert!(!has_table(&copy, "_sqlx_migrations"));
        let flag: String = copy
            .query_row("SELECT typeof(is_manually_categorized) FROM transactions", [], |r| r.get(0))
            .unwrap();
        assert_eq!(flag, "text", "taken before the repair too");
        assert!(!dir.join(MIGRATION_BACKUP_DIR).join(format!("{}.tmp", name)).exists());
    }

    #[test]
    fn an_interrupted_backup_is_replaced() {
        let dir = unique_dir("stale-tmp");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        let backups = dir.join(MIGRATION_BACKUP_DIR);
        fs::create_dir_all(&backups).unwrap();
        let tmp = backups.join("profile_0a1b2c3d-before-v17.db.tmp");
        fs::write(&tmp, b"half a copy").unwrap();

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report.catch_up_error, None);
        assert!(!tmp.exists());
        let copy = open(&backups.join("profile_0a1b2c3d-before-v17.db"));
        assert!(has_table(&copy, "transactions"));
    }

    #[test]
    fn a_complete_backup_is_kept() {
        let dir = unique_dir("kept");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        let backups = dir.join(MIGRATION_BACKUP_DIR);
        fs::create_dir_all(&backups).unwrap();
        let existing = backups.join("profile_0a1b2c3d-before-v17.db");
        fs::write(&existing, b"earlier copy").unwrap();

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report.catch_up.unwrap().applied, vec![17]);
        assert_eq!(fs::read(&existing).unwrap(), b"earlier copy");
    }

    #[test]
    fn a_failed_backup_writes_nothing() {
        let dir = unique_dir("no-backup");
        let path = create_db(&dir, SECONDARY, &[CONSOLIDATED_V0_10_0]);
        // A file where the folder should be: create_dir_all fails.
        fs::write(dir.join(MIGRATION_BACKUP_DIR), b"not a folder").unwrap();

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        let error = report.catch_up_error.unwrap();
        assert!(error.starts_with("backup_failed"), "{}", error);
        assert!(!error.contains(dir.to_str().unwrap()), "{}", error);
        let conn = open(&path);
        assert!(!has_table(&conn, "_sqlx_migrations"));
        assert!(!column_exists(&conn, "import_sources", "header_signature").unwrap());
    }

    #[test]
    fn remove_backups_takes_only_this_profile() {
        let dir = unique_dir("cleanup");
        for name in [
            "profile_0a1b2c3d-before-v7.db",
            "profile_0a1b2c3d-before-v17.db.tmp",
            "profile_99999999-before-v17.db",
            "profile_0a1b2c3d-notes.txt",
        ] {
            fs::write(dir.join(name), b"x").unwrap();
        }

        assert_eq!(remove_backups(&dir, "profile_0a1b2c3d"), Ok(2));
        let mut left: Vec<String> = fs::read_dir(&dir)
            .unwrap()
            .map(|e| e.unwrap().file_name().into_string().unwrap())
            .collect();
        left.sort();
        assert_eq!(left, vec!["profile_0a1b2c3d-notes.txt", "profile_99999999-before-v17.db"]);
        assert_eq!(remove_backups(&dir.join("absent"), "profile_0a1b2c3d"), Ok(0));
    }

    // -- the default profile, missing files and the repair ------------------

    #[test]
    fn the_default_profile_is_only_repaired() {
        let dir = unique_dir("default");
        let path = create_db(&dir, DEFAULT_PROFILE_DB, &[CONSOLIDATED_V0_3_0, SEED_V0_3_0]);
        seed_user_rows(&open(&path));

        let report = prepare_at(&path, DEFAULT_PROFILE_DB, &migrations()).unwrap();

        assert_eq!(report.catch_up, None);
        assert_eq!(report.catch_up_error, None);
        assert_eq!(report.repaired_flags, 1);
        let conn = open(&path);
        assert!(!has_table(&conn, "_sqlx_migrations"), "tauri-plugin-sql migrates the default profile");
        assert!(!dir.join(MIGRATION_BACKUP_DIR).exists());
    }

    #[test]
    fn a_missing_database_is_not_created() {
        let dir = unique_dir("missing");
        let path = dir.join(SECONDARY);

        let report = prepare_at(&path, SECONDARY, &migrations()).unwrap();

        assert_eq!(report, ProfilePreparation { elapsed_ms: report.elapsed_ms, ..ProfilePreparation::default() });
        assert!(!path.exists());
    }

    #[test]
    fn repair_turns_text_flags_into_integers_and_leaves_integers_alone() {
        let dir = unique_dir("repair");
        let path = create_db(&dir, SECONDARY, &[crate::database::CONSOLIDATED_SCHEMA]);
        let conn = open(&path);
        for flag in ["'true'", "1", "0", "'false'", "'yes'"] {
            conn.execute(
                &format!(
                    "INSERT INTO transactions (date, description, amount, is_manually_categorized) \
                     VALUES ('2026-01-01', 'x', -1, {})",
                    flag
                ),
                [],
            )
            .unwrap();
        }

        assert_eq!(repair_manual_flags(&conn), Ok(3));

        let flags: Vec<(i64, String)> = conn
            .prepare("SELECT is_manually_categorized, typeof(is_manually_categorized) FROM transactions ORDER BY id")
            .unwrap()
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        let integer = "integer".to_string();
        assert_eq!(
            flags,
            vec![(1, integer.clone()), (1, integer.clone()), (0, integer.clone()), (0, integer.clone()), (0, integer)]
        );
        assert_eq!(repair_manual_flags(&conn), Ok(0));
    }

    #[test]
    fn repair_skips_a_database_without_transactions() {
        let conn = Connection::open_in_memory().unwrap();
        assert_eq!(repair_manual_flags(&conn), Ok(0));
    }

    // -- the migration list itself ------------------------------------------

    #[test]
    fn migration_checksums_match_the_released_ones() {
        // SHA-384 of every migration's SQL, computed on main before
        // `migrations()` was extracted from `run()` (#390). sqlx refuses to
        // open the default profile when a recorded checksum changes, and a
        // secondary profile's history must hold the same values.
        let pinned: [(i64, &str); 17] = [
            (1, "a2f964547e0d0ebf141a978b1f2b5bfd030be6618e13dcdc6fa16504a269b6a036d3dbf7e82700db07c6aa85948bfedf"),
            (2, "59d9c344713108fa6a1fb5c63635d5baab1d78bd8d5aa438a28dafdcb0e78c4cb66a11a6a2fec4358b61c45a9e055c96"),
            (3, "fcb7bc6a921a68cb682f6441655c2152e3e131af6cacac8fec69b5ff73bd9c5a95bdf44a74fcfdfc5e322976f1c7b4d3"),
            (4, "610fc4d778e3507bad1e3bfad5d682a534e753c326db6cf7d7af95a6772e37773518af0a7363bc0e47bbf6e9c49a9531"),
            (5, "131c4b1e0fd6265e5afda9d4f9e505290b8b7108d71b33a957646068f8b116b0cddfca9f285742a93dc7101f4ac8794c"),
            (6, "329e36c0cc63309292c10c92a9137a8f7150c2e5f84e4660ccf41fd4a85a3531e72b74f4ecd321c0236eb37bad934d8f"),
            (7, "4c57798526796c44abe39116e4d498d6a6b330b58e134b67d23c7c1aad7da9e7856f8ac4296b9177591aaf127a2c54bc"),
            (8, "a8763846054027b38f03f9d29219b08891ce931be07ceb221fe6e346d81a037cb9f77953ba230850810029bf25fd4cec"),
            (9, "0f826db3a4bb2910e92c76be1de83a2d3858df783782b0278cbafcb0ae11b40ff1af540478e59a4cf1f7b4ba7933307a"),
            (10, "646ae1048a55a04fbdd40eee045c75950f28f11492c31d94d2f0a32999e8f488650b3548c1d33e02b4b403ff91bf4077"),
            (11, "037129d61869502d377ef0723dee830ad6bf023ef49564dfad9e4a1af1a342bee4161805b6b460bfac3f9748fb01cfd1"),
            (12, "e4ad66f1b7fe8baa174d3fa3f898a98efb44293d35bae665fe4844b090fdfc57383f114f5f756066f09264cc30723f31"),
            (13, "a85e83cea40c9bd70b435ead61db91eecd7bf4de2f80149e2cbf525f65fc1bbcf17eb5b29e6741f41bc8cb699d18b5e9"),
            (14, "e1ef785df3777651f13bae14ebe8da3696f1b793a7676dffe1aaeed14e66f8214eece72466c8df002e746352c333c8b5"),
            (15, "ee608cf4174c8127975aab5039348c1278c365c25c640186a6719b1cf056007772d0296a2efd9a05f5ecf351ef3a128f"),
            (16, "e1be444fca469235871b89dfc264e6020a8ef1dbafff7bc487597eee29971baadd0bd694a352a4eea7e7b9d4ae752d99"),
            (17, "0c60796250708aa2e4339640c99836f5cbdc52e1901231e3fbdc5668772b4d1e09c4fc697c94698523f56151492749b5"),
        ];
        let actual: Vec<(i64, String)> = migrations()
            .iter()
            .map(|m| (m.version, checksum(m.sql).iter().map(|b| format!("{:02x}", b)).collect()))
            .collect();
        let expected: Vec<(i64, String)> = pinned.iter().map(|(v, h)| (*v, h.to_string())).collect();
        assert_eq!(actual, expected);
    }
}