jellytau_lib/storage/

mod.rs

//! Offline storage module using SQLite
//!
//! Provides local caching of Jellyfin metadata, download management,
//! and offline mutation queue for sync-back operations.

pub mod db_service;
pub mod models;
pub mod schema;

use crate::utils::lock::MutexSafe;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};

use log::{debug, error, info};
use rusqlite::{Connection, Result as SqliteResult};

pub use db_service::{DatabaseService, RusqliteService};
use schema::MIGRATIONS;

/// How long a connection retries a locked database before erroring — covers a
/// reader meeting a WAL checkpoint, or the writer meeting a reader's snapshot.
const BUSY_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);

/// How many read-only connections serve queries alongside the writer. Reads
/// are short; a few cover a library page's parallel fetches plus background
/// work without holding many file handles.
const READER_CONNECTIONS: usize = 3;

/// The database: opened once at startup and owned for the life of the app.
///
/// All access goes through [`Database::service`], which hands out clones of one
/// [`RusqliteService`] — a writer thread plus a pool of read-only connections
/// (see `db_service`). Nothing else opens the database file.
pub struct Database {
    /// The read-write connection. Owned by the service's writer thread; kept
    /// here only for migrations (which run before that thread starts taking
    /// work) and for tests.
    #[cfg_attr(not(test), allow(dead_code))]
    conn: Arc<Mutex<Connection>>,
    service: RusqliteService,
    path: PathBuf,
}

