koan-core 0.60.3

use rusqlite::Connection;

/// Bumped whenever the schema changes. Stored in `PRAGMA user_version` so an
/// older build refuses a database it does not understand rather than writing to it.
pub const SCHEMA_VERSION: i64 = 21;

/// Create all tables. Idempotent — safe to call on every startup.
pub fn create_tables(conn: &Connection) -> rusqlite::Result<()> {
    // Before any DDL: the ORDER BY clauses that use it are everywhere, and a
    // connection without it fails them rather than sorting differently.
    super::connection::register_library_collation(conn)?;
    super::connection::register_shuffle_function(conn)?;
    super::connection::register_fold_function(conn)?;
    let found: i64 = conn.query_row("PRAGMA user_version", [], |r| r.get(0))?;
    if found > SCHEMA_VERSION {
        return Err(rusqlite::Error::SqliteFailure(
            rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_ERROR),
            Some(format!(
                "database schema version {found} is newer than this build understands \
                 ({SCHEMA_VERSION}) — upgrade koan rather than downgrading the library"
            )),
        ));
    }

    // Every statement past here takes the write lock, whether or not it
    // changes a row, and an open waits behind any scan or sync holding it. A
    // database already at this version has nothing for them to do. The column
    // checks are reads, and catch a column added without a version bump.
    if found == SCHEMA_VERSION {
        add_missing_columns(conn)?;
        return crate::db::queries::auth::adopt_local_rows(conn);
    }

    // The whole upgrade, version included, is one transaction: another
    // process reading the database, such as the server this one is
    // replacing, sees the old schema or the new one and nothing between,
    // and an upgrade that fails part way leaves the database as it was.
    // The table rebuilds need foreign keys off, and that pragma does nothing
    // inside a transaction, so it brackets it.
    conn.pragma_update(None, "foreign_keys", "off")?;
    let upgraded = crate::db::queries::atomically(conn, || upgrade(conn, found));
    let restored = conn.pragma_update(None, "foreign_keys", "on");
    upgraded?;
    restored
}

