jellytau_lib/storage/
mod.rs

1//! Offline storage module using SQLite
2//!
3//! Provides local caching of Jellyfin metadata, download management,
4//! and offline mutation queue for sync-back operations.
5
6pub mod db_service;
7pub mod models;
8pub mod schema;
9
10use crate::utils::lock::MutexSafe;
11use std::path::PathBuf;
12use std::sync::{Arc, Mutex};
13
14use log::{debug, error, info};
15use rusqlite::{Connection, Result as SqliteResult};
16
17pub use db_service::{DatabaseService, RusqliteService};
18use schema::MIGRATIONS;
19
20/// Database connection wrapper with thread-safe access
21pub struct Database {
22    conn: Arc<Mutex<Connection>>,
23    path: PathBuf,
24}
25
26impl Database {
27    /// Open or create the database at a specific path
28    pub fn open(path: &PathBuf) -> SqliteResult<Self> {
29        // Ensure parent directory exists
30        if let Some(parent) = path.parent() {
31            std::fs::create_dir_all(parent).ok();
32        }
33
34        let conn = Connection::open(path)?;
35
36        // Enable foreign keys
37        conn.execute_batch("PRAGMA foreign_keys = ON;")?;
38
39        // Enable WAL mode for better concurrent access
40        conn.execute_batch("PRAGMA journal_mode = WAL;")?;
41
42        let db = Self {
43            conn: Arc::new(Mutex::new(conn)),
44            path: path.clone(),
45        };
46
47        // Run migrations
48        db.migrate()?;
49
50        Ok(db)
51    }
52
53    /// Open an in-memory database (for testing)
54    #[cfg(test)]
55    pub fn open_in_memory() -> SqliteResult<Self> {
56        let conn = Connection::open_in_memory()?;
57
58        // Enable foreign keys
59        conn.execute_batch("PRAGMA foreign_keys = ON;")?;
60
61        let db = Self {
62            conn: Arc::new(Mutex::new(conn)),
63            path: PathBuf::from(":memory:"),
64        };
65
66        // Run migrations
67        db.migrate()?;
68
69        Ok(db)
70    }
71
72    /// Get connection (for testing)
73    #[cfg(test)]
74    pub fn connection(&self) -> Arc<Mutex<Connection>> {
75        Arc::clone(&self.conn)
76    }
77
78    /// Run all pending migrations
79    pub fn migrate(&self) -> SqliteResult<()> {
80        info!("Starting database migrations...");
81        let conn = self.conn.lock_safe();
82
83        // Create migrations table if it doesn't exist
84        debug!("Creating _migrations table if it doesn't exist...");
85        match conn.execute(
86            "CREATE TABLE IF NOT EXISTS _migrations (
87                id INTEGER PRIMARY KEY,
88                name TEXT NOT NULL UNIQUE,
89                applied_at TEXT DEFAULT CURRENT_TIMESTAMP
90            )",
91            [],
92        ) {
93            Ok(_) => debug!("_migrations table ready"),
94            Err(e) => {
95                error!("Failed to create _migrations table: {}", e);
96                return Err(e);
97            }
98        }
99
100        // Get applied migrations
101        debug!("Querying applied migrations...");
102        let mut stmt = conn.prepare("SELECT name FROM _migrations")?;
103        let applied: Vec<String> = stmt
104            .query_map([], |row: &rusqlite::Row| row.get(0))?
105            .filter_map(|r| r.ok())
106            .collect();
107        debug!("Found {} applied migrations", applied.len());
108
109        // Apply pending migrations
110        for (name, sql) in MIGRATIONS {
111            if !applied.contains(&name.to_string()) {
112                info!("Applying migration: {}", name);
113                match conn.execute_batch(sql) {
114                    Ok(_) => {
115                        info!("Successfully applied migration: {}", name);
116                        match conn.execute("INSERT INTO _migrations (name) VALUES (?1)", [name]) {
117                            Ok(_) => debug!("Recorded migration: {}", name),
118                            Err(e) => {
119                                error!("Failed to record migration {}: {}", name, e);
120                                return Err(e);
121                            }
122                        }
123                    }
124                    Err(e) => {
125                        error!("Failed to apply migration {}: {}", name, e);
126                        return Err(e);
127                    }
128                }
129            } else {
130                debug!("Skipping already applied migration: {}", name);
131            }
132        }
133
134        info!("All migrations completed successfully");
135        Ok(())
136    }
137
138    /// Get a database service for async-safe operations
139    ///
140    /// This wraps all blocking database operations in spawn_blocking to prevent
141    /// freezing the async runtime.
142    pub fn service(&self) -> RusqliteService {
143        RusqliteService::new(Arc::clone(&self.conn))
144    }
145
146    /// Get the database file path
147    pub fn path(&self) -> &PathBuf {
148        &self.path
149    }
150
151    /// Get database file size in bytes
152    pub fn file_size(&self) -> Option<u64> {
153        std::fs::metadata(&self.path).ok().map(|m| m.len())
154    }
155}
156
157// 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
158#[cfg(test)]
159mod tests {
160    use super::*;
161    use rusqlite::params;
162
163    #[test]
164    fn test_open_in_memory() {
165        let db = Database::open_in_memory().unwrap();
166        assert_eq!(db.path().to_str(), Some(":memory:"));
167    }
168
169    #[test]
170    fn test_migrations_run() {
171        let db = Database::open_in_memory().unwrap();
172        let conn = db.connection();
173        let conn = conn.lock_safe();
174
175        // Check that tables exist
176        let mut stmt = conn
177            .prepare("SELECT name FROM sqlite_master WHERE type='table' AND name='items'")
178            .unwrap();
179        let exists: Option<String> = stmt.query_row([], |row: &rusqlite::Row| row.get(0)).ok();
180        assert!(exists.is_some());
181    }
182
183    #[test]
184    fn test_all_tables_created() {
185        let db = Database::open_in_memory().unwrap();
186        let conn = db.connection();
187        let conn = conn.lock_safe();
188
189        let expected_tables = [
190            "servers",
191            "users",
192            "libraries",
193            "items",
194            "media_streams",
195            "user_data",
196            "downloads",
197            "sync_queue",
198            "thumbnails",
199            "playlists",
200            "playlist_items",
201            "genres",
202        ];
203
204        for table in expected_tables {
205            let exists: Option<String> = conn
206                .query_row(
207                    "SELECT name FROM sqlite_master WHERE type='table' AND name=?1",
208                    [table],
209                    |row: &rusqlite::Row| row.get(0),
210                )
211                .ok();
212            assert!(exists.is_some(), "Table '{}' should exist", table);
213        }
214    }
215
216    #[test]
217    fn test_fts_table_created() {
218        let db = Database::open_in_memory().unwrap();
219        let conn = db.connection();
220        let conn = conn.lock_safe();
221
222        let exists: Option<String> = conn
223            .query_row(
224                "SELECT name FROM sqlite_master WHERE type='table' AND name='items_fts'",
225                [],
226                |row: &rusqlite::Row| row.get(0),
227            )
228            .ok();
229        assert!(exists.is_some(), "FTS table 'items_fts' should exist");
230    }
231
232    #[test]
233    fn test_server_crud() {
234        let db = Database::open_in_memory().unwrap();
235        let conn = db.connection();
236        let conn = conn.lock_safe();
237
238        // Insert a server
239        conn.execute(
240            "INSERT INTO servers (id, name, url, version) VALUES (?1, ?2, ?3, ?4)",
241            params!["server1", "My Server", "http://localhost:8096", "10.8.0"],
242        )
243        .unwrap();
244
245        // Read it back
246        let (name, url): (String, String) = conn
247            .query_row(
248                "SELECT name, url FROM servers WHERE id = ?1",
249                ["server1"],
250                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
251            )
252            .unwrap();
253        assert_eq!(name, "My Server");
254        assert_eq!(url, "http://localhost:8096");
255
256        // Update it
257        conn.execute(
258            "UPDATE servers SET name = ?1 WHERE id = ?2",
259            params!["Updated Server", "server1"],
260        )
261        .unwrap();
262
263        let name: String = conn
264            .query_row(
265                "SELECT name FROM servers WHERE id = ?1",
266                ["server1"],
267                |row: &rusqlite::Row| row.get(0),
268            )
269            .unwrap();
270        assert_eq!(name, "Updated Server");
271
272        // Delete it
273        conn.execute("DELETE FROM servers WHERE id = ?1", ["server1"])
274            .unwrap();
275
276        let count: i32 = conn
277            .query_row("SELECT COUNT(*) FROM servers", [], |row: &rusqlite::Row| {
278                row.get(0)
279            })
280            .unwrap();
281        assert_eq!(count, 0);
282    }
283
284    #[test]
285    fn test_user_crud() {
286        let db = Database::open_in_memory().unwrap();
287        let conn = db.connection();
288        let conn = conn.lock_safe();
289
290        // Create a server first (foreign key)
291        conn.execute(
292            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
293            params!["server1", "Test Server", "http://localhost:8096"],
294        )
295        .unwrap();
296
297        // Insert a user
298        conn.execute(
299            "INSERT INTO users (id, server_id, username, is_active)
300             VALUES (?1, ?2, ?3, ?4)",
301            params!["user1", "server1", "admin", 1],
302        )
303        .unwrap();
304
305        // Read it back
306        let (username, is_active): (String, i32) = conn
307            .query_row(
308                "SELECT username, is_active FROM users WHERE id = ?1",
309                ["user1"],
310                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
311            )
312            .unwrap();
313        assert_eq!(username, "admin");
314        assert_eq!(is_active, 1);
315
316        // Update is_active
317        conn.execute("UPDATE users SET is_active = 0 WHERE id = ?1", ["user1"])
318            .unwrap();
319
320        let is_active: i32 = conn
321            .query_row(
322                "SELECT is_active FROM users WHERE id = ?1",
323                ["user1"],
324                |row: &rusqlite::Row| row.get(0),
325            )
326            .unwrap();
327        assert_eq!(is_active, 0);
328    }
329
330    #[test]
331    fn test_cascade_delete_server_removes_users() {
332        let db = Database::open_in_memory().unwrap();
333        let conn = db.connection();
334        let conn = conn.lock_safe();
335
336        // Create server and user
337        conn.execute(
338            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
339            params!["server1", "Test Server", "http://localhost:8096"],
340        )
341        .unwrap();
342
343        conn.execute(
344            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
345            params!["user1", "server1", "admin"],
346        )
347        .unwrap();
348
349        // Verify user exists
350        let count: i32 = conn
351            .query_row(
352                "SELECT COUNT(*) FROM users WHERE server_id = ?1",
353                ["server1"],
354                |row: &rusqlite::Row| row.get(0),
355            )
356            .unwrap();
357        assert_eq!(count, 1);
358
359        // Delete server
360        conn.execute("DELETE FROM servers WHERE id = ?1", ["server1"])
361            .unwrap();
362
363        // User should be deleted via CASCADE
364        let count: i32 = conn
365            .query_row("SELECT COUNT(*) FROM users", [], |row: &rusqlite::Row| {
366                row.get(0)
367            })
368            .unwrap();
369        assert_eq!(count, 0);
370    }
371
372    #[test]
373    fn test_item_insert_and_fts_search() {
374        let db = Database::open_in_memory().unwrap();
375        let conn = db.connection();
376        let conn = conn.lock_safe();
377
378        // Create server first
379        conn.execute(
380            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
381            params!["server1", "Test Server", "http://localhost:8096"],
382        )
383        .unwrap();
384
385        // Insert an item
386        conn.execute(
387            "INSERT INTO items (id, server_id, name, item_type, overview, album_name, artists)
388             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
389            params![
390                "item1",
391                "server1",
392                "Bohemian Rhapsody",
393                "Audio",
394                "A legendary rock song",
395                "A Night at the Opera",
396                "[\"Queen\"]"
397            ],
398        )
399        .unwrap();
400
401        // Search via FTS
402        let mut stmt = conn
403            .prepare(
404                "SELECT i.name FROM items i
405                 JOIN items_fts ON i.rowid = items_fts.rowid
406                 WHERE items_fts MATCH ?1",
407            )
408            .unwrap();
409
410        // Search by song name
411        let result: Option<String> = stmt
412            .query_row(["Bohemian"], |row: &rusqlite::Row| row.get(0))
413            .ok();
414        assert_eq!(result, Some("Bohemian Rhapsody".to_string()));
415
416        // Search by album name
417        let result: Option<String> = stmt
418            .query_row(["Opera"], |row: &rusqlite::Row| row.get(0))
419            .ok();
420        assert_eq!(result, Some("Bohemian Rhapsody".to_string()));
421
422        // Search by artist
423        let result: Option<String> = stmt
424            .query_row(["Queen"], |row: &rusqlite::Row| row.get(0))
425            .ok();
426        assert_eq!(result, Some("Bohemian Rhapsody".to_string()));
427    }
428
429    #[test]
430    fn test_user_data_playback_position() {
431        let db = Database::open_in_memory().unwrap();
432        let conn = db.connection();
433        let conn = conn.lock_safe();
434
435        // Setup: server, user, item
436        conn.execute(
437            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
438            params!["server1", "Test", "http://localhost"],
439        )
440        .unwrap();
441
442        conn.execute(
443            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
444            params!["user1", "server1", "admin"],
445        )
446        .unwrap();
447
448        conn.execute(
449            "INSERT INTO items (id, server_id, name, item_type) VALUES (?1, ?2, ?3, ?4)",
450            params!["item1", "server1", "Test Movie", "Movie"],
451        )
452        .unwrap();
453
454        // Insert user data with playback position
455        conn.execute(
456            "INSERT INTO user_data (user_id, item_id, playback_position_ticks, is_played)
457             VALUES (?1, ?2, ?3, ?4)",
458            params!["user1", "item1", 12345678900_i64, 0],
459        )
460        .unwrap();
461
462        // Read back
463        let (position, is_played): (i64, i32) = conn
464            .query_row(
465                "SELECT playback_position_ticks, is_played FROM user_data
466                 WHERE user_id = ?1 AND item_id = ?2",
467                ["user1", "item1"],
468                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
469            )
470            .unwrap();
471        assert_eq!(position, 12345678900);
472        assert_eq!(is_played, 0);
473
474        // Update to mark as played
475        conn.execute(
476            "UPDATE user_data SET is_played = 1, playback_position_ticks = 0
477             WHERE user_id = ?1 AND item_id = ?2",
478            ["user1", "item1"],
479        )
480        .unwrap();
481
482        let is_played: i32 = conn
483            .query_row(
484                "SELECT is_played FROM user_data WHERE user_id = ?1 AND item_id = ?2",
485                ["user1", "item1"],
486                |row: &rusqlite::Row| row.get(0),
487            )
488            .unwrap();
489        assert_eq!(is_played, 1);
490    }
491
492    #[test]
493    fn test_sync_queue_operations() {
494        let db = Database::open_in_memory().unwrap();
495        let conn = db.connection();
496        let conn = conn.lock_safe();
497
498        // Setup
499        conn.execute(
500            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
501            params!["server1", "Test", "http://localhost"],
502        )
503        .unwrap();
504
505        conn.execute(
506            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
507            params!["user1", "server1", "admin"],
508        )
509        .unwrap();
510
511        // Queue a sync operation
512        conn.execute(
513            "INSERT INTO sync_queue (user_id, operation, item_id, payload, status)
514             VALUES (?1, ?2, ?3, ?4, ?5)",
515            params![
516                "user1",
517                "mark_favorite",
518                "item123",
519                r#"{"favorite": true}"#,
520                "pending"
521            ],
522        )
523        .unwrap();
524
525        // Get pending operations
526        let mut stmt = conn
527            .prepare("SELECT operation, item_id FROM sync_queue WHERE status = 'pending'")
528            .unwrap();
529
530        let ops: Vec<(String, String)> = stmt
531            .query_map([], |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)))
532            .unwrap()
533            .filter_map(|r| r.ok())
534            .collect();
535
536        assert_eq!(ops.len(), 1);
537        assert_eq!(ops[0].0, "mark_favorite");
538        assert_eq!(ops[0].1, "item123");
539
540        // Mark as completed
541        conn.execute(
542            "UPDATE sync_queue SET status = 'completed' WHERE item_id = ?1",
543            ["item123"],
544        )
545        .unwrap();
546
547        let pending_count: i32 = conn
548            .query_row(
549                "SELECT COUNT(*) FROM sync_queue WHERE status = 'pending'",
550                [],
551                |row: &rusqlite::Row| row.get(0),
552            )
553            .unwrap();
554        assert_eq!(pending_count, 0);
555    }
556
557    #[test]
558    fn test_downloads_table() {
559        let db = Database::open_in_memory().unwrap();
560        let conn = db.connection();
561        let conn = conn.lock_safe();
562
563        // Setup
564        conn.execute(
565            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
566            params!["server1", "Test", "http://localhost"],
567        )
568        .unwrap();
569
570        conn.execute(
571            "INSERT INTO users (id, server_id, username) VALUES (?1, ?2, ?3)",
572            params!["user1", "server1", "admin"],
573        )
574        .unwrap();
575
576        conn.execute(
577            "INSERT INTO items (id, server_id, name, item_type) VALUES (?1, ?2, ?3, ?4)",
578            params!["item1", "server1", "Test Song", "Audio"],
579        )
580        .unwrap();
581
582        // Queue a download
583        conn.execute(
584            "INSERT INTO downloads (item_id, user_id, file_path, status, progress)
585             VALUES (?1, ?2, ?3, ?4, ?5)",
586            params!["item1", "user1", "/data/downloads/test.mp3", "pending", 0.0],
587        )
588        .unwrap();
589
590        // Update progress
591        conn.execute(
592            "UPDATE downloads SET status = 'downloading', progress = 0.5
593             WHERE item_id = ?1",
594            ["item1"],
595        )
596        .unwrap();
597
598        let (status, progress): (String, f64) = conn
599            .query_row(
600                "SELECT status, progress FROM downloads WHERE item_id = ?1",
601                ["item1"],
602                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
603            )
604            .unwrap();
605        assert_eq!(status, "downloading");
606        assert!((progress - 0.5).abs() < 0.001);
607
608        // Complete download
609        conn.execute(
610            "UPDATE downloads SET status = 'completed', progress = 1.0
611             WHERE item_id = ?1",
612            ["item1"],
613        )
614        .unwrap();
615
616        let status: String = conn
617            .query_row(
618                "SELECT status FROM downloads WHERE item_id = ?1",
619                ["item1"],
620                |row: &rusqlite::Row| row.get(0),
621            )
622            .unwrap();
623        assert_eq!(status, "completed");
624    }
625
626    #[test]
627    fn test_migrations_idempotent() {
628        let db = Database::open_in_memory().unwrap();
629
630        // Run migrations again - should not fail
631        let result = db.migrate();
632        assert!(result.is_ok());
633
634        // Tables should still exist
635        let conn = db.connection();
636        let conn = conn.lock_safe();
637
638        let count: i32 = conn
639            .query_row(
640                "SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='items'",
641                [],
642                |row: &rusqlite::Row| row.get(0),
643            )
644            .unwrap();
645        assert_eq!(count, 1);
646    }
647
648    #[test]
649    fn test_global_active_user_deactivation() {
650        let db = Database::open_in_memory().unwrap();
651        let conn = db.connection();
652        let conn = conn.lock_safe();
653
654        // Create two servers
655        conn.execute(
656            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
657            params!["server1", "Server 1", "http://server1.com"],
658        )
659        .unwrap();
660
661        conn.execute(
662            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
663            params!["server2", "Server 2", "http://server2.com"],
664        )
665        .unwrap();
666
667        // Create users on different servers
668        conn.execute(
669            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
670             VALUES (?1, ?2, ?3, 1, '2024-01-01 10:00:00')",
671            params!["user1", "server1", "admin"],
672        )
673        .unwrap();
674
675        conn.execute(
676            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
677             VALUES (?1, ?2, ?3, 1, '2024-01-01 11:00:00')",
678            params!["user2", "server2", "admin"],
679        )
680        .unwrap();
681
682        // Initially both users are active (simulating the old bug)
683        let active_count: i32 = conn
684            .query_row(
685                "SELECT COUNT(*) FROM users WHERE is_active = 1",
686                [],
687                |row: &rusqlite::Row| row.get(0),
688            )
689            .unwrap();
690        assert_eq!(active_count, 2);
691
692        // Now simulate setting user1 as active (global deactivation)
693        conn.execute("UPDATE users SET is_active = 0", []).unwrap();
694        conn.execute(
695            "UPDATE users SET is_active = 1, last_login_at = CURRENT_TIMESTAMP WHERE id = ?1",
696            ["user1"],
697        )
698        .unwrap();
699
700        // Only one user should be active now
701        let active_count: i32 = conn
702            .query_row(
703                "SELECT COUNT(*) FROM users WHERE is_active = 1",
704                [],
705                |row: &rusqlite::Row| row.get(0),
706            )
707            .unwrap();
708        assert_eq!(active_count, 1);
709
710        // And it should be user1
711        let active_user: String = conn
712            .query_row(
713                "SELECT id FROM users WHERE is_active = 1",
714                [],
715                |row: &rusqlite::Row| row.get(0),
716            )
717            .unwrap();
718        assert_eq!(active_user, "user1");
719    }
720
721    #[test]
722    fn test_active_session_query_ordering() {
723        let db = Database::open_in_memory().unwrap();
724        let conn = db.connection();
725        let conn = conn.lock_safe();
726
727        // Create server
728        conn.execute(
729            "INSERT INTO servers (id, name, url) VALUES (?1, ?2, ?3)",
730            params!["server1", "Test Server", "http://localhost:8096"],
731        )
732        .unwrap();
733
734        // Create users with different login times
735        conn.execute(
736            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
737             VALUES (?1, ?2, ?3, 0, '2024-01-01 10:00:00')",
738            params!["user1", "server1", "old_user"],
739        )
740        .unwrap();
741
742        conn.execute(
743            "INSERT INTO users (id, server_id, username, is_active, last_login_at)
744             VALUES (?1, ?2, ?3, 1, '2024-01-01 12:00:00')",
745            params!["user2", "server1", "recent_user"],
746        )
747        .unwrap();
748
749        // Query for active user ordered by last_login_at DESC (simulating storage_get_active_session)
750        let (user_id, username): (String, String) = conn
751            .query_row(
752                "SELECT u.id, u.username FROM users u
753                 JOIN servers s ON u.server_id = s.id
754                 WHERE u.is_active = 1
755                 ORDER BY u.last_login_at DESC
756                 LIMIT 1",
757                [],
758                |row: &rusqlite::Row| Ok((row.get(0)?, row.get(1)?)),
759            )
760            .unwrap();
761
762        assert_eq!(user_id, "user2");
763        assert_eq!(username, "recent_user");
764    }
765}
jellytau_lib/storage/mod.rs

jellytau_lib/storage/
mod.rs