impl Database {
    /// Open or create the database at a specific path
    pub fn open(path: &PathBuf) -> SqliteResult<Self> {
        // Ensure parent directory exists
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent).ok();
        }

        let conn = Connection::open(path)?;

        // Enable foreign keys
        conn.execute_batch("PRAGMA foreign_keys = ON;")?;

        // WAL lets the reader connections run alongside the writer.
        conn.execute_batch("PRAGMA journal_mode = WAL;")?;
        // In WAL mode NORMAL is corruption-safe and skips the fsync FULL pays on
        // every commit (a power cut can lose the last commits, an app crash
        // cannot). On Android flash that fsync dominated every small write.
        conn.execute_batch("PRAGMA synchronous = NORMAL;")?;
        conn.busy_timeout(BUSY_TIMEOUT)?;

        let conn = Arc::new(Mutex::new(conn));
        Self::migrate_connection(&conn, MIGRATIONS)?;

        // Planner statistics. Without them SQLite guesses between indexes, and
        // guessed badly for the listing query (see 08-database-design.md →
        // "Listing query shape"). `optimize` only analyses what is missing or
        // stale; `analysis_limit` bounds each table's scan so this stays in the
        // milliseconds on a large catalogue. Failure is not fatal.
        if let Err(e) = conn
            .lock_safe()
            .execute_batch("PRAGMA analysis_limit = 400; PRAGMA optimize = 0x10002;")
        {
            error!("PRAGMA optimize failed: {}", e);
        }

        // Readers open after migrations, so they only ever see the final schema.
        let readers = (0..READER_CONNECTIONS)
            .map(|_| Self::open_reader(path))
            .collect::<SqliteResult<Vec<_>>>()?;

        Ok(Self {
            service: RusqliteService::with_readers(Arc::clone(&conn), readers),
            conn,
            path: path.clone(),
        })
    }

    /// A connection that can only read. `query_only` makes an accidental write
    /// routed to the pool fail loudly instead of racing the writer.
    fn open_reader(path: &PathBuf) -> SqliteResult<Connection> {
        let conn = Connection::open(path)?;
        conn.busy_timeout(BUSY_TIMEOUT)?;
        conn.execute_batch("PRAGMA query_only = ON;")?;
        Ok(conn)
    }

    /// Open an in-memory database (for testing)
    #[cfg(test)]
    pub fn open_in_memory() -> SqliteResult<Self> {
        let conn = Connection::open_in_memory()?;

        // Enable foreign keys
        conn.execute_batch("PRAGMA foreign_keys = ON;")?;

        let conn = Arc::new(Mutex::new(conn));
        Self::migrate_connection(&conn, MIGRATIONS)?;

        // An in-memory database cannot be shared between connections, so this
        // one has no reader pool: reads go through the writer.
        Ok(Self {
            service: RusqliteService::new(Arc::clone(&conn)),
            conn,
            path: PathBuf::from(":memory:"),
        })
    }

    /// Get connection (for testing)
    #[cfg(test)]
    pub fn connection(&self) -> Arc<Mutex<Connection>> {
        Arc::clone(&self.conn)
    }

    /// Re-run all migrations against an open database (tests only; `open` runs
    /// them before the service starts).
    #[cfg(test)]
    pub fn migrate(&self) -> SqliteResult<()> {
        self.migrate_with(MIGRATIONS)
    }

    /// Test seam: inject a failing migration. See [`Self::migrate_connection`].
    #[cfg(test)]
    fn migrate_with(&self, migrations: &[(&str, &str)]) -> SqliteResult<()> {
        Self::migrate_connection(&self.conn, migrations)
    }

    /// Apply `migrations` in order, skipping ones `_migrations` already records.
    ///
    /// **Each migration is one transaction, and the `_migrations` row is written
    /// inside it.** SQLite autocommits every statement otherwise, so a migration
    /// that failed partway — low disk, an OOM kill, the process dying mid-boot —
    /// used to leave its earlier statements applied while recording nothing.
    /// `execute_batch` aborts on the first error, so the retry on the next launch
    /// then failed at statement 1 ("duplicate column name") and kept failing
    /// forever; `Database::open` turns that into a panic, so the app never
    /// started again and the only fix was clearing app data. Committing the
    /// schema change and the bookkeeping together makes a migration all-or-nothing
    /// and a retry always safe.
    ///
    /// Every migration is pure DDL/DML, which SQLite runs transactionally — a
    /// `PRAGMA` or `VACUUM` added to one would not roll back and must not be.
    ///
    /// Runs on the bare connection, before the writer thread takes it, so
    /// tests can also inject a failing migration through `migrate_with`.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    fn migrate_connection(
        conn: &Mutex<Connection>,
        migrations: &[(&str, &str)],
    ) -> SqliteResult<()> {
        info!("Starting database migrations...");
        let conn = conn.lock_safe();

        // Create migrations table if it doesn't exist
        debug!("Creating _migrations table if it doesn't exist...");
        match conn.execute(
            "CREATE TABLE IF NOT EXISTS _migrations (
                id INTEGER PRIMARY KEY,
                name TEXT NOT NULL UNIQUE,
                applied_at TEXT DEFAULT CURRENT_TIMESTAMP
            )",
            [],
        ) {
            Ok(_) => debug!("_migrations table ready"),
            Err(e) => {
                error!("Failed to create _migrations table: {}", e);
                return Err(e);
            }
        }

        // Get applied migrations
        debug!("Querying applied migrations...");
        let mut stmt = conn.prepare("SELECT name FROM _migrations")?;
        let applied: Vec<String> = stmt
            .query_map([], |row: &rusqlite::Row| row.get(0))?
            .filter_map(|r| r.ok())
            .collect();
        debug!("Found {} applied migrations", applied.len());

        // Apply pending migrations
        for (name, sql) in migrations {
            if applied.contains(&name.to_string()) {
                debug!("Skipping already applied migration: {}", name);
                continue;
            }

            info!("Applying migration: {}", name);

            // `unchecked_transaction` because the connection is reached through a
            // shared guard rather than `&mut`. Dropping the transaction without
            // committing rolls it back, which is exactly what the `?`s below do.
            let tx = conn.unchecked_transaction()?;

            if let Err(e) = tx.execute_batch(sql) {
                error!("Failed to apply migration {} (rolled back): {}", name, e);
                return Err(e);
            }
            if let Err(e) = tx.execute("INSERT INTO _migrations (name) VALUES (?1)", [name]) {
                error!("Failed to record migration {} (rolled back): {}", name, e);
                return Err(e);
            }
            tx.commit()?;

            info!("Successfully applied migration: {}", name);
        }

        info!("All migrations completed successfully");
        Ok(())
    }

    /// A handle to the database service. Cheap: every call returns a clone of
    /// the same writer thread and reader pool.
    pub fn service(&self) -> RusqliteService {
        self.service.clone()
    }

    /// Get the database file path
    pub fn path(&self) -> &PathBuf {
        &self.path
    }

    /// Get database file size in bytes
    pub fn file_size(&self) -> Option<u64> {
        std::fs::metadata(&self.path).ok().map(|m| m.len())
    }
}