/// Everything [`create_tables`] does to bring a database from version
/// `found` to [`SCHEMA_VERSION`], inside its transaction.
fn upgrade(conn: &Connection, found: i64) -> rusqlite::Result<()> {
    conn.execute_batch(
        "
        CREATE TABLE IF NOT EXISTS artists (
            id          INTEGER PRIMARY KEY,
            name        TEXT NOT NULL,
            sort_name   TEXT,
            mbid        TEXT,
            remote_id   TEXT,
            UNIQUE(name)
        );

        CREATE TABLE IF NOT EXISTS albums (
            id           INTEGER PRIMARY KEY,
            title        TEXT NOT NULL,
            artist_id    INTEGER REFERENCES artists(id),
            date         TEXT,
            total_discs  INTEGER,
            total_tracks INTEGER,
            codec        TEXT,
            label        TEXT,
            remote_id    TEXT,
            added_at     TEXT
        );

        CREATE TABLE IF NOT EXISTS tracks (
            id            INTEGER PRIMARY KEY,
            album_id      INTEGER REFERENCES albums(id),
            artist_id     INTEGER REFERENCES artists(id),
            disc          INTEGER,
            track_number  INTEGER,
            title         TEXT NOT NULL,
            duration_ms   INTEGER,
            path          TEXT,
            codec         TEXT,
            sample_rate   INTEGER,
            bit_depth     INTEGER,
            channels      INTEGER,
            bitrate       INTEGER,
            size_bytes    INTEGER,
            mtime         INTEGER,
            genre         TEXT,
            source        TEXT NOT NULL DEFAULT 'local' CHECK (source IN ('local', 'remote', 'cached')),
            remote_id     TEXT,
            remote_url    TEXT,
            cached_path   TEXT,
            UNIQUE(path)
        );

        CREATE INDEX IF NOT EXISTS idx_tracks_album ON tracks(album_id);
        CREATE INDEX IF NOT EXISTS idx_tracks_artist ON tracks(artist_id);
        CREATE INDEX IF NOT EXISTS idx_tracks_source ON tracks(source);
        CREATE INDEX IF NOT EXISTS idx_tracks_remote_id ON tracks(remote_id);
        CREATE INDEX IF NOT EXISTS idx_albums_artist ON albums(artist_id);
        CREATE INDEX IF NOT EXISTS idx_tracks_album_order ON tracks(album_id, disc, track_number);
        -- Favourites are keyed by path, and a track can be reached by three of
        -- them. Without these, matching a favourite to its track means reading
        -- every row in the library: the query planner said SCAN, and finding
        -- a hundred favourites among fifty thousand tracks took fifty
        -- milliseconds, on every listing that wanted to know what was starred.
        -- Partial, because both columns are null for anything purely local.
        CREATE INDEX IF NOT EXISTS idx_tracks_cached_path ON tracks(cached_path)
            WHERE cached_path IS NOT NULL;
        CREATE INDEX IF NOT EXISTS idx_tracks_remote_url ON tracks(remote_url)
            WHERE remote_url IS NOT NULL;
        -- A remote sync enriches every album and every artist it paged through,
        -- matched on the server's id. Without these that is one full table read
        -- per record, so a library twice the size costs four times as much to
        -- sync. Partial, because a locally-scanned record has no remote id.
        CREATE INDEX IF NOT EXISTS idx_albums_remote_id ON albums(remote_id)
            WHERE remote_id IS NOT NULL;
        CREATE INDEX IF NOT EXISTS idx_artists_remote_id ON artists(remote_id)
            WHERE remote_id IS NOT NULL;
        -- The genre list and the genre filter both match case-insensitively,
        -- and both want the albums a genre spans.
        CREATE INDEX IF NOT EXISTS idx_tracks_genre
            ON tracks(genre COLLATE NOCASE, album_id);

        CREATE VIRTUAL TABLE IF NOT EXISTS tracks_fts USING fts5(
            title,
            artist_name,
            album_title,
            genre
        );

        CREATE TABLE IF NOT EXISTS library_folders (
            id        INTEGER PRIMARY KEY,
            path      TEXT NOT NULL UNIQUE,
            last_scan INTEGER
        );

        CREATE TABLE IF NOT EXISTS scan_cache (
            path      TEXT PRIMARY KEY,
            mtime     INTEGER NOT NULL,
            size      INTEGER NOT NULL,
            track_id  INTEGER REFERENCES tracks(id)
        );

        -- Forgetting a track deletes its scan cache entry by track id, and the
        -- primary key is the path.
        CREATE INDEX IF NOT EXISTS idx_scan_cache_track ON scan_cache(track_id);

        CREATE TABLE IF NOT EXISTS remote_servers (
            id        INTEGER PRIMARY KEY,
            url       TEXT NOT NULL UNIQUE,
            username  TEXT NOT NULL,
            last_sync INTEGER,
            library_version INTEGER
        );

        CREATE TABLE IF NOT EXISTS organize_log (
            id         INTEGER PRIMARY KEY,
            batch_id   TEXT NOT NULL,
            track_id   INTEGER,
            from_path  TEXT NOT NULL,
            to_path    TEXT NOT NULL,
            size_bytes INTEGER,
            mtime      INTEGER,
            created_at TEXT DEFAULT (datetime('now'))
        );

        -- Undo reads back one batch at a time.
        CREATE INDEX IF NOT EXISTS idx_organize_log_batch ON organize_log(batch_id);
        -- Merging two track rows repoints the log from one to the other.
        CREATE INDEX IF NOT EXISTS idx_organize_log_track ON organize_log(track_id);

        CREATE TABLE IF NOT EXISTS lyrics_cache (
            id          INTEGER PRIMARY KEY,
            track_id    INTEGER REFERENCES tracks(id),
            source      TEXT NOT NULL,
            synced      INTEGER DEFAULT 0,
            content     TEXT NOT NULL,
            fetched_at  INTEGER NOT NULL,
            UNIQUE(track_id)
        );

        -- Rows deleted from albums, artists and tracks, for front ends that
        -- cache by id: SQLite hands a freed id to the next row, and a cover
        -- cached under it would be shown for the wrong record. Rows under a
        -- rescanned folder are named too, since a cover image beside them
        -- may have changed.
        CREATE TABLE IF NOT EXISTS art_evictions (
            seq   INTEGER PRIMARY KEY AUTOINCREMENT,
            kind  TEXT NOT NULL,
            id    INTEGER NOT NULL
        );
        CREATE TRIGGER IF NOT EXISTS evict_album_art AFTER DELETE ON albums
            BEGIN INSERT INTO art_evictions (kind, id) VALUES ('album', old.id); END;
        CREATE TRIGGER IF NOT EXISTS evict_artist_art AFTER DELETE ON artists
            BEGIN INSERT INTO art_evictions (kind, id) VALUES ('artist', old.id); END;
        CREATE TRIGGER IF NOT EXISTS evict_track_art AFTER DELETE ON tracks
            BEGIN INSERT INTO art_evictions (kind, id) VALUES ('track', old.id); END;

        -- A server's record of the koan apps that have linked to it, and what
        -- waits for each while it is away: see koan-server's clients.rs.
        CREATE TABLE IF NOT EXISTS link_devices (
            device     TEXT NOT NULL,
            username   TEXT NOT NULL,
            name       TEXT NOT NULL,
            platform   TEXT NOT NULL,
            last_seen  INTEGER NOT NULL,
            -- The address it last linked from, which lets a device of another
            -- account behind the same router wake it.
            addr       TEXT,
            PRIMARY KEY (device, username)
        );

        -- Where Apple's push service reaches each linked iOS app, kept after
        -- its socket closes: reaching a suspended app is what it is for.
        CREATE TABLE IF NOT EXISTS link_push (
            device      TEXT NOT NULL,
            username    TEXT NOT NULL,
            token       TEXT NOT NULL,
            sandbox     INTEGER NOT NULL,
            updated_at  INTEGER NOT NULL,
            PRIMARY KEY (device, username)
        );

        -- Devices an owner has let other accounts on this server control,
        -- granted from the device itself: see koan-server's clients.rs.
        CREATE TABLE IF NOT EXISTS link_grants (
            device      TEXT NOT NULL,
            owner       TEXT NOT NULL,
            grantee     TEXT NOT NULL,
            created_at  INTEGER NOT NULL,
            PRIMARY KEY (device, owner, grantee)
        );

        CREATE TABLE IF NOT EXISTS link_orders (
            id          TEXT PRIMARY KEY,
            body        TEXT NOT NULL,
            created_at  INTEGER NOT NULL
        );

        CREATE TABLE IF NOT EXISTS link_outbox (
            id          INTEGER PRIMARY KEY,
            device      TEXT NOT NULL,
            username    TEXT NOT NULL,
            command     TEXT NOT NULL,
            created_at  INTEGER NOT NULL
        );

        -- One row per artist looked up, misses included: an empty row is the
        -- answer that nothing was found, which stops a page asking again.
        CREATE TABLE IF NOT EXISTS artist_info (
            artist_id     INTEGER PRIMARY KEY REFERENCES artists(id) ON DELETE CASCADE,
            bio           TEXT,
            bio_url       TEXT,
            image_url     TEXT,
            image_credit  TEXT,
            fetched_at    INTEGER NOT NULL
        );

        -- Favourites, playlists, play history and shares belong to a user:
        -- an account's id, or 0 for the implicit user of an install with no
        -- admin account (see `queries::auth::LOCAL_USER`). 0 names no row in
        -- `users`, so there is no foreign key; the `users_personal_data`
        -- trigger does the cascading one would.
        -- Favourites name rows. A rebuilt index re-reads its sources into the
        -- rows it has rather than making new ones, so they survive it.
        CREATE TABLE IF NOT EXISTS favourites (
            user_id     INTEGER NOT NULL DEFAULT 0,
            track_id    INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
            created_at  TEXT DEFAULT (datetime('now')),
            PRIMARY KEY (user_id, track_id)
        );

        CREATE TABLE IF NOT EXISTS favourite_albums (
            user_id     INTEGER NOT NULL DEFAULT 0,
            album_id    INTEGER NOT NULL REFERENCES albums(id) ON DELETE CASCADE,
            created_at  TEXT DEFAULT (datetime('now')),
            PRIMARY KEY (user_id, album_id)
        );

        CREATE TABLE IF NOT EXISTS favourite_artists (
            user_id     INTEGER NOT NULL DEFAULT 0,
            artist_id   INTEGER NOT NULL REFERENCES artists(id) ON DELETE CASCADE,
            created_at  TEXT DEFAULT (datetime('now')),
            PRIMARY KEY (user_id, artist_id)
        );

        -- One to five, per account. See `queries::ratings`.
        CREATE TABLE IF NOT EXISTS track_ratings (
            user_id     INTEGER NOT NULL DEFAULT 0,
            track_id    INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
            rating      INTEGER NOT NULL CHECK (rating BETWEEN 1 AND 5),
            changed_at  TEXT DEFAULT (datetime('now')),
            PRIMARY KEY (user_id, track_id)
        );

        CREATE TABLE IF NOT EXISTS album_ratings (
            user_id     INTEGER NOT NULL DEFAULT 0,
            album_id    INTEGER NOT NULL REFERENCES albums(id) ON DELETE CASCADE,
            rating      INTEGER NOT NULL CHECK (rating BETWEEN 1 AND 5),
            changed_at  TEXT DEFAULT (datetime('now')),
            PRIMARY KEY (user_id, album_id)
        );

        CREATE TABLE IF NOT EXISTS artist_ratings (
            user_id     INTEGER NOT NULL DEFAULT 0,
            artist_id   INTEGER NOT NULL REFERENCES artists(id) ON DELETE CASCADE,
            rating      INTEGER NOT NULL CHECK (rating BETWEEN 1 AND 5),
            changed_at  TEXT DEFAULT (datetime('now')),
            PRIMARY KEY (user_id, artist_id)
        );

        -- The queue a Subsonic client saved for its account. See
        -- `queries::play_queues`. The current entry is held by its place in
        -- the order as saved.
        CREATE TABLE IF NOT EXISTS play_queues (
            user_id           INTEGER PRIMARY KEY,
            current_position  INTEGER,
            position_ms       INTEGER NOT NULL,
            changed_at        INTEGER NOT NULL,
            changed_by        TEXT NOT NULL
        );

        CREATE TABLE IF NOT EXISTS play_queue_entries (
            user_id   INTEGER NOT NULL,
            position  INTEGER NOT NULL,
            track_id  INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
            PRIMARY KEY (user_id, position)
        );

        -- Where an account is in a track. See `queries::bookmarks`.
        CREATE TABLE IF NOT EXISTS bookmarks (
            user_id      INTEGER NOT NULL DEFAULT 0,
            track_id     INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
            position_ms  INTEGER NOT NULL,
            comment      TEXT,
            created_at   INTEGER NOT NULL,
            changed_at   INTEGER NOT NULL,
            PRIMARY KEY (user_id, track_id)
        );

        CREATE TABLE IF NOT EXISTS playback_state (
            id          INTEGER PRIMARY KEY CHECK (id = 1),
            queue_json  TEXT NOT NULL DEFAULT '[]',
            cursor_id   TEXT,
            position_ms INTEGER NOT NULL DEFAULT 0,
            updated_at  TEXT DEFAULT (datetime('now'))
        );

        -- Where the saved queue is up to, written every second while music
        -- plays. A row of its own because SQLite rewrites a record whenever
        -- its size changes, and in `playback_state` that record carries the
        -- whole queue.
        CREATE TABLE IF NOT EXISTS playback_position (
            id            INTEGER PRIMARY KEY CHECK (id = 1),
            cursor_id     TEXT,
            position_ms   INTEGER NOT NULL DEFAULT 0,
            was_playing   INTEGER NOT NULL DEFAULT 0,
            updated_at    TEXT DEFAULT (datetime('now'))
        );

        -- `AUTOINCREMENT` because ids are the cursor devices page a
        -- server's history by: one handed out again would never reach them.
        CREATE TABLE IF NOT EXISTS play_history (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            track_id    INTEGER REFERENCES tracks(id) ON DELETE CASCADE,
            played_at   INTEGER NOT NULL,
            duration_ms INTEGER,
            source      TEXT DEFAULT 'local'
        );

        -- Plays forgotten on a server, kept so the account's devices forget
        -- them too: see `queries::history`. A row without a track forgets
        -- every play up to `played_at`.
        CREATE TABLE IF NOT EXISTS play_history_forgotten (
            id          INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id     INTEGER NOT NULL,
            track_uid   TEXT,
            played_at   INTEGER NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_play_history_forgotten_user
            ON play_history_forgotten(user_id, id);

        -- Plays and forgets waiting to reach the signed-in server, sent in
        -- order once it answers. `remote_id` is the track's id there; a
        -- `clear` has none and forgets every play up to `at_ms`.
        CREATE TABLE IF NOT EXISTS history_outbox (
            id          INTEGER PRIMARY KEY,
            kind        TEXT NOT NULL CHECK (kind IN ('scrobble', 'forget', 'clear')),
            remote_id   TEXT,
            at_ms       INTEGER NOT NULL
        );

        -- With `played_at`, a track's last play is read off the index.
        CREATE INDEX IF NOT EXISTS idx_play_history_track_played
            ON play_history(track_id, played_at);
        CREATE INDEX IF NOT EXISTS idx_play_history_time ON play_history(played_at);

        -- Playlists carry what Subsonic carries and nothing else, so a
        -- playlist made here and one made on the server are the same object.
        -- `sort_order` and `grouped` are the exceptions: where a playlist sits
        -- in your sidebar and how you like to look at it are facts about this
        -- machine, and no server has anywhere to put them.
        CREATE TABLE IF NOT EXISTS playlists (
            id         INTEGER PRIMARY KEY,
            name       TEXT NOT NULL,
            comment    TEXT,
            public     INTEGER NOT NULL DEFAULT 0,
            owner      TEXT,
            remote_id  TEXT UNIQUE,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            changed_at TEXT NOT NULL DEFAULT (datetime('now')),
            sort_order INTEGER NOT NULL DEFAULT 0,
            grouped    INTEGER
        );

        -- One row per entry, with an id of its own.
        --
        -- A playlist may hold the same track twice, so a track id names neither
        -- a row nor a place. The entry id does: it survives a reorder, it is
        -- what a queue item remembers it came from, and it is how the two
        -- copies of a song are told apart when one of them is playing.
        CREATE TABLE IF NOT EXISTS playlist_tracks (
            id          INTEGER PRIMARY KEY,
            playlist_id INTEGER NOT NULL REFERENCES playlists(id) ON DELETE CASCADE,
            position    INTEGER NOT NULL,
            track_id    INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
            UNIQUE (playlist_id, position)
        );

        CREATE INDEX IF NOT EXISTS idx_playlist_tracks_track ON playlist_tracks(track_id);

        -- Auth tables
        CREATE TABLE IF NOT EXISTS users (
            id            INTEGER PRIMARY KEY AUTOINCREMENT,
            username      TEXT NOT NULL UNIQUE,
            password_hash TEXT NOT NULL,
            role          TEXT NOT NULL DEFAULT 'user' CHECK (role IN ('admin', 'user', 'readonly')),
            created_at    TEXT DEFAULT (datetime('now'))
        );

        CREATE TABLE IF NOT EXISTS refresh_tokens (
            id          TEXT PRIMARY KEY,
            user_id     INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            expires_at  INTEGER NOT NULL,
            revoked     INTEGER NOT NULL DEFAULT 0,
            created_at  TEXT DEFAULT (datetime('now'))
        );

        CREATE INDEX IF NOT EXISTS idx_refresh_tokens_user ON refresh_tokens(user_id);

        -- Subsonic API keys. Only `sha256(key)` is kept: a key is 32 random
        -- bytes, so a fast hash is enough, and a database read yields nothing
        -- that signs in.
        CREATE TABLE IF NOT EXISTS api_keys (
            id           INTEGER PRIMARY KEY,
            user_id      INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            name         TEXT NOT NULL,
            key_hash     TEXT NOT NULL UNIQUE,
            created_at   INTEGER NOT NULL,
            last_used_at INTEGER
        );

        CREATE INDEX IF NOT EXISTS idx_api_keys_user ON api_keys(user_id);

        -- Generated passwords for Subsonic clients that only speak token auth,
        -- sealed under a key derived from the server's signing key: see
        -- queries/app_passwords.rs.
        CREATE TABLE IF NOT EXISTS app_passwords (
            id           INTEGER PRIMARY KEY,
            user_id      INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            name         TEXT NOT NULL,
            sealed       BLOB NOT NULL,
            created_at   INTEGER NOT NULL,
            last_used_at INTEGER
        );

        CREATE INDEX IF NOT EXISTS idx_app_passwords_user ON app_passwords(user_id);

        -- Listening services an account forwards its plays to, with the
        -- credential each takes. `error` is set when the service refuses the
        -- credential; nothing is sent for the account until it connects again.
        -- An account's EQ profiles kept everywhere, as its devices last sent
        -- them: the profile as JSON, or NULL once deleted. `rev` counts the
        -- account's changes, profiles and dismissals together, so a device
        -- reads what came after the last it saw.
        CREATE TABLE IF NOT EXISTS dsp_profiles (
            user_id    INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            uid        TEXT NOT NULL,
            rev        INTEGER NOT NULL,
            edited_at  INTEGER NOT NULL,
            doc        TEXT,
            PRIMARY KEY (user_id, uid)
        );
        -- Impulse responses and the like, by content, once per account.
        CREATE TABLE IF NOT EXISTS dsp_files (
            user_id  INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            sha256   TEXT NOT NULL,
            data     BLOB NOT NULL,
            PRIMARY KEY (user_id, sha256)
        );
        -- Outputs whose AutoEQ suggestion the account turned down.
        CREATE TABLE IF NOT EXISTS dsp_dismissed (
            user_id  INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            output   TEXT NOT NULL,
            rev      INTEGER NOT NULL,
            PRIMARY KEY (user_id, output)
        );
        -- A device's side: how far it has read each server's profiles, the
        -- revision and content of each profile as last synced, and when each
        -- was last changed here.
        CREATE TABLE IF NOT EXISTS dsp_sync_cursor (
            url     TEXT PRIMARY KEY,
            cursor  INTEGER NOT NULL
        );
        CREATE TABLE IF NOT EXISTS dsp_synced (
            url   TEXT NOT NULL,
            uid   TEXT NOT NULL,
            rev   INTEGER NOT NULL,
            hash  TEXT NOT NULL,
            PRIMARY KEY (url, uid)
        );
        CREATE TABLE IF NOT EXISTS dsp_local (
            uid        TEXT PRIMARY KEY,
            hash       TEXT NOT NULL,
            edited_at  INTEGER NOT NULL,
            refused    TEXT,
            note       TEXT
        );

        CREATE TABLE IF NOT EXISTS scrobble_services (
            user_id       INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
            service       TEXT NOT NULL,
            token         TEXT NOT NULL,
            account_name  TEXT NOT NULL,
            connected_at  INTEGER NOT NULL,
            error         TEXT,
            PRIMARY KEY (user_id, service)
        );

        -- Plays waiting to reach a service. A row is deleted once the service
        -- has accepted the play, so what is here survives a restart or an
        -- outage and is sent when the service answers again.
        CREATE TABLE IF NOT EXISTS scrobble_outbox (
            id          INTEGER PRIMARY KEY,
            user_id     INTEGER NOT NULL,
            service     TEXT NOT NULL,
            history_id  INTEGER NOT NULL REFERENCES play_history(id) ON DELETE CASCADE,
            FOREIGN KEY (user_id, service)
                REFERENCES scrobble_services(user_id, service) ON DELETE CASCADE
        );
        CREATE INDEX IF NOT EXISTS idx_scrobble_outbox_service
            ON scrobble_outbox(user_id, service);
        CREATE INDEX IF NOT EXISTS idx_scrobble_outbox_history
            ON scrobble_outbox(history_id);

        -- Share links this koan serves itself. The tracks are an explicit list,
        -- not a query: what a link names is all an anonymous visitor can play,
        -- so it must not grow when the library does.
        CREATE TABLE IF NOT EXISTS shares (
            id           TEXT PRIMARY KEY,
            description  TEXT,
            created_at   INTEGER NOT NULL,
            expires_at   INTEGER,
            visits       INTEGER NOT NULL DEFAULT 0,
            last_visited INTEGER
        );

        CREATE TABLE IF NOT EXISTS share_tracks (
            share_id TEXT NOT NULL REFERENCES shares(id) ON DELETE CASCADE,
            position INTEGER NOT NULL,
            track_id INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
            PRIMARY KEY (share_id, position)
        );
        -- Deleting or merging a track looks for the shares that name it.
        CREATE INDEX IF NOT EXISTS idx_share_tracks_track ON share_tracks(track_id);
        -- Expiry was indexed and never used: the only query that reads it is the
        -- cleanup sweep, whose `revoked = 1 OR expires_at <= ?` spans two columns
        -- and reads the table either way. An index nothing reads is a cost paid
        -- on every sign-in.
        DROP INDEX IF EXISTS idx_refresh_tokens_expires;
        ",
    )?;
    conn.execute_batch(crate::db::queries::sources::SOURCE_TABLES)?;
    apply_migrations(conn, found)?;
    // After the migrations: rebuilding `play_history` drops its triggers.
    //
    // A play a client reports is queued for every service its account
    // forwards to, in the transaction that records it. koan's own playback
    // records a play when the track starts, before it is known to have been
    // heard, so only reported plays are forwarded as they happen.
    conn.execute_batch(
        "CREATE TRIGGER IF NOT EXISTS scrobble_reported_play AFTER INSERT ON play_history
             WHEN new.source = 'subsonic'
             BEGIN
                 INSERT INTO scrobble_outbox (user_id, service, history_id)
                 SELECT user_id, service, new.id FROM scrobble_services
                  WHERE user_id = new.user_id;
             END;",
    )?;
    cascade_orphans(conn)?;
    refuse_dangling_references(conn)?;
    conn.pragma_update(None, "user_version", SCHEMA_VERSION)?;

    Ok(())
}

/// Do what `ON DELETE CASCADE` would have done for the rows the upgrade
/// deleted. Foreign keys are off for the whole upgrade, since the table
/// rebuilds need them off and the pragma cannot change inside a transaction,
/// so deleting an album or an artist left its favourites and ratings behind.
/// For every key declared to cascade, the rows naming a parent that is gone
/// are deleted, until a pass deletes nothing: a deleted track takes its
/// playlist entries, and so on down.
fn cascade_orphans(conn: &Connection) -> rusqlite::Result<()> {
    let tables: Vec<String> = conn
        .prepare(
            "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'",
        )?
        .query_map([], |r| r.get(0))?
        .collect::<rusqlite::Result<_>>()?;
    let mut cascades = Vec::new();
    for table in &tables {
        let mut stmt = conn.prepare(&format!("PRAGMA foreign_key_list(\"{table}\")"))?;
        let keys = stmt.query_map([], |r| {
            Ok((
                r.get::<_, String>("table")?,
                r.get::<_, String>("from")?,
                r.get::<_, Option<String>>("to")?,
                r.get::<_, String>("on_delete")?,
            ))
        })?;
        for key in keys {
            let (parent, from, to, on_delete) = key?;
            if on_delete.eq_ignore_ascii_case("CASCADE") {
                let to = to.unwrap_or_else(|| "rowid".to_owned());
                cascades.push(format!(
                    "DELETE FROM \"{table}\" WHERE \"{from}\" IS NOT NULL
                       AND \"{from}\" NOT IN (SELECT \"{to}\" FROM \"{parent}\")"
                ));
            }
        }
    }
    loop {
        let mut deleted = 0;
        for sql in &cascades {
            deleted += conn.execute(sql, [])?;
        }
        if deleted == 0 {
            return Ok(());
        }
    }
}

/// Fail the upgrade, and with it roll everything back, if any row names a
/// parent that is not there: a key that does not cascade, whose child the
/// upgrade should have dealt with and did not.
fn refuse_dangling_references(conn: &Connection) -> rusqlite::Result<()> {
    let dangling: Vec<(String, i64, String)> = conn
        .prepare(r#"SELECT "table", rowid, parent FROM pragma_foreign_key_check LIMIT 5"#)?
        .query_map([], |r| {
            Ok((
                r.get(0)?,
                r.get::<_, Option<i64>>(1)?.unwrap_or(0),
                r.get(2)?,
            ))
        })?
        .collect::<rusqlite::Result<_>>()?;
    if dangling.is_empty() {
        return Ok(());
    }
    let rows: Vec<String> = dangling
        .iter()
        .map(|(table, rowid, parent)| format!("{table} row {rowid} names a missing {parent}"))
        .collect();
    Err(rusqlite::Error::SqliteFailure(
        rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_CONSTRAINT_FOREIGNKEY),
        Some(format!(
            "the upgrade would leave rows naming what is not there: {}",
            rows.join("; ")
        )),
    ))
}

/// Columns added after the initial schema. Applied when absent, so a database
/// created by any earlier version converges on the current shape.
///
/// `organize_log.size_bytes`/`mtime` are checked against the file before undo
/// moves it back, so a file replaced since the organize is left alone.
const ADDED_COLUMNS: &[(&str, &str, &str)] = &[
    // The device a key signed in, and the public key that device proves
    // itself with to the account's other devices on the local network. On the
    // key's row, so revoking the key drops it. See `api_keys::device_keys`.
    ("api_keys", "device", "TEXT"),
    ("api_keys", "device_key", "TEXT"),
    ("tracks", "cache_size_bytes", "INTEGER"),
    ("tracks", "cache_download_date", "INTEGER"),
    // Downloaded because someone asked for the file, not to play it: evicted
    // only once everything fetched for playback has gone.
    ("tracks", "cache_pinned", "INTEGER NOT NULL DEFAULT 0"),
    ("organize_log", "size_bytes", "INTEGER"),
    ("organize_log", "mtime", "INTEGER"),
    // When the album entered the library, so clients can offer a
    // recently-added ordering. Remote sync supplies the server's own `created`;
    // a local scan the earliest mtime among the album's files.
    ("albums", "added_at", "TEXT"),
    // Whether playback was running when the session was saved, so reopening can
    // pick up where it left off rather than always paused.
    (
        "playback_state",
        "was_playing",
        "INTEGER NOT NULL DEFAULT 0",
    ),
    // The play mode, beside the position rather than the queue so that it
    // outlives a queue: it is the player's.
    ("playback_position", "shuffle", "INTEGER NOT NULL DEFAULT 0"),
    ("playback_position", "repeat", "TEXT NOT NULL DEFAULT 'off'"),
    // MusicBrainz ids are the join key for anything that wants to look a
    // release or a recording up elsewhere. The server hands them over on every
    // album and every song.
    ("albums", "mbid", "TEXT"),
    ("tracks", "mbid", "TEXT"),
    // The server's own sort key, which is what it orders by.
    ("albums", "sort_name", "TEXT"),
    // What a share is a slice of, so its page shows an album or an artist as
    // one. The track list stays authoritative; shares made before are loose
    // tracks.
    ("shares", "kind", "TEXT NOT NULL DEFAULT 'tracks'"),
    ("shares", "subject_id", "INTEGER"),
    ("shares", "start_track_id", "INTEGER"),
    // Whose it is. See the note above `favourites`.
    ("play_history", "user_id", "INTEGER NOT NULL DEFAULT 0"),
    ("playlists", "user_id", "INTEGER NOT NULL DEFAULT 0"),
    ("shares", "user_id", "INTEGER NOT NULL DEFAULT 0"),
    // The id every surface publishes; see `queries::uids`. Row ids are
    // numbered per table and per database, so album 5 and song 5 are the same
    // id to an endpoint that takes either, and neither means anything on
    // another device.
    ("artists", "uid", "TEXT"),
    ("albums", "uid", "TEXT"),
    ("tracks", "uid", "TEXT"),
    ("playlists", "uid", "TEXT"),
    // Two-way playlist sync. `revision` counts local edits, `synced_revision`
    // is the one the last push or pull covered, `remote_changed` the server's
    // `changed` as of then, and `remote_account` whose server `remote_id`
    // names a playlist on. Each side is judged against its own record, which
    // is what lets a sync tell "the server moved" from "we moved" without
    // comparing two machines' clocks.
    ("playlists", "revision", "INTEGER NOT NULL DEFAULT 0"),
    ("playlists", "synced_revision", "INTEGER"),
    ("playlists", "remote_changed", "TEXT"),
    ("playlists", "remote_account", "TEXT"),
    ("remote_servers", "library_version", "INTEGER"),
    // OAuth grants, for reuse detection; see `queries::auth::revoke_replayed_grant`.
    ("refresh_tokens", "grant_id", "TEXT"),
    ("refresh_tokens", "client_name", "TEXT"),
    ("refresh_tokens", "used_at", "INTEGER"),
    // Names as matching compares them; see `queries::sources::fold`. Kept
    // by triggers, so a row has its key however it was written.
    ("artists", "name_key", "TEXT"),
    ("albums", "title_key", "TEXT"),
    // Where a linked device last connected from: see koan-server's clients.rs.
    ("link_devices", "addr", "TEXT"),
    // Smart playlists: the rules as JSON (see `crate::smart`), when they were
    // last evaluated (unix seconds), and the file they were read from, if
    // any. `readonly` is a server's word that a playlist there takes no
    // edits, kept by clients that sync from it.
    ("playlists", "rules", "TEXT"),
    ("playlists", "refreshed_at", "INTEGER"),
    ("playlists", "source_path", "TEXT"),
    ("playlists", "readonly", "INTEGER NOT NULL DEFAULT 0"),
    // How far this device has read the server's play history: see
    // `remote::history`.
    ("remote_servers", "history_cursor", "TEXT"),
];

/// A UUIDv7 in SQL, for the triggers that give every new row its `uid`: a
/// trigger covers every insert, however it is written, where a Rust-side
/// default would have to be remembered at each one.
const SQL_UUID7: &str = "(SELECT substr(t, 1, 8) || '-' || substr(t, 9, 4) || '-7' || substr(r, 1, 3)
        || '-' || substr('89ab', 1 + (random() & 3), 1) || substr(r, 4, 3) || '-' || substr(r, 7, 12)
    FROM (SELECT printf('%012x', CAST((julianday('now') - 2440587.5) * 86400000 AS INTEGER)) AS t,
                 lower(hex(randomblob(9))) AS r))";

