Duncan Tourolle 6866f03c55 Architecture remediation A/B/F: poison-tolerant locks, graceful backend init, doc fixes
Workstream A — poison-tolerant locking:
- Add utils/lock.rs with MutexSafe/RwLockSafe extension traits that recover a
  poisoned std::sync lock instead of panicking, plus unit tests.
- Replace all 153 .lock().unwrap() and 4 .read()/.write().unwrap() production
  sites with _safe variants across 14 files, eliminating the player
  crash-cascade class. Tokio async mutexes are unchanged.

Workstream B — graceful backend init:
- create_player_backend no longer panics when MPV/ExoPlayer fail to initialize;
  it falls back to NullBackend and emits a backend-init-failed event so the UI
  can show "playback unavailable" instead of the app crashing. Fatal DB-setup
  panics are kept.

Workstream F — doc reconciliation:
- Rewrite software-architecture.md's inaccurate "thin UI / ~800 lines" claims to
  reflect reality (~20.5k non-test frontend) and document the events+polling
  hybrid plus the new locking/backend-init behavior.
2026-06-20 16:03:54 +02:00

758 lines
23 KiB
Rust

//! 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};
use schema::MIGRATIONS;
pub use db_service::{DatabaseService, RusqliteService};
/// Database connection wrapper with thread-safe access
pub struct Database {
conn: Arc<Mutex<Connection>>,
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;")?;
// Enable WAL mode for better concurrent access
conn.execute_batch("PRAGMA journal_mode = WAL;")?;
let db = Self {
conn: Arc::new(Mutex::new(conn)),
path: path.clone(),
};
// Run migrations
db.migrate()?;
Ok(db)
}
/// 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 db = Self {
conn: Arc::new(Mutex::new(conn)),
path: PathBuf::from(":memory:"),
};
// Run migrations
db.migrate()?;
Ok(db)
}
/// Get connection (for testing)
#[cfg(test)]
pub fn connection(&self) -> Arc<Mutex<Connection>> {
Arc::clone(&self.conn)
}
/// Run all pending migrations
pub fn migrate(&self) -> SqliteResult<()> {
info!("Starting database migrations...");
let conn = self.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()) {
info!("Applying migration: {}", name);
match conn.execute_batch(sql) {
Ok(_) => {
info!("Successfully applied migration: {}", name);
match conn.execute(
"INSERT INTO _migrations (name) VALUES (?1)",
[name],
) {
Ok(_) => debug!("Recorded migration: {}", name),
Err(e) => {
error!("Failed to record migration {}: {}", name, e);
return Err(e);
}
}
}
Err(e) => {
error!("Failed to apply migration {}: {}", name, e);
return Err(e);
}
}
} else {
debug!("Skipping already applied migration: {}", name);
}
}
info!("All migrations completed successfully");
Ok(())
}
/// Get a database service for async-safe operations
///
/// This wraps all blocking database operations in spawn_blocking to prevent
/// freezing the async runtime.
pub fn service(&self) -> RusqliteService {
RusqliteService::new(Arc::clone(&self.conn))
}
/// 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:"));
}
#[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",
];
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");
}
}