//! Migration-log tests: schema shape, replay, idempotence and the M018/M019 rewrites.
//!
//! Extracted from the former `db.rs` inline test module.

use crate::db::migrations::{MIGRATION_034, MIGRATION_035, MIGRATIONS};
use crate::db::*;

#[test]
fn migration_034_normalises_legacy_key_spellings() {
    let db = Database::open_in_memory().unwrap();
    // Rows written before the detector normalised its output. Inserted
    // post-migration and re-run explicitly, since an in-memory DB starts
    // empty and the migration would otherwise have nothing to rewrite.
    db.conn()
        .execute_batch(
            "INSERT INTO samples
                    (hash, original_name, file_extension, file_size, import_date, last_modified)
                 VALUES ('a', 'a.wav', 'wav', 1, 0, 0), ('b', 'b.wav', 'wav', 1, 0, 0),
                        ('c', 'c.wav', 'wav', 1, 0, 0), ('d', 'd.wav', 'wav', 1, 0, 0);
                 INSERT INTO audio_analysis
                    (hash, musical_key, duration, sample_rate, channels, analyzed_at)
                 VALUES ('a', 'Am', 1.0, 44100, 2, 0), ('b', 'C#m', 1.0, 44100, 2, 0),
                        ('c', 'F#', 1.0, 44100, 2, 0), ('d', 'A minor', 1.0, 44100, 2, 0);",
        )
        .unwrap();
    db.conn().execute_batch(MIGRATION_034).unwrap();

    let mut stmt = db
        .conn()
        .prepare("SELECT hash, musical_key FROM audio_analysis ORDER BY hash")
        .unwrap();
    let got: Vec<(String, String)> = stmt
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))
        .unwrap()
        .map(Result::unwrap)
        .collect();

    assert_eq!(got[0].1, "A minor");
    assert_eq!(got[1].1, "C# minor");
    assert_eq!(got[2].1, "F# major");
    // Already canonical, must not be rewritten to "A minor major".
    assert_eq!(got[3].1, "A minor");

    // Idempotent: a second application changes nothing.
    db.conn().execute_batch(MIGRATION_034).unwrap();
    let after: String = db
        .conn()
        .query_row(
            "SELECT musical_key FROM audio_analysis WHERE hash = 'a'",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(after, "A minor");
}

#[test]
fn migration_035_rewrites_legacy_key_tags() {
    let db = Database::open_in_memory().unwrap();
    db.conn()
        .execute_batch(
            "INSERT INTO samples
                    (hash, original_name, file_extension, file_size, import_date, last_modified)
                 VALUES ('a', 'a.wav', 'wav', 1, 0, 0), ('b', 'b.wav', 'wav', 1, 0, 0),
                        ('c', 'c.wav', 'wav', 1, 0, 0);
                 INSERT INTO tags (sample_hash, tag) VALUES
                    ('a', 'key.am'),
                    ('a', 'genre.techno'),
                    ('b', 'key.c-sharpm'),
                    ('b', 'key.f-sharp'),
                    -- already migrated, plus its legacy twin: must not collide
                    ('c', 'key.a-minor'),
                    ('c', 'key.am');",
        )
        .unwrap();
    db.conn().execute_batch(MIGRATION_035).unwrap();

    let tags = |hash: &str| -> Vec<String> {
        let mut stmt = db
            .conn()
            .prepare("SELECT tag FROM tags WHERE sample_hash = ?1 ORDER BY tag")
            .unwrap();
        let v: Vec<String> = stmt
            .query_map([hash], |r| r.get(0))
            .unwrap()
            .map(Result::unwrap)
            .collect();
        v
    };

    assert_eq!(tags("a"), vec!["genre.techno", "key.a-minor"]);
    assert_eq!(tags("b"), vec!["key.c-sharp-minor", "key.f-sharp-major"]);
    // The collision collapses to the single canonical tag rather than
    // failing the migration on the primary key.
    assert_eq!(tags("c"), vec!["key.a-minor"]);

    // Idempotent: canonical tags match no legacy spelling.
    db.conn().execute_batch(MIGRATION_035).unwrap();
    assert_eq!(tags("a"), vec!["genre.techno", "key.a-minor"]);
}