/// Add whichever of `ADDED_COLUMNS` the database lacks. Reads alone when none
/// are missing.
fn add_missing_columns(conn: &Connection) -> rusqlite::Result<()> {
    let had_remote_account = column_exists(conn, "playlists", "remote_account")?;
    for (table, column, ty) in ADDED_COLUMNS {
        if !column_exists(conn, table, column)? {
            conn.execute(&format!("ALTER TABLE {table} ADD COLUMN {column} {ty}"), [])?;
        }
    }

    // Playlists synced before `remote_account` existed belong to the server
    // last synced with. Recorded now, before a sync against a different one
    // can take their ids for its own.
    if !had_remote_account {
        conn.execute(
            "UPDATE playlists SET remote_account =
               (SELECT username || '@' || rtrim(url, '/') FROM remote_servers
                ORDER BY last_sync DESC LIMIT 1)
             WHERE remote_id IS NOT NULL",
            [],
        )?;
    }
    Ok(())
}

fn apply_migrations(conn: &Connection, found: i64) -> rusqlite::Result<()> {
    add_missing_columns(conn)?;

    // Left by the radio mode koan used to have: its similar-artist cache,
    // acoustic vectors and saved toggle. Nothing reads them.
    if found < 11 {
        conn.execute_batch(
            "DROP TABLE IF EXISTS similar_artists;
             DROP TABLE IF EXISTS track_vectors;
             DROP INDEX IF EXISTS idx_artists_mbid;",
        )?;
        for table in ["playback_state", "playback_position"] {
            if column_exists(conn, table, "radio_enabled")? {
                conn.execute(
                    &format!("ALTER TABLE {table} DROP COLUMN radio_enabled"),
                    [],
                )?;
            }
        }
    }

    // Cross-source dedup looks tracks up by recording id.
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_tracks_mbid ON tracks(mbid) WHERE mbid IS NOT NULL",
        [],
    )?;

    // Scans before version 3 never read MusicBrainz ids from tags. Forgetting
    // the files that came through without one has the next scan read them
    // again, which is what pairs them with the server's copy by id. Files that
    // have none are read once more and then left alone.
    if found < 3 {
        conn.execute(
            "DELETE FROM scan_cache WHERE track_id IN
               (SELECT id FROM tracks WHERE path IS NOT NULL AND mbid IS NULL)",
            [],
        )?;
    }

    // Locally-scanned albums were briefly stamped with the time the scan ran,
    // which pinned every one of them to the top of recently-added and buried
    // whatever the server actually considered new. Clearing the scan-time
    // values lets the next scan refill them from the files themselves; the
    // server's own ISO 8601 dates are left alone.
    conn.execute(
        "UPDATE albums SET added_at = NULL
           WHERE added_at IS NOT NULL AND added_at NOT LIKE '%T%Z'",
        [],
    )?;

    // Syncs before 0.36.2 stored a server's "no id" (`musicBrainzId: ""`) as
    // an empty string, which everything downstream took for an id: artist
    // info looked "" up instead of resolving one, and never showed anything.
    // Sort names, labels and genres came through the same way. Found-nothing
    // artist info is no longer cached, and what was is dropped. Idempotent,
    // and cheap once there are none.
    conn.execute_batch(
        "DELETE FROM artist_info WHERE bio IS NULL AND image_url IS NULL;
         UPDATE artists SET mbid = NULL WHERE mbid = '';
         UPDATE artists SET sort_name = NULL WHERE sort_name = '';
         UPDATE albums SET sort_name = NULL WHERE sort_name = '';
         UPDATE albums SET label = NULL WHERE label = '';
         UPDATE tracks SET genre = NULL WHERE genre = '';
         UPDATE albums SET mbid = NULL WHERE mbid = '';
         UPDATE tracks SET mbid = NULL WHERE mbid = '';",
    )?;
    // Stale-track removal once left the album of a moved or deleted record
    // behind with nothing in it, and a server lists it to everyone who syncs.
    conn.execute(
        "DELETE FROM albums WHERE NOT EXISTS
           (SELECT 1 FROM tracks WHERE tracks.album_id = albums.id)",
        [],
    )?;

    autoincrement_user_ids(conn)?;
    // Passwords were once also kept encrypted, for Subsonic token auth. A
    // password the server can read back is one an admin can read too.
    if column_exists(conn, "users", "sealed_password")? {
        conn.execute("ALTER TABLE users DROP COLUMN sealed_password", [])?;
    }
    // A client more than this far behind purges its whole cache instead.
    conn.execute(
        "DELETE FROM art_evictions WHERE seq < (SELECT MAX(seq) FROM art_evictions) - 50000",
        [],
    )?;
    cascade_play_history(conn)?;
    autoincrement_play_history(conn)?;
    snapshots_to_playlists(conn)?;
    per_user_favourites(conn)?;
    name_keys(conn)?;
    albums_without_name_uniqueness(conn)?;
    favourites_by_id(conn)?;
    merge_folded_artists(conn)?;
    merge_folded_albums(conn)?;
    // After the rebuilds above, which drop a table's indexes with it.
    // `idx_play_history_track` is a prefix of `idx_play_history_track_played`.
    // The favourites key leads with the user, and merging two tracks repoints
    // every user's favourite by path.
    conn.execute_batch(
        "CREATE INDEX IF NOT EXISTS idx_play_history_user ON play_history(user_id, played_at);
         CREATE INDEX IF NOT EXISTS idx_play_history_track_played ON play_history(track_id, played_at);
         DROP INDEX IF EXISTS idx_play_history_track;
         DROP INDEX IF EXISTS idx_favourites_path;
         CREATE INDEX IF NOT EXISTS idx_favourites_track ON favourites(track_id);
         CREATE INDEX IF NOT EXISTS idx_favourite_albums_album ON favourite_albums(album_id);
         CREATE INDEX IF NOT EXISTS idx_favourite_artists_artist ON favourite_artists(artist_id);
         CREATE INDEX IF NOT EXISTS idx_playlists_user ON playlists(user_id);
         CREATE TRIGGER IF NOT EXISTS users_personal_data AFTER DELETE ON users BEGIN
             DELETE FROM favourites WHERE user_id = OLD.id;
             DELETE FROM favourite_albums WHERE user_id = OLD.id;
             DELETE FROM favourite_artists WHERE user_id = OLD.id;
             DELETE FROM play_history WHERE user_id = OLD.id;
             DELETE FROM play_history_forgotten WHERE user_id = OLD.id;
             DELETE FROM playlists WHERE user_id = OLD.id;
             DELETE FROM shares WHERE user_id = OLD.id;
         END;",
    )?;
    conn.execute_batch(
        "CREATE INDEX IF NOT EXISTS idx_track_ratings_track ON track_ratings(track_id);
         CREATE INDEX IF NOT EXISTS idx_album_ratings_album ON album_ratings(album_id);
         CREATE INDEX IF NOT EXISTS idx_artist_ratings_artist ON artist_ratings(artist_id);
         CREATE TRIGGER IF NOT EXISTS users_ratings AFTER DELETE ON users BEGIN
             DELETE FROM track_ratings WHERE user_id = OLD.id;
             DELETE FROM album_ratings WHERE user_id = OLD.id;
             DELETE FROM artist_ratings WHERE user_id = OLD.id;
         END;",
    )?;
    conn.execute_batch(
        "CREATE INDEX IF NOT EXISTS idx_play_queue_entries_track ON play_queue_entries(track_id);
         CREATE TRIGGER IF NOT EXISTS users_play_queues AFTER DELETE ON users BEGIN
             DELETE FROM play_queue_entries WHERE user_id = OLD.id;
             DELETE FROM play_queues WHERE user_id = OLD.id;
         END;",
    )?;
    conn.execute_batch(
        "CREATE INDEX IF NOT EXISTS idx_bookmarks_track ON bookmarks(track_id);
         CREATE TRIGGER IF NOT EXISTS users_bookmarks AFTER DELETE ON users BEGIN
             DELETE FROM bookmarks WHERE user_id = OLD.id;
         END;",
    )?;
    crate::db::queries::auth::adopt_local_rows(conn)?;

    // The position moved out of the queue's row; the last one saved comes too.
    if found < 9 {
        conn.execute(
            "INSERT OR IGNORE INTO playback_position
                 (id, cursor_id, position_ms, was_playing)
             SELECT 1, cursor_id, position_ms, was_playing
               FROM playback_state WHERE id = 1",
            [],
        )?;
    }

    for table in ["artists", "albums", "tracks", "playlists"] {
        conn.execute_batch(&format!(
            "CREATE UNIQUE INDEX IF NOT EXISTS idx_{table}_uid ON {table}(uid);
             CREATE TRIGGER IF NOT EXISTS {table}_uid AFTER INSERT ON {table}
               WHEN NEW.uid IS NULL
             BEGIN
               UPDATE {table} SET uid = {SQL_UUID7} WHERE id = NEW.id;
             END;"
        ))?;
        backfill_uids(conn, table)?;
    }

    // Each file and each server entry became a source row of its own. Built
    // once from what the tracks hold, which is one source's names, then
    // replaced by each source's own: the next scan reads every file again and
    // the next sync walks the whole server.
    if found < 14 {
        if found < 3 {
            crate::db::queries::tracks::clear_zero_discs(conn)?;
        }
        crate::db::queries::tracks::merge_spelling_twins(conn)?;
        crate::db::queries::sources::build_from_tracks(conn).map_err(|e| match e {
            crate::db::connection::DbError::Sqlite(e) => e,
            other => rusqlite::Error::ToSqlConversionFailure(Box::new(other)),
        })?;
        conn.execute_batch(
            "DELETE FROM scan_cache;
             UPDATE remote_servers SET library_version = NULL;",
        )?;
    }

    Ok(())
}

