//! 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>, service: RusqliteService, path: PathBuf, } impl Database { /// Open or create the database at a specific path pub fn open(path: &PathBuf) -> SqliteResult { // 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::>>()?; 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 { 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 { 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> { 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, 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 = 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 { 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 = 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 = 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 = 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 = 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 = 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 = 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 = 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(); } }