#[test]
fn key_migrations_do_not_enqueue_sync_changelog() {
    // Both key migrations normalise a local format that every device fixes
    // for itself. Replicating them would push canonical values to peers
    // still running the old detector, which would keep writing the compact
    // spelling and leave the vault holding both.
    let db = Database::open_in_memory().unwrap();
    db.conn()
        .execute_batch(
            "INSERT INTO sync_state (key, value) VALUES ('applying_remote', '0')
                 ON CONFLICT(key) DO UPDATE SET value = '0';
                 INSERT INTO samples
                    (hash, original_name, file_extension, file_size, import_date, last_modified)
                 VALUES ('a', 'a.wav', 'wav', 1, 0, 0);
                 INSERT INTO audio_analysis
                    (hash, musical_key, duration, sample_rate, channels, analyzed_at)
                 VALUES ('a', 'Am', 1.0, 44100, 2, 0);",
        )
        .unwrap();
    let before: i64 = db
        .conn()
        .query_row("SELECT count(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();

    db.conn().execute_batch(MIGRATION_034).unwrap();
    db.conn().execute_batch(MIGRATION_035).unwrap();

    let after: i64 = db
        .conn()
        .query_row("SELECT count(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();
    assert_eq!(
        before, after,
        "key migrations must not enqueue changelog rows"
    );

    // Control: the same write outside the migration must enqueue, otherwise
    // the assertion above would hold even if the triggers never fired here
    // and would prove nothing.
    db.conn()
        .execute_batch("UPDATE audio_analysis SET musical_key = 'B minor' WHERE hash = 'a';")
        .unwrap();
    let control: i64 = db
        .conn()
        .query_row("SELECT count(*) FROM sync_changelog", [], |r| r.get(0))
        .unwrap();
    assert!(
        control > after,
        "sync trigger never fired, so the suppression assertion is vacuous"
    );

    // And the flag is left back where it started, not stuck at '1', which
    // would silently stop capturing every later user edit.
    let flag: String = db
        .conn()
        .query_row(
            "SELECT value FROM sync_state WHERE key = 'applying_remote'",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(flag, "0");
}

#[test]
fn open_in_memory_creates_all_tables() {
    let db = Database::open_in_memory().unwrap();

    // Exclude the FTS5 shadow tables (vfs_nodes_fts, _data, _idx, _docsize,
    // _config) created by M027, this asserts the set of logical tables.
    let tables: Vec<String> = db
            .conn()
            .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' AND name NOT LIKE 'vfs_nodes_fts%' ORDER BY name")
            .unwrap()
            .query_map([], |row| row.get(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();

    let expected = vec![
        "audio_analysis",
        "classifier_exemplars",
        "classifier_layer_rules",
        "classifier_layers",
        "cluster_members",
        "clusters",
        "collection_members",
        "collections",
        "config_key_policy",
        "edit_history",
        "fingerprints",
        "hlc_ledger",
        "neighbour_graph_dirty",
        "neighbour_graph_meta",
        "sample_features",
        "sample_neighbours",
        "samples",
        "sync_changelog",
        "sync_state",
        "tag_policy",
        "tag_provenance",
        "tag_rules",
        "tags",
        "trained_head",
        "user_config",
        "vfs",
        "vfs_nodes",
        "waveform_data",
    ];
    assert_eq!(tables, expected);
}

#[test]
fn migration_sets_user_version() {
    let db = Database::open_in_memory().unwrap();
    let version: i32 = db
        .conn()
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(version, SCHEMA_VERSION);
}

#[test]
fn migration_is_idempotent() {
    let db = Database::open_in_memory().unwrap();
    // Opening again on the same connection shouldn't fail
    let version: i32 = db
        .conn()
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(version, SCHEMA_VERSION);
}

#[test]
fn audio_analysis_sync_triggers_carry_all_columns() {
    // Regression guard for the M018 -> M032 fix: M018 recreated these
    // triggers with the pre-M011 column list, so edits to the columns below
    // stopped propagating to sync_changelog (and thus across devices).
    // Assert the live trigger bodies emit every later-added analysis column.
    // (classification_confidence was in this list until it was retired.)
    let db = Database::open_in_memory().unwrap();
    let later_columns = [
        "spectral_bandwidth",
        "centroid_variance",
        "crest_factor",
        "attack_time",
    ];
    for trigger in ["sync_audio_analysis_insert", "sync_audio_analysis_update"] {
        let sql: String = db
            .conn()
            .query_row(
                "SELECT sql FROM sqlite_master WHERE type = 'trigger' AND name = ?1",
                [trigger],
                |row| row.get(0),
            )
            .unwrap();
        for col in later_columns {
            assert!(sql.contains(col), "{trigger} is missing column {col}");
        }
    }
}

/// The retired sample-class columns must not come back. They were removed from
/// the migration bodies that added them (M003, M011) rather than dropped by a
/// later migration, which is only safe while nothing is deployed, so this is
/// the guard that the edit stays coherent: absent from the table, and absent
/// from the changelog payload a peer would receive.
#[test]
fn retired_class_columns_are_absent() {
    let db = Database::open_in_memory().unwrap();
    let columns: Vec<String> = db
        .conn()
        .prepare("SELECT name FROM pragma_table_info('audio_analysis')")
        .unwrap()
        .query_map([], |row| row.get(0))
        .unwrap()
        .collect::<Result<_, _>>()
        .unwrap();
    for retired in ["classification", "classification_confidence"] {
        assert!(
            !columns.iter().any(|c| c == retired),
            "audio_analysis still has {retired}"
        );
        for trigger in ["sync_audio_analysis_insert", "sync_audio_analysis_update"] {
            let sql: String = db
                .conn()
                .query_row(
                    "SELECT sql FROM sqlite_master WHERE type = 'trigger' AND name = ?1",
                    [trigger],
                    |row| row.get(0),
                )
                .unwrap();
            assert!(!sql.contains(retired), "{trigger} still emits {retired}");
        }
    }
}

/// Open a fresh file-backed DB, close, reopen. The second open re-enters
/// `migrate()`; with `user_version=17` no migration body runs, but the
/// shape verifies our open/close cycle is clean (no locks, no WAL leak).
#[test]
fn migration_replay_from_file_no_op() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("audiofiles.db");

    let db = Database::open(&path).unwrap();
    drop(db);

    let db = Database::open(&path).unwrap();
    let version: i32 = db
        .conn()
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(version, SCHEMA_VERSION);
}

/// Simulates the worst-case recovery path: a prior partial migration left
/// every object in place but `user_version` rolled back. Re-running
/// `migrate()` against the pre-populated schema must succeed without
/// silent failure. This catches the "silent failure → bump user_version"
/// bug class for every migration past the inherently-one-shot ones.
///
/// The inherently-one-shot migrations are excluded from this replay
/// loop:
/// * M001, initial schema; bare CREATE TABLEs, runs against an empty DB.
/// * M002, `DROP TABLE tags; ALTER tags_v2 RENAME TO tags` rebuild dance.
/// * M015, adds `collections.filter_json` and backfills from
///   `smart_folders`, then drops `smart_folders`. The backfill SELECT
///   references a table that no longer exists after the migration runs,
///   so it cannot parse on replay against a post-M015 schema. None of
///   these need replay safety: SQLite's atomic-transaction guarantee
///   means each migration either fully commits or fully rolls back, so
///   the realistic recovery scenario is "re-apply the one migration
///   that crashed", not "re-apply every migration from scratch".
///
/// Every migration from M003 onward (excluding M015) MUST be
/// replay-safe against a populated schema; if you add a new one that
/// isn't, this test fails and you should add `IF NOT EXISTS` /
/// `DROP IF EXISTS` / `INSERT OR IGNORE` accordingly, or add it to the
/// one-shot list above with a clear rationale.
#[test]
fn migration_replay_from_version_fifteen_against_full_schema() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("audiofiles.db");

    Database::open(&path).unwrap();

    {
        let conn = Connection::open(&path).unwrap();
        conn.execute_batch("PRAGMA user_version = 15").unwrap();
    }

    let db = Database::open(&path).unwrap();
    let version: i32 = db
        .conn()
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(version, SCHEMA_VERSION);
}

/// `SCHEMA_VERSION` is derived from `MIGRATIONS`, and the migration runner
/// keys the version it writes off the same list. Pinning the number here
/// means adding a migration without meaning to shows up as a failure.
#[test]
fn schema_version_matches_the_migration_list() {
    assert_eq!(SCHEMA_VERSION, 39);
    assert_eq!(SCHEMA_VERSION as usize, MIGRATIONS.len());
}

/// M037 contract: the browse-list sort must not build a temp B-tree.
///
/// Asserting the index exists would be the weaker test, because an index
/// SQLite declines to use buys nothing. What actually regressed here was the
/// PLAN: `SCAN n` plus `USE TEMP B-TREE FOR ORDER BY` sorted the whole
/// library to return 500 rows, which cost 63 ms at 40k nodes against 0.81 ms
/// once the sort could be walked from the index. So this pins the plan.
///
/// It breaks if someone changes the ORDER BY in `search_global` without
/// moving the index with it, which is the failure that would silently
/// restore the full sort.
#[test]
fn m037_browse_sort_uses_the_index_and_not_a_temp_btree() {
    let db = Database::open_in_memory().unwrap();
    let plan: Vec<String> = db
        .conn()
        .prepare(
            "EXPLAIN QUERY PLAN
                 SELECT n.id, n.name FROM vfs_nodes n
                 LEFT JOIN audio_analysis a ON n.sample_hash = a.hash
                 LEFT JOIN samples s ON n.sample_hash = s.hash
                 WHERE s.deleted_at IS NULL
                 ORDER BY n.node_type ASC, n.name ASC LIMIT 500",
        )
        .unwrap()
        .query_map([], |row| row.get::<_, String>(3))
        .unwrap()
        .collect::<std::result::Result<Vec<_>, _>>()
        .unwrap();
    let plan = plan.join("\n");

    assert!(
        !plan.to_uppercase().contains("TEMP B-TREE"),
        "browse sort fell back to a full sort:\n{plan}"
    );
    assert!(
        plan.contains("idx_vfs_nodes_sort"),
        "browse sort is not walking the sort index:\n{plan}"
    );
}

/// M018 contract: the `sync_changelog.row_id` for sensitive tables must
/// be a 64-hex SHA-256 (per `hash_row_id`), NOT the cleartext content
/// fingerprint or tag string. The cleartext key lives only in `data`.
/// This test is the regression gate for the upload audit fix.
#[test]
fn m018_hashes_sensitive_row_ids() {
    let db = Database::open_in_memory().unwrap();
    let conn = db.conn();

    // Seed: insert a sample and a tag. Both should fire triggers that
    // write to sync_changelog with a hashed row_id.
    conn.execute(
        "INSERT INTO samples (hash, original_name, file_extension, file_size, \
             import_date, last_modified) VALUES \
             ('abc123', 'kick.wav', 'wav', 100, 0, 0)",
        [],
    )
    .unwrap();
    conn.execute(
        "INSERT INTO tags (sample_hash, tag) VALUES ('abc123', 'drums')",
        [],
    )
    .unwrap();

    // samples row_id: 64-hex hash, NOT "abc123".
    let row_id: String = conn
        .query_row(
            "SELECT row_id FROM sync_changelog WHERE table_name = 'samples' AND op = 'INSERT'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(row_id.len(), 64, "row_id should be SHA-256 hex");
    assert!(row_id.chars().all(|c| c.is_ascii_hexdigit()));
    assert_ne!(row_id, "abc123", "cleartext sample hash must not leak");

    // tags row_id: 64-hex hash, NOT "abc123:drums".
    let row_id: String = conn
        .query_row(
            "SELECT row_id FROM sync_changelog WHERE table_name = 'tags' AND op = 'INSERT'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_eq!(row_id.len(), 64);
    assert_ne!(row_id, "abc123:drums", "cleartext tag string must not leak");

    // Salted: hash depends on the per-user salt, so two fresh DBs see
    // different row_ids for the same logical key.
    let db2 = Database::open_in_memory().unwrap();
    let conn2 = db2.conn();
    conn2
        .execute(
            "INSERT INTO samples (hash, original_name, file_extension, file_size, \
                 import_date, last_modified) VALUES \
                 ('abc123', 'kick.wav', 'wav', 100, 0, 0)",
            [],
        )
        .unwrap();
    let row_id2: String = conn2
        .query_row(
            "SELECT row_id FROM sync_changelog WHERE table_name = 'samples' AND op = 'INSERT'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    assert_ne!(row_id, row_id2, "salt should differ between DBs");
}

/// M018 contract: DELETE rows must carry the canonical PK in `data` so
/// the receiving device's `resolve::apply_delete` can reconstruct the
/// WHERE clause without parsing the (now-hashed) row_id.
#[test]
fn m018_delete_triggers_emit_canonical_key_in_data() {
    let db = Database::open_in_memory().unwrap();
    let conn = db.conn();

    conn.execute(
        "INSERT INTO samples (hash, original_name, file_extension, file_size, \
             import_date, last_modified) VALUES \
             ('abc', 'k.wav', 'wav', 1, 0, 0)",
        [],
    )
    .unwrap();
    conn.execute(
        "INSERT INTO tags (sample_hash, tag) VALUES ('abc', 'kick')",
        [],
    )
    .unwrap();
    conn.execute(
        "DELETE FROM tags WHERE sample_hash = 'abc' AND tag = 'kick'",
        [],
    )
    .unwrap();

    let data: String = conn
        .query_row(
            "SELECT data FROM sync_changelog WHERE table_name = 'tags' AND op = 'DELETE'",
            [],
            |row| row.get(0),
        )
        .unwrap();
    let parsed: serde_json::Value = serde_json::from_str(&data).unwrap();
    assert_eq!(parsed["sample_hash"], "abc");
    assert_eq!(parsed["tag"], "kick");
}

/// M019 contract: `samples.deleted_at` column exists, the partial
/// index is in place, and the read-path filter actually hides
/// tombstoned rows from `sample_extension` (and by extension every
/// other query that uses the `query_sample_field` helper).
///
/// This test is the regression gate that proves Phase 1 of the
/// tombstone design (docs/design-sample-deletion.md) lands the
/// promised infrastructure. Phase 2 will wire the UPDATE path that
/// sets deleted_at; today nothing in app code sets it, so every
/// query continues to return all rows in practice, but tests can
/// set it directly and observe the filter working.
#[test]
fn m019_tombstone_column_and_read_filter() {
    let db = Database::open_in_memory().unwrap();
    let conn = db.conn();

    // Column exists with default NULL.
    conn.execute(
        "INSERT INTO samples (hash, original_name, file_extension, file_size, \
             import_date, last_modified) VALUES \
             ('live', 'k.wav', 'wav', 1, 0, 0)",
        [],
    )
    .unwrap();
    conn.execute(
        "INSERT INTO samples (hash, original_name, file_extension, file_size, \
             import_date, last_modified) VALUES \
             ('tomb', 't.wav', 'wav', 1, 0, 0)",
        [],
    )
    .unwrap();
    conn.execute(
        "UPDATE samples SET deleted_at = 1700000000 WHERE hash = 'tomb'",
        [],
    )
    .unwrap();

    // sample_extension reads via query_sample_field, which now filters
    // out tombstoned rows.
    let live_ext =
        crate::store::sample_extension(&db, &crate::SampleHash::from_trusted("live")).unwrap();
    assert_eq!(live_ext, "wav");

    let tomb_ext = crate::store::sample_extension(&db, &crate::SampleHash::from_trusted("tomb"));
    assert!(
        matches!(tomb_ext, Err(crate::error::CoreError::SampleNotFound(_))),
        "tombstoned sample should be hidden from sample_extension; got {tomb_ext:?}"
    );

    // storage_stats also applies the read-path filter: only the live sample
    // (file_size 1) is counted, not the tombstoned one.
    let (count, bytes) = db.storage_stats().unwrap();
    assert_eq!(count, 1, "tombstoned sample should not be counted");
    assert_eq!(bytes, 1, "tombstoned sample's bytes should be excluded");

    // Default retain-days seed is present.
    let retain: String = conn
        .query_row(
            "SELECT value FROM user_config WHERE key = 'sample_tombstone_retain_days'",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(retain, "30");

    // Partial index exists.
    let idx_count: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM sqlite_master \
                 WHERE type = 'index' AND name = 'idx_samples_deleted_at'",
            [],
            |r| r.get(0),
        )
        .unwrap();
    assert_eq!(idx_count, 1);
}

/// Recovery branch contract: when the non-ALTER batch fails for a
/// reason OTHER than "already exists", `migrate()` must roll back and
/// surface the error, NOT bump `user_version` past the failed
/// migration. Prior behavior was a silent `tracing::warn!` followed by
/// a `user_version` bump, which left a partially applied schema
/// invisible to future open() calls.
///
/// Simulates the failure mode by:
/// 1. Bringing the DB up to current version.
/// 2. Injecting an inline migration (M999) whose non-ALTER body
///    references a non-existent table, AND prepending an ALTER on a
///    column that already exists, that's the duplicate-column trip
///    wire that funnels execution into the recovery branch.
/// 3. Setting user_version back to 18 so the runner attempts M019.
/// 4. Asserting migrate() returns Err and user_version stays at 18.
///
/// We can't easily inject a new migration into the const array, so we
/// drive the recovery branch by calling the runner inline.
#[test]
fn migrate_recovery_branch_fails_fast_on_non_alter_error() {
    use rusqlite::Connection;

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("audiofiles.db");
    let db = Database::open(&path).unwrap();
    drop(db);

    // Reopen with a raw Connection so we can hand-craft the recovery
    // scenario without going through migrate().
    let conn = Connection::open(&path).unwrap();
    register_hash_row_id(&conn).unwrap();
    conn.execute_batch("PRAGMA foreign_keys = ON;").unwrap();

    // Simulate the recovery-branch logic directly: try a migration
    // batch that fails with "duplicate column" (forcing recovery),
    // and whose non-ALTER body references a missing table (forcing
    // the failure that previously got swallowed).
    let bad_sql = "ALTER TABLE samples ADD COLUMN cloud_only INTEGER NOT NULL DEFAULT 0;\n\
                       INSERT INTO no_such_table_exists (k) VALUES ('x');";
    let initial_version: i32 = conn
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(initial_version, SCHEMA_VERSION);

    let batch = format!("BEGIN;\n{bad_sql}\nPRAGMA user_version = 999;\nCOMMIT;");
    let first_err = conn.execute_batch(&batch).unwrap_err();
    assert!(
        first_err.to_string().contains("duplicate column"),
        "expected duplicate-column trip wire, got: {first_err}"
    );

    // Recovery: ALTER tolerated, non-ALTER must fail loudly.
    let _ = conn.execute_batch("ROLLBACK");
    conn.execute_batch("BEGIN").unwrap();
    // ALTER passes (column exists; tolerated).
    let alter = "ALTER TABLE samples ADD COLUMN cloud_only INTEGER NOT NULL DEFAULT 0";
    let alter_res = conn.execute_batch(alter);
    assert!(alter_res.is_err());
    assert!(
        alter_res
            .unwrap_err()
            .to_string()
            .contains("duplicate column")
    );

    // Non-ALTER: fail-fast, return error, do not bump user_version.
    let non_alter = "INSERT INTO no_such_table_exists (k) VALUES ('x')";
    let na_res = conn.execute_batch(non_alter);
    assert!(na_res.is_err());
    let msg = na_res.unwrap_err().to_string();
    assert!(
        !msg.contains("already exists"),
        "expected a real failure (no such table), got: {msg}"
    );

    // The fail-fast path rolls back and never reaches the
    // user_version bump. Confirm.
    conn.execute_batch("ROLLBACK").unwrap();
    let after: i32 = conn
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(
        after, initial_version,
        "user_version must not bump when recovery non-ALTER fails for a real reason"
    );
}

/// Companion to `migration_replay_from_version_fifteen_against_full_schema`:
/// rolling user_version back and re-opening must heal to version 18 without
/// silent partial-state. Identical setup; kept as a contract-specific name
/// so a failing test points the reader at the recovery-branch design rather
/// than the broader replay-safety claim.
#[test]
fn migrate_recovery_branch_tolerates_already_exists() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("audiofiles.db");

    Database::open(&path).unwrap();
    {
        let conn = rusqlite::Connection::open(&path).unwrap();
        conn.execute_batch("PRAGMA user_version = 15").unwrap();
    }
    let db = Database::open(&path).unwrap();
    let version: i32 = db
        .conn()
        .query_row("PRAGMA user_version", [], |row| row.get(0))
        .unwrap();
    assert_eq!(version, SCHEMA_VERSION);
}