/// Turn saved queues into playlists, then drop the table they lived in.
///
/// Snapshots were playlists with a resume position — a whole second feature to
/// maintain for one number, and one the server had no idea about. The track
/// lists are real work someone did, so they come across; only the position is
/// lost. The Subsonic API already served snapshots as playlists, so its clients
/// see the same names either side of this.
fn snapshots_to_playlists(conn: &Connection) -> rusqlite::Result<()> {
    if !table_exists(conn, "queue_snapshots")? {
        return Ok(());
    }

    let mut saved: Vec<(String, String, String)> = Vec::new();
    {
        let mut stmt = conn.prepare("SELECT name, queue_json, created_at FROM queue_snapshots")?;
        let rows = stmt.query_map([], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, String>(1)?,
                row.get::<_, Option<String>>(2)?.unwrap_or_default(),
            ))
        })?;
        for row in rows {
            saved.push(row?);
        }
    }

    for (name, json, created_at) in saved {
        let paths: Vec<String> = serde_json::from_str::<Vec<serde_json::Value>>(&json)
            .unwrap_or_default()
            .into_iter()
            .filter_map(|item| {
                item.get("path")
                    .and_then(|p| p.as_str())
                    .map(|p| p.to_string())
            })
            .collect();

        conn.execute(
            "INSERT INTO playlists (name, created_at, changed_at)
             VALUES (?1, COALESCE(NULLIF(?2, ''), datetime('now')), datetime('now'))",
            rusqlite::params![name, created_at],
        )?;
        let playlist_id = conn.last_insert_rowid();

        let mut position = 0i64;
        for path in paths {
            let track_id: Option<i64> = conn
                .query_row("SELECT id FROM tracks WHERE path = ?1", [&path], |r| {
                    r.get(0)
                })
                .ok();
            // A snapshot held enough metadata to play a file that was never
            // indexed. A playlist points at library rows, so anything with no
            // row behind it cannot come across.
            if let Some(track_id) = track_id {
                conn.execute(
                    "INSERT INTO playlist_tracks (playlist_id, position, track_id)
                     VALUES (?1, ?2, ?3)",
                    rusqlite::params![playlist_id, position, track_id],
                )?;
                position += 1;
            }
        }
    }

    conn.execute("DROP TABLE queue_snapshots", [])?;
    Ok(())
}

/// Give every row of `table` that has none a `uid`, in row order so the ids
/// keep the order the rows were added in. Rows written since the trigger
/// existed already have one, so after the first run this finds nothing.
fn backfill_uids(conn: &Connection, table: &str) -> rusqlite::Result<()> {
    let ids: Vec<i64> = conn
        .prepare(&format!(
            "SELECT id FROM {table} WHERE uid IS NULL ORDER BY id"
        ))?
        .query_map([], |r| r.get(0))?
        .collect::<rusqlite::Result<_>>()?;
    if ids.is_empty() {
        return Ok(());
    }
    conn.execute_batch("SAVEPOINT backfill_uids")?;
    let written = (|| {
        let mut update = conn.prepare(&format!("UPDATE {table} SET uid = ?1 WHERE id = ?2"))?;
        for id in ids {
            update.execute(rusqlite::params![uuid::Uuid::now_v7().to_string(), id])?;
        }
        Ok(())
    })();
    match written {
        Ok(()) => conn.execute_batch("RELEASE backfill_uids"),
        Err(e) => {
            conn.execute_batch("ROLLBACK TO backfill_uids; RELEASE backfill_uids")?;
            Err(e)
        }
    }
}

/// Rebuild the favourite tables keyed by user, keeping every row.
///
/// They were keyed by what was favourited alone, and SQLite cannot change a
/// primary key in place. Existing rows come across as the implicit local user;
/// `adopt_local_rows` then hands them to the first admin, if there is one.
fn per_user_favourites(conn: &Connection) -> rusqlite::Result<()> {
    if column_exists(conn, "favourites", "user_id")? {
        return Ok(());
    }
    // A trigger naming a table mid-rebuild fails the rename that completes
    // it. `apply_migrations` recreates it afterwards.
    conn.execute_batch(
        "DROP TRIGGER IF EXISTS users_personal_data;
         SAVEPOINT rebuild;
         CREATE TABLE favourites_new (
             user_id     INTEGER NOT NULL DEFAULT 0,
             track_path  TEXT NOT NULL,
             created_at  TEXT DEFAULT (datetime('now')),
             PRIMARY KEY (user_id, track_path)
         );
         INSERT INTO favourites_new (track_path, created_at)
             SELECT track_path, created_at FROM favourites;
         DROP TABLE favourites;
         ALTER TABLE favourites_new RENAME TO favourites;

         CREATE TABLE favourite_albums_new (
             user_id     INTEGER NOT NULL DEFAULT 0,
             artist_name TEXT NOT NULL,
             album_title TEXT NOT NULL,
             created_at  TEXT DEFAULT (datetime('now')),
             PRIMARY KEY (user_id, artist_name, album_title)
         );
         INSERT INTO favourite_albums_new (artist_name, album_title, created_at)
             SELECT artist_name, album_title, created_at FROM favourite_albums;
         DROP TABLE favourite_albums;
         ALTER TABLE favourite_albums_new RENAME TO favourite_albums;

         CREATE TABLE favourite_artists_new (
             user_id     INTEGER NOT NULL DEFAULT 0,
             artist_name TEXT NOT NULL,
             created_at  TEXT DEFAULT (datetime('now')),
             PRIMARY KEY (user_id, artist_name)
         );
         INSERT INTO favourite_artists_new (artist_name, created_at)
             SELECT artist_name, created_at FROM favourite_artists;
         DROP TABLE favourite_artists;
         ALTER TABLE favourite_artists_new RENAME TO favourite_artists;
         RELEASE rebuild;",
    )
}

/// Whether `table` exists.
fn table_exists(conn: &Connection, table: &str) -> rusqlite::Result<bool> {
    let found: i64 = conn.query_row(
        "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
        [table],
        |r| r.get(0),
    )?;
    Ok(found > 0)
}

/// Keep `artists.name_key` and `albums.title_key` filled, and indexed for the
/// lookups that match on them.
fn name_keys(conn: &Connection) -> rusqlite::Result<()> {
    conn.execute_batch(
        "CREATE TRIGGER IF NOT EXISTS artists_name_key AFTER INSERT ON artists
           WHEN NEW.name_key IS NULL
         BEGIN UPDATE artists SET name_key = koan_fold(NEW.name) WHERE id = NEW.id; END;
         CREATE TRIGGER IF NOT EXISTS artists_name_key_renamed AFTER UPDATE OF name ON artists
         BEGIN UPDATE artists SET name_key = koan_fold(NEW.name) WHERE id = NEW.id; END;
         UPDATE artists SET name_key = koan_fold(name) WHERE name_key IS NULL;
         CREATE INDEX IF NOT EXISTS idx_artists_name_key ON artists(name_key);
         DROP INDEX IF EXISTS idx_artists_name_nocase;",
    )?;
    album_title_keys(conn)
}

fn album_title_keys(conn: &Connection) -> rusqlite::Result<()> {
    conn.execute_batch(
        "CREATE TRIGGER IF NOT EXISTS albums_title_key AFTER INSERT ON albums
           WHEN NEW.title_key IS NULL
         BEGIN UPDATE albums SET title_key = koan_fold(NEW.title) WHERE id = NEW.id; END;
         CREATE TRIGGER IF NOT EXISTS albums_title_key_renamed AFTER UPDATE OF title ON albums
         BEGIN UPDATE albums SET title_key = koan_fold(NEW.title) WHERE id = NEW.id; END;
         UPDATE albums SET title_key = koan_fold(title) WHERE title_key IS NULL;
         CREATE INDEX IF NOT EXISTS idx_albums_title_key ON albums(title_key, artist_id);",
    )
}

/// Rebuild `albums` without `UNIQUE(title, artist_id)`. Two editions with the
/// same title and album artist are two albums when their release ids differ;
/// see `queries::get_or_create_album`.
fn albums_without_name_uniqueness(conn: &Connection) -> rusqlite::Result<()> {
    let sql: String = conn.query_row(
        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'albums'",
        [],
        |r| r.get(0),
    )?;
    if !sql.contains("UNIQUE(title, artist_id)") {
        return Ok(());
    }
    // Pragma changes are no-ops inside a transaction, so this must bracket it.
    conn.pragma_update(None, "foreign_keys", "off")?;
    let rebuild = conn.execute_batch(
        "SAVEPOINT rebuild;
         CREATE TABLE albums_new (
             id           INTEGER PRIMARY KEY,
             title        TEXT NOT NULL,
             artist_id    INTEGER REFERENCES artists(id),
             date         TEXT,
             total_discs  INTEGER,
             total_tracks INTEGER,
             codec        TEXT,
             label        TEXT,
             remote_id    TEXT,
             added_at     TEXT,
             mbid         TEXT,
             sort_name    TEXT,
             uid          TEXT,
             title_key    TEXT
         );
         INSERT INTO albums_new (id, title, artist_id, date, total_discs, total_tracks, codec,
                                 label, remote_id, added_at, mbid, sort_name, uid, title_key)
             SELECT id, title, artist_id, date, total_discs, total_tracks, codec,
                    label, remote_id, added_at, mbid, sort_name, uid, title_key FROM albums;
         DROP TABLE albums;
         ALTER TABLE albums_new RENAME TO albums;
         CREATE INDEX IF NOT EXISTS idx_albums_artist ON albums(artist_id);
         CREATE INDEX IF NOT EXISTS idx_albums_remote_id ON albums(remote_id)
             WHERE remote_id IS NOT NULL;
         CREATE TRIGGER IF NOT EXISTS evict_album_art AFTER DELETE ON albums
             BEGIN INSERT INTO art_evictions (kind, id) VALUES ('album', old.id); END;
         RELEASE rebuild;",
    );
    conn.pragma_update(None, "foreign_keys", "on")?;
    rebuild?;
    album_title_keys(conn)
}