// TRACES: UR-002, UR-012, UR-019, UR-025 | DR-012 | UT-014, UT-015, UT-016, UT-017, UT-018, UT-019, UT-020, UT-021, UT-022, UT-023, UT-025
#[cfg(test)]
mod tests {
    use super::*;
    use rusqlite::params;

    #[test]
    fn test_open_in_memory() {
        let db = Database::open_in_memory().unwrap();
        assert_eq!(db.path().to_str(), Some(":memory:"));
    }

    /// A migration that dies partway must leave *nothing* behind.
    ///
    /// SQLite autocommits each statement, so before migrations were wrapped in a
    /// transaction the first `ADD COLUMN` of a failing batch stuck while the
    /// `_migrations` row was never written. `execute_batch` aborts on the first
    /// error, so the retry on the next launch failed at statement 1 with
    /// "duplicate column name" and kept failing forever — and `Database::open`
    /// panics on that, so the app never started again.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    #[test]
    fn test_failed_migration_rolls_back_and_stays_retryable() {
        let db = Database::open_in_memory().unwrap();

        // Statement 1 succeeds, statement 2 fails, statement 3 never runs.
        let poisoned = &[(
            "900_partially_failing",
            "ALTER TABLE downloads ADD COLUMN audit_a TEXT;
             ALTER TABLE downloads ADD COLUMN audit_b TEXT FROM NOWHERE;
             ALTER TABLE downloads ADD COLUMN audit_c TEXT;",
        )][..];

        let first = db.migrate_with(poisoned).unwrap_err();

        // Nothing from the batch may survive, or the retry cannot re-run it.
        assert!(
            !has_column(&db, "downloads", "audit_a"),
            "statement 1 of a failed migration was left applied: the batch did not roll back"
        );
        assert!(!has_column(&db, "downloads", "audit_c"));

        // And it must not be recorded as applied.
        let recorded: i64 = db
            .connection()
            .lock_safe()
            .query_row(
                "SELECT COUNT(*) FROM _migrations WHERE name = ?1",
                ["900_partially_failing"],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(recorded, 0, "a failed migration must not be recorded");

        // The retry must fail the same way it did the first time — reaching the
        // real error — rather than tripping over its own leftovers.
        let second = db.migrate_with(poisoned).unwrap_err();
        assert!(
            !second.to_string().contains("duplicate column"),
            "the retry hit leftovers from the failed run instead of the real error: {second}"
        );
        assert_eq!(first.to_string(), second.to_string());

        // A corrected migration under the same name then applies cleanly.
        let fixed = &[(
            "900_partially_failing",
            "ALTER TABLE downloads ADD COLUMN audit_a TEXT;
             ALTER TABLE downloads ADD COLUMN audit_c TEXT;",
        )][..];
        db.migrate_with(fixed).unwrap();
        assert!(has_column(&db, "downloads", "audit_a"));
        assert!(has_column(&db, "downloads", "audit_c"));
    }

    /// A committed migration is recorded, so it is never applied twice.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    #[test]
    fn test_successful_migration_is_recorded_in_the_same_transaction() {
        let db = Database::open_in_memory().unwrap();
        let m = &[(
            "901_adds_a_column",
            "ALTER TABLE downloads ADD COLUMN audit_d TEXT;",
        )][..];

        db.migrate_with(m).unwrap();
        // Re-running must be a no-op, not a "duplicate column" failure.
        db.migrate_with(m).unwrap();
        assert!(has_column(&db, "downloads", "audit_d"));
    }

    fn has_column(db: &Database, table: &str, column: &str) -> bool {
        let conn = db.connection();
        let conn = conn.lock_safe();
        let mut stmt = conn
            .prepare(&format!("PRAGMA table_info({table})"))
            .unwrap();
        let mut names = stmt
            .query_map([], |row| row.get::<_, String>(1))
            .unwrap()
            .filter_map(|r| r.ok());
        names.any(|n| n == column)
    }

    #[test]
    fn test_migrations_run() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Check that tables exist
        let mut stmt = conn
            .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='items'")
            .unwrap();
        let exists: Option<String> = stmt.query_row([], |row: &rusqlite::Row| row.get(0)).ok();
        assert!(exists.is_some());
    }

