//! Tauri commands for database/storage operations //! //! TRACES: UR-002, UR-011, UR-012, UR-017, UR-019, UR-025, UR-047 | IR-013 | DR-012, DR-013, DR-022, DR-060 use std::sync::{Arc, Mutex}; use log::{debug, error, info, warn}; use serde::{Deserialize, Serialize}; use tauri::State; use crate::credentials::CredentialStore; use crate::storage::db_service::{DatabaseService, Query, QueryParam}; use crate::storage::Database; use crate::thumbnail::ThumbnailCache; use super::SmartCacheWrapper; // Cohesive command clusters in their own submodules, re-exported so the command // names remain at `commands::storage::*` (invoke_handler unchanged). mod people; mod series_prefs; mod thumbnails; pub use people::*; pub use series_prefs::*; pub use thumbnails::*; /// Wrapper for thread-safe database access pub struct DatabaseWrapper(pub Mutex); /// Wrapper for thread-safe credential store access pub struct CredentialStoreWrapper(pub Mutex); /// Wrapper for thread-safe thumbnail cache access pub struct ThumbnailCacheWrapper(pub Arc); /// Server info returned to frontend #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] pub struct ServerInfo { pub id: String, pub name: String, pub url: String, pub version: Option, } /// User info returned to frontend #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct UserInfo { pub id: String, pub server_id: String, pub username: String, pub is_active: bool, } /// Active session info (for session restoration) #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct ActiveSession { pub user_id: String, pub username: String, pub server_id: String, pub server_url: String, pub server_name: String, pub access_token: String, } /// Security status info #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct SecurityStatus { pub using_keyring: bool, pub storage_type: String, } /// Initialize the database and run migrations #[tauri::command] #[specta::specta] pub fn storage_init(db: State) -> Result { let database = db.0.lock().map_err(|e| e.to_string())?; Ok(database.path().to_string_lossy().to_string()) } /// A playable URL for a downloaded file on disk. /// /// Local media is served over a loopback HTTP server rather than handed to the /// webview as a `file://`/asset URL, because the asset protocol cannot stream a /// large file — it answers a range-less request with the whole thing, which /// Chromium abandons. See `media_server` for why real HTTP is used. /// /// The returned URL carries the server's per-session token, so it is only valid /// for this run of the app and must not be persisted. /// /// TRACES: UR-071 | DR-137 #[tauri::command] #[specta::specta] pub fn media_local_url( server: State, path: String, ) -> Result { server .0 .as_ref() .map(|s| s.url_for(&path)) .ok_or_else(|| "Local media server is not running".to_string()) } /// The stream selection for a downloaded file. /// /// The local-playback counterpart to `repository_get_stream_selection`. A file /// on disk needs no negotiation — it is a direct play over a local transport, /// with no quality ladder, because nothing about it can be re-negotiated — but /// the *frontend must not be the one to say so*. It gets the same /// [`StreamSelection`] shape as a streamed source so the player has one contract /// to consume rather than two, and so no caller has to infer a transport from a /// loopback URL. /// /// TRACES: UR-071, UR-079 | DR-225 #[tauri::command] #[specta::specta] pub fn media_local_selection( server: State, path: String, ) -> Result { server .0 .as_ref() .map(|s| crate::repository::StreamSelection::local_file(s.url_for(&path))) .ok_or_else(|| "Local media server is not running".to_string()) } /// Get storage directory path (parent directory of the database file) #[tauri::command] #[specta::specta] pub fn storage_get_path(db: State) -> Result { let database = db.0.lock().map_err(|e| e.to_string())?; let db_path = database.path(); // Return the parent directory instead of the database file path let storage_dir = db_path .parent() .ok_or_else(|| "Database path has no parent directory".to_string())?; Ok(storage_dir.to_string_lossy().to_string()) } /// Get database file size in bytes #[tauri::command] #[specta::specta] pub fn storage_get_size(db: State) -> Result, String> { let database = db.0.lock().map_err(|e| e.to_string())?; Ok(database.file_size()) } /// Get security status (keyring vs encrypted file fallback) #[tauri::command] #[specta::specta] pub fn storage_get_security_status( creds: State, ) -> Result { let store = creds.0.lock().map_err(|e| e.to_string())?; let using_keyring = store.is_using_keyring(); Ok(SecurityStatus { using_keyring, storage_type: if using_keyring { "system_keyring".to_string() } else { "encrypted_file".to_string() }, }) } /// Save a server connection /// Uses INSERT ... ON CONFLICT to avoid triggering CASCADE DELETE on users #[tauri::command] #[specta::specta] pub async fn storage_save_server( db: State<'_, DatabaseWrapper>, id: String, name: String, url: String, version: Option, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // IMPORTANT: Use ON CONFLICT ... DO UPDATE instead of INSERT OR REPLACE // INSERT OR REPLACE triggers DELETE + INSERT, which cascades to delete all users! let query = Query::with_params( "INSERT INTO servers (id, name, url, version, last_connected_at) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(id) DO UPDATE SET name = excluded.name, url = excluded.url, version = excluded.version, last_connected_at = CURRENT_TIMESTAMP", vec![ QueryParam::String(id), QueryParam::String(name), QueryParam::String(url), version.map(QueryParam::String).unwrap_or(QueryParam::Null), ], ); db_service.execute(query).await.map_err(|e| e.to_string())?; Ok(()) } /// Get all saved servers #[tauri::command] #[specta::specta] pub async fn storage_get_servers( db: State<'_, DatabaseWrapper>, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::new("SELECT id, name, url, version FROM servers ORDER BY last_connected_at DESC"); let servers = db_service .query_many(query, |row| { Ok(ServerInfo { id: row.get(0)?, name: row.get(1)?, url: row.get(2)?, version: row.get(3)?, }) }) .await .map_err(|e| e.to_string())?; Ok(servers) } /// Delete a server and all associated data #[tauri::command] #[specta::specta] pub async fn storage_delete_server( db: State<'_, DatabaseWrapper>, creds: State<'_, CredentialStoreWrapper>, server_id: String, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Get all user IDs for this server to delete their tokens let user_query = Query::with_params( "SELECT id FROM users WHERE server_id = ?", vec![QueryParam::String(server_id.clone())], ); let user_ids: Vec = db_service .query_many(user_query, |row| row.get(0)) .await .map_err(|e| e.to_string())?; // Delete tokens from secure storage { let store = creds.0.lock().map_err(|e| e.to_string())?; for user_id in user_ids { let _ = store.delete_token(&user_id); // Ignore errors, user might not have token } } // Drop store lock here // Delete server (cascades to users via foreign key) let delete_query = Query::with_params( "DELETE FROM servers WHERE id = ?", vec![QueryParam::String(server_id)], ); db_service .execute(delete_query) .await .map_err(|e| e.to_string())?; Ok(()) } /// Save a user account (token stored in secure storage, not database) #[tauri::command] #[specta::specta] pub async fn storage_save_user( db: State<'_, DatabaseWrapper>, creds: State<'_, CredentialStoreWrapper>, id: String, server_id: String, username: String, access_token: Option, ) -> Result { info!( "storage_save_user called: id={}, server_id={}, username={}", id, server_id, username ); let (db_service, db_path) = { let database = db.0.lock().map_err(|e| { error!("Failed to lock database: {}", e); e.to_string() })?; let path = database.path().to_path_buf(); (Arc::new(database.service()), path) }; // Save user metadata to database (without token) // IMPORTANT: Use ON CONFLICT ... DO UPDATE instead of INSERT OR REPLACE // INSERT OR REPLACE triggers DELETE + INSERT, which cascades to delete user_data! debug!("Executing INSERT INTO users with ON CONFLICT..."); let insert_query = Query::with_params( "INSERT INTO users (id, server_id, username, last_login_at) VALUES (?, ?, ?, CURRENT_TIMESTAMP) ON CONFLICT(id) DO UPDATE SET server_id = excluded.server_id, username = excluded.username, last_login_at = CURRENT_TIMESTAMP", vec![ QueryParam::String(id.clone()), QueryParam::String(server_id), QueryParam::String(username), ], ); db_service.execute(insert_query).await.map_err(|e| { error!("Failed to save user: {}", e); e.to_string() })?; info!("User saved to database successfully"); // Verify the user was actually saved let verify_query = Query::with_params( "SELECT COUNT(*) FROM users WHERE id = ?", vec![QueryParam::String(id.clone())], ); let verify_count: i32 = db_service .query_one(verify_query, |row| row.get(0)) .await .unwrap_or(-1); debug!("VERIFY: {} users with id={} after insert", verify_count, id); debug!("Database path: {:?}", db_path); // Force WAL checkpoint to ensure user data is persisted to disk let checkpoint_query = Query::new("PRAGMA wal_checkpoint(PASSIVE)"); match db_service.execute(checkpoint_query).await { Ok(_) => debug!("WAL checkpoint completed after save_user"), Err(e) => warn!("WAL checkpoint failed: {}", e), } // Save token to secure storage if provided let using_keyring = if let Some(token) = access_token { let store = creds.0.lock().map_err(|e| e.to_string())?; let result = store.save_token(&id, &token).map_err(|e| e.to_string())?; let is_keyring = matches!(result, crate::credentials::CredentialResult::Keyring); is_keyring } else { // No token provided, check current status let store = creds.0.lock().map_err(|e| e.to_string())?; store.is_using_keyring() }; // Return whether we're using the secure keyring Ok(using_keyring) } /// Get users for a server #[tauri::command] #[specta::specta] pub async fn storage_get_users( db: State<'_, DatabaseWrapper>, server_id: String, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "SELECT id, server_id, username, is_active FROM users WHERE server_id = ? ORDER BY last_login_at DESC", vec![QueryParam::String(server_id)], ); let users = db_service .query_many(query, |row| { Ok(UserInfo { id: row.get(0)?, server_id: row.get(1)?, username: row.get(2)?, is_active: row.get::<_, i32>(3)? != 0, }) }) .await .map_err(|e| e.to_string())?; Ok(users) } /// Set a user as active (and deactivate all other users globally) #[tauri::command] #[specta::specta] pub async fn storage_set_active_user( db: State<'_, DatabaseWrapper>, user_id: String, _server_id: String, ) -> Result<(), String> { info!("storage_set_active_user called: user_id={}", user_id); let (db_service, db_path) = { let database = db.0.lock().map_err(|e| e.to_string())?; let path = database.path().to_path_buf(); (Arc::new(database.service()), path) }; // Deactivate ALL users globally (since we only connect to one server at a time) let deactivate_query = Query::new("UPDATE users SET is_active = 0"); db_service .execute(deactivate_query) .await .map_err(|e| e.to_string())?; // Activate the specified user and update last_login_at let activate_query = Query::with_params( "UPDATE users SET is_active = 1, last_login_at = CURRENT_TIMESTAMP WHERE id = ?", vec![QueryParam::String(user_id.clone())], ); let rows_affected = db_service .execute(activate_query) .await .map_err(|e| e.to_string())?; debug!("storage_set_active_user: {} rows affected", rows_affected); if rows_affected == 0 { warn!("No user found with id={} to set as active!", user_id); } // Verify the user is now active let verify_query = Query::new("SELECT COUNT(*) FROM users WHERE is_active = 1"); let verify_count: i32 = db_service .query_one(verify_query, |row| row.get(0)) .await .unwrap_or(-1); debug!("VERIFY: {} active users after set_active", verify_count); debug!("Database path: {:?}", db_path); // Force WAL checkpoint to ensure data is persisted to disk let checkpoint_query = Query::new("PRAGMA wal_checkpoint(PASSIVE)"); match db_service.execute(checkpoint_query).await { Ok(_) => debug!("WAL checkpoint completed"), Err(e) => warn!("WAL checkpoint failed: {}", e), } Ok(()) } /// Get the active user for a server #[tauri::command] #[specta::specta] pub async fn storage_get_active_user( db: State<'_, DatabaseWrapper>, server_id: String, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "SELECT id, server_id, username, is_active FROM users WHERE server_id = ? AND is_active = 1", vec![QueryParam::String(server_id)], ); db_service .query_optional(query, |row| { Ok(UserInfo { id: row.get(0)?, server_id: row.get(1)?, username: row.get(2)?, is_active: true, }) }) .await .map_err(|e| e.to_string()) } /// Get the active session (user + server + token) for session restoration #[tauri::command] #[specta::specta] pub async fn storage_get_active_session( db: State<'_, DatabaseWrapper>, creds: State<'_, CredentialStoreWrapper>, ) -> Result, String> { info!("storage_get_active_session called"); let (db_service, db_path) = { let database = db.0.lock().map_err(|e| e.to_string())?; let path = database.path().to_path_buf(); (Arc::new(database.service()), path) }; // Debug: count total users and active users let total_query = Query::new("SELECT COUNT(*) FROM users"); let total_users: i32 = db_service .query_one(total_query, |row| row.get(0)) .await .unwrap_or(-1); let active_query = Query::new("SELECT COUNT(*) FROM users WHERE is_active = 1"); let active_users: i32 = db_service .query_one(active_query, |row| row.get(0)) .await .unwrap_or(-1); debug!( "Database state: {} total users, {} active users", total_users, active_users ); debug!("Database path: {:?}", db_path); // Find active user with their server info, ordered by most recently logged in let session_query = Query::new( "SELECT u.id, u.username, u.server_id, s.url, s.name 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", ); let result = db_service .query_optional(session_query, |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, String>(3)?, row.get::<_, String>(4)?, )) }) .await .map_err(|e| e.to_string())?; match result { Some((user_id, username, server_id, server_url, server_name)) => { info!("Found active user: {} ({})", username, user_id); // Get token from secure storage let store = creds.0.lock().map_err(|e| e.to_string())?; match store.get_token(&user_id) { Ok(access_token) => { debug!("Successfully retrieved token from secure storage"); Ok(Some(ActiveSession { user_id, username, server_id, server_url, server_name, access_token, })) } Err(e) => { // Token not found or error - session is invalid warn!("Failed to get token from secure storage: {:?}", e); Ok(None) } } } None => { info!("No active user found in database"); Ok(None) } } } /// Get user's access token from secure storage #[tauri::command] #[specta::specta] pub fn storage_get_access_token( creds: State, user_id: String, ) -> Result, String> { let store = creds.0.lock().map_err(|e| e.to_string())?; match store.get_token(&user_id) { Ok(token) => Ok(Some(token)), Err(crate::credentials::CredentialError::NotFound) => Ok(None), Err(e) => Err(e.to_string()), } } /// Delete a user account and their token from secure storage #[tauri::command] #[specta::specta] pub async fn storage_delete_user( db: State<'_, DatabaseWrapper>, creds: State<'_, CredentialStoreWrapper>, user_id: String, ) -> Result<(), String> { info!("storage_delete_user called: user_id={}", user_id); debug!("STACK TRACE: This is where the user is being deleted!"); // Delete token from secure storage first { let store = creds.0.lock().map_err(|e| e.to_string())?; let _ = store.delete_token(&user_id); // Ignore errors, token might not exist } // Delete user from database let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "DELETE FROM users WHERE id = ?", vec![QueryParam::String(user_id)], ); db_service.execute(query).await.map_err(|e| e.to_string())?; info!("User deleted successfully"); Ok(()) } /// Playback progress info #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct PlaybackProgress { pub item_id: String, /// Resume position in milliseconds. Stored as Jellyfin ticks in the DB; /// converted here so the frontend never sees ticks. pub position_ms: i64, pub is_played: bool, pub is_favorite: bool, pub play_count: i32, } /// Update playback progress in local database /// This stores the progress locally for offline access and "continue watching" #[tauri::command] #[specta::specta] pub async fn storage_update_playback_progress( db: State<'_, DatabaseWrapper>, user_id: String, item_id: String, position_ms: i64, ) -> Result<(), String> { // The frontend speaks milliseconds; ticks are a Jellyfin storage detail that // stays on this side of the boundary. 10_000 ticks = 1 ms. let position_ticks = position_ms * 10_000; let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Note: user_id and item_id are Jellyfin IDs (strings) // The user_data table uses them directly as strings, not foreign keys to integer IDs // Insert or update playback position let query = Query::with_params( "INSERT INTO user_data (user_id, item_id, playback_position_ticks, last_played_at, pending_sync) VALUES (?, ?, ?, CURRENT_TIMESTAMP, 1) ON CONFLICT(user_id, item_id) DO UPDATE SET playback_position_ticks = excluded.playback_position_ticks, last_played_at = excluded.last_played_at, pending_sync = 1", vec![ QueryParam::String(user_id), QueryParam::String(item_id.clone()), QueryParam::Int64(position_ticks), ], ); match db_service.execute(query).await { Ok(_) => Ok(()), Err(e) if e.contains("constraint") || e.contains("UNIQUE") => { // Foreign key constraint failed - this can happen if: // 1. The item isn't synced locally yet (item_id not in items table) // 2. Database migration 003 hasn't been applied (old database) // // In either case, we silently succeed - playback progress will still // be reported to the server, and local progress will be tracked // once the item is synced. debug!( "Skipping local playback progress for item {} (not cached locally)", item_id ); Ok(()) } Err(e) => Err(format!("Failed to update playback progress: {}", e)), } } /// Update playback progress with context in local database /// This stores the progress along with playback context (container vs single) #[tauri::command] #[specta::specta] pub async fn storage_update_playback_context( db: State<'_, DatabaseWrapper>, user_id: String, item_id: String, position_ms: i64, context_type: Option, context_id: Option, ) -> Result<(), String> { use crate::storage::db_service::{Query, QueryParam}; // Milliseconds in, Jellyfin ticks stored. 10_000 ticks = 1 ms. let position_ticks = position_ms * 10_000; let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "INSERT INTO user_data (user_id, item_id, playback_position_ticks, last_played_at, playback_context_type, playback_context_id, pending_sync) VALUES (?, ?, ?, CURRENT_TIMESTAMP, ?, ?, 1) ON CONFLICT(user_id, item_id) DO UPDATE SET playback_position_ticks = excluded.playback_position_ticks, last_played_at = excluded.last_played_at, playback_context_type = excluded.playback_context_type, playback_context_id = excluded.playback_context_id, pending_sync = 1", vec![ QueryParam::String(user_id.clone()), QueryParam::String(item_id.clone()), QueryParam::Int64(position_ticks), context_type .map(QueryParam::String) .unwrap_or(QueryParam::Null), context_id .map(QueryParam::String) .unwrap_or(QueryParam::Null), ], ); match db_service.execute(query).await { Ok(_) => Ok(()), Err(e) if e.contains("constraint") || e.contains("UNIQUE") => { debug!( "Skipping local playback context for item {} (not cached locally)", item_id ); Ok(()) } Err(e) => Err(format!("Failed to update playback context: {}", e)), } } /// Mark item as played in local database #[tauri::command] #[specta::specta] pub async fn storage_mark_played( db: State<'_, DatabaseWrapper>, smart_cache: State<'_, SmartCacheWrapper>, user_id: String, item_id: String, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Get album_id for this item (if it's an audio track) let album_query = Query::with_params( "SELECT album_id, item_type FROM items WHERE id = ?", vec![QueryParam::String(item_id.clone())], ); let album_info: Option<(Option, String)> = db_service .query_optional(album_query, |row| Ok((row.get(0)?, row.get(1)?))) .await .unwrap_or(None); // Track play in SmartCache (for audio tracks with albums) if let Some((Some(album_id), item_type)) = album_info.clone() { if item_type == "Audio" { // Clone cache to avoid holding lock across async operations let should_cache = { let cache = smart_cache.0.lock().map_err(|e| e.to_string())?; cache.track_play(&item_id, Some(&album_id)); cache.should_cache_album(&album_id) }; // Check if we should cache this album if let Some(true) = should_cache { info!("Album affinity threshold reached for album {}!", album_id); // Auto-queue remaining album tracks for download // Get tracks from this album that aren't already downloaded let tracks_query = Query::with_params( "SELECT i.id, i.name, i.artists, i.album_name FROM items i LEFT JOIN downloads d ON d.item_id = i.id AND d.user_id = ? WHERE i.album_id = ? AND i.item_type = 'Audio' AND (d.id IS NULL OR d.status NOT IN ('completed', 'downloading', 'pending')) ORDER BY i.index_number", vec![ QueryParam::String(user_id.clone()), QueryParam::String(album_id.clone()), ], ); let tracks: Vec<(String, String, Option, Option)> = db_service .query_many(tracks_query, |row| { Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)) }) .await .unwrap_or_else(|e| { warn!("Failed to query album tracks: {}", e); Vec::new() }); if !tracks.is_empty() { info!( "Auto-queueing {} tracks from album for download", tracks.len() ); // Queue each track with high priority (50) and mark as auto-downloaded for (track_id, track_name, artist_name, album_name) in tracks { // Generate a sanitized file path (simplified version) let sanitized_name = track_name .chars() .map(|c| { if c.is_alphanumeric() || c == ' ' || c == '-' || c == '_' { c } else { '_' } }) .collect::(); let file_path = format!("downloads/{}/{}.mp3", album_id, sanitized_name); let insert_query = Query::with_params( "INSERT INTO downloads (item_id, user_id, file_path, status, priority, queued_at, item_name, artist_name, album_name, download_source) VALUES (?, ?, ?, 'pending', 50, CURRENT_TIMESTAMP, ?, ?, ?, 'auto') ON CONFLICT(item_id, user_id) DO UPDATE SET priority = 50, status = 'pending', download_source = 'auto'", vec![ QueryParam::String(track_id.clone()), QueryParam::String(user_id.clone()), QueryParam::String(file_path), QueryParam::String(track_name), artist_name.map(QueryParam::String).unwrap_or(QueryParam::Null), album_name.map(QueryParam::String).unwrap_or(QueryParam::Null), ], ); if let Err(e) = db_service.execute(insert_query).await { warn!("Failed to queue track {}: {}", track_id, e); } } info!("Album tracks queued for automatic download"); } else { info!("All album tracks already downloaded or queued"); } } } } let query = Query::with_params( "INSERT INTO user_data (user_id, item_id, is_played, play_count, last_played_at, pending_sync) VALUES (?, ?, 1, 1, CURRENT_TIMESTAMP, 1) ON CONFLICT(user_id, item_id) DO UPDATE SET is_played = 1, play_count = play_count + 1, last_played_at = CURRENT_TIMESTAMP, pending_sync = 1", vec![QueryParam::String(user_id), QueryParam::String(item_id.clone())], ); match db_service.execute(query).await { Ok(_) => Ok(()), Err(e) if e.contains("constraint") => { // Foreign key constraint failed - item not cached locally debug!( "Skipping local mark_played for item {} (not cached locally)", item_id ); Ok(()) } Err(e) => Err(e.to_string()), } } /// Set the watched flag locally for an item **and everything inside it**. /// /// This backs the watched toggle, and is deliberately separate from /// [`storage_mark_played`] — which reports a single track/episode finishing and /// increments `play_count` — because the toggle has two directions and applies /// to containers. /// /// The recursion is what makes the toggle honest offline. Jellyfin applies /// `POST`/`DELETE /PlayedItems/{id}` recursively over a season or series, so /// online the server fixes up the children on the next read; with no server to /// ask, marking a season watched would otherwise tick the season and leave every /// episode inside it unwatched. Targets are drawn from `items` by the same link /// columns the rest of the offline layer uses, so an id that is not cached /// selects nothing and the statement is a no-op rather than a foreign-key error. /// /// Un-marking clears the resume position too, matching the server, so an item /// un-marked offline does not come back offering to resume from a position it is /// no longer meant to have. /// /// `pending_sync = 1` hands the rows to the sync drain. /// /// TRACES: UR-073 | DR-158 #[tauri::command] #[specta::specta] pub async fn storage_set_watched( db: State<'_, DatabaseWrapper>, user_id: String, item_id: String, watched: bool, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // The item itself plus its descendants: a season's episodes reach it by // season_id, a series' by series_id, its seasons by parent_id, an album's // tracks by album_id. let targets = "SELECT id FROM items WHERE id = ? OR parent_id = ? OR album_id = ? OR season_id = ? OR series_id = ?"; let sql = if watched { format!( "INSERT INTO user_data (user_id, item_id, is_played, play_count, last_played_at, pending_sync) SELECT ?, id, 1, 1, CURRENT_TIMESTAMP, 1 FROM ({targets}) ON CONFLICT(user_id, item_id) DO UPDATE SET is_played = 1, play_count = MAX(user_data.play_count, 1), last_played_at = CURRENT_TIMESTAMP, pending_sync = 1" ) } else { format!( "INSERT INTO user_data (user_id, item_id, is_played, play_count, playback_position_ticks, pending_sync) SELECT ?, id, 0, 0, 0, 1 FROM ({targets}) ON CONFLICT(user_id, item_id) DO UPDATE SET is_played = 0, play_count = 0, playback_position_ticks = 0, pending_sync = 1" ) }; let query = Query::with_params( sql, vec![ QueryParam::String(user_id), QueryParam::String(item_id.clone()), QueryParam::String(item_id.clone()), QueryParam::String(item_id.clone()), QueryParam::String(item_id.clone()), QueryParam::String(item_id.clone()), ], ); db_service.execute(query).await.map_err(|e| e.to_string())?; Ok(()) } /// Get playback progress for an item #[tauri::command] #[specta::specta] pub async fn storage_get_playback_progress( db: State<'_, DatabaseWrapper>, user_id: String, item_id: String, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "SELECT item_id, playback_position_ticks, is_played, is_favorite, play_count FROM user_data WHERE user_id = ? AND item_id = ?", vec![QueryParam::String(user_id), QueryParam::String(item_id)], ); db_service .query_optional(query, |row| { let position_ticks: i64 = row.get(1)?; Ok(PlaybackProgress { item_id: row.get(0)?, position_ms: position_ticks / 10_000, is_played: row.get::<_, i32>(2)? != 0, is_favorite: row.get::<_, i32>(3)? != 0, play_count: row.get(4)?, }) }) .await .map_err(|e| e.to_string()) } /// Mark pending sync as completed for an item #[tauri::command] #[specta::specta] pub async fn storage_mark_synced( db: State<'_, DatabaseWrapper>, user_id: String, item_id: String, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "UPDATE user_data SET pending_sync = 0, synced_at = CURRENT_TIMESTAMP WHERE user_id = ? AND item_id = ?", vec![QueryParam::String(user_id), QueryParam::String(item_id)], ); db_service.execute(query).await.map_err(|e| e.to_string())?; Ok(()) } /// Toggle favorite status for an item in local database /// This updates the is_favorite field and marks it for sync to Jellyfin #[tauri::command] #[specta::specta] pub async fn storage_toggle_favorite( db: State<'_, DatabaseWrapper>, user_id: String, item_id: String, is_favorite: bool, ) -> Result { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Insert or update favorite status let query = Query::with_params( "INSERT INTO user_data (user_id, item_id, is_favorite, pending_sync) VALUES (?, ?, ?, 1) ON CONFLICT(user_id, item_id) DO UPDATE SET is_favorite = excluded.is_favorite, pending_sync = 1", vec![ QueryParam::String(user_id), QueryParam::String(item_id.clone()), QueryParam::Int(is_favorite as i32), ], ); match db_service.execute(query).await { Ok(_) => Ok(is_favorite), Err(e) if e.contains("constraint") => { // Foreign key constraint failed - item not cached locally // Still return the requested state, server update will work separately debug!( "Skipping local favorite toggle for item {} (not cached locally)", item_id ); Ok(is_favorite) } Err(e) => Err(format!("Failed to toggle favorite: {}", e)), } } // ============================================================================= // Offline Data Queries // ============================================================================= /// Cached library info returned to frontend #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct CachedLibrary { pub id: String, pub server_id: String, pub name: String, pub collection_type: Option, pub image_tag: Option, } /// Cached media item returned to frontend #[derive(specta::Type, Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct CachedItem { pub id: String, pub name: String, pub item_type: String, pub parent_id: Option, pub library_id: Option, pub overview: Option, pub genres: Option, pub runtime_ticks: Option, pub production_year: Option, pub community_rating: Option, pub official_rating: Option, pub primary_image_tag: Option, // Music-specific pub album_id: Option, pub album_name: Option, pub album_artist: Option, pub artists: Option, pub index_number: Option, // TV-specific pub series_id: Option, pub series_name: Option, pub season_id: Option, pub season_name: Option, pub parent_index_number: Option, } /// Get cached libraries for a server #[tauri::command] #[specta::specta] pub async fn storage_get_libraries( db: State<'_, DatabaseWrapper>, server_id: String, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "SELECT id, server_id, name, collection_type, image_tag FROM libraries WHERE server_id = ? ORDER BY sort_order ASC, name ASC", vec![QueryParam::String(server_id)], ); let libraries = db_service .query_many(query, |row| { Ok(CachedLibrary { id: row.get(0)?, server_id: row.get(1)?, name: row.get(2)?, collection_type: row.get(3)?, image_tag: row.get(4)?, }) }) .await .map_err(|e| e.to_string())?; Ok(libraries) } fn row_to_cached_item(row: &rusqlite::Row) -> rusqlite::Result { Ok(CachedItem { id: row.get(0)?, name: row.get(1)?, item_type: row.get(2)?, parent_id: row.get(3)?, library_id: row.get(4)?, overview: row.get(5)?, genres: row.get(6)?, runtime_ticks: row.get(7)?, production_year: row.get(8)?, community_rating: row.get(9)?, official_rating: row.get(10)?, primary_image_tag: row.get(11)?, album_id: row.get(12)?, album_name: row.get(13)?, album_artist: row.get(14)?, artists: row.get(15)?, index_number: row.get(16)?, series_id: row.get(17)?, series_name: row.get(18)?, season_id: row.get(19)?, season_name: row.get(20)?, parent_index_number: row.get(21)?, }) } /// Get cached items with optional filtering #[tauri::command] #[specta::specta] pub async fn storage_get_items( db: State<'_, DatabaseWrapper>, server_id: String, parent_id: Option, library_id: Option, item_type: Option, limit: Option, offset: Option, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Build base query let base_select = "SELECT id, name, item_type, parent_id, library_id, overview, genres, runtime_ticks, production_year, community_rating, official_rating, primary_image_tag, album_id, album_name, album_artist, artists, index_number, series_id, series_name, season_id, season_name, parent_index_number FROM items WHERE server_id = ?"; let mut conditions = Vec::new(); let mut params = vec![QueryParam::String(server_id)]; if let Some(pid) = parent_id { conditions.push("parent_id = ?"); params.push(QueryParam::String(pid)); } if let Some(lid) = library_id { conditions.push("library_id = ?"); params.push(QueryParam::String(lid)); } if let Some(itype) = item_type { conditions.push("item_type = ?"); params.push(QueryParam::String(itype)); } let mut sql = base_select.to_string(); for cond in &conditions { sql.push_str(&format!(" AND {}", cond)); } sql.push_str(" ORDER BY sort_name ASC, name ASC"); if let Some(lim) = limit { sql.push_str(&format!(" LIMIT {}", lim)); } if let Some(off) = offset { sql.push_str(&format!(" OFFSET {}", off)); } let query = Query::with_params(sql, params); let items = db_service .query_many(query, row_to_cached_item) .await .map_err(|e| e.to_string())?; Ok(items) } /// Get a single cached item by ID #[tauri::command] #[specta::specta] pub async fn storage_get_item( db: State<'_, DatabaseWrapper>, item_id: String, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "SELECT id, name, item_type, parent_id, library_id, overview, genres, runtime_ticks, production_year, community_rating, official_rating, primary_image_tag, album_id, album_name, album_artist, artists, index_number, series_id, series_name, season_id, season_name, parent_index_number FROM items WHERE id = ?", vec![QueryParam::String(item_id)], ); db_service .query_optional(query, row_to_cached_item) .await .map_err(|e| e.to_string()) } /// Search cached items using FTS #[tauri::command] #[specta::specta] pub async fn storage_search_items( db: State<'_, DatabaseWrapper>, server_id: String, query: String, limit: Option, ) -> Result, String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Escape FTS special characters and add prefix matching let fts_query = format!("{}*", query.replace('"', "\"\"")); let limit_clause = limit.map(|l| format!(" LIMIT {}", l)).unwrap_or_default(); let sql = format!( "SELECT i.id, i.name, i.item_type, i.parent_id, i.library_id, i.overview, i.genres, i.runtime_ticks, i.production_year, i.community_rating, i.official_rating, i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists, i.index_number, i.series_id, i.series_name, i.season_id, i.season_name, i.parent_index_number FROM items i JOIN items_fts fts ON fts.rowid = i.rowid WHERE i.server_id = ? AND items_fts MATCH ? ORDER BY rank{}", limit_clause ); let query_obj = Query::with_params( sql, vec![QueryParam::String(server_id), QueryParam::String(fts_query)], ); let items = db_service .query_many(query_obj, row_to_cached_item) .await .map_err(|e| e.to_string())?; Ok(items) } /// Save a library to the cache #[tauri::command] #[specta::specta] pub async fn storage_save_library( db: State<'_, DatabaseWrapper>, id: String, server_id: String, name: String, collection_type: Option, image_tag: Option, sort_order: Option, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "INSERT OR REPLACE INTO libraries (id, server_id, name, collection_type, image_tag, sort_order, synced_at) VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)", vec![ QueryParam::String(id), QueryParam::String(server_id), QueryParam::String(name), collection_type.map(QueryParam::String).unwrap_or(QueryParam::Null), image_tag.map(QueryParam::String).unwrap_or(QueryParam::Null), QueryParam::Int(sort_order.unwrap_or(0)), ], ); db_service.execute(query).await.map_err(|e| e.to_string())?; Ok(()) } /// Save an item to the cache #[tauri::command] #[specta::specta] pub async fn storage_save_item( db: State<'_, DatabaseWrapper>, item: CachedItem, server_id: String, ) -> Result<(), String> { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; // Generate sort_name from name (remove leading "The ", "A ", etc.) let sort_name = item .name .strip_prefix("The ") .or_else(|| item.name.strip_prefix("A ")) .or_else(|| item.name.strip_prefix("An ")) .unwrap_or(&item.name) .to_string(); let query = Query::with_params( "INSERT OR REPLACE INTO items ( id, server_id, library_id, parent_id, name, sort_name, item_type, overview, genres, runtime_ticks, production_year, community_rating, official_rating, primary_image_tag, album_id, album_name, album_artist, artists, index_number, series_id, series_name, season_id, season_name, parent_index_number, synced_at ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP )", vec![ QueryParam::String(item.id), QueryParam::String(server_id), item.library_id .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.parent_id .map(QueryParam::String) .unwrap_or(QueryParam::Null), QueryParam::String(item.name), QueryParam::String(sort_name), QueryParam::String(item.item_type), item.overview .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.genres .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.runtime_ticks .map(QueryParam::Int64) .unwrap_or(QueryParam::Null), item.production_year .map(QueryParam::Int) .unwrap_or(QueryParam::Null), item.community_rating .map(QueryParam::Float) .unwrap_or(QueryParam::Null), item.official_rating .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.primary_image_tag .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.album_id .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.album_name .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.album_artist .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.artists .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.index_number .map(QueryParam::Int) .unwrap_or(QueryParam::Null), item.series_id .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.series_name .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.season_id .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.season_name .map(QueryParam::String) .unwrap_or(QueryParam::Null), item.parent_index_number .map(QueryParam::Int) .unwrap_or(QueryParam::Null), ], ); db_service.execute(query).await.map_err(|e| e.to_string())?; Ok(()) } /// Get count of pending sync operations for a user #[tauri::command] #[specta::specta] pub async fn storage_get_pending_sync_count( db: State<'_, DatabaseWrapper>, user_id: String, ) -> Result { let db_service = { let database = db.0.lock().map_err(|e| e.to_string())?; Arc::new(database.service()) }; let query = Query::with_params( "SELECT COUNT(*) FROM user_data WHERE user_id = ? AND pending_sync = 1", vec![QueryParam::String(user_id)], ); let count: i32 = db_service .query_one(query, |row| row.get(0)) .await .map_err(|e| e.to_string())?; Ok(count) } #[cfg(test)] mod tests { use super::*; #[test] fn test_server_info_serialization() { let server = ServerInfo { id: "server-123".to_string(), name: "My Server".to_string(), url: "https://jellyfin.example.com".to_string(), version: Some("10.8.0".to_string()), }; let json = serde_json::to_string(&server); assert!(json.is_ok()); let serialized = json.unwrap(); assert!(serialized.contains("server-123")); assert!(serialized.contains("My Server")); } #[test] fn test_server_info_without_version() { let server = ServerInfo { id: "server-456".to_string(), name: "Test Server".to_string(), url: "https://test.local".to_string(), version: None, }; let json = serde_json::to_string(&server).unwrap(); assert!(json.contains("null") || json.contains("\"version\":null")); } #[test] fn test_server_info_roundtrip() { let original = ServerInfo { id: "srv-999".to_string(), name: "Production".to_string(), url: "https://prod.jellyfin.example.com:8096".to_string(), version: Some("10.9.0".to_string()), }; let json = serde_json::to_string(&original).unwrap(); let deserialized: ServerInfo = serde_json::from_str(&json).unwrap(); assert_eq!(original.id, deserialized.id); assert_eq!(original.name, deserialized.name); assert_eq!(original.url, deserialized.url); assert_eq!(original.version, deserialized.version); } #[test] fn test_user_info_serialization() { let user = UserInfo { id: "user-123".to_string(), server_id: "server-456".to_string(), username: "john_doe".to_string(), is_active: true, }; let json = serde_json::to_string(&user); assert!(json.is_ok()); let serialized = json.unwrap(); assert!(serialized.contains("user-123")); assert!(serialized.contains("john_doe")); } #[test] fn test_user_info_inactive() { let user = UserInfo { id: "user-inactive".to_string(), server_id: "server-789".to_string(), username: "jane_doe".to_string(), is_active: false, }; let json = serde_json::to_string(&user).unwrap(); assert!(json.contains("false")); let deserialized: UserInfo = serde_json::from_str(&json).unwrap(); assert!(!deserialized.is_active); } #[test] fn test_active_session_serialization() { let session = ActiveSession { user_id: "user-001".to_string(), username: "alice".to_string(), server_id: "server-001".to_string(), server_url: "https://jellyfin.example.com".to_string(), server_name: "Home Jellyfin".to_string(), access_token: "very-long-token-string-abc123".to_string(), }; let json = serde_json::to_string(&session); assert!(json.is_ok()); let serialized = json.unwrap(); assert!(serialized.contains("alice")); assert!(serialized.contains("Home Jellyfin")); } #[test] fn test_active_session_roundtrip() { let original = ActiveSession { user_id: "u999".to_string(), username: "testuser".to_string(), server_id: "s999".to_string(), server_url: "https://test.example.com:8096".to_string(), server_name: "Test Server".to_string(), access_token: "token-xyz".to_string(), }; let json = serde_json::to_string(&original).unwrap(); let deserialized: ActiveSession = serde_json::from_str(&json).unwrap(); assert_eq!(original.user_id, deserialized.user_id); assert_eq!(original.username, deserialized.username); assert_eq!(original.access_token, deserialized.access_token); } #[test] fn test_security_status_with_keyring() { let status = SecurityStatus { using_keyring: true, storage_type: "system_keyring".to_string(), }; let json = serde_json::to_string(&status).unwrap(); assert!(json.contains("true")); assert!(json.contains("system_keyring")); } #[test] fn test_security_status_with_encrypted_file() { let status = SecurityStatus { using_keyring: false, storage_type: "encrypted_file".to_string(), }; let json = serde_json::to_string(&status).unwrap(); assert!(json.contains("false")); assert!(json.contains("encrypted_file")); } #[test] fn test_playback_progress_serialization() { let progress = PlaybackProgress { item_id: "item-123".to_string(), position_ms: 150_000_000, is_played: true, is_favorite: false, play_count: 3, }; let json = serde_json::to_string(&progress); assert!(json.is_ok()); let serialized = json.unwrap(); assert!(serialized.contains("item-123")); assert!(serialized.contains("150000000")); } #[test] fn test_playback_progress_played_status() { let progress = PlaybackProgress { item_id: "item-456".to_string(), position_ms: 0, is_played: true, is_favorite: true, play_count: 1, }; let json = serde_json::to_string(&progress).unwrap(); let deserialized: PlaybackProgress = serde_json::from_str(&json).unwrap(); assert!(deserialized.is_played); assert!(deserialized.is_favorite); assert_eq!(deserialized.play_count, 1); } #[test] fn test_playback_progress_not_played() { let progress = PlaybackProgress { item_id: "item-789".to_string(), position_ms: 30_000_000, is_played: false, is_favorite: false, play_count: 0, }; let json = serde_json::to_string(&progress).unwrap(); let deserialized: PlaybackProgress = serde_json::from_str(&json).unwrap(); assert!(!deserialized.is_played); assert_eq!(deserialized.play_count, 0); } #[test] fn test_database_wrapper_structure() { // Verify DatabaseWrapper can be created and holds Mutex assert!(std::mem::size_of::() > 0); } #[test] fn test_credential_store_wrapper_structure() { // Verify CredentialStoreWrapper can be created assert!(std::mem::size_of::() > 0); } #[test] fn test_thumbnail_cache_wrapper_structure() { // Verify ThumbnailCacheWrapper holds Arc assert!(std::mem::size_of::() > 0); } #[test] fn test_user_info_camel_case() { let user = UserInfo { id: "u1".to_string(), server_id: "s1".to_string(), username: "user1".to_string(), is_active: true, }; let json = serde_json::to_string(&user).unwrap(); // Verify camelCase serialization assert!(json.contains("serverId")); assert!(json.contains("isActive")); } #[test] fn test_active_session_camel_case() { let session = ActiveSession { user_id: "u1".to_string(), username: "user1".to_string(), server_id: "s1".to_string(), server_url: "url1".to_string(), server_name: "name1".to_string(), access_token: "token1".to_string(), }; let json = serde_json::to_string(&session).unwrap(); // Verify camelCase serialization assert!(json.contains("userId")); assert!(json.contains("serverId")); assert!(json.contains("serverUrl")); assert!(json.contains("serverName")); assert!(json.contains("accessToken")); } #[test] fn test_playback_progress_camel_case() { let progress = PlaybackProgress { item_id: "i1".to_string(), position_ms: 100, is_played: true, is_favorite: false, play_count: 1, }; let json = serde_json::to_string(&progress).unwrap(); // Verify camelCase serialization assert!(json.contains("itemId")); assert!(json.contains("positionMs")); assert!(json.contains("isPlayed")); assert!(json.contains("isFavorite")); assert!(json.contains("playCount")); } }