/// Favourites from names and paths to row ids. A favourite whose track,
/// album or artist is not in the library names nothing any more, and goes.
fn favourites_by_id(conn: &Connection) -> rusqlite::Result<()> {
    if !column_exists(conn, "favourites", "track_path")? {
        return Ok(());
    }
    // A trigger naming a table mid-rebuild fails the rename that completes
    // it. `apply_migrations` recreates it afterwards.
    conn.execute_batch(
        "DROP TRIGGER IF EXISTS users_personal_data;
         SAVEPOINT rebuild;
         CREATE TABLE favourites_new (
             user_id     INTEGER NOT NULL DEFAULT 0,
             track_id    INTEGER NOT NULL REFERENCES tracks(id) ON DELETE CASCADE,
             created_at  TEXT DEFAULT (datetime('now')),
             PRIMARY KEY (user_id, track_id)
         );
         INSERT OR IGNORE INTO favourites_new (user_id, track_id, created_at)
             SELECT f.user_id, t.id, f.created_at FROM favourites f JOIN tracks t ON t.path = f.track_path
             UNION ALL
             SELECT f.user_id, t.id, f.created_at FROM favourites f JOIN tracks t ON t.cached_path = f.track_path
             UNION ALL
             SELECT f.user_id, t.id, f.created_at FROM favourites f JOIN tracks t ON t.remote_url = f.track_path;
         DROP TABLE favourites;
         ALTER TABLE favourites_new RENAME TO favourites;

         CREATE TABLE favourite_albums_new (
             user_id     INTEGER NOT NULL DEFAULT 0,
             album_id    INTEGER NOT NULL REFERENCES albums(id) ON DELETE CASCADE,
             created_at  TEXT DEFAULT (datetime('now')),
             PRIMARY KEY (user_id, album_id)
         );
         INSERT OR IGNORE INTO favourite_albums_new (user_id, album_id, created_at)
             SELECT f.user_id, al.id, f.created_at FROM favourite_albums f
               JOIN artists ar ON ar.name_key = koan_fold(f.artist_name)
               JOIN albums al ON al.artist_id = ar.id AND al.title_key = koan_fold(f.album_title);
         DROP TABLE favourite_albums;
         ALTER TABLE favourite_albums_new RENAME TO favourite_albums;

         CREATE TABLE favourite_artists_new (
             user_id     INTEGER NOT NULL DEFAULT 0,
             artist_id   INTEGER NOT NULL REFERENCES artists(id) ON DELETE CASCADE,
             created_at  TEXT DEFAULT (datetime('now')),
             PRIMARY KEY (user_id, artist_id)
         );
         INSERT OR IGNORE INTO favourite_artists_new (user_id, artist_id, created_at)
             SELECT f.user_id, ar.id, f.created_at FROM favourite_artists f
               JOIN artists ar ON ar.name_key = koan_fold(f.artist_name);
         DROP TABLE favourite_artists;
         ALTER TABLE favourite_artists_new RENAME TO favourite_artists;
         RELEASE rebuild;",
    )
}

/// Merge artists whose names fold to the same key, which matching treats as
/// one: tags spell the same act "The Squire Of Gothos" on one record and
/// "of" on the next, or a server in capitals. The row with the most albums,
/// then tracks, keeps its name.
fn merge_folded_artists(conn: &Connection) -> rusqlite::Result<()> {
    let groups: Vec<String> = conn
        .prepare("SELECT name_key FROM artists GROUP BY name_key HAVING COUNT(*) > 1")?
        .query_map([], |r| r.get(0))?
        .collect::<rusqlite::Result<_>>()?;
    for key in groups {
        let ids: Vec<i64> = conn
            .prepare(
                "SELECT a.id FROM artists a WHERE a.name_key = ?1
                 ORDER BY (SELECT COUNT(*) FROM albums WHERE artist_id = a.id) DESC,
                          (SELECT COUNT(*) FROM tracks WHERE artist_id = a.id) DESC,
                          a.id",
            )?
            .query_map([&key], |r| r.get(0))?
            .collect::<rusqlite::Result<_>>()?;
        let Some((&keep, rest)) = ids.split_first() else {
            continue;
        };
        for &gone in rest {
            for sql in [
                "UPDATE albums SET artist_id = ?1 WHERE artist_id = ?2",
                "UPDATE tracks SET artist_id = ?1 WHERE artist_id = ?2",
                "UPDATE OR IGNORE favourite_artists SET artist_id = ?1 WHERE artist_id = ?2",
                "UPDATE shares SET subject_id = ?1 WHERE kind = 'artist' AND subject_id = ?2",
                "UPDATE artists SET
                     remote_id = COALESCE(remote_id, (SELECT remote_id FROM artists WHERE id = ?2)),
                     mbid = COALESCE(mbid, (SELECT mbid FROM artists WHERE id = ?2)),
                     sort_name = COALESCE(sort_name, (SELECT sort_name FROM artists WHERE id = ?2))
                 WHERE id = ?1",
            ] {
                conn.execute(sql, [keep, gone])?;
            }
            for sql in [
                "DELETE FROM favourite_artists WHERE artist_id = ?1",
                "DELETE FROM artist_info WHERE artist_id = ?1",
                "DELETE FROM artists WHERE id = ?1",
            ] {
                conn.execute(sql, [gone])?;
            }
        }
    }
    Ok(())
}

/// Merge albums that are one by `get_or_create_album`'s reckoning: the same
/// folded title and album artist, and the same release or none. Arises where
/// the old key compared titles bytewise, or once folded artists merged.
fn merge_folded_albums(conn: &Connection) -> rusqlite::Result<()> {
    let twins: Vec<(i64, i64)> = conn
        .prepare(
            "SELECT MIN(k.id), g.id FROM albums g JOIN albums k
                ON k.title_key = g.title_key AND k.artist_id IS g.artist_id AND k.id < g.id
               AND (k.mbid IS g.mbid OR k.mbid IS NULL OR g.mbid IS NULL)
             GROUP BY g.id",
        )?
        .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))?
        .collect::<rusqlite::Result<_>>()?;
    for (keep, gone) in twins {
        let both: bool = conn.query_row(
            "SELECT COUNT(*) = 2 FROM albums WHERE id IN (?1, ?2)",
            [keep, gone],
            |r| r.get(0),
        )?;
        if both {
            crate::db::queries::merge_albums(conn, keep, gone)?;
        }
    }
    Ok(())
}

/// Give `play_history.track_id` its `ON DELETE CASCADE`.
///
/// The column shipped as a bare `REFERENCES`, which under `foreign_keys = ON`
/// makes a track with history undeletable unless the caller remembers to clear
/// the history first. One caller does; the constraint should not depend on the
/// next one remembering. SQLite cannot alter a constraint in place, so the
/// table is rebuilt.
fn cascade_play_history(conn: &Connection) -> rusqlite::Result<()> {
    if fk_cascades(conn, "play_history")? {
        return Ok(());
    }

    // Pragma changes are no-ops inside a transaction, so this must bracket it.
    conn.pragma_update(None, "foreign_keys", "off")?;
    // Recreated by `apply_migrations`; see `per_user_favourites`.
    let rebuild = conn.execute_batch(
        "DROP TRIGGER IF EXISTS users_personal_data;
         SAVEPOINT rebuild;
         CREATE TABLE play_history_new (
             id          INTEGER PRIMARY KEY,
             track_id    INTEGER REFERENCES tracks(id) ON DELETE CASCADE,
             played_at   INTEGER NOT NULL,
             duration_ms INTEGER,
             source      TEXT DEFAULT 'local',
             user_id     INTEGER NOT NULL DEFAULT 0
         );
         -- Entries whose track has already gone would violate the new
         -- constraint the moment it is enforced. They are unreachable anyway.
         INSERT INTO play_history_new (id, track_id, played_at, duration_ms, source, user_id)
             SELECT id, track_id, played_at, duration_ms, source, user_id FROM play_history
             WHERE track_id IS NULL OR track_id IN (SELECT id FROM tracks);
         DROP TABLE play_history;
         ALTER TABLE play_history_new RENAME TO play_history;
         CREATE INDEX IF NOT EXISTS idx_play_history_track_played
             ON play_history(track_id, played_at);
         CREATE INDEX IF NOT EXISTS idx_play_history_time ON play_history(played_at);
         RELEASE rebuild;",
    );
    conn.pragma_update(None, "foreign_keys", "on")?;
    rebuild
}

/// Give `play_history.id` `AUTOINCREMENT`, keeping every row and id.
///
/// A server's history is paged by id, so a device that has read up to an id
/// asks only for those after it. Without the keyword SQLite hands the highest
/// id out again once that play is forgotten, and the play given it would never
/// reach a device that had already read past it.
fn autoincrement_play_history(conn: &Connection) -> rusqlite::Result<()> {
    let sql: String = conn.query_row(
        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'play_history'",
        [],
        |r| r.get(0),
    )?;
    if sql.to_ascii_uppercase().contains("AUTOINCREMENT") {
        return Ok(());
    }

    // Pragma changes are no-ops inside a transaction, so this must bracket it.
    conn.pragma_update(None, "foreign_keys", "off")?;
    // Recreated by `apply_migrations`; see `per_user_favourites`. So are the
    // indexes, which go with the table.
    let rebuild = conn.execute_batch(
        "DROP TRIGGER IF EXISTS users_personal_data;
         SAVEPOINT rebuild;
         CREATE TABLE play_history_new (
             id          INTEGER PRIMARY KEY AUTOINCREMENT,
             track_id    INTEGER REFERENCES tracks(id) ON DELETE CASCADE,
             played_at   INTEGER NOT NULL,
             duration_ms INTEGER,
             source      TEXT DEFAULT 'local',
             user_id     INTEGER NOT NULL DEFAULT 0
         );
         INSERT INTO play_history_new (id, track_id, played_at, duration_ms, source, user_id)
             SELECT id, track_id, played_at, duration_ms, source, user_id FROM play_history;
         DROP TABLE play_history;
         ALTER TABLE play_history_new RENAME TO play_history;
         CREATE INDEX IF NOT EXISTS idx_play_history_time ON play_history(played_at);
         RELEASE rebuild;",
    );
    // Back to before the rebuild and no further: inside an upgrade, a bare
    // ROLLBACK would end the upgrade's transaction and hide this error behind
    // the failure of the rollback that follows.
    if rebuild.is_err() {
        let _ = conn.execute_batch("ROLLBACK TO rebuild; RELEASE rebuild");
    }
    conn.pragma_update(None, "foreign_keys", "on")?;
    rebuild
}

/// Give `users.id` `AUTOINCREMENT`, keeping every row and id.
///
/// Without it SQLite hands the highest id out again once that account is
/// deleted, and whatever still names the old account by id — a live access
/// token, a link — would then name the new one. SQLite cannot add the keyword
/// in place, so the table is rebuilt. Ids deleted before this ran may still be
/// reused once; none after.
fn autoincrement_user_ids(conn: &Connection) -> rusqlite::Result<()> {
    let sql: String = conn.query_row(
        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'users'",
        [],
        |r| r.get(0),
    )?;
    if sql.to_ascii_uppercase().contains("AUTOINCREMENT") {
        return Ok(());
    }

    // Off, or dropping `users` would cascade into every table that names it.
    // Pragma changes are no-ops inside a transaction, so this must bracket it.
    conn.pragma_update(None, "foreign_keys", "off")?;
    // Recreated by `apply_migrations`; see `per_user_favourites`.
    let rebuild = conn.execute_batch(
        "DROP TRIGGER IF EXISTS users_personal_data;
         SAVEPOINT rebuild;
         CREATE TABLE users_new (
             id              INTEGER PRIMARY KEY AUTOINCREMENT,
             username        TEXT NOT NULL UNIQUE,
             password_hash   TEXT NOT NULL,
             role            TEXT NOT NULL DEFAULT 'user' CHECK (role IN ('admin', 'user', 'readonly')),
             created_at      TEXT DEFAULT (datetime('now'))
         );
         INSERT INTO users_new (id, username, password_hash, role, created_at)
             SELECT id, username, password_hash, role, created_at FROM users;
         DROP TABLE users;
         ALTER TABLE users_new RENAME TO users;
         RELEASE rebuild;",
    );
    // Back to before the rebuild and no further: inside an upgrade, a bare
    // ROLLBACK would end the upgrade's transaction and hide this error behind
    // the failure of the rollback that follows.
    if rebuild.is_err() {
        let _ = conn.execute_batch("ROLLBACK TO rebuild; RELEASE rebuild");
    }
    conn.pragma_update(None, "foreign_keys", "on")?;
    rebuild
}

/// Whether every foreign key on `table` deletes its rows with the parent.
fn fk_cascades(conn: &Connection, table: &str) -> rusqlite::Result<bool> {
    let mut stmt = conn.prepare(&format!("PRAGMA foreign_key_list({table})"))?;
    let mut rows = stmt.query([])?;
    let mut any = false;
    while let Some(row) = rows.next()? {
        any = true;
        // Column 6 is `on_delete`.
        if !row.get::<_, String>(6)?.eq_ignore_ascii_case("CASCADE") {
            return Ok(false);
        }
    }
    Ok(any)
}

/// Whether `table` already has `column`.
///
/// PRAGMA cannot take a bound parameter for the table name, so the name is
/// interpolated — every caller passes a literal from `ADDED_COLUMNS`, never
/// user input.
fn column_exists(conn: &Connection, table: &str, column: &str) -> rusqlite::Result<bool> {
    let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
    let mut rows = stmt.query([])?;
    while let Some(row) = rows.next()? {
        if row.get::<_, String>(1)? == column {
            return Ok(true);
        }
    }
    Ok(false)
}