    #[test]
    fn test_all_tables_created() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        let expected_tables = [
            "servers",
            "users",
            "libraries",
            "items",
            "media_streams",
            "user_data",
            "downloads",
            "sync_queue",
            "thumbnails",
            "playlists",
            "playlist_items",
            "genres",
        ];

        for table in expected_tables {
            let exists: Option<String> = conn
                .query_row(
                    "SELECT name FROM sqlite_master WHERE type='table' AND name=?1",
                    [table],
                    |row: &rusqlite::Row| row.get(0),
                )
                .ok();
            assert!(exists.is_some(), "Table '{}' should exist", table);
        }
    }

    #[test]
    fn test_fts_table_created() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        let exists: Option<String> = conn
            .query_row(
                "SELECT name FROM sqlite_master WHERE type='table' AND name='items_fts'",
                [],
                |row: &rusqlite::Row| row.get(0),
            )
            .ok();
        assert!(exists.is_some(), "FTS table 'items_fts' should exist");
    }

    #[test]
    fn test_server_crud() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Insert a server
        conn.execute(
            "INSERT INTO servers (id, name, url, version) VALUES (?1, ?2, ?3, ?4)",
            params!["server1", "My Server", "http://localhost:8096", "10.8.0"],
        )
        .unwrap();

        // Read it back
        let (name, url): (String, String) = conn
            .query_row(
                "SELECT name, url FROM servers WHERE id = ?1",
                ["server1"],
                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(name, "My Server");
        assert_eq!(url, "http://localhost:8096");

        // Update it
        conn.execute(
            "UPDATE servers SET name = ?1 WHERE id = ?2",
            params!["Updated Server", "server1"],
        )
        .unwrap();

        let name: String = conn
            .query_row(
                "SELECT name FROM servers WHERE id = ?1",
                ["server1"],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(name, "Updated Server");

        // Delete it
        conn.execute("DELETE FROM servers WHERE id = ?1", ["server1"])
            .unwrap();

        let count: i32 = conn
            .query_row("SELECT COUNT(*) FROM servers", [], |row: &rusqlite::Row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_user_crud() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Create a server first (foreign key)
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test Server", "http://localhost:8096"],
        )
        .unwrap();

        // Insert a user
        conn.execute(
            "INSERT INTO users (id, server_id, username, is_active)
             VALUES (?1, ?2, ?3, ?4)",
            params!["user1", "server1", "admin", 1],
        )
        .unwrap();

        // Read it back
        let (username, is_active): (String, i32) = conn
            .query_row(
                "SELECT username, is_active FROM users WHERE id = ?1",
                ["user1"],
                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(username, "admin");
        assert_eq!(is_active, 1);

        // Update is_active
        conn.execute("UPDATE users SET is_active = 0 WHERE id = ?1", ["user1"])
            .unwrap();

        let is_active: i32 = conn
            .query_row(
                "SELECT is_active FROM users WHERE id = ?1",
                ["user1"],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(is_active, 0);
    }

    #[test]
    fn test_cascade_delete_server_removes_users() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Create server and user
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test Server", "http://localhost:8096"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
            params!["user1", "server1", "admin"],
        )
        .unwrap();

        // Verify user exists
        let count: i32 = conn
            .query_row(
                "SELECT COUNT(*) FROM users WHERE server_id = ?1",
                ["server1"],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);

        // Delete server
        conn.execute("DELETE FROM servers WHERE id = ?1", ["server1"])
            .unwrap();

        // User should be deleted via CASCADE
        let count: i32 = conn
            .query_row("SELECT COUNT(*) FROM users", [], |row: &rusqlite::Row| {
                row.get(0)
            })
            .unwrap();
        assert_eq!(count, 0);
    }

    #[test]
    fn test_item_insert_and_fts_search() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Create server first
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test Server", "http://localhost:8096"],
        )
        .unwrap();

        // Insert an item
        conn.execute(
            "INSERT INTO items (id, server_id, name, item_type, overview, album_name, artists)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
            params![
                "item1",
                "server1",
                "Bohemian Rhapsody",
                "Audio",
                "A legendary rock song",
                "A Night at the Opera",
                "[\"Queen\"]"
            ],
        )
        .unwrap();

        // Search via FTS
        let mut stmt = conn
            .prepare(
                "SELECT i.name FROM items i
                 JOIN items_fts ON i.rowid = items_fts.rowid
                 WHERE items_fts MATCH ?1",
            )
            .unwrap();

        // Search by song name
        let result: Option<String> = stmt
            .query_row(["Bohemian"], |row: &rusqlite::Row| row.get(0))
            .ok();
        assert_eq!(result, Some("Bohemian Rhapsody".to_string()));

        // Search by album name
        let result: Option<String> = stmt
            .query_row(["Opera"], |row: &rusqlite::Row| row.get(0))
            .ok();
        assert_eq!(result, Some("Bohemian Rhapsody".to_string()));

        // Search by artist
        let result: Option<String> = stmt
            .query_row(["Queen"], |row: &rusqlite::Row| row.get(0))
            .ok();
        assert_eq!(result, Some("Bohemian Rhapsody".to_string()));
    }

    #[test]
    fn test_user_data_playback_position() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Setup: server, user, item
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test", "http://localhost"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
            params!["user1", "server1", "admin"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO items (id, server_id, name, item_type) VALUES (?1, ?2, ?3, ?4)",
            params!["item1", "server1", "Test Movie", "Movie"],
        )
        .unwrap();

        // Insert user data with playback position
        conn.execute(
            "INSERT INTO user_data (user_id, item_id, playback_position_ticks, is_played)
             VALUES (?1, ?2, ?3, ?4)",
            params!["user1", "item1", 12345678900_i64, 0],
        )
        .unwrap();

        // Read back
        let (position, is_played): (i64, i32) = conn
            .query_row(
                "SELECT playback_position_ticks, is_played FROM user_data
                 WHERE user_id = ?1 AND item_id = ?2",
                ["user1", "item1"],
                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(position, 12345678900);
        assert_eq!(is_played, 0);

        // Update to mark as played
        conn.execute(
            "UPDATE user_data SET is_played = 1, playback_position_ticks = 0
             WHERE user_id = ?1 AND item_id = ?2",
            ["user1", "item1"],
        )
        .unwrap();

        let is_played: i32 = conn
            .query_row(
                "SELECT is_played FROM user_data WHERE user_id = ?1 AND item_id = ?2",
                ["user1", "item1"],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(is_played, 1);
    }

    #[test]
    fn test_sync_queue_operations() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Setup
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test", "http://localhost"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
            params!["user1", "server1", "admin"],
        )
        .unwrap();

        // Queue a sync operation
        conn.execute(
            "INSERT INTO sync_queue (user_id, operation, item_id, payload, status)
             VALUES (?1, ?2, ?3, ?4, ?5)",
            params![
                "user1",
                "mark_favorite",
                "item123",
                r#"{"favorite": true}"#,
                "pending"
            ],
        )
        .unwrap();

        // Get pending operations
        let mut stmt = conn
            .prepare("SELECT operation, item_id FROM sync_queue WHERE status = 'pending'")
            .unwrap();

        let ops: Vec<(String, String)> = stmt
            .query_map([], |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)))
            .unwrap()
            .filter_map(|r| r.ok())
            .collect();

        assert_eq!(ops.len(), 1);
        assert_eq!(ops[0].0, "mark_favorite");
        assert_eq!(ops[0].1, "item123");

        // Mark as completed
        conn.execute(
            "UPDATE sync_queue SET status = 'completed' WHERE item_id = ?1",
            ["item123"],
        )
        .unwrap();

        let pending_count: i32 = conn
            .query_row(
                "SELECT COUNT(*) FROM sync_queue WHERE status = 'pending'",
                [],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(pending_count, 0);
    }

    #[test]
    fn test_downloads_table() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Setup
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test", "http://localhost"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
            params!["user1", "server1", "admin"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO items (id, server_id, name, item_type) VALUES (?1, ?2, ?3, ?4)",
            params!["item1", "server1", "Test Song", "Audio"],
        )
        .unwrap();

        // Queue a download
        conn.execute(
            "INSERT INTO downloads (item_id, user_id, file_path, status, progress)
             VALUES (?1, ?2, ?3, ?4, ?5)",
            params!["item1", "user1", "/data/downloads/test.mp3", "pending", 0.0],
        )
        .unwrap();

        // Update progress
        conn.execute(
            "UPDATE downloads SET status = 'downloading', progress = 0.5
             WHERE item_id = ?1",
            ["item1"],
        )
        .unwrap();

        let (status, progress): (String, f64) = conn
            .query_row(
                "SELECT status, progress FROM downloads WHERE item_id = ?1",
                ["item1"],
                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();
        assert_eq!(status, "downloading");
        assert!((progress - 0.5).abs() < 0.001);

        // Complete download
        conn.execute(
            "UPDATE downloads SET status = 'completed', progress = 1.0
             WHERE item_id = ?1",
            ["item1"],
        )
        .unwrap();

        let status: String = conn
            .query_row(
                "SELECT status FROM downloads WHERE item_id = ?1",
                ["item1"],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(status, "completed");
    }

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

        // Run migrations again - should not fail
        let result = db.migrate();
        assert!(result.is_ok());

        // Tables should still exist
        let conn = db.connection();
        let conn = conn.lock_safe();

        let count: i32 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='items'",
                [],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(count, 1);
    }

    #[test]
    fn test_global_active_user_deactivation() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Create two servers
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Server 1", "http://server1.com"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server2", "Server 2", "http://server2.com"],
        )
        .unwrap();

        // Create users on different servers
        conn.execute(
            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
             VALUES (?1, ?2, ?3, 1, '2024-01-01 10:00:00')",
            params!["user1", "server1", "admin"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
             VALUES (?1, ?2, ?3, 1, '2024-01-01 11:00:00')",
            params!["user2", "server2", "admin"],
        )
        .unwrap();

        // Initially both users are active (simulating the old bug)
        let active_count: i32 = conn
            .query_row(
                "SELECT COUNT(*) FROM users WHERE is_active = 1",
                [],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(active_count, 2);

        // Now simulate setting user1 as active (global deactivation)
        conn.execute("UPDATE users SET is_active = 0", []).unwrap();
        conn.execute(
            "UPDATE users SET is_active = 1, last_login_at = CURRENT_TIMESTAMP WHERE id = ?1",
            ["user1"],
        )
        .unwrap();

        // Only one user should be active now
        let active_count: i32 = conn
            .query_row(
                "SELECT COUNT(*) FROM users WHERE is_active = 1",
                [],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(active_count, 1);

        // And it should be user1
        let active_user: String = conn
            .query_row(
                "SELECT id FROM users WHERE is_active = 1",
                [],
                |row: &rusqlite::Row| row.get(0),
            )
            .unwrap();
        assert_eq!(active_user, "user1");
    }

    #[test]
    fn test_active_session_query_ordering() {
        let db = Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        // Create server
        conn.execute(
            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
            params!["server1", "Test Server", "http://localhost:8096"],
        )
        .unwrap();

        // Create users with different login times
        conn.execute(
            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
             VALUES (?1, ?2, ?3, 0, '2024-01-01 10:00:00')",
            params!["user1", "server1", "old_user"],
        )
        .unwrap();

        conn.execute(
            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
             VALUES (?1, ?2, ?3, 1, '2024-01-01 12:00:00')",
            params!["user2", "server1", "recent_user"],
        )
        .unwrap();

        // Query for active user ordered by last_login_at DESC (simulating storage_get_active_session)
        let (user_id, username): (String, String) = conn
            .query_row(
                "SELECT u.id, u.username FROM users u
                 JOIN servers s ON u.server_id = s.id
                 WHERE u.is_active = 1
                 ORDER BY u.last_login_at DESC
                 LIMIT 1",
                [],
                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
            )
            .unwrap();

        assert_eq!(user_id, "user2");
        assert_eq!(username, "recent_user");
    }

    /// A read must not queue behind a long write.
    ///
    /// The database is in WAL mode precisely so readers can run alongside a
    /// writer, but every query used to go through one connection behind one
    /// mutex — so a big catalog-cache transaction stalled every library page,
    /// thumbnail lookup and settings read in the app until it committed.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn reads_do_not_wait_for_an_in_flight_write() {
        use crate::storage::db_service::{DatabaseService, Query};
        use std::time::{Duration, Instant};

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

        let writer = service.clone();
        let write = tokio::spawn(async move {
            writer
                .transaction(|_tx| {
                    std::thread::sleep(Duration::from_millis(600));
                    Ok(())
                })
                .await
        });
        // Let the write take the connection first.
        tokio::time::sleep(Duration::from_millis(100)).await;

        let started = Instant::now();
        let servers: i64 = service
            .query_one(Query::new("SELECT COUNT(*) FROM servers"), |r| r.get(0))
            .await
            .unwrap();
        let waited = started.elapsed();

        assert_eq!(servers, 0);
        assert!(
            waited < Duration::from_millis(250),
            "a read waited {waited:?} for an unrelated write to commit"
        );
        write.await.unwrap().unwrap();
    }

    /// Commit cost: in WAL mode `synchronous = NORMAL` is corruption-safe and
    /// skips the per-commit fsync that FULL (the default) pays — on Android
    /// flash that is the dominant cost of every small write. A busy timeout
    /// lets the reader connections ride out a checkpoint instead of failing.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    #[test]
    fn open_configures_wal_for_interactive_use() {
        let dir = tempfile::tempdir().unwrap();
        let db = Database::open(&dir.path().join("jellytau.db")).unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();

        let mode: String = conn
            .query_row("PRAGMA journal_mode", [], |r| r.get(0))
            .unwrap();
        let synchronous: i64 = conn
            .query_row("PRAGMA synchronous", [], |r| r.get(0))
            .unwrap();
        let busy_timeout: i64 = conn
            .query_row("PRAGMA busy_timeout", [], |r| r.get(0))
            .unwrap();

        assert_eq!(mode, "wal");
        assert_eq!(synchronous, 1, "expected synchronous = NORMAL");
        assert!(busy_timeout > 0, "expected a busy timeout");
    }

    /// The planner gets statistics: the app used to never run `ANALYZE`, so
    /// SQLite guessed between indexes — and for the listing query guessed the
    /// `server_id` index, which every row shares, turning an index lookup into
    /// a walk of the whole catalogue. `PRAGMA optimize` at open refreshes
    /// whatever statistics are missing or stale, bounded by `analysis_limit`.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    #[test]
    fn open_gives_the_planner_statistics() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("jellytau.db");
        {
            let db = Database::open(&path).unwrap();
            let conn = db.connection();
            let conn = conn.lock_safe();
            conn.execute_batch(
                "INSERT INTO servers (id, name, url) VALUES ('s', 'S', 'http://s');",
            )
            .unwrap();
            for i in 0..2000 {
                conn.execute(
                    "INSERT INTO items (id, server_id, name, item_type) VALUES (?1, 's', 'n', 'Audio')",
                    [format!("i{i}")],
                )
                .unwrap();
            }
        }

        let db = Database::open(&path).unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();
        let analysed: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_master WHERE name = 'sqlite_stat1'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(analysed, 1, "the planner has no statistics");
        let items_stats: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM sqlite_stat1 WHERE tbl = 'items'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert!(items_stats > 0, "no statistics for items");
    }

    /// Writes a phone-sized catalogue to `$JELLYTAU_BENCH_DB` for timing
    /// queries with the `sqlite3` CLI. Not a test; run explicitly with
    /// `--ignored`.
    #[test]
    #[ignore]
    fn write_bench_database() {
        let Ok(path) = std::env::var("JELLYTAU_BENCH_DB") else {
            return;
        };
        let _ = std::fs::remove_file(&path);
        let db = Database::open(&PathBuf::from(&path)).unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();
        conn.execute_batch(
            "BEGIN;
             INSERT INTO servers (id, name, url) VALUES ('srv', 'S', 'http://s');
             INSERT INTO users (id, server_id, username) VALUES ('u', 'srv', 'u');
             INSERT INTO libraries (id, server_id, name, collection_type) VALUES ('tv', 'srv', 'TV', 'tvshows');
             INSERT INTO libraries (id, server_id, name, collection_type) VALUES ('music', 'srv', 'Music', 'music');",
        )
        .unwrap();
        let now = "2026-09-23T00:00:00Z";
        let mut item = conn
            .prepare(
                "INSERT INTO items (id, server_id, library_id, parent_id, name, sort_name, item_type,
                                    series_id, season_id, album_id, synced_at)
                 VALUES (?1, 'srv', ?2, ?3, ?4, ?4, ?5, ?6, ?7, ?8, ?9)",
            )
            .unwrap();
        let none: Option<String> = None;
        for s in 0..300 {
            let series = format!("series-{s}");
            item.execute(rusqlite::params![
                series,
                "tv",
                none,
                format!("Show {s}"),
                "Series",
                none,
                none,
                none,
                now
            ])
            .unwrap();
            for n in 0..11 {
                let season = format!("{series}-s{n}");
                item.execute(rusqlite::params![
                    season,
                    "tv",
                    series,
                    format!("Season {n}"),
                    "Season",
                    series,
                    none,
                    none,
                    now
                ])
                .unwrap();
                for e in 0..24 {
                    let ep = format!("{season}-e{e}");
                    item.execute(rusqlite::params![
                        ep,
                        "tv",
                        season,
                        format!("A Title {e}"),
                        "Episode",
                        series,
                        season,
                        none,
                        now
                    ])
                    .unwrap();
                    conn.execute(
                        "INSERT INTO user_data (user_id, item_id, playback_position_ticks) VALUES ('u', ?1, 5)",
                        [&ep],
                    )
                    .unwrap();
                }
            }
        }
        for a in 0..2000 {
            let album = format!("album-{a}");
            item.execute(rusqlite::params![
                album,
                "music",
                none,
                format!("Album {a}"),
                "MusicAlbum",
                none,
                none,
                none,
                now
            ])
            .unwrap();
            for t in 0..12 {
                item.execute(rusqlite::params![
                    format!("{album}-t{t}"),
                    "music",
                    album,
                    format!("Track {t}"),
                    "Audio",
                    none,
                    none,
                    album,
                    now
                ])
                .unwrap();
            }
        }
        for d in 0..400 {
            conn.execute(
                "INSERT INTO downloads (item_id, user_id, file_path, status) VALUES (?1, 'u', '/x', 'completed')",
                [format!("series-{}-s1-e{}", d % 300, d % 24)],
            )
            .unwrap();
        }
        drop(item);
        // No ANALYZE: the app never runs it, so the planner works without stats.
        conn.execute_batch("COMMIT;").unwrap();
    }
}