#[cfg(test)]
mod tests {
    #[test]
    fn case_duplicate_artists_are_merged_on_open() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let conn = rusqlite::Connection::open(&path).unwrap();
            super::create_tables(&conn).unwrap();
            conn.execute_batch(
                "INSERT INTO artists (id, name) VALUES (1, 'The Squire of Gothos'), (2, 'The Squire Of Gothos');
                 INSERT INTO albums (id, title, artist_id) VALUES (1, 'We Do Scorpion Things', 1);
                 INSERT INTO tracks (title, album_id, artist_id, path) VALUES ('Dark Ting', 1, 2, '/a.flac');
                 PRAGMA user_version = 8;",
            )
            .unwrap();
        }
        let conn = rusqlite::Connection::open(&path).unwrap();
        super::create_tables(&conn).unwrap();
        let count = |sql: &str| -> i64 { conn.query_row(sql, [], |r| r.get(0)).unwrap() };
        assert_eq!(count("SELECT COUNT(*) FROM artists"), 1);
        assert_eq!(
            count("SELECT artist_id FROM tracks"),
            1,
            "onto the album owner's spelling"
        );
    }

    #[test]
    fn an_album_both_spellings_hold_becomes_one_album() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let conn = rusqlite::Connection::open(&path).unwrap();
            super::create_tables(&conn).unwrap();
            conn.execute_batch(
                "INSERT INTO artists (id, name) VALUES (101, 'Tove Lo'), (102, 'TOVE LO');
                 INSERT INTO albums (id, title, artist_id) VALUES (101, 'Habits', 101), (102, 'Habits', 102), (103, 'Other', 102);
                 INSERT INTO tracks (title, album_id, artist_id, path) VALUES
                   ('a', 101, 101, '/a.flac'), ('b', 102, 102, '/b.flac'), ('c', 103, 102, '/c.flac');
                 PRAGMA user_version = 8;",
            )
            .unwrap();
        }
        let conn = rusqlite::Connection::open(&path).unwrap();
        super::create_tables(&conn).expect("opens despite the clash");
        let count = |sql: &str| -> i64 { conn.query_row(sql, [], |r| r.get(0)).unwrap() };
        assert_eq!(
            count("SELECT COUNT(*) FROM artists WHERE lower(name) = 'tove lo'"),
            1
        );
        assert_eq!(
            count("SELECT COUNT(*) FROM albums WHERE title IN ('Habits', 'Other')"),
            2
        );
        assert_eq!(
            count(
                "SELECT COUNT(*) FROM tracks t JOIN albums al ON al.id = t.album_id
                  WHERE al.title = 'Habits'"
            ),
            2,
            "Habits once, holding both copies' tracks"
        );
    }

    #[test]
    fn deleted_ids_are_logged_for_caches() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.execute_batch(
            "INSERT INTO artists (id, name) VALUES (7, 'A');
             INSERT INTO albums (id, title, artist_id) VALUES (9, 'B', 7);
             DELETE FROM albums WHERE id = 9;",
        )
        .unwrap();
        let logged: (String, i64) = conn
            .query_row("SELECT kind, id FROM art_evictions", [], |r| {
                Ok((r.get(0)?, r.get(1)?))
            })
            .unwrap();
        assert_eq!(logged, ("album".to_string(), 9));
    }

    #[test]
    fn empty_musicbrainz_ids_are_cleared_on_open() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let conn = rusqlite::Connection::open(&path).unwrap();
            super::create_tables(&conn).unwrap();
            conn.execute("INSERT INTO artists (name, mbid) VALUES ('Crass', '')", [])
                .unwrap();
            conn.execute(
                "INSERT INTO artist_info (artist_id, fetched_at)
                   SELECT id, 1 FROM artists WHERE name = 'Crass'",
                [],
            )
            .unwrap();
            conn.pragma_update(None, "user_version", 8).unwrap();
        }
        let conn = rusqlite::Connection::open(&path).unwrap();
        super::create_tables(&conn).unwrap();
        let mbid: Option<String> = conn
            .query_row("SELECT mbid FROM artists WHERE name = 'Crass'", [], |r| {
                r.get(0)
            })
            .unwrap();
        assert_eq!(mbid, None);
        let misses: i64 = conn
            .query_row("SELECT COUNT(*) FROM artist_info", [], |r| r.get(0))
            .unwrap();
        assert_eq!(misses, 0, "cached misses are dropped");
    }

    use super::*;
    use crate::db::connection::Database;

    /// Queries that answer a question about a handful of rows, and must not
    /// read the library to do it.
    ///
    /// The planner will happily fall back to a full scan when a query is
    /// written in a shape no index can serve — an `OR` spanning two columns, a
    /// `LIKE` pattern, a collation the index does not use — and nothing about
    /// the result says it happened. It costs, it does not fail, and it gets
    /// worse with the size of somebody's library. So the plans are asserted.
    #[test]
    fn hot_queries_do_not_scan() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();

        let plan = |sql: &str| -> Vec<String> {
            let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
            let nulls = vec![rusqlite::types::Null; stmt.parameter_count()];
            stmt.query_map(rusqlite::params_from_iter(nulls), |r| r.get::<_, String>(3))
                .unwrap()
                .map(Result::unwrap)
                .collect()
        };

        let cases: &[(&str, &str)] = &[
            (
                "a track by any of its three paths",
                "SELECT id FROM tracks WHERE path = ?1 OR cached_path = ?1 OR remote_url = ?1",
            ),
            (
                "an artist's tracks, own or album credit",
                "SELECT t.id FROM tracks t LEFT JOIN albums al ON t.album_id = al.id
                  WHERE t.artist_id = ?1
                     OR t.album_id IN (SELECT id FROM albums WHERE artist_id = ?1)",
            ),
            (
                "every track a user favourited",
                "SELECT track_id FROM favourites WHERE user_id = ?1",
            ),
            (
                "tracks under a folder",
                "SELECT id FROM tracks WHERE path >= ?1 AND path < ?2",
            ),
            (
                "an album by the server's id",
                "SELECT id FROM albums WHERE remote_id = ?1",
            ),
            (
                "an artist by the server's id",
                "SELECT id FROM artists WHERE remote_id = ?1",
            ),
            (
                "an artist by name, however it is spelled",
                "SELECT id FROM artists WHERE name_key = ?1",
            ),
            (
                "a scan cache entry by track",
                "SELECT path FROM scan_cache WHERE track_id = ?1",
            ),
            (
                "one organize batch",
                "SELECT id FROM organize_log WHERE batch_id = ?1",
            ),
            (
                "the albums a genre spans",
                "SELECT DISTINCT album_id FROM tracks WHERE genre = ?1 COLLATE NOCASE",
            ),
            (
                "organize history repointed to a merged track",
                "UPDATE organize_log SET track_id = ?1 WHERE track_id = ?2",
            ),
            (
                "shares repointed to a merged track",
                "UPDATE share_tracks SET track_id = ?1 WHERE track_id = ?2",
            ),
            (
                "favourites repointed to a merged track",
                "UPDATE OR IGNORE favourites SET track_id = ?1 WHERE track_id = ?2",
            ),
            (
                "a track's last play",
                "SELECT MAX(played_at) FROM play_history WHERE track_id = ?1",
            ),
        ];

        for (what, sql) in cases {
            let steps = plan(sql);
            assert!(
                !steps.iter().any(|s| s.starts_with("SCAN")),
                "{what}: reads the whole table\n  {}",
                steps.join("\n  ")
            );
        }
    }

    #[test]
    fn clears_scan_time_added_at_but_keeps_the_servers() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.execute_batch(
            "INSERT INTO artists (id, name) VALUES (1, 'Klaxons');
             INSERT INTO albums (id, title, artist_id, added_at)
               VALUES (1, 'Local', 1, '2026-08-23 12:14:57'),
                      (2, 'Remote', 1, '2026-08-06T22:53:14.851697506Z'),
                      (3, 'Neither', 1, NULL);
             INSERT INTO tracks (title, album_id, artist_id, path)
               VALUES ('a', 1, 1, '/a'), ('b', 2, 1, '/b'), ('c', 3, 1, '/c');
             PRAGMA user_version = 8;",
        )
        .unwrap();

        create_tables(&conn).unwrap();

        let added = |id: i64| -> Option<String> {
            conn.query_row("SELECT added_at FROM albums WHERE id = ?1", [id], |r| {
                r.get(0)
            })
            .unwrap()
        };
        assert_eq!(added(1), None, "scan-time stamp cleared");
        assert_eq!(
            added(2).as_deref(),
            Some("2026-08-06T22:53:14.851697506Z"),
            "the server's own date is left alone"
        );
        assert_eq!(added(3), None);
    }

    #[test]
    fn saved_queues_become_playlists() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        // The table as it stood before playlists existed.
        conn.execute_batch(
            "CREATE TABLE queue_snapshots (
                 id          INTEGER PRIMARY KEY,
                 name        TEXT NOT NULL UNIQUE,
                 queue_json  TEXT NOT NULL DEFAULT '[]',
                 cursor_path TEXT,
                 position_ms INTEGER NOT NULL DEFAULT 0,
                 created_at  TEXT DEFAULT (datetime('now'))
             );
             INSERT INTO artists (id, name) VALUES (1, 'Klaxons');
             INSERT INTO albums (id, title, artist_id) VALUES (1, 'Myths', 1);
             INSERT INTO tracks (id, album_id, artist_id, title, path)
               VALUES (1, 1, 1, 'Atlantis', '/music/atlantis.flac'),
                      (2, 1, 1, 'Golden Skans', '/music/golden.flac');
             INSERT INTO queue_snapshots (name, queue_json, created_at) VALUES
               ('techno',
                '[{\"path\":\"/music/golden.flac\"},{\"path\":\"/music/nowhere.flac\"},{\"path\":\"/music/atlantis.flac\"}]',
                '2026-01-01 10:00:00');
             PRAGMA user_version = 8;",
        )
        .unwrap();

        create_tables(&conn).unwrap();

        assert!(!table_exists(&conn, "queue_snapshots").unwrap());
        let (id, name, created): (i64, String, String) = conn
            .query_row("SELECT id, name, created_at FROM playlists", [], |r| {
                Ok((r.get(0)?, r.get(1)?, r.get(2)?))
            })
            .unwrap();
        assert_eq!(name, "techno");
        assert_eq!(created, "2026-01-01 10:00:00", "when it was saved is kept");

        let mut stmt = conn
            .prepare(
                "SELECT track_id FROM playlist_tracks WHERE playlist_id = ?1 ORDER BY position",
            )
            .unwrap();
        let members: Vec<i64> = stmt
            .query_map([id], |r| r.get(0))
            .unwrap()
            .map(Result::unwrap)
            .collect();
        assert_eq!(
            members,
            vec![2, 1],
            "order is kept, and a file with no library row cannot come across"
        );
    }

    /// `create_tables` runs its ALTER TABLE migrations unconditionally and
    /// detects the already-migrated case from SQLite's "duplicate column" error
    /// text. On an existing database that is the *normal* path, taken on every
    /// open, so a change in SQLite's wording would stop koan starting.
    #[test]
    fn sqlite_still_reports_duplicate_column() {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch("CREATE TABLE t (a INTEGER, b INTEGER);")
            .unwrap();
        let err = conn
            .execute("ALTER TABLE t ADD COLUMN b INTEGER", [])
            .unwrap_err();
        assert!(
            err.to_string().contains("duplicate column"),
            "SQLite error wording moved, create_tables no longer detects \
             already-applied migrations: {err}"
        );
    }

    /// Opening a database already at this version must not need the write
    /// lock: every background task opens one, and each would otherwise wait
    /// behind whatever scan or sync holds it.
    #[test]
    fn opening_a_current_database_writes_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        Database::open(&path).unwrap();

        let writer = Connection::open(&path).unwrap();
        writer.execute_batch("BEGIN IMMEDIATE").unwrap();

        let conn = Connection::open(&path).unwrap();
        conn.busy_timeout(std::time::Duration::from_millis(50))
            .unwrap();
        create_tables(&conn).expect("opened while another connection holds the write lock");

        // The whole open, pragmas and all, against its usual 30 s timeout.
        let started = std::time::Instant::now();
        Database::open(&path).unwrap();
        assert!(started.elapsed() < std::time::Duration::from_secs(5));
        writer.execute_batch("ROLLBACK").unwrap();
    }

    #[test]
    fn the_sweeps_wait_for_an_older_version() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        let conn = Connection::open(&path).unwrap();
        create_tables(&conn).unwrap();
        conn.execute("INSERT INTO artists (name, mbid) VALUES ('Crass', '')", [])
            .unwrap();
        let mbid = |conn: &Connection| -> Option<String> {
            conn.query_row("SELECT mbid FROM artists", [], |r| r.get(0))
                .unwrap()
        };

        create_tables(&conn).unwrap();
        assert_eq!(mbid(&conn).as_deref(), Some(""), "current: left alone");

        conn.pragma_update(None, "user_version", SCHEMA_VERSION - 1)
            .unwrap();
        create_tables(&conn).unwrap();
        assert_eq!(mbid(&conn), None, "older: swept");
        let v: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(v, SCHEMA_VERSION);
    }

    /// An upgrade that fails part way leaves the database as it was found:
    /// the steps before the failure are rolled back with it, and the version
    /// is not bumped.
    #[test]
    fn a_failed_upgrade_changes_nothing() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        // An empty album, which the upgrade deletes early on, and a table in
        // the way of an index it creates late.
        conn.execute_batch(&format!(
            "INSERT INTO albums (title) VALUES ('Empty');
             DROP INDEX idx_bookmarks_track;
             CREATE TABLE idx_bookmarks_track (x);
             PRAGMA user_version = {};",
            SCHEMA_VERSION - 1
        ))
        .unwrap();
        assert!(create_tables(&conn).is_err());
        let version: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(version, SCHEMA_VERSION - 1);
        let albums: i64 = conn
            .query_row("SELECT COUNT(*) FROM albums", [], |r| r.get(0))
            .unwrap();
        assert_eq!(albums, 1, "the album deleted before the failure is back");
        assert!(conn.is_autocommit(), "no transaction left open");
        let fk: i64 = conn
            .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
            .unwrap();
        assert_eq!(fk, 1);
    }

    /// A second connection, as another process has, sees the old version
    /// until the upgrade commits, and the new one after.
    #[test]
    fn a_reader_sees_one_version_or_the_other() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        let conn = Connection::open(&path).unwrap();
        conn.pragma_update(None, "journal_mode", "wal").unwrap();
        create_tables(&conn).unwrap();
        conn.pragma_update(None, "user_version", SCHEMA_VERSION - 1)
            .unwrap();
        let reader = Connection::open(&path).unwrap();
        let version = |c: &Connection| -> i64 {
            c.query_row("PRAGMA user_version", [], |r| r.get(0))
                .unwrap()
        };
        // A read under way while the upgrade runs, as a request in the
        // older server is.
        reader.execute_batch("BEGIN").unwrap();
        assert_eq!(version(&reader), SCHEMA_VERSION - 1);
        create_tables(&conn).unwrap();
        assert_eq!(
            version(&reader),
            SCHEMA_VERSION - 1,
            "a read begun before the commit keeps the old schema"
        );
        reader.execute_batch("COMMIT").unwrap();
        assert_eq!(version(&reader), SCHEMA_VERSION);
    }

    /// What the upgrade deletes takes what hangs off it, as it would with
    /// foreign keys on: an empty album's favourite and rating.
    #[test]
    fn an_upgrade_leaves_nothing_naming_what_it_deleted() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        crate::db::queries::auth::create_user(&conn, "jo", "pw", crate::auth::Role::User).unwrap();
        conn.execute_batch(&format!(
            "INSERT INTO albums (id, title) VALUES (7, 'Empty');
             INSERT INTO favourite_albums (user_id, album_id)
                 SELECT id, 7 FROM users WHERE username = 'jo';
             INSERT INTO album_ratings (user_id, album_id, rating)
                 SELECT id, 7, 5 FROM users WHERE username = 'jo';
             PRAGMA user_version = {};",
            SCHEMA_VERSION - 1
        ))
        .unwrap();
        create_tables(&conn).unwrap();
        let count = |sql: &str| -> i64 { conn.query_row(sql, [], |r| r.get(0)).unwrap() };
        assert_eq!(count("SELECT COUNT(*) FROM albums WHERE id = 7"), 0);
        assert_eq!(count("SELECT COUNT(*) FROM favourite_albums"), 0);
        assert_eq!(count("SELECT COUNT(*) FROM album_ratings"), 0);
        assert_eq!(count("SELECT COUNT(*) FROM pragma_foreign_key_check"), 0);
    }

    /// A reference no cascade covers, left dangling, fails the upgrade and
    /// rolls it back.
    #[test]
    fn an_upgrade_that_would_leave_a_dangling_reference_fails() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.pragma_update(None, "foreign_keys", "off").unwrap();
        conn.execute_batch(&format!(
            "INSERT INTO tracks (title, album_id, path) VALUES ('Archangel', 99, '/a.flac');
             PRAGMA user_version = {};",
            SCHEMA_VERSION - 1
        ))
        .unwrap();
        conn.pragma_update(None, "foreign_keys", "on").unwrap();
        let err = create_tables(&conn).unwrap_err().to_string();
        assert!(err.contains("missing albums"), "{err}");
        let version: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(version, SCHEMA_VERSION - 1);
        assert!(conn.is_autocommit());
    }

    /// A rebuild that fails reports its own error, not the failure of a
    /// rollback after its transaction was already gone.
    #[test]
    fn a_failed_rebuild_reports_its_own_error() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.pragma_update(None, "foreign_keys", "off").unwrap();
        conn.execute_batch(&format!(
            "CREATE TABLE users_plain (
                 id INTEGER PRIMARY KEY, username TEXT NOT NULL UNIQUE,
                 password_hash TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'user',
                 created_at TEXT
             );
             INSERT INTO users_plain SELECT id, username, password_hash, role, created_at FROM users;
             DROP TABLE users;
             ALTER TABLE users_plain RENAME TO users;
             CREATE TABLE users_new (x);
             PRAGMA user_version = {};",
            SCHEMA_VERSION - 1
        ))
        .unwrap();
        conn.pragma_update(None, "foreign_keys", "on").unwrap();
        let err = create_tables(&conn).unwrap_err().to_string();
        assert!(err.contains("users_new"), "{err}");
        assert!(!err.contains("no transaction"), "{err}");
        let version: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(version, SCHEMA_VERSION - 1);
        assert!(conn.is_autocommit());
    }

    #[test]
    fn the_saved_position_moves_to_its_own_table() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.execute_batch(
            "DROP TABLE playback_position;
             INSERT INTO playback_state
                 (id, queue_json, cursor_id, position_ms, was_playing)
             VALUES (1, '[]', '/music/a.flac', 61000, 1);
             PRAGMA user_version = 8;",
        )
        .unwrap();

        create_tables(&conn).unwrap();

        let moved: (String, i64, bool) = conn
            .query_row(
                "SELECT cursor_id, position_ms, was_playing
                   FROM playback_position WHERE id = 1",
                [],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
            )
            .unwrap();
        assert_eq!(moved, ("/music/a.flac".into(), 61000, true));
    }

    #[test]
    fn radio_leaves_nothing_behind() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.execute_batch(
            "CREATE TABLE similar_artists (artist_id INTEGER, similar_id INTEGER);
             CREATE TABLE track_vectors (track_id INTEGER PRIMARY KEY, embedding BLOB);
             CREATE INDEX idx_artists_mbid ON artists(mbid) WHERE mbid IS NOT NULL;
             ALTER TABLE playback_state ADD COLUMN radio_enabled INTEGER NOT NULL DEFAULT 0;
             ALTER TABLE playback_position ADD COLUMN radio_enabled INTEGER NOT NULL DEFAULT 0;
             PRAGMA user_version = 10;",
        )
        .unwrap();

        create_tables(&conn).unwrap();

        for name in ["similar_artists", "track_vectors", "idx_artists_mbid"] {
            let left: bool = conn
                .query_row(
                    "SELECT EXISTS (SELECT 1 FROM sqlite_master WHERE name = ?1)",
                    [name],
                    |r| r.get(0),
                )
                .unwrap();
            assert!(!left, "{name}");
        }
        for table in ["playback_state", "playback_position"] {
            assert!(
                !column_exists(&conn, table, "radio_enabled").unwrap(),
                "{table}"
            );
        }
    }

    #[test]
    fn reopening_a_migrated_database_succeeds() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        Database::open(&path).unwrap();
        Database::open(&path).unwrap();
        Database::open(&path).unwrap();
    }

    #[test]
    fn synced_playlists_are_tied_to_the_server_last_synced() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let db = Database::open(&path).unwrap();
            db.conn
                .execute_batch(
                    "INSERT INTO remote_servers (url, username, last_sync)
                       VALUES ('https://old/', 'ann', 1), ('https://new', 'bob', 2);
                     INSERT INTO playlists (name, remote_id) VALUES ('Synced', 'p1');
                     INSERT INTO playlists (name) VALUES ('Local');
                     ALTER TABLE playlists DROP COLUMN remote_account;",
                )
                .unwrap();
        }

        let db = Database::open(&path).unwrap();
        let accounts: Vec<Option<String>> = db
            .conn
            .prepare("SELECT remote_account FROM playlists ORDER BY name DESC")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(accounts, [Some("bob@https://new".into()), None]);
    }

    #[test]
    fn pre_migration_database_gains_the_new_columns() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            // Build the current schema, then strip the migrated columns back off
            // to reproduce a database written by an older koan.
            let db = Database::open(&path).unwrap();
            db.conn
                .execute_batch(
                    "ALTER TABLE tracks DROP COLUMN cache_size_bytes;
                     ALTER TABLE tracks DROP COLUMN cache_download_date;
                     ALTER TABLE shares DROP COLUMN kind;
                     ALTER TABLE shares DROP COLUMN subject_id;
                     ALTER TABLE shares DROP COLUMN start_track_id;",
                )
                .unwrap();
        }

        let db = Database::open(&path).unwrap();
        for (table, column) in [
            ("tracks", "cache_size_bytes"),
            ("tracks", "cache_download_date"),
            ("shares", "kind"),
            ("shares", "subject_id"),
            ("shares", "start_track_id"),
        ] {
            let found: i64 = db
                .conn
                .query_row(
                    &format!(
                        "SELECT COUNT(*) FROM pragma_table_info('{table}') WHERE name = '{column}'"
                    ),
                    [],
                    |row| row.get(0),
                )
                .unwrap();
            assert_eq!(found, 1, "{table}.{column} was not migrated");
        }
    }

    #[test]
    fn favourites_from_before_ids_name_their_rows() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let db = Database::open(&path).unwrap();
            db.conn
                .execute_batch(
                    "DROP TRIGGER users_personal_data;
                     DROP TABLE favourites; DROP TABLE favourite_albums; DROP TABLE favourite_artists;
                     CREATE TABLE favourites (user_id INTEGER NOT NULL DEFAULT 0, track_path TEXT NOT NULL,
                         created_at TEXT, PRIMARY KEY (user_id, track_path));
                     CREATE TABLE favourite_albums (user_id INTEGER NOT NULL DEFAULT 0,
                         artist_name TEXT NOT NULL, album_title TEXT NOT NULL, created_at TEXT,
                         PRIMARY KEY (user_id, artist_name, album_title));
                     CREATE TABLE favourite_artists (user_id INTEGER NOT NULL DEFAULT 0,
                         artist_name TEXT NOT NULL, created_at TEXT, PRIMARY KEY (user_id, artist_name));
                     INSERT INTO artists (id, name) VALUES (1, 'Burial');
                     INSERT INTO albums (id, title, artist_id) VALUES (1, 'Untrue', 1);
                     INSERT INTO tracks (id, title, album_id, artist_id, path) VALUES (1, 'Archangel', 1, 1, '/a.flac');
                     INSERT INTO tracks (id, title, album_id, artist_id, remote_id, remote_url, source)
                         VALUES (2, 'Near Dark', 1, 1, 'sub-2', 'http://server/2', 'remote');
                     INSERT INTO favourites (track_path) VALUES ('/a.flac'), ('http://server/2'), ('/gone.flac');
                     INSERT INTO favourite_albums (artist_name, album_title) VALUES ('BURIAL', 'untrue');
                     INSERT INTO favourite_artists (artist_name) VALUES ('Burial');
                     PRAGMA user_version = 13;",
                )
                .unwrap();
        }

        let db = Database::open(&path).unwrap();
        let ids = |sql: &str| -> Vec<i64> {
            db.conn
                .prepare(sql)
                .unwrap()
                .query_map([], |r| r.get(0))
                .unwrap()
                .collect::<rusqlite::Result<_>>()
                .unwrap()
        };
        assert_eq!(ids("SELECT track_id FROM favourites ORDER BY 1"), [1, 2]);
        assert_eq!(ids("SELECT album_id FROM favourite_albums"), [1]);
        assert_eq!(ids("SELECT artist_id FROM favourite_artists"), [1]);
    }

    #[test]
    fn a_library_from_before_source_rows_gets_them() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let db = Database::open(&path).unwrap();
            // A local file and the server's copy, left apart by the old
            // matching because the server spelled the credit differently.
            db.conn
                .execute_batch(
                    "DROP TABLE local_files; DROP TABLE remote_entries;
                     INSERT INTO artists (id, name) VALUES (1, 'Petrol Girls'), (2, 'Petrol Girls • Ren Aldridge');
                     INSERT INTO albums (id, title, artist_id) VALUES (1, 'Talk of Violence', 1);
                     INSERT INTO tracks (id, title, album_id, artist_id, track_number, path) VALUES
                         (1, 'Treading Water', 1, 1, 1, '/a.flac');
                     INSERT INTO tracks (id, title, album_id, artist_id, track_number, remote_id, source) VALUES
                         (2, 'Treading Water', 1, 2, 1, 'sub-1', 'remote');
                     INSERT INTO play_history (track_id, played_at) VALUES (2, 1);
                     INSERT INTO scan_cache (path, mtime, size, track_id) VALUES ('/a.flac', 1, 1, 1);
                     INSERT INTO remote_servers (url, username, library_version)
                         VALUES ('https://server', 'me', 5);
                     PRAGMA user_version = 13;",
                )
                .unwrap();
        }

        let db = Database::open(&path).unwrap();
        let one = |sql: &str| -> i64 { db.conn.query_row(sql, [], |r| r.get(0)).unwrap() };
        assert_eq!(one("SELECT COUNT(*) FROM tracks"), 1);
        assert_eq!(
            one(
                "SELECT COUNT(*) FROM local_files f JOIN remote_entries r ON r.track_id = f.track_id"
            ),
            1
        );
        assert_eq!(
            one("SELECT COUNT(*) FROM play_history WHERE track_id = 1"),
            1
        );
        // Read every file and walk the server again, for each source's own tags.
        assert_eq!(one("SELECT COUNT(*) FROM scan_cache"), 0);
        assert_eq!(
            one("SELECT COUNT(*) FROM remote_servers WHERE library_version IS NULL"),
            1
        );
    }

    #[test]
    fn rows_from_before_uids_are_given_one_and_new_rows_get_their_own() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        {
            let db = Database::open(&path).unwrap();
            db.conn
                .execute_batch(
                    "DROP TRIGGER artists_uid; DROP TRIGGER albums_uid; DROP TRIGGER tracks_uid;
                     DROP INDEX idx_artists_uid; DROP INDEX idx_albums_uid; DROP INDEX idx_tracks_uid;
                     ALTER TABLE artists DROP COLUMN uid;
                     ALTER TABLE albums DROP COLUMN uid;
                     ALTER TABLE tracks DROP COLUMN uid;
                     INSERT INTO artists (id, name) VALUES (5, 'Burial');
                     INSERT INTO albums (id, title, artist_id) VALUES (5, 'Untrue', 5);
                     INSERT INTO tracks (id, title, album_id, artist_id, path) VALUES
                         (5, 'Archangel', 5, 5, '/a.flac'), (6, 'Near Dark', 5, 5, '/b.flac');
                     PRAGMA user_version = 6;",
                )
                .unwrap();
        }

        let db = Database::open(&path).unwrap();
        db.conn
            .execute(
                "INSERT INTO tracks (title, album_id, artist_id, path) VALUES ('Ghost Hardware', 5, 5, '/c.flac')",
                [],
            )
            .unwrap();
        let uids: Vec<String> = db
            .conn
            .prepare(
                "SELECT uid FROM artists UNION ALL SELECT uid FROM albums
                 UNION ALL SELECT uid FROM tracks",
            )
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .collect::<rusqlite::Result<_>>()
            .unwrap();
        assert_eq!(uids.len(), 5);
        for uid in &uids {
            let parsed = uuid::Uuid::parse_str(uid).unwrap();
            assert_eq!(parsed.get_version_num(), 7, "{uid}");
            assert_eq!(parsed.to_string(), *uid, "stored hyphenated and lower case");
        }
        let distinct: std::collections::HashSet<_> = uids.iter().collect();
        assert_eq!(distinct.len(), uids.len());

        // Backfilled in row order.
        let tracks: Vec<String> = db
            .conn
            .prepare("SELECT uid FROM tracks WHERE id IN (5, 6) ORDER BY id")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .collect::<rusqlite::Result<_>>()
            .unwrap();
        let mut sorted = tracks.clone();
        sorted.sort();
        assert_eq!(tracks, sorted);
    }

    /// A users table from before `AUTOINCREMENT`: rebuilt with its rows, ids
    /// and the tables that refer to it intact, and a deleted id not reissued.
    #[test]
    fn user_ids_are_never_reused_after_the_rebuild() {
        let conn = Connection::open_in_memory().unwrap();
        conn.pragma_update(None, "foreign_keys", "on").unwrap();
        create_tables(&conn).unwrap();
        conn.execute_batch(
            "DROP TRIGGER users_personal_data;
             DROP TABLE users;
             CREATE TABLE users (
                 id            INTEGER PRIMARY KEY,
                 username      TEXT NOT NULL UNIQUE,
                 password_hash TEXT NOT NULL,
                 role          TEXT NOT NULL DEFAULT 'user' CHECK (role IN ('admin', 'user', 'readonly')),
                 created_at    TEXT DEFAULT (datetime('now')),
                 sealed_password BLOB
             );
             INSERT INTO users (id, username, password_hash, role, sealed_password) VALUES
                 (3, 'mate', 'h', 'user', x'01'), (5, 'owner', 'h', 'admin', NULL);
             INSERT INTO api_keys (user_id, name, key_hash, created_at) VALUES (3, 'k', 'kh', 0);
             INSERT INTO refresh_tokens (id, user_id, expires_at) VALUES ('t', 5, 9999);
             PRAGMA user_version = 8;",
        )
        .unwrap();

        create_tables(&conn).unwrap();

        let sql: String = conn
            .query_row(
                "SELECT sql FROM sqlite_master WHERE name = 'users'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(sql.contains("AUTOINCREMENT"), "{sql}");
        assert!(!column_exists(&conn, "users", "sealed_password").unwrap());
        let fk: i64 = conn
            .query_row("PRAGMA foreign_keys", [], |r| r.get(0))
            .unwrap();
        assert_eq!(fk, 1);

        // References still resolve, and still cascade.
        conn.execute("DELETE FROM users WHERE id = 3", []).unwrap();
        let keys: i64 = conn
            .query_row("SELECT COUNT(*) FROM api_keys", [], |r| r.get(0))
            .unwrap();
        assert_eq!(keys, 0);
        conn.execute("DELETE FROM users WHERE id = 5", []).unwrap();
        let tokens: i64 = conn
            .query_row("SELECT COUNT(*) FROM refresh_tokens", [], |r| r.get(0))
            .unwrap();
        assert_eq!(tokens, 0);

        // The highest id, deleted, is not handed out again.
        conn.execute(
            "INSERT INTO users (username, password_hash) VALUES ('new', 'h')",
            [],
        )
        .unwrap();
        assert_eq!(conn.last_insert_rowid(), 6);
        assert!(conn.execute_batch("PRAGMA foreign_key_check").is_ok());
    }

    #[test]
    fn foreign_keys_are_enforced() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("koan.db");
        let db = Database::open(&path).unwrap();

        db.conn
            .execute(
                "INSERT INTO users (id, username, password_hash) VALUES (1, 'u', 'h')",
                [],
            )
            .unwrap();
        db.conn
            .execute(
                "INSERT INTO refresh_tokens (id, user_id, expires_at) VALUES ('t', 1, 9999)",
                [],
            )
            .unwrap();
        assert!(
            db.conn
                .execute(
                    "INSERT INTO refresh_tokens (id, user_id, expires_at) VALUES ('t2', 999, 9999)",
                    [],
                )
                .is_err(),
            "foreign key constraint did not fire"
        );

        db.conn
            .execute("DELETE FROM users WHERE id = 1", [])
            .unwrap();
        let remaining: i64 = db
            .conn
            .query_row("SELECT COUNT(*) FROM refresh_tokens", [], |row| row.get(0))
            .unwrap();
        assert_eq!(remaining, 0, "ON DELETE CASCADE did not fire");
    }
    #[test]
    fn play_history_from_before_the_cascade_is_rebuilt_keeping_its_rows() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();

        // Seeded with enforcement off so the deliberately-orphaned entry lands.
        conn.pragma_update(None, "foreign_keys", "off").unwrap();
        // Put back the original constraint-free table and refill it.
        conn.execute_batch(
            "DROP TABLE play_history;
             CREATE TABLE play_history (
                 id          INTEGER PRIMARY KEY,
                 track_id    INTEGER REFERENCES tracks(id),
                 played_at   INTEGER NOT NULL,
                 duration_ms INTEGER,
                 source      TEXT DEFAULT 'local'
             );
             INSERT INTO artists (id, name) VALUES (1, 'A');
             INSERT INTO tracks (id, artist_id, title, source) VALUES (7, 1, 'T', 'local');
             INSERT INTO play_history (id, track_id, played_at, duration_ms, source)
                 VALUES (1, 7, 100, 5000, 'local'),
                        (2, 999, 200, NULL, 'local');",
        )
        .unwrap();
        assert!(!fk_cascades(&conn, "play_history").unwrap());

        apply_migrations(&conn, SCHEMA_VERSION).unwrap();

        assert!(fk_cascades(&conn, "play_history").unwrap());
        let kept: Vec<(i64, i64, Option<i64>)> = conn
            .prepare("SELECT id, played_at, duration_ms FROM play_history ORDER BY id")
            .unwrap()
            .query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(
            kept,
            vec![(1, 100, Some(5000))],
            "the live entry survives; the one pointing at a track that is gone does not"
        );

        // And the constraint now does the work the callers were doing by hand.
        conn.pragma_update(None, "foreign_keys", "on").unwrap();
        conn.execute("DELETE FROM tracks WHERE id = 7", []).unwrap();
        let left: i64 = conn
            .query_row("SELECT COUNT(*) FROM play_history", [], |r| r.get(0))
            .unwrap();
        assert_eq!(left, 0);
    }

    #[test]
    fn play_history_ids_are_never_handed_out_again_after_the_rebuild() {
        let conn = rusqlite::Connection::open_in_memory().unwrap();
        super::create_tables(&conn).unwrap();
        conn.execute_batch(
            "DROP TABLE play_history;
             CREATE TABLE play_history (
                 id          INTEGER PRIMARY KEY,
                 track_id    INTEGER REFERENCES tracks(id) ON DELETE CASCADE,
                 played_at   INTEGER NOT NULL,
                 duration_ms INTEGER,
                 source      TEXT DEFAULT 'local',
                 user_id     INTEGER NOT NULL DEFAULT 0
             );
             INSERT INTO play_history (id, track_id, played_at) VALUES (1, NULL, 10), (2, NULL, 20);
             PRAGMA user_version = 15;",
        )
        .unwrap();
        super::create_tables(&conn).unwrap();
        let ids: Vec<i64> = conn
            .prepare("SELECT id FROM play_history ORDER BY id")
            .unwrap()
            .query_map([], |r| r.get(0))
            .unwrap()
            .collect::<Result<_, _>>()
            .unwrap();
        assert_eq!(ids, [1, 2]);
        conn.execute_batch(
            "DELETE FROM play_history WHERE id = 2;
             INSERT INTO play_history (track_id, played_at) VALUES (NULL, 30);",
        )
        .unwrap();
        let newest: i64 = conn
            .query_row("SELECT MAX(id) FROM play_history", [], |r| r.get(0))
            .unwrap();
        assert_eq!(newest, 3);
    }

    #[test]
    fn cascading_play_history_is_idempotent() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        cascade_play_history(&conn).unwrap();
        cascade_play_history(&conn).unwrap();
        assert!(fk_cascades(&conn, "play_history").unwrap());
    }

    #[test]
    fn fresh_database_is_stamped_with_the_current_version() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        let v: i64 = conn
            .query_row("PRAGMA user_version", [], |r| r.get(0))
            .unwrap();
        assert_eq!(v, SCHEMA_VERSION);
    }

    #[test]
    fn create_tables_is_idempotent_across_repeated_opens() {
        let conn = Connection::open_in_memory().unwrap();
        for _ in 0..3 {
            create_tables(&conn).unwrap();
        }
        assert!(column_exists(&conn, "tracks", "cache_size_bytes").unwrap());
        assert!(column_exists(&conn, "organize_log", "mtime").unwrap());
    }

    #[test]
    fn a_database_missing_added_columns_is_migrated() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        // Rebuild `organize_log` without the columns added after the initial
        // schema, so the file looks like one written by an earlier version.
        conn.execute_batch(
            "DROP TABLE organize_log;
             CREATE TABLE organize_log (
                 id         INTEGER PRIMARY KEY,
                 batch_id   TEXT NOT NULL,
                 track_id   INTEGER,
                 from_path  TEXT NOT NULL,
                 to_path    TEXT NOT NULL,
                 created_at TEXT DEFAULT (datetime('now'))
             );
             PRAGMA user_version = 0;",
        )
        .unwrap();
        assert!(!column_exists(&conn, "organize_log", "size_bytes").unwrap());

        create_tables(&conn).unwrap();

        assert!(column_exists(&conn, "organize_log", "size_bytes").unwrap());
        assert!(column_exists(&conn, "organize_log", "mtime").unwrap());
    }

    #[test]
    fn migration_does_not_depend_on_sqlite_error_text() {
        // The previous implementation swallowed a duplicate-column ALTER by
        // string-matching SQLite's message, so a wording change in a bundled
        // SQLite upgrade would have failed every open. Adding a column that is
        // already present must now be a no-op decided by schema inspection.
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        apply_migrations(&conn, SCHEMA_VERSION).unwrap();
        apply_migrations(&conn, SCHEMA_VERSION).unwrap();
    }

    #[test]
    fn a_newer_database_is_refused_rather_than_written_to() {
        let conn = Connection::open_in_memory().unwrap();
        create_tables(&conn).unwrap();
        conn.pragma_update(None, "user_version", SCHEMA_VERSION + 1)
            .unwrap();

        let err = create_tables(&conn).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("newer than this build"), "unexpected: {msg}");
    }

    /// A database from before favourites, playlists, history and shares were
    /// per user: every row goes to the first admin, and nothing is lost.
    #[test]
    fn shared_data_from_before_accounts_had_their_own_goes_to_the_first_admin() {
        let conn = Connection::open_in_memory().unwrap();
        conn.pragma_update(None, "foreign_keys", "on").unwrap();
        create_tables(&conn).unwrap();
        conn.execute_batch(
            "DROP TRIGGER users_personal_data;
             DROP TRIGGER scrobble_reported_play;
             DROP INDEX idx_play_history_user;
             DROP INDEX idx_playlists_user;
             ALTER TABLE play_history DROP COLUMN user_id;
             ALTER TABLE playlists DROP COLUMN user_id;
             ALTER TABLE shares DROP COLUMN user_id;
             DROP TABLE favourites;
             DROP TABLE favourite_albums;
             DROP TABLE favourite_artists;
             CREATE TABLE favourites (
                 track_path TEXT PRIMARY KEY,
                 created_at TEXT DEFAULT (datetime('now'))
             );
             CREATE TABLE favourite_albums (
                 artist_name TEXT NOT NULL,
                 album_title TEXT NOT NULL,
                 created_at  TEXT DEFAULT (datetime('now')),
                 PRIMARY KEY (artist_name, album_title)
             );
             CREATE TABLE favourite_artists (
                 artist_name TEXT PRIMARY KEY,
                 created_at  TEXT DEFAULT (datetime('now'))
             );
             INSERT INTO users (id, username, password_hash, role) VALUES
                 (3, 'mate', 'h', 'user'), (5, 'owner', 'h', 'admin'), (7, 'late', 'h', 'admin');
             INSERT INTO artists (id, name) VALUES (1, 'A');
             INSERT INTO albums (id, title, artist_id) VALUES (1, 'B', 1);
             INSERT INTO tracks (id, artist_id, album_id, title, source, path) VALUES
                 (1, 1, 1, 'T', 'local', '/a.flac'), (2, 1, 1, 'U', 'local', '/b.flac');
             INSERT INTO favourites (track_path) VALUES ('/a.flac'), ('/b.flac');
             INSERT INTO favourite_albums (artist_name, album_title) VALUES ('A', 'B');
             INSERT INTO favourite_artists (artist_name) VALUES ('A');
             INSERT INTO play_history (track_id, played_at) VALUES (1, 100);
             INSERT INTO playlists (name) VALUES ('Mix');
             INSERT INTO shares (id, created_at) VALUES ('s', 0);
             PRAGMA user_version = 6;",
        )
        .unwrap();

        create_tables(&conn).unwrap();

        for (table, rows) in [
            ("favourites", 2),
            ("favourite_albums", 1),
            ("favourite_artists", 1),
            ("play_history", 1),
            ("playlists", 1),
            ("shares", 1),
        ] {
            let owned: i64 = conn
                .query_row(
                    &format!("SELECT COUNT(*) FROM {table} WHERE user_id = 5"),
                    [],
                    |r| r.get(0),
                )
                .unwrap();
            assert_eq!(owned, rows, "{table} did not go to the first admin");
        }
        // Keyed by user now: the same favourite for someone else is its own row.
        conn.execute(
            "INSERT INTO favourites (user_id, track_id) VALUES (3, 1)",
            [],
        )
        .unwrap();
    }
}