1use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use log::debug;
9
10use super::{types::*, MediaRepository};
11use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};
12
13static INCLUDE_CATALOG_BROWSE: AtomicBool = AtomicBool::new(true);
26
27pub fn set_include_catalog_browse(include: bool) {
32 INCLUDE_CATALOG_BROWSE.store(include, Ordering::Relaxed);
33}
34
35pub fn include_catalog_browse() -> bool {
45 INCLUDE_CATALOG_BROWSE.load(Ordering::Relaxed)
46}
47
48fn build_fts_prefix_query(query: &str) -> Option<String> {
64 let tokens: Vec<String> = query
65 .split_whitespace()
66 .filter(|token| token.chars().any(char::is_alphanumeric))
67 .map(|token| format!("\"{}\"", token.replace('"', "\"\"")))
68 .collect();
69
70 let last = tokens.len().checked_sub(1)?;
71 Some(
72 tokens
73 .iter()
74 .enumerate()
75 .map(|(i, token)| {
76 if i == last {
77 format!("{}*", token)
78 } else {
79 token.clone()
80 }
81 })
82 .collect::<Vec<_>>()
83 .join(" "),
84 )
85}
86
87pub struct OfflineRepository {
88 db_service: Arc<RusqliteService>,
89 server_id: String,
90 user_id: String,
91}
92
93impl OfflineRepository {
94 pub fn new(db_service: Arc<RusqliteService>, server_id: String, user_id: String) -> Self {
95 Self {
96 db_service,
97 server_id,
98 user_id,
99 }
100 }
101
102 fn cached_item_to_media_item(item: CachedItem, user_data: Option<UserData>) -> MediaItem {
104 let artists_vec = item
105 .artists
106 .as_ref()
107 .and_then(|s| serde_json::from_str::<Vec<String>>(s).ok())
108 .unwrap_or_default();
109
110 let kind = crate::domain::kind_from_jellyfin(&item.item_type, item.is_folder);
111
112 MediaItem {
113 id: item.id.clone(),
114 name: item.name,
115 item_type: item.item_type,
116 kind,
117 is_folder: item.is_folder,
118 server_id: item.server_id,
119 parent_id: item.parent_id,
120 library_id: item.library_id,
121 overview: item.overview,
122 genres: item
123 .genres
124 .as_ref()
125 .and_then(|s| serde_json::from_str::<Vec<String>>(s).ok()),
126 runtime_ticks: item.runtime_ticks,
127 duration_ms: item.runtime_ticks.map(crate::domain::ticks_to_ms),
128 production_year: item.production_year,
129 premiere_date: item.premiere_date,
130 community_rating: item.community_rating,
131 official_rating: item.official_rating,
132 primary_image_tag: item.primary_image_tag.clone(),
133 image_id: item.primary_image_tag,
134 backdrop_image_tags: item.backdrop_image_tags,
135 parent_backdrop_image_tags: item.parent_backdrop_image_tags,
136 album_id: item.album_id,
137 album_name: item.album_name,
138 album_artist: item.album_artist,
139 artists: Some(artists_vec),
140 artist_items: None, index_number: item.index_number,
142 series_id: item.series_id,
143 series_name: item.series_name,
144 season_id: item.season_id,
145 season_name: item.season_name,
146 parent_index_number: item.parent_index_number,
147 user_data,
148 media_streams: None, media_sources: None, people: None, }
152 }
153
154 async fn get_user_data(&self, item_id: &str) -> Option<UserData> {
156 let query = Query::with_params(
157 "SELECT playback_position_ticks, is_played, is_favorite, play_count, last_played_at, playback_context_type, playback_context_id
158 FROM user_data WHERE user_id = ? AND item_id = ?",
159 vec![
160 QueryParam::String(self.user_id.clone()),
161 QueryParam::String(item_id.to_string()),
162 ],
163 );
164
165 self.db_service
166 .query_optional(query, |row| {
167 let playback_position_ticks: Option<i64> = row.get(0).ok();
168 Ok(UserData {
169 playback_position_ticks,
170 playback_position_ms: playback_position_ticks.map(crate::domain::ticks_to_ms),
171 is_played: row.get::<_, Option<i32>>(1).ok().flatten().map(|v| v != 0),
172 is_favorite: row.get::<_, Option<i32>>(2).ok().flatten().map(|v| v != 0),
173 play_count: row.get(3).ok(),
174 last_played_date: row.get(4).ok(),
175 playback_context_type: row.get(5).ok(),
176 playback_context_id: row.get(6).ok(),
177 })
178 })
179 .await
180 .ok()
181 .flatten()
182 }
183}
184
185#[derive(Debug)]
187struct CachedItem {
188 id: String,
189 name: String,
190 item_type: String,
191 is_folder: bool,
192 server_id: String,
193 parent_id: Option<String>,
194 library_id: Option<String>,
195 overview: Option<String>,
196 genres: Option<String>,
197 runtime_ticks: Option<i64>,
198 production_year: Option<i32>,
199 premiere_date: Option<String>,
200 community_rating: Option<f64>,
201 official_rating: Option<String>,
202 primary_image_tag: Option<String>,
203 backdrop_image_tags: Option<Vec<String>>,
204 parent_backdrop_image_tags: Option<Vec<String>>,
205 album_id: Option<String>,
206 album_name: Option<String>,
207 album_artist: Option<String>,
208 artists: Option<String>,
209 index_number: Option<i32>,
210 series_id: Option<String>,
211 series_name: Option<String>,
212 season_id: Option<String>,
213 season_name: Option<String>,
214 parent_index_number: Option<i32>,
215}
216
217fn row_to_cached_item(row: &rusqlite::Row) -> rusqlite::Result<CachedItem> {
218 Ok(CachedItem {
219 id: row.get(0)?,
220 name: row.get(1)?,
221 item_type: row.get(2)?,
222 server_id: row.get(3)?,
223 parent_id: row.get(4)?,
224 library_id: row.get(5)?,
225 overview: row.get(6)?,
226 genres: row.get(7)?,
227 runtime_ticks: row.get(8)?,
228 production_year: row.get(9)?,
229 community_rating: row.get(10)?,
230 official_rating: row.get(11)?,
231 primary_image_tag: row.get(12)?,
232 backdrop_image_tags: None, parent_backdrop_image_tags: None, album_id: row.get(13)?,
235 album_name: row.get(14)?,
236 album_artist: row.get(15)?,
237 artists: row.get(16)?,
238 index_number: row.get(17)?,
239 series_id: row.get(18)?,
240 series_name: row.get(19)?,
241 season_id: row.get(20)?,
242 season_name: row.get(21)?,
243 parent_index_number: row.get(22)?,
244 is_folder: row.get::<_, Option<i64>>(23)?.unwrap_or(0) != 0,
246 premiere_date: row.get(24)?,
247 })
248}
249
250impl OfflineRepository {
251 pub async fn prune_stale_catalog(
278 &self,
279 cutoff: &str,
280 item_types: &[String],
281 ) -> Result<usize, RepoError> {
282 if item_types.is_empty() {
283 return Ok(0);
284 }
285 let placeholders = vec!["?"; item_types.len()].join(",");
286 let sql = format!(
287 "DELETE FROM items
288 WHERE server_id = ?
289 AND synced_at IS NOT NULL
290 AND synced_at < ?
291 AND item_type IN ({})
292 AND id NOT IN (
293 -- Playable items with completed downloads
294 SELECT i.id
295 FROM items i
296 INNER JOIN downloads d ON i.id = d.item_id
297 WHERE d.status = 'completed'
298
299 UNION
300
301 -- Containers with downloaded children
302 SELECT i.id
303 FROM items i
304 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
305 INNER JOIN downloads d ON children.id = d.item_id
306 WHERE d.status = 'completed'
307 )",
308 placeholders
309 );
310
311 let mut params = vec![
312 QueryParam::String(self.server_id.clone()),
313 QueryParam::String(cutoff.to_string()),
314 ];
315 params.extend(item_types.iter().cloned().map(QueryParam::String));
316
317 let removed = self
318 .db_service
319 .execute(Query::with_params(sql, params))
320 .await
321 .map_err(|e| RepoError::Database { message: e })?;
322
323 Ok(removed)
324 }
325
326 async fn search_people(
335 &self,
336 fts_query: &str,
337 limit: usize,
338 ) -> Result<Vec<MediaItem>, RepoError> {
339 let sql = format!(
340 "SELECT p.id, p.name, p.overview, p.primary_image_tag, p.premiere_date
341 FROM people p
342 JOIN people_fts fts ON fts.rowid = p.rowid
343 WHERE p.server_id = ? AND people_fts MATCH ?
344 ORDER BY rank
345 LIMIT {}",
346 limit
347 );
348
349 let rows = self
350 .db_service
351 .query_many(
352 Query::with_params(
353 sql,
354 vec![
355 QueryParam::String(self.server_id.clone()),
356 QueryParam::String(fts_query.to_string()),
357 ],
358 ),
359 |row| {
360 Ok((
361 row.get::<_, String>(0)?,
362 row.get::<_, String>(1)?,
363 row.get::<_, Option<String>>(2)?,
364 row.get::<_, Option<String>>(3)?,
365 row.get::<_, Option<String>>(4)?,
366 ))
367 },
368 )
369 .await
370 .map_err(|e| RepoError::Database { message: e })?;
371
372 Ok(rows
373 .into_iter()
374 .map(
375 |(id, name, overview, primary_image_tag, premiere_date)| MediaItem {
376 id,
377 name,
378 item_type: "Person".to_string(),
379 kind: crate::domain::MediaKind::Person,
380 is_folder: false,
381 server_id: self.server_id.clone(),
382 overview,
383 primary_image_tag,
384 premiere_date,
385 ..Default::default()
386 },
387 )
388 .collect())
389 }
390
391 pub async fn save_to_cache(
396 &self,
397 parent_id: &str,
398 items: &[MediaItem],
399 ) -> Result<usize, RepoError> {
400 if items.is_empty() {
401 return Ok(0);
402 }
403
404 let now = chrono::Utc::now().to_rfc3339();
405
406 self.db_service
409 .execute(Query::new("PRAGMA foreign_keys = OFF"))
410 .await
411 .map_err(|e| RepoError::Database { message: e })?;
412
413 let result = self.save_to_cache_impl(parent_id, items, &now).await;
415
416 let _ = self
418 .db_service
419 .execute(Query::new("PRAGMA foreign_keys = ON"))
420 .await;
421
422 result
423 }
424
425 async fn save_to_cache_impl(
426 &self,
427 parent_id: &str,
428 items: &[MediaItem],
429 now: &str,
430 ) -> Result<usize, RepoError> {
431 let mut parent_ids = std::collections::HashSet::new();
433 parent_ids.insert(parent_id.to_string());
434
435 for item in items {
436 if let Some(pid) = &item.parent_id {
437 parent_ids.insert(pid.clone());
438 }
439 }
440
441 #[cfg(test)]
443 println!("Creating stub parents for: {:?}", parent_ids);
444
445 for pid in parent_ids {
446 let parent_query = Query::with_params(
447 "INSERT OR IGNORE INTO items (id, server_id, name, item_type, synced_at)
448 VALUES (?1, ?2, ?3, ?4, ?5)",
449 vec![
450 QueryParam::String(pid.clone()),
451 QueryParam::String(self.server_id.clone()),
452 QueryParam::String("Parent".to_string()),
453 QueryParam::String("Folder".to_string()),
454 QueryParam::String(now.to_string()),
455 ],
456 );
457
458 let _stub_rows = self
459 .db_service
460 .execute(parent_query)
461 .await
462 .map_err(|e| RepoError::Database { message: e })?;
463 #[cfg(test)]
464 println!(
465 " Created stub parent {} (rows affected: {})",
466 pid, _stub_rows
467 );
468 }
469
470 let mut count = 0;
471
472 for item in items {
473 let genres_json = item
475 .genres
476 .as_ref()
477 .map(|g| serde_json::to_string(g).unwrap_or_else(|_| "[]".to_string()));
478 let artists_json = item
479 .artists
480 .as_ref()
481 .map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string()));
482 let backdrop_tags_json = item
483 .backdrop_image_tags
484 .as_ref()
485 .map(|b| serde_json::to_string(b).unwrap_or_else(|_| "[]".to_string()));
486
487 let query = Query::with_params(
503 "INSERT INTO items (
504 id, server_id, library_id, parent_id,
505 name, item_type, is_folder, overview,
506 genres, series_id, series_name,
507 season_id, season_name, index_number, parent_index_number,
508 album_id, album_name, album_artist, artists,
509 production_year, premiere_date, runtime_ticks,
510 primary_image_tag, backdrop_image_tags,
511 community_rating, official_rating,
512 synced_at
513 ) VALUES (
514 ?1, ?2, ?3, ?4,
515 ?5, ?6, ?7, ?8,
516 ?9, ?10, ?11,
517 ?12, ?13, ?14, ?15,
518 ?16, ?17, ?18, ?19,
519 ?20, ?21, ?22,
520 ?23, ?24,
521 ?25, ?26,
522 ?27
523 )
524 ON CONFLICT(id) DO UPDATE SET
525 server_id = excluded.server_id,
526 -- This call site never supplies library_id (it is always
527 -- bound NULL), so keep whatever another path recorded rather
528 -- than clearing it the way REPLACE did.
529 library_id = COALESCE(excluded.library_id, items.library_id),
530 parent_id = excluded.parent_id,
531 name = excluded.name,
532 item_type = excluded.item_type,
533 is_folder = excluded.is_folder,
534 overview = excluded.overview,
535 genres = excluded.genres,
536 series_id = excluded.series_id,
537 series_name = excluded.series_name,
538 season_id = excluded.season_id,
539 season_name = excluded.season_name,
540 index_number = excluded.index_number,
541 parent_index_number = excluded.parent_index_number,
542 album_id = excluded.album_id,
543 album_name = excluded.album_name,
544 album_artist = excluded.album_artist,
545 artists = excluded.artists,
546 production_year = excluded.production_year,
547 premiere_date = excluded.premiere_date,
548 runtime_ticks = excluded.runtime_ticks,
549 primary_image_tag = excluded.primary_image_tag,
550 backdrop_image_tags = excluded.backdrop_image_tags,
551 community_rating = excluded.community_rating,
552 official_rating = excluded.official_rating,
553 synced_at = excluded.synced_at",
554 vec![
555 QueryParam::String(item.id.clone()),
556 QueryParam::String(self.server_id.clone()),
557 QueryParam::Null, match &item.parent_id {
561 Some(pid) => QueryParam::String(pid.clone()),
562 None => QueryParam::Null,
563 },
564 QueryParam::String(item.name.clone()),
565 QueryParam::String(item.item_type.clone()),
566 QueryParam::Int(if item.is_folder { 1 } else { 0 }),
567 match &item.overview {
568 Some(o) => QueryParam::String(o.clone()),
569 None => QueryParam::Null,
570 },
571 match genres_json {
572 Some(g) => QueryParam::String(g),
573 None => QueryParam::Null,
574 },
575 match &item.series_id {
576 Some(s) => QueryParam::String(s.clone()),
577 None => QueryParam::Null,
578 },
579 match &item.series_name {
580 Some(s) => QueryParam::String(s.clone()),
581 None => QueryParam::Null,
582 },
583 match &item.season_id {
584 Some(s) => QueryParam::String(s.clone()),
585 None => QueryParam::Null,
586 },
587 match &item.season_name {
588 Some(s) => QueryParam::String(s.clone()),
589 None => QueryParam::Null,
590 },
591 match item.index_number {
592 Some(i) => QueryParam::Int(i),
593 None => QueryParam::Null,
594 },
595 match item.parent_index_number {
596 Some(i) => QueryParam::Int(i),
597 None => QueryParam::Null,
598 },
599 match &item.album_id {
600 Some(a) => QueryParam::String(a.clone()),
601 None => QueryParam::Null,
602 },
603 match &item.album_name {
604 Some(a) => QueryParam::String(a.clone()),
605 None => QueryParam::Null,
606 },
607 match &item.album_artist {
608 Some(a) => QueryParam::String(a.clone()),
609 None => QueryParam::Null,
610 },
611 match artists_json {
612 Some(a) => QueryParam::String(a),
613 None => QueryParam::Null,
614 },
615 match item.production_year {
616 Some(y) => QueryParam::Int(y),
617 None => QueryParam::Null,
618 },
619 match &item.premiere_date {
620 Some(d) => QueryParam::String(d.clone()),
621 None => QueryParam::Null,
622 },
623 match item.runtime_ticks {
624 Some(r) => QueryParam::Int64(r),
625 None => QueryParam::Null,
626 },
627 match &item.primary_image_tag {
628 Some(t) => QueryParam::String(t.clone()),
629 None => QueryParam::Null,
630 },
631 match backdrop_tags_json {
632 Some(b) => QueryParam::String(b),
633 None => QueryParam::Null,
634 },
635 match item.community_rating {
636 Some(r) => QueryParam::Float(r),
637 None => QueryParam::Null,
638 },
639 match &item.official_rating {
640 Some(r) => QueryParam::String(r.clone()),
641 None => QueryParam::Null,
642 },
643 QueryParam::String(now.to_string()),
644 ],
645 );
646
647 let _rows_affected =
648 self.db_service
649 .execute(query)
650 .await
651 .map_err(|e| RepoError::Database {
652 message: format!("Failed to insert item {}: {}", item.id, e),
653 })?;
654 #[cfg(test)]
655 println!(
656 " [save_to_cache] Saved item {} (rows affected: {})",
657 item.id, _rows_affected
658 );
659
660 self.mirror_user_data(item, now).await?;
661 count += 1;
662 }
663
664 Ok(count)
665 }
666
667 async fn mirror_user_data(&self, item: &MediaItem, now: &str) -> Result<(), RepoError> {
690 let user_data = item.user_data.as_ref();
691 let is_favorite = user_data.and_then(|ud| ud.is_favorite);
692 let position_ticks = user_data.and_then(|ud| ud.playback_position_ticks);
693
694 if is_favorite.is_none() && position_ticks.is_none() {
696 return Ok(());
697 }
698
699 let query = Query::with_params(
700 "INSERT INTO user_data
701 (user_id, item_id, is_favorite, playback_position_ticks, synced_at, pending_sync)
702 VALUES (?1, ?2, ?3, ?4, ?5, 0)
703 ON CONFLICT(user_id, item_id) DO UPDATE SET
704 is_favorite = COALESCE(excluded.is_favorite, user_data.is_favorite),
705 playback_position_ticks = COALESCE(
706 excluded.playback_position_ticks, user_data.playback_position_ticks),
707 synced_at = excluded.synced_at
708 WHERE user_data.pending_sync = 0",
709 vec![
710 QueryParam::String(self.user_id.clone()),
711 QueryParam::String(item.id.clone()),
712 is_favorite
713 .map(|f| QueryParam::Int(if f { 1 } else { 0 }))
714 .unwrap_or(QueryParam::Null),
715 position_ticks
716 .map(QueryParam::Int64)
717 .unwrap_or(QueryParam::Null),
718 QueryParam::String(now.to_string()),
719 ],
720 );
721
722 if let Err(e) = self.db_service.execute(query).await {
726 debug!(
727 "[OfflineRepo] user_data mirror skipped for {}: {}",
728 item.id, e
729 );
730 }
731
732 Ok(())
733 }
734
735 pub async fn save_libraries_to_cache(&self, libraries: &[Library]) -> Result<usize, RepoError> {
740 if libraries.is_empty() {
741 return Ok(0);
742 }
743
744 let mut count = 0;
745 for (idx, lib) in libraries.iter().enumerate() {
746 let query = Query::with_params(
747 "INSERT OR REPLACE INTO libraries (id, server_id, name, collection_type, image_tag, sort_order, synced_at)
748 VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)",
749 vec![
750 QueryParam::String(lib.id.clone()),
751 QueryParam::String(self.server_id.clone()),
752 QueryParam::String(lib.name.clone()),
753 QueryParam::String(lib.collection_type.clone()),
754 lib.image_tag.clone().map(QueryParam::String).unwrap_or(QueryParam::Null),
755 QueryParam::Int(idx as i32),
756 ],
757 );
758 self.db_service
759 .execute(query)
760 .await
761 .map_err(|e| RepoError::Database { message: e })?;
762 count += 1;
763 }
764 Ok(count)
765 }
766
767 pub async fn save_genres_to_cache(
772 &self,
773 parent_id: Option<&str>,
774 genres: &[Genre],
775 ) -> Result<usize, RepoError> {
776 if genres.is_empty() {
777 return Ok(0);
778 }
779
780 let library_id = parent_id.unwrap_or("").to_string();
783 let server_id = self.server_id.clone();
784 let genres: Vec<(String, String, Option<u32>)> = genres
785 .iter()
786 .map(|g| (g.id.clone(), g.name.clone(), g.album_count))
787 .collect();
788 let saved = genres.len();
789
790 self.db_service
791 .transaction(move |tx| {
792 use crate::storage::db_service::{Query, QueryParam};
793
794 tx.execute(Query::with_params(
796 "DELETE FROM genres WHERE server_id = ? AND library_id = ?",
797 vec![
798 QueryParam::String(server_id.clone()),
799 QueryParam::String(library_id.clone()),
800 ],
801 ))?;
802
803 for (id, name, album_count) in &genres {
804 tx.execute(Query::with_params(
805 "INSERT OR REPLACE INTO genres (id, server_id, library_id, name, album_count, synced_at)
806 VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)",
807 vec![
808 QueryParam::String(id.clone()),
809 QueryParam::String(server_id.clone()),
810 QueryParam::String(library_id.clone()),
811 QueryParam::String(name.clone()),
812 album_count.map(|c| QueryParam::Int(c as i32)).unwrap_or(QueryParam::Null),
813 ],
814 ))?;
815 }
816
817 Ok(())
818 })
819 .await
820 .map_err(|e| RepoError::Database { message: e })?;
821
822 Ok(saved)
823 }
824
825 const LIBRARY_HOLDS_ITEM: &'static str = "(
850 (l.collection_type = 'music' AND i.item_type IN ('MusicAlbum', 'MusicArtist', 'Audio'))
851 OR (l.collection_type = 'movies' AND i.item_type = 'Movie')
852 OR (l.collection_type = 'tvshows' AND i.item_type IN ('Series', 'Season', 'Episode'))
853 OR l.collection_type IS NULL
854 OR l.collection_type NOT IN ('music', 'movies', 'tvshows')
855 )";
856
857 const DOWNLOADED_ITEMS_CTE: &'static str = "
859 WITH downloaded_items AS (
860 SELECT DISTINCT i.id
861 FROM items i
862 INNER JOIN downloads d ON i.id = d.item_id
863 WHERE d.status = 'completed'
864 AND i.item_type IN ('Audio', 'Movie', 'Episode')
865
866 UNION
867
868 SELECT DISTINCT i.id
869 FROM items i
870 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
871 INNER JOIN downloads d ON children.id = d.item_id
872 WHERE d.status = 'completed'
873 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
874 )";
875
876 pub async fn get_downloaded_items(
886 &self,
887 parent_id: &str,
888 options: Option<GetItemsOptions>,
889 ) -> Result<SearchResult, RepoError> {
890 let opts = options.unwrap_or_default();
891 let limit = opts.limit.unwrap_or(10000);
892 let start_index = opts.start_index.unwrap_or(0);
893
894 let type_filter = if let Some(include_item_types) = &opts.include_item_types {
895 if !include_item_types.is_empty() {
896 let types = include_item_types
897 .iter()
898 .map(|t| format!("'{}'", t.replace('\'', "''")))
899 .collect::<Vec<_>>()
900 .join(",");
901 format!(" AND i.item_type IN ({})", types)
902 } else {
903 String::new()
904 }
905 } else {
906 String::new()
907 };
908
909 let sql = format!(
922 "{cte}
923 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
924 i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
925 i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
926 i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
927 i.parent_index_number, i.is_folder, i.premiere_date
928 FROM items i
929 INNER JOIN downloaded_items di ON i.id = di.id
930 WHERE i.server_id = ?
931 AND (
932 i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ?
933 OR (
934 EXISTS (
935 SELECT 1 FROM libraries l
936 WHERE l.id = ? AND l.server_id = i.server_id
937 AND {membership}
938 )
939 -- Top-level only: hide leaves whose container is downloaded.
940 AND NOT EXISTS (
941 SELECT 1 FROM downloaded_items parent
942 WHERE parent.id = i.album_id
943 OR parent.id = i.season_id
944 OR parent.id = i.series_id
945 OR parent.id = i.parent_id
946 )
947 )
948 ){type_filter}
949 ORDER BY i.sort_name ASC, i.name ASC
950 LIMIT {limit} OFFSET {start_index}",
951 cte = Self::DOWNLOADED_ITEMS_CTE,
952 membership = Self::LIBRARY_HOLDS_ITEM,
953 );
954
955 let query = Query::with_params(
956 sql,
957 vec![
958 QueryParam::String(self.server_id.clone()),
959 QueryParam::String(parent_id.to_string()),
960 QueryParam::String(parent_id.to_string()),
961 QueryParam::String(parent_id.to_string()),
962 QueryParam::String(parent_id.to_string()),
963 QueryParam::String(parent_id.to_string()),
964 ],
965 );
966
967 let cached_items: Vec<CachedItem> = self
968 .db_service
969 .query_many(query, row_to_cached_item)
970 .await
971 .map_err(|e| RepoError::Database { message: e })?;
972
973 let mut items = Vec::new();
974 for cached in cached_items {
975 let user_data = self.get_user_data(&cached.id).await;
976 items.push(Self::cached_item_to_media_item(cached, user_data));
977 }
978
979 let total_record_count = items.len();
980 Ok(SearchResult {
981 items,
982 total_record_count,
983 })
984 }
985
986 pub async fn get_downloaded_libraries(&self) -> Result<Vec<Library>, RepoError> {
992 let query = Query::with_params(
997 format!(
998 "{cte}
999 SELECT l.id, l.name, l.collection_type, l.image_tag
1000 FROM libraries l
1001 WHERE l.server_id = ?
1002 AND EXISTS (
1003 SELECT 1 FROM items i
1004 INNER JOIN downloaded_items di ON i.id = di.id
1005 WHERE i.server_id = l.server_id
1006 AND {membership}
1007 )
1008 ORDER BY l.sort_order ASC, l.name ASC",
1009 cte = Self::DOWNLOADED_ITEMS_CTE,
1010 membership = Self::LIBRARY_HOLDS_ITEM,
1011 ),
1012 vec![QueryParam::String(self.server_id.clone())],
1013 );
1014
1015 self.db_service
1016 .query_many(query, |row| {
1017 Ok(Library::new(
1018 row.get(0)?,
1019 row.get(1)?,
1020 row.get::<_, Option<String>>(2)?
1021 .unwrap_or_else(|| "unknown".to_string()),
1022 row.get(3)?,
1023 ))
1024 })
1025 .await
1026 .map_err(|e| RepoError::Database { message: e })
1027 }
1028
1029 pub async fn get_download_disk_usage(&self) -> Result<DownloadDiskUsage, RepoError> {
1038 let leaf_query = Query::with_params(
1040 "SELECT d.item_id, COALESCE(d.file_size, 0)
1041 FROM downloads d
1042 INNER JOIN items i ON i.id = d.item_id
1043 WHERE d.status = 'completed'
1044 AND i.server_id = ?
1045 AND i.item_type IN ('Audio', 'Movie', 'Episode')",
1046 vec![QueryParam::String(self.server_id.clone())],
1047 );
1048 let leaves: Vec<(String, i64)> = self
1049 .db_service
1050 .query_many(leaf_query, |row| Ok((row.get(0)?, row.get(1)?)))
1051 .await
1052 .map_err(|e| RepoError::Database { message: e })?;
1053
1054 let container_query = Query::with_params(
1056 "SELECT c.id, COALESCE(SUM(d.file_size), 0)
1057 FROM items c
1058 INNER JOIN items children
1059 ON (children.parent_id = c.id OR children.album_id = c.id OR children.season_id = c.id OR children.series_id = c.id)
1060 INNER JOIN downloads d ON children.id = d.item_id
1061 WHERE d.status = 'completed'
1062 AND c.server_id = ?
1063 AND c.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
1064 GROUP BY c.id",
1065 vec![QueryParam::String(self.server_id.clone())],
1066 );
1067 let containers: Vec<(String, i64)> = self
1068 .db_service
1069 .query_many(container_query, |row| Ok((row.get(0)?, row.get(1)?)))
1070 .await
1071 .map_err(|e| RepoError::Database { message: e })?;
1072
1073 let partial_query = Query::with_params(
1084 "WITH downloaded_containers AS (
1085 SELECT DISTINCT c.id
1086 FROM items c
1087 INNER JOIN items children
1088 ON (children.parent_id = c.id OR children.album_id = c.id OR children.season_id = c.id OR children.series_id = c.id)
1089 INNER JOIN downloads d ON children.id = d.item_id
1090 WHERE d.status = 'completed'
1091 AND c.server_id = ?
1092 AND c.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
1093 )
1094 SELECT c.id,
1095 COUNT(children.id) AS total_children,
1096 SUM(CASE WHEN d.status = 'completed' THEN 1 ELSE 0 END) AS downloaded_children
1097 FROM items c
1098 INNER JOIN downloaded_containers dc ON dc.id = c.id
1099 INNER JOIN items children
1100 ON (children.parent_id = c.id OR children.album_id = c.id OR children.season_id = c.id OR children.series_id = c.id)
1101 LEFT JOIN downloads d ON children.id = d.item_id AND d.status = 'completed'
1102 WHERE children.item_type IN ('Audio', 'Movie', 'Episode', 'Season')
1103 GROUP BY c.id",
1104 vec![QueryParam::String(self.server_id.clone())],
1105 );
1106 let partial_rows: Vec<(String, i64, i64)> = self
1107 .db_service
1108 .query_many(partial_query, |row| {
1109 Ok((
1110 row.get(0)?,
1111 row.get(1)?,
1112 row.get::<_, Option<i64>>(2)?.unwrap_or(0),
1113 ))
1114 })
1115 .await
1116 .map_err(|e| RepoError::Database { message: e })?;
1117
1118 let mut partial_containers = std::collections::HashMap::new();
1119 for (id, total, downloaded) in partial_rows {
1120 if downloaded > 0 && downloaded < total {
1123 partial_containers.insert(id, true);
1124 }
1125 }
1126
1127 let item_count = leaves.len() as u32;
1128 let device_total_bytes: i64 = leaves.iter().map(|(_, b)| *b).sum();
1129
1130 let mut sizes = std::collections::HashMap::new();
1131 for (id, bytes) in leaves.into_iter().chain(containers) {
1132 *sizes.entry(id).or_insert(0) += bytes;
1135 }
1136
1137 Ok(DownloadDiskUsage {
1138 sizes,
1139 partial_containers,
1140 device_total_bytes,
1141 item_count,
1142 })
1143 }
1144
1145 pub async fn save_playlist_items_to_cache(
1148 &self,
1149 playlist_id: &str,
1150 entries: &[PlaylistEntry],
1151 ) -> Result<(), RepoError> {
1152 let playlist_id = playlist_id.to_string();
1153 let user_id = self.user_id.clone();
1154 let entries: Vec<(String, String, usize)> = entries
1155 .iter()
1156 .enumerate()
1157 .map(|(i, e)| (e.playlist_item_id.clone(), e.item.id.clone(), i))
1158 .collect();
1159
1160 self.db_service
1161 .transaction(move |tx| {
1162 use crate::storage::db_service::{Query, QueryParam};
1163
1164 tx.execute(Query::with_params(
1166 "INSERT OR IGNORE INTO playlists (id, user_id, name, is_local) VALUES (?1, ?2, '', 0)",
1167 vec![QueryParam::String(playlist_id.clone()), QueryParam::String(user_id)],
1168 ))?;
1169
1170 tx.execute(Query::with_params(
1172 "DELETE FROM playlist_items WHERE playlist_id = ?",
1173 vec![QueryParam::String(playlist_id.clone())],
1174 ))?;
1175
1176 for (_, item_id, sort_order) in &entries {
1177 tx.execute(Query::with_params(
1178 "INSERT OR IGNORE INTO playlist_items (playlist_id, item_id, sort_order) VALUES (?1, ?2, ?3)",
1179 vec![
1180 QueryParam::String(playlist_id.clone()),
1181 QueryParam::String(item_id.clone()),
1182 QueryParam::Int(*sort_order as i32),
1183 ],
1184 ))?;
1185 }
1186
1187 Ok(())
1188 })
1189 .await
1190 .map_err(|e| RepoError::Database {
1191 message: format!("Failed to cache playlist items: {}", e),
1192 })
1193 }
1194}
1195
1196#[async_trait]
1197impl MediaRepository for OfflineRepository {
1198 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
1199 let query = Query::with_params(
1207 "SELECT l.id, l.name, l.collection_type, l.image_tag
1208 FROM libraries l
1209 WHERE l.server_id = ?
1210 ORDER BY l.sort_order ASC, l.name ASC",
1211 vec![QueryParam::String(self.server_id.clone())],
1212 );
1213
1214 self.db_service
1215 .query_many(query, |row| {
1216 Ok(Library::new(
1217 row.get(0)?,
1218 row.get(1)?,
1219 row.get::<_, Option<String>>(2)?
1220 .unwrap_or_else(|| "unknown".to_string()),
1221 row.get(3)?,
1222 ))
1223 })
1224 .await
1225 .map_err(|e| RepoError::Database { message: e })
1226 }
1227
1228 async fn get_items(
1229 &self,
1230 parent_id: &str,
1231 options: Option<GetItemsOptions>,
1232 ) -> Result<SearchResult, RepoError> {
1233 debug!(
1234 "[OfflineRepo] get_items called for parent_id: {}",
1235 &parent_id[..8.min(parent_id.len())]
1236 );
1237 let opts = options.unwrap_or_default();
1238 let limit = opts.limit.unwrap_or(10000); let start_index = opts.start_index.unwrap_or(0);
1240
1241 let order_by = match opts.sort_by.as_deref() {
1244 Some("Random") => "RANDOM()",
1245 _ => "i.sort_name ASC, i.name ASC",
1246 };
1247
1248 let type_values: &[String] = opts
1255 .include_item_types
1256 .as_deref()
1257 .filter(|types| !types.is_empty())
1258 .unwrap_or(&[]);
1259 let type_filter = if type_values.is_empty() {
1260 String::new()
1261 } else {
1262 let placeholders = vec!["?"; type_values.len()].join(",");
1263 format!(" AND i.item_type IN ({})", placeholders)
1264 };
1265
1266 let favorites_filter = if opts.favorites_only == Some(true) {
1271 " AND EXISTS (
1272 SELECT 1 FROM user_data ud
1273 WHERE ud.item_id = i.id AND ud.user_id = ? AND ud.is_favorite = 1
1274 )"
1275 } else {
1276 ""
1277 };
1278
1279 let catalog_branch = if include_catalog_browse() {
1287 "UNION
1288
1289 -- Cached items for fast browsing (online) or the offline catalog view
1290 SELECT DISTINCT i.id
1291 FROM items i
1292 WHERE i.synced_at IS NOT NULL"
1293 } else {
1294 ""
1295 };
1296 let sql = format!(
1297 "WITH available_items AS (
1298 -- Playable items with completed downloads
1299 SELECT DISTINCT i.id
1300 FROM items i
1301 INNER JOIN downloads d ON i.id = d.item_id
1302 WHERE d.status = 'completed'
1303 AND i.item_type IN ('Audio', 'Movie', 'Episode')
1304
1305 UNION
1306
1307 -- Containers with downloaded children
1308 SELECT DISTINCT i.id
1309 FROM items i
1310 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
1311 INNER JOIN downloads d ON children.id = d.item_id
1312 WHERE d.status = 'completed'
1313 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
1314
1315 {catalog_branch}
1316 )
1317 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
1318 i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
1319 i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
1320 i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
1321 i.parent_index_number, i.is_folder, i.premiere_date
1322 FROM items i
1323 INNER JOIN available_items ai ON i.id = ai.id
1324 WHERE i.server_id = ?
1325 AND (
1326 i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ?
1327 -- When the requested parent is a LIBRARY, there is no per-item
1328 -- link back to it (library_id/parent_id are NULL in the cache),
1329 -- so match every item on the server and let the type filter
1330 -- (e.g. MusicAlbum / Movie / Series) narrow it. This is what
1331 -- makes library landing pages show albums/movies/shows offline.
1332 OR EXISTS (
1333 SELECT 1 FROM libraries l
1334 WHERE l.id = ? AND l.server_id = i.server_id
1335 )
1336 ){}{}
1337 ORDER BY {}
1338 LIMIT {} OFFSET {}",
1339 type_filter, favorites_filter, order_by, limit, start_index
1340 );
1341
1342 let mut params = vec![
1348 QueryParam::String(self.server_id.clone()),
1349 QueryParam::String(parent_id.to_string()), QueryParam::String(parent_id.to_string()), QueryParam::String(parent_id.to_string()), QueryParam::String(parent_id.to_string()), QueryParam::String(parent_id.to_string()), ];
1355 params.extend(type_values.iter().cloned().map(QueryParam::String));
1360 if !favorites_filter.is_empty() {
1361 params.push(QueryParam::String(self.user_id.clone())); }
1363 let query = Query::with_params(sql, params);
1364
1365 let cached_items: Vec<CachedItem> = self
1366 .db_service
1367 .query_many(query, row_to_cached_item)
1368 .await
1369 .map_err(|e| RepoError::Database { message: e })?;
1370
1371 debug!(
1372 "[OfflineRepo] Found {} cached items for parent {}",
1373 cached_items.len(),
1374 &parent_id[..8.min(parent_id.len())]
1375 );
1376
1377 let mut items = Vec::new();
1379 for cached in cached_items {
1380 let user_data = self.get_user_data(&cached.id).await;
1381 items.push(Self::cached_item_to_media_item(cached, user_data));
1382 }
1383
1384 let total_record_count = items.len();
1385
1386 debug!(
1387 "[OfflineRepo] Returning {} items for parent {}",
1388 total_record_count,
1389 &parent_id[..8.min(parent_id.len())]
1390 );
1391
1392 Ok(SearchResult {
1393 items,
1394 total_record_count,
1395 })
1396 }
1397
1398 async fn get_item(&self, item_id: &str) -> Result<MediaItem, RepoError> {
1399 let query = Query::with_params(
1401 "WITH downloaded_items AS (
1402 -- Playable items with completed downloads
1403 SELECT DISTINCT i.id
1404 FROM items i
1405 INNER JOIN downloads d ON i.id = d.item_id
1406 WHERE d.status = 'completed'
1407 AND i.item_type IN ('Audio', 'Movie', 'Episode')
1408
1409 UNION
1410
1411 -- Containers with downloaded children
1412 SELECT DISTINCT i.id
1413 FROM items i
1414 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
1415 INNER JOIN downloads d ON children.id = d.item_id
1416 WHERE d.status = 'completed'
1417 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
1418 )
1419 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
1420 i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
1421 i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
1422 i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
1423 i.parent_index_number, i.is_folder, i.premiere_date
1424 FROM items i
1425 INNER JOIN downloaded_items di ON i.id = di.id
1426 WHERE i.id = ?",
1427 vec![QueryParam::String(item_id.to_string())],
1428 );
1429
1430 let cached = self
1431 .db_service
1432 .query_optional(query, row_to_cached_item)
1433 .await
1434 .map_err(|e| RepoError::Database { message: e })?
1435 .ok_or_else(|| RepoError::NotFound {
1436 message: format!(
1437 "Item {} not found in offline cache or not downloaded",
1438 item_id
1439 ),
1440 })?;
1441
1442 let user_data = self.get_user_data(item_id).await;
1443 Ok(Self::cached_item_to_media_item(cached, user_data))
1444 }
1445
1446 async fn get_latest_items(
1447 &self,
1448 parent_id: &str,
1449 limit: Option<usize>,
1450 ) -> Result<Vec<MediaItem>, RepoError> {
1451 let limit_val = limit.unwrap_or(16);
1452
1453 let query = Query::with_params(
1454 format!(
1455 "WITH downloaded_items AS (
1456 -- Playable items with completed downloads
1457 SELECT DISTINCT i.id
1458 FROM items i
1459 INNER JOIN downloads d ON i.id = d.item_id
1460 WHERE d.status = 'completed'
1461 AND i.item_type IN ('Audio', 'Movie', 'Episode')
1462
1463 UNION
1464
1465 -- Containers with downloaded children
1466 SELECT DISTINCT i.id
1467 FROM items i
1468 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
1469 INNER JOIN downloads d ON children.id = d.item_id
1470 WHERE d.status = 'completed'
1471 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
1472 )
1473 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
1474 i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
1475 i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
1476 i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
1477 i.parent_index_number, i.is_folder, i.premiere_date
1478 FROM items i
1479 INNER JOIN downloaded_items di ON i.id = di.id
1480 WHERE i.server_id = ? AND i.library_id = ?
1481 -- Collapse leaves into the container that was added: a new
1482 -- 14-track album should read as one album, not 14 songs. Only
1483 -- drops a leaf when its own container is present in the same
1484 -- result, so a standalone track or movie still appears.
1485 AND NOT EXISTS (
1486 SELECT 1 FROM downloaded_items parent
1487 WHERE parent.id IN (i.album_id, i.season_id, i.series_id, i.parent_id)
1488 )
1489 ORDER BY i.synced_at DESC
1490 LIMIT {}", limit_val
1491 ),
1492 vec![
1493 QueryParam::String(self.server_id.clone()),
1494 QueryParam::String(parent_id.to_string()),
1495 ],
1496 );
1497
1498 let cached_items: Vec<CachedItem> = self
1499 .db_service
1500 .query_many(query, row_to_cached_item)
1501 .await
1502 .map_err(|e| RepoError::Database { message: e })?;
1503
1504 let mut items = Vec::new();
1505 for cached in cached_items {
1506 let user_data = self.get_user_data(&cached.id).await;
1507 items.push(Self::cached_item_to_media_item(cached, user_data));
1508 }
1509
1510 Ok(items)
1511 }
1512
1513 async fn get_resume_items(
1514 &self,
1515 parent_id: Option<&str>,
1516 limit: Option<usize>,
1517 ) -> Result<Vec<MediaItem>, RepoError> {
1518 let limit_val = limit.unwrap_or(12);
1519
1520 let (sql, params) = if let Some(pid) = parent_id {
1522 (
1523 format!(
1524 "SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
1525 i.overview, i.genres, i.runtime_ticks, i.production_year,
1526 i.community_rating, i.official_rating, i.primary_image_tag,
1527 i.album_id, i.album_name, i.album_artist, i.artists,
1528 i.index_number, i.series_id, i.series_name, i.season_id,
1529 i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
1530 FROM items i
1531 JOIN user_data ud ON i.id = ud.item_id
1532 INNER JOIN downloads d ON i.id = d.item_id
1533 WHERE i.server_id = ? AND ud.user_id = ? AND i.library_id = ?
1534 AND ud.playback_position_ticks > 0 AND ud.is_played = 0
1535 AND d.status = 'completed'
1536 AND i.item_type IN ('Movie', 'Episode')
1537 ORDER BY ud.last_played_at DESC
1538 LIMIT {}",
1539 limit_val
1540 ),
1541 vec![
1542 QueryParam::String(self.server_id.clone()),
1543 QueryParam::String(self.user_id.clone()),
1544 QueryParam::String(pid.to_string()),
1545 ],
1546 )
1547 } else {
1548 (
1549 format!(
1550 "SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
1551 i.overview, i.genres, i.runtime_ticks, i.production_year,
1552 i.community_rating, i.official_rating, i.primary_image_tag,
1553 i.album_id, i.album_name, i.album_artist, i.artists,
1554 i.index_number, i.series_id, i.series_name, i.season_id,
1555 i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
1556 FROM items i
1557 JOIN user_data ud ON i.id = ud.item_id
1558 INNER JOIN downloads d ON i.id = d.item_id
1559 WHERE i.server_id = ? AND ud.user_id = ?
1560 AND ud.playback_position_ticks > 0 AND ud.is_played = 0
1561 AND d.status = 'completed'
1562 AND i.item_type IN ('Movie', 'Episode')
1563 ORDER BY ud.last_played_at DESC
1564 LIMIT {}",
1565 limit_val
1566 ),
1567 vec![
1568 QueryParam::String(self.server_id.clone()),
1569 QueryParam::String(self.user_id.clone()),
1570 ],
1571 )
1572 };
1573
1574 let query = Query::with_params(sql, params);
1575
1576 let cached_items: Vec<CachedItem> = self
1577 .db_service
1578 .query_many(query, row_to_cached_item)
1579 .await
1580 .map_err(|e| RepoError::Database { message: e })?;
1581
1582 let mut items = Vec::new();
1583 for cached in cached_items {
1584 let user_data = self.get_user_data(&cached.id).await;
1585 items.push(Self::cached_item_to_media_item(cached, user_data));
1586 }
1587
1588 Ok(items)
1589 }
1590
1591 async fn get_next_up_episodes(
1592 &self,
1593 _series_id: Option<&str>,
1594 _limit: Option<usize>,
1595 ) -> Result<Vec<MediaItem>, RepoError> {
1596 Ok(Vec::new())
1599 }
1600
1601 async fn get_recently_played_audio(
1602 &self,
1603 limit: Option<usize>,
1604 ) -> Result<Vec<MediaItem>, RepoError> {
1605 let limit_val = limit.unwrap_or(12);
1606
1607 let query = Query::with_params(
1612 format!(
1613 "WITH downloaded_items AS (
1614 -- Playable items with completed downloads (Audio tracks)
1615 SELECT DISTINCT i.id
1616 FROM items i
1617 INNER JOIN downloads d ON i.id = d.item_id
1618 WHERE d.status = 'completed'
1619 AND i.item_type = 'Audio'
1620
1621 UNION
1622
1623 -- Containers with downloaded children (Albums)
1624 SELECT DISTINCT i.id
1625 FROM items i
1626 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
1627 INNER JOIN downloads d ON children.id = d.item_id
1628 WHERE d.status = 'completed'
1629 AND i.item_type = 'MusicAlbum'
1630 ),
1631 ranked_plays AS (
1632 SELECT
1633 CASE
1634 WHEN ud.playback_context_type = 'container' THEN ud.playback_context_id
1635 WHEN ud.playback_context_type = 'single' THEN ud.item_id
1636 ELSE COALESCE(i.album_id, ud.item_id)
1637 END AS display_id,
1638 MAX(ud.last_played_at) AS most_recent_play
1639 FROM user_data ud
1640 JOIN items i ON ud.item_id = i.id
1641 WHERE ud.user_id = ? AND i.server_id = ?
1642 AND i.item_type = 'Audio'
1643 AND ud.last_played_at IS NOT NULL
1644 GROUP BY display_id
1645 ORDER BY most_recent_play DESC
1646 LIMIT {}
1647 )
1648 SELECT DISTINCT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
1649 i.overview, i.genres, i.runtime_ticks, i.production_year,
1650 i.community_rating, i.official_rating, i.primary_image_tag,
1651 i.album_id, i.album_name, i.album_artist, i.artists,
1652 i.index_number, i.series_id, i.series_name, i.season_id,
1653 i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
1654 FROM ranked_plays rp
1655 JOIN items i ON rp.display_id = i.id
1656 INNER JOIN downloaded_items di ON i.id = di.id
1657 ORDER BY rp.most_recent_play DESC",
1658 limit_val
1659 ),
1660 vec![
1661 QueryParam::String(self.user_id.clone()),
1662 QueryParam::String(self.server_id.clone()),
1663 ],
1664 );
1665
1666 let cached_items: Vec<CachedItem> = self
1667 .db_service
1668 .query_many(query, row_to_cached_item)
1669 .await
1670 .map_err(|e| RepoError::Database { message: e })?;
1671
1672 let mut items = Vec::new();
1673 for cached in cached_items {
1674 let user_data = self.get_user_data(&cached.id).await;
1675 items.push(Self::cached_item_to_media_item(cached, user_data));
1676 }
1677
1678 Ok(items)
1679 }
1680
1681 async fn get_rediscover_albums(
1682 &self,
1683 _parent_id: Option<&str>,
1684 _limit: Option<usize>,
1685 ) -> Result<Vec<MediaItem>, RepoError> {
1686 Ok(Vec::new())
1690 }
1691
1692 async fn get_resume_movies(&self, limit: Option<usize>) -> Result<Vec<MediaItem>, RepoError> {
1693 let limit_val = limit.unwrap_or(12);
1694
1695 let query = Query::with_params(
1697 format!(
1698 "SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
1699 i.overview, i.genres, i.runtime_ticks, i.production_year,
1700 i.community_rating, i.official_rating, i.primary_image_tag,
1701 i.album_id, i.album_name, i.album_artist, i.artists,
1702 i.index_number, i.series_id, i.series_name, i.season_id,
1703 i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
1704 FROM items i
1705 JOIN user_data ud ON i.id = ud.item_id
1706 INNER JOIN downloads d ON i.id = d.item_id
1707 WHERE i.server_id = ? AND ud.user_id = ? AND i.item_type = 'Movie'
1708 AND ud.playback_position_ticks > 0 AND ud.is_played = 0
1709 AND d.status = 'completed'
1710 ORDER BY ud.last_played_at DESC
1711 LIMIT {}",
1712 limit_val
1713 ),
1714 vec![
1715 QueryParam::String(self.server_id.clone()),
1716 QueryParam::String(self.user_id.clone()),
1717 ],
1718 );
1719
1720 let cached_items: Vec<CachedItem> = self
1721 .db_service
1722 .query_many(query, row_to_cached_item)
1723 .await
1724 .map_err(|e| RepoError::Database { message: e })?;
1725
1726 let mut items = Vec::new();
1727 for cached in cached_items {
1728 let user_data = self.get_user_data(&cached.id).await;
1729 items.push(Self::cached_item_to_media_item(cached, user_data));
1730 }
1731
1732 Ok(items)
1733 }
1734
1735 async fn get_genres(&self, parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
1736 let library_id = parent_id.unwrap_or("").to_string();
1741
1742 let query = Query::with_params(
1743 "SELECT id, name, album_count FROM genres WHERE server_id = ? AND library_id = ?",
1744 vec![
1745 QueryParam::String(self.server_id.clone()),
1746 QueryParam::String(library_id),
1747 ],
1748 );
1749
1750 let genres: Vec<Genre> = self
1751 .db_service
1752 .query_many(query, |row| {
1753 Ok(Genre {
1754 id: row.get(0)?,
1755 name: row.get(1)?,
1756 album_count: row.get::<_, Option<i64>>(2)?.map(|c| c as u32),
1757 })
1758 })
1759 .await
1760 .map_err(|e| RepoError::Database { message: e })?;
1761
1762 Ok(genres)
1763 }
1764
1765 async fn search(
1766 &self,
1767 query: &str,
1768 options: Option<SearchOptions>,
1769 ) -> Result<SearchResult, RepoError> {
1770 let opts = options.unwrap_or_default();
1771 let limit = opts.limit.unwrap_or(20);
1772
1773 let Some(fts_query) = build_fts_prefix_query(query) else {
1776 return Ok(SearchResult {
1777 items: Vec::new(),
1778 total_record_count: 0,
1779 });
1780 };
1781
1782 let type_values: &[String] = opts
1786 .include_item_types
1787 .as_deref()
1788 .filter(|types| !types.is_empty())
1789 .unwrap_or(&[]);
1790 let type_filter = if type_values.is_empty() {
1791 String::new()
1792 } else {
1793 let placeholders = vec!["?"; type_values.len()].join(",");
1794 format!(" AND i.item_type IN ({})", placeholders)
1795 };
1796
1797 let catalog_branch = if include_catalog_browse() {
1803 "UNION
1804
1805 -- Synced catalog: fast online search, or the offline
1806 -- 'Show all server media' view. See set_include_catalog_browse.
1807 SELECT DISTINCT i.id
1808 FROM items i
1809 WHERE i.synced_at IS NOT NULL"
1810 } else {
1811 ""
1812 };
1813
1814 let sql = format!(
1815 "WITH available_items AS (
1816 -- Playable items with completed downloads
1817 SELECT DISTINCT i.id
1818 FROM items i
1819 INNER JOIN downloads d ON i.id = d.item_id
1820 WHERE d.status = 'completed'
1821 AND i.item_type IN ('Audio', 'Movie', 'Episode')
1822
1823 UNION
1824
1825 -- Containers with downloaded children
1826 SELECT DISTINCT i.id
1827 FROM items i
1828 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
1829 INNER JOIN downloads d ON children.id = d.item_id
1830 WHERE d.status = 'completed'
1831 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
1832
1833 {}
1834 )
1835 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
1836 i.overview, i.genres, i.runtime_ticks, i.production_year,
1837 i.community_rating, i.official_rating, i.primary_image_tag,
1838 i.album_id, i.album_name, i.album_artist, i.artists,
1839 i.index_number, i.series_id, i.series_name, i.season_id,
1840 i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
1841 FROM items i
1842 JOIN items_fts fts ON fts.rowid = i.rowid
1843 INNER JOIN available_items ai ON i.id = ai.id
1844 WHERE i.server_id = ? AND items_fts MATCH ?{}
1845 ORDER BY rank
1846 LIMIT {}",
1847 catalog_branch, type_filter, limit
1848 );
1849
1850 let mut params = vec![
1851 QueryParam::String(self.server_id.clone()),
1852 QueryParam::String(fts_query.clone()),
1853 ];
1854 params.extend(type_values.iter().cloned().map(QueryParam::String));
1855
1856 let db_query = Query::with_params(sql, params);
1857
1858 let cached_items: Vec<CachedItem> = self
1859 .db_service
1860 .query_many(db_query, row_to_cached_item)
1861 .await
1862 .map_err(|e| RepoError::Database { message: e })?;
1863
1864 let mut items = Vec::new();
1865 for cached in cached_items {
1866 let user_data = self.get_user_data(&cached.id).await;
1867 items.push(Self::cached_item_to_media_item(cached, user_data));
1868 }
1869
1870 if type_values.is_empty() {
1876 items.extend(self.search_people(&fts_query, limit).await?);
1877 }
1878
1879 let total_record_count = items.len();
1880
1881 Ok(SearchResult {
1882 items,
1883 total_record_count,
1884 })
1885 }
1886
1887 async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
1888 Err(RepoError::Offline)
1890 }
1891
1892 async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
1893 Err(RepoError::Offline)
1895 }
1896
1897 async fn get_audio_only_stream_url_for_video(
1898 &self,
1899 _item_id: &str,
1900 _media_source_id: Option<&str>,
1901 _start_time_seconds: Option<f64>,
1902 _audio_stream_index: Option<i32>,
1903 ) -> Result<String, RepoError> {
1904 Err(RepoError::Offline)
1906 }
1907
1908 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
1909 Err(RepoError::Offline)
1911 }
1912
1913 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
1914 Err(RepoError::Offline)
1916 }
1917
1918 async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
1919 Err(RepoError::Offline)
1921 }
1922
1923 async fn report_playback_start(
1924 &self,
1925 _item_id: &str,
1926 _position_ticks: i64,
1927 ) -> Result<(), RepoError> {
1928 Err(RepoError::Offline)
1930 }
1931
1932 async fn report_playback_progress(
1933 &self,
1934 _item_id: &str,
1935 _position_ticks: i64,
1936 ) -> Result<(), RepoError> {
1937 Err(RepoError::Offline)
1939 }
1940
1941 async fn report_playback_stopped(
1942 &self,
1943 _item_id: &str,
1944 _position_ticks: i64,
1945 ) -> Result<(), RepoError> {
1946 Err(RepoError::Offline)
1948 }
1949
1950 fn get_image_url(
1951 &self,
1952 item_id: &str,
1953 image_type: ImageType,
1954 options: Option<ImageOptions>,
1955 ) -> String {
1956 let type_str = match image_type {
1959 ImageType::Primary => "Primary",
1960 ImageType::Backdrop => "Backdrop",
1961 ImageType::Logo => "Logo",
1962 ImageType::Thumb => "Thumb",
1963 ImageType::Banner => "Banner",
1964 };
1965
1966 if let Some(opts) = options {
1967 if let Some(tag) = opts.tag {
1968 return format!("offline://{}/{}/{}", item_id, type_str, tag);
1969 }
1970 }
1971
1972 format!("offline://{}/{}", item_id, type_str)
1973 }
1974
1975 fn get_subtitle_url(
1976 &self,
1977 _item_id: &str,
1978 _media_source_id: &str,
1979 _stream_index: i32,
1980 _format: &str,
1981 ) -> String {
1982 String::new()
1984 }
1985
1986 fn get_video_download_url(
1987 &self,
1988 _item_id: &str,
1989 _quality: &str,
1990 _media_source_id: Option<&str>,
1991 _source_audio_codec: Option<&str>,
1992 ) -> String {
1993 String::new()
1995 }
1996
1997 async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1998 Err(RepoError::Offline)
2000 }
2001
2002 async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
2003 Err(RepoError::Offline)
2005 }
2006
2007 async fn get_favorites(
2016 &self,
2017 scope: SearchScope,
2018 options: Option<GetItemsOptions>,
2019 ) -> Result<SearchResult, RepoError> {
2020 let opts = options.unwrap_or_default();
2021 let limit = opts.limit.unwrap_or(10000);
2022 let start_index = opts.start_index.unwrap_or(0);
2023
2024 let type_filter = match scope.item_types() {
2027 Some(types) if !types.is_empty() => {
2028 let placeholders = vec!["?"; types.len()].join(",");
2029 format!(" AND i.item_type IN ({})", placeholders)
2030 }
2031 _ => String::new(),
2032 };
2033
2034 let catalog_branch = if include_catalog_browse() {
2035 "UNION
2036
2037 SELECT DISTINCT i.id
2038 FROM items i
2039 WHERE i.synced_at IS NOT NULL"
2040 } else {
2041 ""
2042 };
2043
2044 let sql = format!(
2045 "WITH available_items AS (
2046 SELECT DISTINCT i.id
2047 FROM items i
2048 INNER JOIN downloads d ON i.id = d.item_id
2049 WHERE d.status = 'completed'
2050 AND i.item_type IN ('Audio', 'Movie', 'Episode')
2051
2052 UNION
2053
2054 SELECT DISTINCT i.id
2055 FROM items i
2056 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
2057 INNER JOIN downloads d ON children.id = d.item_id
2058 WHERE d.status = 'completed'
2059 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
2060
2061 {catalog_branch}
2062 )
2063 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
2064 i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
2065 i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
2066 i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
2067 i.parent_index_number, i.is_folder, i.premiere_date
2068 FROM items i
2069 INNER JOIN available_items ai ON i.id = ai.id
2070 INNER JOIN user_data ud ON ud.item_id = i.id
2071 WHERE i.server_id = ?
2072 AND ud.user_id = ?
2073 AND ud.is_favorite = 1{}
2074 ORDER BY i.sort_name ASC, i.name ASC
2075 LIMIT {} OFFSET {}",
2076 type_filter, limit, start_index
2077 );
2078
2079 let mut params = vec![
2080 QueryParam::String(self.server_id.clone()),
2081 QueryParam::String(self.user_id.clone()),
2082 ];
2083 if let Some(types) = scope.item_types() {
2084 params.extend(types.into_iter().map(QueryParam::String));
2085 }
2086
2087 let cached_items: Vec<CachedItem> = self
2088 .db_service
2089 .query_many(Query::with_params(sql, params), row_to_cached_item)
2090 .await
2091 .map_err(|e| RepoError::Database { message: e })?;
2092
2093 let mut items = Vec::new();
2094 for cached in cached_items {
2095 let user_data = self.get_user_data(&cached.id).await;
2096 items.push(Self::cached_item_to_media_item(cached, user_data));
2097 }
2098
2099 let total_record_count = items.len();
2100 debug!("[OfflineRepo] Returning {} favourites", total_record_count);
2101
2102 Ok(SearchResult {
2103 items,
2104 total_record_count,
2105 })
2106 }
2107
2108 async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
2109 Err(RepoError::Offline)
2112 }
2113
2114 async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> {
2115 Err(RepoError::Offline)
2118 }
2119
2120 async fn get_person(&self, person_id: &str) -> Result<MediaItem, RepoError> {
2121 let query = Query::with_params(
2122 "SELECT id, name, overview, primary_image_tag
2123 FROM people WHERE id = ?",
2124 vec![QueryParam::String(person_id.to_string())],
2125 );
2126
2127 let person_data = self
2128 .db_service
2129 .query_optional(query, |row| {
2130 Ok((
2131 row.get::<_, String>(0)?,
2132 row.get::<_, String>(1)?,
2133 row.get::<_, Option<String>>(2)?,
2134 row.get::<_, Option<String>>(3)?,
2135 ))
2136 })
2137 .await
2138 .map_err(|e| RepoError::Database { message: e })?
2139 .ok_or_else(|| RepoError::NotFound {
2140 message: format!("Person {} not found in cache", person_id),
2141 })?;
2142
2143 Ok(MediaItem {
2144 id: person_data.0,
2145 name: person_data.1,
2146 item_type: "Person".to_string(),
2147 kind: crate::domain::MediaKind::Person,
2148 is_folder: false,
2149 server_id: self.server_id.clone(),
2150 parent_id: None,
2151 library_id: None,
2152 overview: person_data.2,
2153 genres: None,
2154 runtime_ticks: None,
2155 duration_ms: None,
2156 production_year: None,
2157 premiere_date: None,
2158 community_rating: None,
2159 official_rating: None,
2160 primary_image_tag: person_data.3.clone(),
2161 image_id: person_data.3,
2162 backdrop_image_tags: None,
2163 parent_backdrop_image_tags: None,
2164 album_id: None,
2165 album_name: None,
2166 album_artist: None,
2167 artists: None,
2168 artist_items: None,
2169 index_number: None,
2170 series_id: None,
2171 series_name: None,
2172 season_id: None,
2173 season_name: None,
2174 parent_index_number: None,
2175 user_data: None,
2176 media_streams: None,
2177 media_sources: None,
2178 people: None,
2179 })
2180 }
2181
2182 async fn get_items_by_person(
2183 &self,
2184 person_id: &str,
2185 options: Option<GetItemsOptions>,
2186 ) -> Result<SearchResult, RepoError> {
2187 let opts = options.unwrap_or_default();
2188 let limit = opts.limit.unwrap_or(10000); let query = Query::with_params(
2192 format!(
2193 "WITH downloaded_items AS (
2194 -- Playable items with completed downloads
2195 SELECT DISTINCT i.id
2196 FROM items i
2197 INNER JOIN downloads d ON i.id = d.item_id
2198 WHERE d.status = 'completed'
2199 AND i.item_type IN ('Audio', 'Movie', 'Episode')
2200
2201 UNION
2202
2203 -- Containers with downloaded children
2204 SELECT DISTINCT i.id
2205 FROM items i
2206 INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
2207 INNER JOIN downloads d ON children.id = d.item_id
2208 WHERE d.status = 'completed'
2209 AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
2210 )
2211 SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
2212 i.overview, i.genres, i.runtime_ticks, i.production_year,
2213 i.community_rating, i.official_rating, i.primary_image_tag,
2214 i.album_id, i.album_name, i.album_artist, i.artists,
2215 i.index_number, i.series_id, i.series_name, i.season_id,
2216 i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
2217 FROM items i
2218 JOIN item_people ip ON i.id = ip.item_id
2219 INNER JOIN downloaded_items di ON i.id = di.id
2220 WHERE i.server_id = ? AND ip.person_id = ?
2221 ORDER BY i.production_year DESC, i.sort_name ASC
2222 LIMIT {}", limit
2223 ),
2224 vec![
2225 QueryParam::String(self.server_id.clone()),
2226 QueryParam::String(person_id.to_string()),
2227 ],
2228 );
2229
2230 let cached_items: Vec<CachedItem> = self
2231 .db_service
2232 .query_many(query, row_to_cached_item)
2233 .await
2234 .map_err(|e| RepoError::Database { message: e })?;
2235
2236 let mut items = Vec::new();
2237 for cached in cached_items {
2238 let user_data = self.get_user_data(&cached.id).await;
2239 items.push(Self::cached_item_to_media_item(cached, user_data));
2240 }
2241
2242 let total_record_count = items.len();
2243
2244 Ok(SearchResult {
2245 items,
2246 total_record_count,
2247 })
2248 }
2249
2250 async fn get_similar_items(
2251 &self,
2252 _item_id: &str,
2253 _limit: Option<usize>,
2254 ) -> Result<SearchResult, RepoError> {
2255 Err(RepoError::Offline)
2257 }
2258
2259 async fn create_playlist(
2262 &self,
2263 name: &str,
2264 item_ids: &[String],
2265 ) -> Result<PlaylistCreatedResult, RepoError> {
2266 let playlist_id = uuid::Uuid::new_v4().to_string();
2267 let user_id = self.user_id.clone();
2268 let name = name.to_string();
2269 let item_ids = item_ids.to_vec();
2270 let pid = playlist_id.clone();
2271
2272 self.db_service
2273 .transaction(move |tx| {
2274 use crate::storage::db_service::{Query, QueryParam};
2275
2276 tx.execute(Query::with_params(
2277 "INSERT INTO playlists (id, user_id, name, is_local) VALUES (?1, ?2, ?3, 1)",
2278 vec![QueryParam::String(pid.clone()), QueryParam::String(user_id), QueryParam::String(name)],
2279 ))?;
2280
2281 for (i, item_id) in item_ids.iter().enumerate() {
2282 tx.execute(Query::with_params(
2283 "INSERT OR IGNORE INTO playlist_items (playlist_id, item_id, sort_order) VALUES (?1, ?2, ?3)",
2284 vec![QueryParam::String(pid.clone()), QueryParam::String(item_id.clone()), QueryParam::Int(i as i32)],
2285 ))?;
2286 }
2287
2288 Ok(())
2289 })
2290 .await
2291 .map_err(|e| RepoError::Database {
2292 message: format!("Failed to create playlist: {}", e),
2293 })?;
2294
2295 Ok(PlaylistCreatedResult { id: playlist_id })
2296 }
2297
2298 async fn delete_playlist(&self, playlist_id: &str) -> Result<(), RepoError> {
2299 let query = Query::with_params(
2300 "DELETE FROM playlists WHERE id = ?",
2301 vec![QueryParam::String(playlist_id.to_string())],
2302 );
2303 self.db_service
2304 .execute(query)
2305 .await
2306 .map_err(|e| RepoError::Database {
2307 message: format!("Failed to delete playlist: {}", e),
2308 })?;
2309 Ok(())
2310 }
2311
2312 async fn rename_playlist(&self, playlist_id: &str, name: &str) -> Result<(), RepoError> {
2313 let query = Query::with_params(
2314 "UPDATE playlists SET name = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
2315 vec![
2316 QueryParam::String(name.to_string()),
2317 QueryParam::String(playlist_id.to_string()),
2318 ],
2319 );
2320 self.db_service
2321 .execute(query)
2322 .await
2323 .map_err(|e| RepoError::Database {
2324 message: format!("Failed to rename playlist: {}", e),
2325 })?;
2326 Ok(())
2327 }
2328
2329 async fn get_playlist_items(&self, playlist_id: &str) -> Result<Vec<PlaylistEntry>, RepoError> {
2330 let query = Query::with_params(
2331 "SELECT pi.id, \
2332 i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres, \
2333 i.runtime_ticks, i.production_year, i.community_rating, i.official_rating, \
2334 i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists, \
2335 i.index_number, i.series_id, i.series_name, i.season_id, i.season_name, \
2336 i.parent_index_number, i.is_folder, i.premiere_date \
2337 FROM playlist_items pi \
2338 JOIN items i ON pi.item_id = i.id \
2339 WHERE pi.playlist_id = ? \
2340 ORDER BY pi.sort_order ASC",
2341 vec![QueryParam::String(playlist_id.to_string())],
2342 );
2343
2344 let items = self
2345 .db_service
2346 .query_many(query, |row| {
2347 let entry_id: i64 = row.get(0)?;
2348 let cached = CachedItem {
2350 id: row.get(1)?,
2351 name: row.get(2)?,
2352 item_type: row.get(3)?,
2353 server_id: row.get(4)?,
2354 parent_id: row.get(5)?,
2355 library_id: row.get(6)?,
2356 overview: row.get(7)?,
2357 genres: row.get(8)?,
2358 runtime_ticks: row.get(9)?,
2359 production_year: row.get(10)?,
2360 community_rating: row.get(11)?,
2361 official_rating: row.get(12)?,
2362 primary_image_tag: row.get(13)?,
2363 backdrop_image_tags: None,
2364 parent_backdrop_image_tags: None,
2365 album_id: row.get(14)?,
2366 album_name: row.get(15)?,
2367 album_artist: row.get(16)?,
2368 artists: row.get(17)?,
2369 index_number: row.get(18)?,
2370 series_id: row.get(19)?,
2371 series_name: row.get(20)?,
2372 season_id: row.get(21)?,
2373 season_name: row.get(22)?,
2374 parent_index_number: row.get(23)?,
2375 is_folder: row.get::<_, Option<i64>>(24)?.unwrap_or(0) != 0,
2376 premiere_date: row.get(25)?,
2377 };
2378 Ok((entry_id.to_string(), cached))
2379 })
2380 .await
2381 .map_err(|e| RepoError::Database {
2382 message: format!("Failed to get playlist items: {}", e),
2383 })?;
2384
2385 Ok(items
2386 .into_iter()
2387 .map(|(entry_id, cached)| PlaylistEntry {
2388 playlist_item_id: entry_id,
2389 item: Self::cached_item_to_media_item(cached, None),
2390 })
2391 .collect())
2392 }
2393
2394 async fn add_to_playlist(
2395 &self,
2396 playlist_id: &str,
2397 item_ids: &[String],
2398 ) -> Result<(), RepoError> {
2399 let max_query = Query::with_params(
2401 "SELECT COALESCE(MAX(sort_order), -1) FROM playlist_items WHERE playlist_id = ?",
2402 vec![QueryParam::String(playlist_id.to_string())],
2403 );
2404 let max_order: i32 = self
2405 .db_service
2406 .query_one(max_query, |row| row.get(0))
2407 .await
2408 .unwrap_or(-1);
2409
2410 let playlist_id = playlist_id.to_string();
2411 let item_ids = item_ids.to_vec();
2412
2413 self.db_service
2414 .transaction(move |tx| {
2415 use crate::storage::db_service::{Query, QueryParam};
2416
2417 for (i, item_id) in item_ids.iter().enumerate() {
2418 tx.execute(Query::with_params(
2419 "INSERT OR IGNORE INTO playlist_items (playlist_id, item_id, sort_order) VALUES (?1, ?2, ?3)",
2420 vec![
2421 QueryParam::String(playlist_id.clone()),
2422 QueryParam::String(item_id.clone()),
2423 QueryParam::Int(max_order + 1 + i as i32),
2424 ],
2425 ))?;
2426 }
2427 Ok(())
2428 })
2429 .await
2430 .map_err(|e| RepoError::Database {
2431 message: format!("Failed to add items to playlist: {}", e),
2432 })?;
2433
2434 Ok(())
2435 }
2436
2437 async fn remove_from_playlist(
2438 &self,
2439 playlist_id: &str,
2440 entry_ids: &[String],
2441 ) -> Result<(), RepoError> {
2442 let playlist_id = playlist_id.to_string();
2443 let entry_ids = entry_ids.to_vec();
2444
2445 self.db_service
2446 .transaction(move |tx| {
2447 use crate::storage::db_service::{Query, QueryParam};
2448
2449 for entry_id in &entry_ids {
2450 tx.execute(Query::with_params(
2451 "DELETE FROM playlist_items WHERE playlist_id = ? AND id = ?",
2452 vec![
2453 QueryParam::String(playlist_id.clone()),
2454 QueryParam::String(entry_id.clone()),
2455 ],
2456 ))?;
2457 }
2458 Ok(())
2459 })
2460 .await
2461 .map_err(|e| RepoError::Database {
2462 message: format!("Failed to remove items from playlist: {}", e),
2463 })?;
2464
2465 Ok(())
2466 }
2467
2468 async fn move_playlist_item(
2469 &self,
2470 playlist_id: &str,
2471 item_id: &str,
2472 new_index: u32,
2473 ) -> Result<(), RepoError> {
2474 let playlist_id = playlist_id.to_string();
2475 let item_id = item_id.to_string();
2476
2477 self.db_service
2478 .transaction(move |tx| {
2479 use crate::storage::db_service::{Query, QueryParam};
2480
2481 let items: Vec<(i64, String)> = tx.query_many(
2483 Query::with_params(
2484 "SELECT id, item_id FROM playlist_items WHERE playlist_id = ? ORDER BY sort_order",
2485 vec![QueryParam::String(playlist_id)],
2486 ),
2487 |row| Ok((row.get(0)?, row.get(1)?)),
2488 )?;
2489
2490 let old_idx = items.iter().position(|(_, iid)| iid == &item_id);
2492 if let Some(old_pos) = old_idx {
2493 let mut ids = items;
2494 let entry = ids.remove(old_pos);
2495 let insert_at = (new_index as usize).min(ids.len());
2496 ids.insert(insert_at, entry);
2497
2498 for (i, (entry_id, _)) in ids.iter().enumerate() {
2500 tx.execute(Query::with_params(
2501 "UPDATE playlist_items SET sort_order = ? WHERE id = ?",
2502 vec![QueryParam::Int(i as i32), QueryParam::Int64(*entry_id)],
2503 ))?;
2504 }
2505 }
2506
2507 Ok(())
2508 })
2509 .await
2510 .map_err(|e| RepoError::Database {
2511 message: format!("Failed to move playlist item: {}", e),
2512 })?;
2513
2514 Ok(())
2515 }
2516}
2517
2518#[cfg(test)]
2519mod tests {
2520 #![allow(clippy::await_holding_lock)]
2529
2530 use super::*;
2531 use crate::storage::db_service::RusqliteService;
2532 use rusqlite::Connection;
2533 use std::sync::{Arc, Mutex};
2534
2535 static CATALOG_BROWSE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
2542
2543 fn lock_catalog_browse() -> std::sync::MutexGuard<'static, ()> {
2544 use crate::utils::lock::MutexSafe;
2545 CATALOG_BROWSE_LOCK.lock_safe()
2546 }
2547
2548 #[test]
2550 fn test_build_fts_prefix_query() {
2551 assert_eq!(build_fts_prefix_query("Arr").as_deref(), Some("\"Arr\"*"));
2553
2554 assert_eq!(
2557 build_fts_prefix_query("parks rec").as_deref(),
2558 Some("\"parks\" \"rec\"*")
2559 );
2560
2561 for query in ["Bob's Burgers", "Spider-Man", "AC/DC", "Wall-E", "9-1-1"] {
2564 let built = build_fts_prefix_query(query).expect("should build");
2565 assert!(
2566 built.starts_with('"') && built.ends_with("*"),
2567 "{query:?} produced {built:?}"
2568 );
2569 }
2570
2571 assert_eq!(
2574 build_fts_prefix_query("say \"hi\"").as_deref(),
2575 Some("\"say\" \"\"\"hi\"\"\"*")
2576 );
2577
2578 assert_eq!(build_fts_prefix_query(""), None);
2581 assert_eq!(build_fts_prefix_query(" "), None);
2582 assert_eq!(build_fts_prefix_query("-"), None);
2583 }
2584
2585 #[tokio::test]
2590 async fn test_search_empty_query_returns_empty_not_error() {
2591 let db_service = create_test_db();
2592 let repo = OfflineRepository::new(
2593 db_service,
2594 "test-server".to_string(),
2595 "test-user".to_string(),
2596 );
2597
2598 let result = repo.search("", None).await;
2599 assert!(
2600 result.is_ok(),
2601 "empty query must not error: {:?}",
2602 result.err()
2603 );
2604 assert!(result.unwrap().items.is_empty());
2605 }
2606
2607 fn create_test_db() -> Arc<RusqliteService> {
2608 let conn = Connection::open_in_memory().unwrap();
2609
2610 conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
2612
2613 conn.execute_batch(
2615 r#"
2616 CREATE TABLE servers (
2617 id TEXT PRIMARY KEY,
2618 name TEXT NOT NULL,
2619 url TEXT NOT NULL UNIQUE
2620 );
2621
2622 CREATE TABLE items (
2623 id TEXT PRIMARY KEY,
2624 server_id TEXT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
2625 library_id TEXT,
2626 parent_id TEXT REFERENCES items(id) ON DELETE CASCADE,
2627 name TEXT NOT NULL,
2628 item_type TEXT NOT NULL,
2629 is_folder INTEGER DEFAULT 0,
2630 overview TEXT,
2631 genres TEXT,
2632 runtime_ticks INTEGER,
2633 production_year INTEGER,
2634 premiere_date TEXT,
2635 community_rating REAL,
2636 official_rating TEXT,
2637 primary_image_tag TEXT,
2638 backdrop_image_tags TEXT,
2639 album_id TEXT,
2640 album_name TEXT,
2641 album_artist TEXT,
2642 artists TEXT,
2643 index_number INTEGER,
2644 series_id TEXT,
2645 series_name TEXT,
2646 season_id TEXT,
2647 season_name TEXT,
2648 parent_index_number INTEGER,
2649 synced_at TEXT,
2650 sort_name TEXT
2651 );
2652
2653 -- Mirrors the real FTS5 index and its triggers (schema.rs migration
2654 -- 001) so search can be exercised in tests at all.
2655 CREATE VIRTUAL TABLE items_fts USING fts5(
2656 name, overview, album_name, album_artist, artists, series_name,
2657 content='items', content_rowid='rowid'
2658 );
2659
2660 CREATE TRIGGER items_ai AFTER INSERT ON items BEGIN
2661 INSERT INTO items_fts(rowid, name, overview, album_name, album_artist, artists, series_name)
2662 VALUES (new.rowid, new.name, new.overview, new.album_name, new.album_artist, new.artists, new.series_name);
2663 END;
2664
2665 CREATE TRIGGER items_ad AFTER DELETE ON items BEGIN
2666 INSERT INTO items_fts(items_fts, rowid, name, overview, album_name, album_artist, artists, series_name)
2667 VALUES('delete', old.rowid, old.name, old.overview, old.album_name, old.album_artist, old.artists, old.series_name);
2668 END;
2669
2670 CREATE TRIGGER items_au AFTER UPDATE ON items BEGIN
2671 INSERT INTO items_fts(items_fts, rowid, name, overview, album_name, album_artist, artists, series_name)
2672 VALUES('delete', old.rowid, old.name, old.overview, old.album_name, old.album_artist, old.artists, old.series_name);
2673 INSERT INTO items_fts(rowid, name, overview, album_name, album_artist, artists, series_name)
2674 VALUES (new.rowid, new.name, new.overview, new.album_name, new.album_artist, new.artists, new.series_name);
2675 END;
2676
2677 CREATE TABLE user_data (
2678 user_id TEXT NOT NULL,
2679 item_id TEXT NOT NULL,
2680 playback_position_ticks INTEGER,
2681 is_played INTEGER,
2682 is_favorite INTEGER,
2683 play_count INTEGER,
2684 last_played_at TEXT,
2685 playback_context_type TEXT,
2686 playback_context_id TEXT,
2687 synced_at TEXT,
2688 pending_sync INTEGER DEFAULT 0,
2689 PRIMARY KEY (user_id, item_id)
2690 );
2691
2692 CREATE TABLE playlists (
2693 id TEXT PRIMARY KEY,
2694 user_id TEXT NOT NULL,
2695 name TEXT NOT NULL,
2696 is_local INTEGER DEFAULT 0,
2697 jellyfin_id TEXT,
2698 created_at TEXT DEFAULT CURRENT_TIMESTAMP,
2699 updated_at TEXT
2700 );
2701
2702 CREATE TABLE playlist_items (
2703 id INTEGER PRIMARY KEY AUTOINCREMENT,
2704 playlist_id TEXT NOT NULL REFERENCES playlists(id) ON DELETE CASCADE,
2705 item_id TEXT NOT NULL REFERENCES items(id) ON DELETE CASCADE,
2706 sort_order INTEGER NOT NULL,
2707 added_at TEXT DEFAULT CURRENT_TIMESTAMP,
2708 UNIQUE(playlist_id, item_id)
2709 );
2710
2711 CREATE INDEX idx_playlist_items_playlist ON playlist_items(playlist_id, sort_order);
2712
2713 CREATE TABLE downloads (
2714 id INTEGER PRIMARY KEY AUTOINCREMENT,
2715 item_id TEXT NOT NULL,
2716 status TEXT NOT NULL,
2717 file_size INTEGER
2718 );
2719
2720 CREATE TABLE libraries (
2721 id TEXT PRIMARY KEY,
2722 server_id TEXT NOT NULL,
2723 name TEXT NOT NULL,
2724 collection_type TEXT,
2725 image_tag TEXT,
2726 sort_order INTEGER DEFAULT 0,
2727 synced_at TEXT
2728 );
2729
2730 -- Mirrors migration 009 + the migration 022 FTS index.
2731 CREATE TABLE people (
2732 id TEXT PRIMARY KEY,
2733 server_id TEXT NOT NULL,
2734 name TEXT NOT NULL,
2735 overview TEXT,
2736 primary_image_tag TEXT,
2737 premiere_date TEXT,
2738 end_date TEXT,
2739 synced_at TEXT DEFAULT CURRENT_TIMESTAMP
2740 );
2741
2742 CREATE VIRTUAL TABLE people_fts USING fts5(
2743 name, overview, content='people', content_rowid='rowid'
2744 );
2745
2746 CREATE TRIGGER people_ai AFTER INSERT ON people BEGIN
2747 INSERT INTO people_fts(rowid, name, overview)
2748 VALUES (new.rowid, new.name, new.overview);
2749 END;
2750
2751 CREATE TRIGGER people_ad AFTER DELETE ON people BEGIN
2752 INSERT INTO people_fts(people_fts, rowid, name, overview)
2753 VALUES('delete', old.rowid, old.name, old.overview);
2754 END;
2755
2756 CREATE TRIGGER people_au AFTER UPDATE ON people BEGIN
2757 INSERT INTO people_fts(people_fts, rowid, name, overview)
2758 VALUES('delete', old.rowid, old.name, old.overview);
2759 INSERT INTO people_fts(rowid, name, overview)
2760 VALUES (new.rowid, new.name, new.overview);
2761 END;
2762
2763 CREATE TABLE genres (
2764 id TEXT NOT NULL,
2765 server_id TEXT NOT NULL,
2766 library_id TEXT,
2767 name TEXT NOT NULL,
2768 album_count INTEGER,
2769 synced_at TEXT DEFAULT CURRENT_TIMESTAMP,
2770 PRIMARY KEY (server_id, library_id, name)
2771 );
2772 "#,
2773 )
2774 .unwrap();
2775
2776 conn.execute(
2778 "INSERT INTO servers (id, name, url) VALUES ('test-server', 'Test Server', 'http://test')",
2779 [],
2780 ).unwrap();
2781
2782 Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
2783 }
2784
2785 fn create_test_item(id: &str, name: &str, parent_id: Option<&str>) -> MediaItem {
2787 MediaItem {
2788 id: id.to_string(),
2789 name: name.to_string(),
2790 item_type: "Audio".to_string(),
2791 kind: crate::domain::MediaKind::Track,
2792 is_folder: false,
2793 server_id: "test-server".to_string(),
2794 parent_id: parent_id.map(|s| s.to_string()),
2795 library_id: None,
2796 overview: None,
2797 genres: None,
2798 runtime_ticks: None,
2799 duration_ms: None,
2800 production_year: None,
2801 premiere_date: None,
2802 community_rating: None,
2803 official_rating: None,
2804 primary_image_tag: None,
2805 image_id: None,
2806 backdrop_image_tags: None,
2807 parent_backdrop_image_tags: None,
2808 album_id: None,
2809 album_name: None,
2810 album_artist: None,
2811 artists: None,
2812 artist_items: None,
2813 index_number: None,
2814 series_id: None,
2815 series_name: None,
2816 season_id: None,
2817 season_name: None,
2818 parent_index_number: None,
2819 user_data: None,
2820 media_streams: None,
2821 media_sources: None,
2822 people: None,
2823 }
2824 }
2825
2826 #[tokio::test]
2827 async fn test_save_to_cache_with_missing_parent_fk() {
2828 let db_service = create_test_db();
2829
2830 let fk_enabled: i32 = db_service
2832 .query_one(Query::new("PRAGMA foreign_keys"), |row| row.get(0))
2833 .await
2834 .unwrap();
2835 println!("Foreign keys enabled: {}", fk_enabled);
2836 assert_eq!(fk_enabled, 1, "Foreign keys should be enabled");
2837
2838 let repo = OfflineRepository::new(
2839 db_service.clone(),
2840 "test-server".to_string(),
2841 "test-user".to_string(),
2842 );
2843
2844 let items = vec![
2848 create_test_item("track-1", "Track 1", Some("album-1")),
2850 create_test_item("track-2", "Track 2", Some("album-1")),
2851 create_test_item("track-3", "Track 3", Some("album-2")),
2853 create_test_item("track-4", "Track 4", Some("album-2")),
2854 create_test_item("album-1", "Album One", Some("library-123")),
2856 create_test_item("album-2", "Album Two", Some("library-123")),
2857 ];
2858
2859 println!("Attempting to save {} items...", items.len());
2860 for (i, item) in items.iter().enumerate() {
2861 println!(" Item {}: {} (parent: {:?})", i, item.id, item.parent_id);
2862 }
2863
2864 let result = repo.save_to_cache("library-123", &items).await;
2869
2870 match &result {
2872 Ok(count) => {
2873 println!("✓ Saved {} items", count);
2874 assert_eq!(*count, 6);
2875
2876 let all_items: Vec<(String, Option<String>)> = db_service
2878 .query_many(
2879 Query::new("SELECT id, parent_id FROM items ORDER BY id"),
2880 |row| Ok((row.get(0)?, row.get(1)?)),
2881 )
2882 .await
2883 .unwrap();
2884
2885 println!("\nAll items in database:");
2886 for (id, parent) in &all_items {
2887 println!(" {} -> parent: {:?}", id, parent);
2888 }
2889
2890 let track1_parent: Option<String> = db_service
2892 .query_optional(
2893 Query::with_params(
2894 "SELECT parent_id FROM items WHERE id = ?",
2895 vec![QueryParam::String("track-1".to_string())],
2896 ),
2897 |row| row.get(0),
2898 )
2899 .await
2900 .unwrap()
2901 .flatten();
2902
2903 println!("\ntrack-1 parent_id in DB: {:?}", track1_parent);
2904 println!("track-1 expected parent_id: Some(\"album-1\")");
2905
2906 assert_eq!(
2908 track1_parent,
2909 Some("album-1".to_string()),
2910 "track-1 should have parent_id='album-1'"
2911 );
2912
2913 let album1_parent: Option<String> = db_service
2915 .query_optional(
2916 Query::with_params(
2917 "SELECT parent_id FROM items WHERE id = ?",
2918 vec![QueryParam::String("album-1".to_string())],
2919 ),
2920 |row| row.get(0),
2921 )
2922 .await
2923 .unwrap()
2924 .flatten();
2925
2926 assert_eq!(
2927 album1_parent,
2928 Some("library-123".to_string()),
2929 "album-1 should have parent_id='library-123'"
2930 );
2931 }
2932 Err(e) => panic!("Unexpected error: {:?}", e),
2933 }
2934 }
2935
2936 #[tokio::test]
2937 async fn test_save_to_cache_simple_case() {
2938 let db_service = create_test_db();
2939 let repo = OfflineRepository::new(
2940 db_service.clone(),
2941 "test-server".to_string(),
2942 "test-user".to_string(),
2943 );
2944
2945 let items = vec![
2947 create_test_item("item-1", "Item 1", Some("parent-123")),
2948 create_test_item("item-2", "Item 2", Some("parent-123")),
2949 create_test_item("item-3", "Item 3", Some("parent-123")),
2950 ];
2951
2952 let result = repo.save_to_cache("parent-123", &items).await;
2953 assert!(result.is_ok(), "Simple case should work: {:?}", result);
2954 assert_eq!(result.unwrap(), 3);
2955 }
2956
2957 #[tokio::test]
2965 async fn test_get_item_album_available_via_album_id_link() {
2966 use crate::storage::db_service::DatabaseService;
2967 let db_service = create_test_db();
2968
2969 for sql in [
2970 "INSERT INTO items (id, server_id, name, item_type, album_id, parent_id) \
2972 VALUES ('album-1', 'test-server', 'Hadestown', 'MusicAlbum', NULL, NULL)",
2973 "INSERT INTO items (id, server_id, name, item_type, album_id, parent_id) \
2975 VALUES ('track-1', 'test-server', 'Wait For Me', 'Audio', 'album-1', NULL)",
2976 "INSERT INTO downloads (item_id, status) VALUES ('track-1', 'completed')",
2977 ] {
2978 db_service.execute(Query::new(sql)).await.unwrap();
2979 }
2980
2981 let repo = OfflineRepository::new(
2982 db_service.clone(),
2983 "test-server".to_string(),
2984 "test-user".to_string(),
2985 );
2986
2987 assert!(
2989 repo.get_item("track-1").await.is_ok(),
2990 "downloaded track should be available offline"
2991 );
2992
2993 let album = repo.get_item("album-1").await;
2996 assert!(
2997 album.is_ok(),
2998 "album with an album_id-linked downloaded track should be available offline, got {:?}",
2999 album.err()
3000 );
3001 assert_eq!(album.unwrap().id, "album-1");
3002
3003 let tracks = repo.get_items("album-1", None).await.unwrap();
3007 assert_eq!(
3008 tracks.items.len(),
3009 1,
3010 "get_items(album_id) should return the track"
3011 );
3012 assert_eq!(tracks.items[0].id, "track-1");
3013 }
3014
3015 #[tokio::test]
3028 async fn test_get_items_toggle_gates_synced_catalog() {
3029 use crate::storage::db_service::DatabaseService;
3030 let _guard = lock_catalog_browse();
3031 let db_service = create_test_db();
3032
3033 for sql in [
3034 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
3036 VALUES ('movie-dl', 'test-server', 'Downloaded', 'Movie', 'lib-1', '2026-01-01')",
3037 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
3038 VALUES ('movie-cat', 'test-server', 'CatalogOnly', 'Movie', 'lib-1', '2026-01-01')",
3039 "INSERT INTO downloads (item_id, status) VALUES ('movie-dl', 'completed')",
3041 "INSERT INTO libraries (id, server_id, name) VALUES ('lib-1', 'test-server', 'Movies')",
3043 ] {
3044 db_service.execute(Query::new(sql)).await.unwrap();
3045 }
3046
3047 let repo = OfflineRepository::new(
3048 db_service.clone(),
3049 "test-server".to_string(),
3050 "test-user".to_string(),
3051 );
3052 let opts = Some(GetItemsOptions {
3053 include_item_types: Some(vec!["Movie".to_string()]),
3054 ..Default::default()
3055 });
3056
3057 set_include_catalog_browse(false);
3059 let local_only = repo.get_items("lib-1", opts.clone()).await.unwrap();
3060 let ids: Vec<&str> = local_only.items.iter().map(|i| i.id.as_str()).collect();
3061 assert_eq!(
3062 ids,
3063 vec!["movie-dl"],
3064 "toggle off should show downloaded media only"
3065 );
3066
3067 set_include_catalog_browse(true);
3069 let full_catalog = repo.get_items("lib-1", opts).await.unwrap();
3070 let mut ids: Vec<&str> = full_catalog.items.iter().map(|i| i.id.as_str()).collect();
3071 ids.sort();
3072 assert_eq!(
3073 ids,
3074 vec!["movie-cat", "movie-dl"],
3075 "toggle on should reveal the full catalog"
3076 );
3077
3078 set_include_catalog_browse(true);
3080 }
3081
3082 #[tokio::test]
3089 async fn test_search_includes_cached_people() {
3090 use crate::storage::db_service::DatabaseService;
3091 let _guard = lock_catalog_browse();
3092 let db_service = create_test_db();
3093
3094 for sql in [
3095 "INSERT INTO people (id, server_id, name, overview) \
3096 VALUES ('p1', 'test-server', 'Tilda Swinton', 'Actor')",
3097 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3100 VALUES ('m1', 'test-server', 'Tilda the Movie', 'Movie', '2026-01-01')",
3101 ] {
3102 db_service.execute(Query::new(sql)).await.unwrap();
3103 }
3104
3105 let repo = OfflineRepository::new(
3106 db_service.clone(),
3107 "test-server".to_string(),
3108 "test-user".to_string(),
3109 );
3110 set_include_catalog_browse(true);
3111
3112 let all = repo.search("Tilda", None).await.unwrap();
3114 let mut ids: Vec<&str> = all.items.iter().map(|i| i.id.as_str()).collect();
3115 ids.sort();
3116 assert_eq!(ids, vec!["m1", "p1"], "unscoped search must include people");
3117
3118 let person = all.items.iter().find(|i| i.id == "p1").unwrap();
3119 assert_eq!(person.item_type, "Person");
3120 assert_eq!(person.kind, crate::domain::MediaKind::Person);
3121
3122 let scoped = repo
3124 .search(
3125 "Tilda",
3126 Some(SearchOptions {
3127 include_item_types: Some(vec!["Movie".to_string()]),
3128 ..Default::default()
3129 }),
3130 )
3131 .await
3132 .unwrap();
3133 let ids: Vec<&str> = scoped.items.iter().map(|i| i.id.as_str()).collect();
3134 assert_eq!(ids, vec!["m1"], "a scoped search must not leak people in");
3135
3136 set_include_catalog_browse(true);
3137 }
3138
3139 #[tokio::test]
3146 async fn test_prune_stale_catalog() {
3147 use crate::storage::db_service::DatabaseService;
3148 let db_service = create_test_db();
3149
3150 let old = "2026-01-01T00:00:00+00:00";
3152 let new = "2026-06-01T00:00:00+00:00";
3153 let cutoff = "2026-03-01T00:00:00+00:00";
3154
3155 for sql in [
3156 "INSERT INTO servers (id, name, url) \
3158 VALUES ('other-server', 'Other', 'http://other')"
3159 .to_string(),
3160 format!(
3162 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3163 VALUES ('keep-fresh', 'test-server', 'Fresh', 'Movie', '{new}')"
3164 ),
3165 format!(
3167 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3168 VALUES ('gone', 'test-server', 'Vanished', 'Movie', '{old}')"
3169 ),
3170 format!(
3172 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3173 VALUES ('keep-dl', 'test-server', 'Downloaded', 'Movie', '{old}')"
3174 ),
3175 "INSERT INTO downloads (item_id, status) VALUES ('keep-dl', 'completed')".to_string(),
3176 format!(
3178 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3179 VALUES ('keep-album', 'test-server', 'Album', 'MusicAlbum', '{old}')"
3180 ),
3181 format!(
3182 "INSERT INTO items (id, server_id, name, item_type, album_id, synced_at) \
3183 VALUES ('keep-track', 'test-server', 'Track', 'Audio', 'keep-album', '{old}')"
3184 ),
3185 "INSERT INTO downloads (item_id, status) VALUES ('keep-track', 'completed')"
3186 .to_string(),
3187 format!(
3190 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3191 VALUES ('keep-artist', 'test-server', 'Artist', 'MusicArtist', '{old}')"
3192 ),
3193 format!(
3195 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3196 VALUES ('keep-other', 'other-server', 'Elsewhere', 'Movie', '{old}')"
3197 ),
3198 ] {
3199 db_service.execute(Query::new(&sql)).await.unwrap();
3200 }
3201
3202 let repo = OfflineRepository::new(
3203 db_service.clone(),
3204 "test-server".to_string(),
3205 "test-user".to_string(),
3206 );
3207
3208 let crawled_types = vec![
3209 "Movie".to_string(),
3210 "MusicAlbum".to_string(),
3211 "Audio".to_string(),
3212 ];
3213 let removed = repo
3214 .prune_stale_catalog(cutoff, &crawled_types)
3215 .await
3216 .unwrap();
3217 assert_eq!(removed, 1, "only the vanished movie should be swept");
3218
3219 let mut surviving: Vec<String> = db_service
3220 .query_many(Query::new("SELECT id FROM items"), |row| row.get(0))
3221 .await
3222 .unwrap();
3223 surviving.sort();
3224 assert_eq!(
3225 surviving,
3226 vec![
3227 "keep-album",
3228 "keep-artist",
3229 "keep-dl",
3230 "keep-fresh",
3231 "keep-other",
3232 "keep-track",
3233 ]
3234 );
3235
3236 assert_eq!(repo.prune_stale_catalog(cutoff, &[]).await.unwrap(), 0);
3238 }
3239
3240 #[tokio::test]
3255 async fn test_repeated_cache_does_not_duplicate_fts_entries() {
3256 use crate::storage::db_service::DatabaseService;
3257 let db_service = create_test_db();
3258 let repo = OfflineRepository::new(
3259 db_service.clone(),
3260 "test-server".to_string(),
3261 "test-user".to_string(),
3262 );
3263
3264 let items = vec![create_test_item("track-1", "Wait For Me", None)];
3265
3266 for _ in 0..3 {
3269 repo.save_to_cache("parent-1", &items).await.unwrap();
3270 }
3271
3272 let item_rows: i64 = db_service
3275 .query_one(
3276 Query::new("SELECT COUNT(*) FROM items WHERE id = 'track-1'"),
3277 |row| row.get(0),
3278 )
3279 .await
3280 .unwrap();
3281 assert_eq!(item_rows, 1, "three passes must leave one item row");
3282
3283 let fts_hits: i64 = db_service
3284 .query_one(
3285 Query::with_params(
3286 "SELECT COUNT(*) FROM items_fts WHERE items_fts MATCH ?",
3287 vec![QueryParam::String("\"Wait\"*".to_string())],
3288 ),
3289 |row| row.get(0),
3290 )
3291 .await
3292 .unwrap();
3293 assert_eq!(
3294 fts_hits, 1,
3295 "the FTS index must hold one entry per item, not one per sync pass"
3296 );
3297 }
3298
3299 #[tokio::test]
3308 async fn test_search_toggle_gates_synced_catalog() {
3309 use crate::storage::db_service::DatabaseService;
3310 let _guard = lock_catalog_browse();
3311 let db_service = create_test_db();
3312
3313 for sql in [
3314 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
3316 VALUES ('movie-dl', 'test-server', 'Arrival', 'Movie', 'lib-1', '2026-01-01')",
3317 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
3318 VALUES ('movie-cat', 'test-server', 'Arrakis', 'Movie', 'lib-1', '2026-01-01')",
3319 "INSERT INTO downloads (item_id, status) VALUES ('movie-dl', 'completed')",
3321 ] {
3322 db_service.execute(Query::new(sql)).await.unwrap();
3323 }
3324
3325 let repo = OfflineRepository::new(
3326 db_service.clone(),
3327 "test-server".to_string(),
3328 "test-user".to_string(),
3329 );
3330
3331 set_include_catalog_browse(true);
3334 let full = repo.search("Arr", None).await.unwrap();
3335 let mut ids: Vec<&str> = full.items.iter().map(|i| i.id.as_str()).collect();
3336 ids.sort();
3337 assert_eq!(
3338 ids,
3339 vec!["movie-cat", "movie-dl"],
3340 "with catalog browse on, search must cover synced-but-not-downloaded items"
3341 );
3342
3343 set_include_catalog_browse(false);
3345 let local_only = repo.search("Arr", None).await.unwrap();
3346 let ids: Vec<&str> = local_only.items.iter().map(|i| i.id.as_str()).collect();
3347 assert_eq!(
3348 ids,
3349 vec!["movie-dl"],
3350 "with catalog browse off, search stays downloads-only"
3351 );
3352
3353 set_include_catalog_browse(true);
3355 }
3356
3357 #[tokio::test]
3364 async fn test_search_type_filter_is_parameterised() {
3365 use crate::storage::db_service::DatabaseService;
3366 let _guard = lock_catalog_browse();
3367 let db_service = create_test_db();
3368
3369 db_service
3370 .execute(Query::new(
3371 "INSERT INTO items (id, server_id, name, item_type, synced_at) \
3372 VALUES ('m1', 'test-server', 'Arrival', 'Movie', '2026-01-01')",
3373 ))
3374 .await
3375 .unwrap();
3376
3377 let repo = OfflineRepository::new(
3378 db_service.clone(),
3379 "test-server".to_string(),
3380 "test-user".to_string(),
3381 );
3382 set_include_catalog_browse(true);
3383
3384 let opts = SearchOptions {
3386 include_item_types: Some(vec!["Movie') OR 1=1 --".to_string()]),
3387 ..Default::default()
3388 };
3389 let result = repo.search("Arr", Some(opts)).await;
3390 assert!(
3391 result.is_ok(),
3392 "a quote in an item type must not break the query: {:?}",
3393 result.err()
3394 );
3395 assert!(
3396 result.unwrap().items.is_empty(),
3397 "an injected type filter must not widen the result set"
3398 );
3399
3400 let opts = SearchOptions {
3402 include_item_types: Some(vec!["Movie".to_string()]),
3403 ..Default::default()
3404 };
3405 assert_eq!(repo.search("Arr", Some(opts)).await.unwrap().items.len(), 1);
3406 }
3407
3408 #[tokio::test]
3413 async fn test_get_item_tv_available_via_season_series_link() {
3414 use crate::storage::db_service::DatabaseService;
3415 let db_service = create_test_db();
3416
3417 for sql in [
3418 "INSERT INTO items (id, server_id, name, item_type, parent_id) \
3419 VALUES ('series-1', 'test-server', 'Gilmore Girls', 'Series', NULL)",
3420 "INSERT INTO items (id, server_id, name, item_type, series_id, parent_id) \
3421 VALUES ('season-1', 'test-server', 'Season 1', 'Season', 'series-1', NULL)",
3422 "INSERT INTO items (id, server_id, name, item_type, season_id, series_id, parent_id) \
3424 VALUES ('ep-1', 'test-server', 'Pilot', 'Episode', 'season-1', 'series-1', NULL)",
3425 "INSERT INTO downloads (item_id, status) VALUES ('ep-1', 'completed')",
3426 ] {
3427 db_service.execute(Query::new(sql)).await.unwrap();
3428 }
3429
3430 let repo = OfflineRepository::new(
3431 db_service.clone(),
3432 "test-server".to_string(),
3433 "test-user".to_string(),
3434 );
3435
3436 assert!(
3437 repo.get_item("ep-1").await.is_ok(),
3438 "downloaded episode available offline"
3439 );
3440 assert!(
3441 repo.get_item("season-1").await.is_ok(),
3442 "season with a season_id-linked downloaded episode should be available offline"
3443 );
3444 assert!(
3445 repo.get_item("series-1").await.is_ok(),
3446 "series with a series_id-linked downloaded episode should be available offline"
3447 );
3448
3449 let season_items = repo.get_items("season-1", None).await.unwrap();
3451 assert!(
3452 season_items.items.iter().any(|i| i.id == "ep-1"),
3453 "get_items(season_id) should return the episode"
3454 );
3455
3456 let series_items = repo.get_items("series-1", None).await.unwrap();
3458 assert!(
3459 series_items.items.iter().any(|i| i.id == "ep-1"),
3460 "get_items(series_id) should surface the downloaded episode"
3461 );
3462 }
3463
3464 #[tokio::test]
3469 async fn test_libraries_cache_roundtrip_available_offline() {
3470 let db_service = create_test_db();
3471 let repo = OfflineRepository::new(
3472 db_service.clone(),
3473 "test-server".to_string(),
3474 "test-user".to_string(),
3475 );
3476
3477 assert!(repo.get_libraries().await.unwrap().is_empty());
3480
3481 let server_libs = vec![
3483 Library::new("music".into(), "Music".into(), "music".into(), None),
3484 Library::new(
3485 "movies".into(),
3486 "Movies".into(),
3487 "movies".into(),
3488 Some("tag".into()),
3489 ),
3490 ];
3491 let saved = repo.save_libraries_to_cache(&server_libs).await.unwrap();
3492 assert_eq!(saved, 2);
3493
3494 let offline_libs = repo.get_libraries().await.unwrap();
3496 let names: Vec<&str> = offline_libs.iter().map(|l| l.name.as_str()).collect();
3497 assert_eq!(
3498 names,
3499 vec!["Music", "Movies"],
3500 "cached libraries available offline in sort order"
3501 );
3502
3503 repo.save_libraries_to_cache(&server_libs).await.unwrap();
3505 assert_eq!(repo.get_libraries().await.unwrap().len(), 2);
3506 }
3507
3508 #[tokio::test]
3515 async fn test_genres_cache_roundtrip_scoped_by_library() {
3516 let db_service = create_test_db();
3517 let repo = OfflineRepository::new(
3518 db_service.clone(),
3519 "test-server".to_string(),
3520 "test-user".to_string(),
3521 );
3522
3523 assert!(repo.get_genres(Some("music-lib")).await.unwrap().is_empty());
3525
3526 let server_genres = vec![
3528 Genre {
3529 id: "g1".into(),
3530 name: "Rock".into(),
3531 album_count: Some(42),
3532 },
3533 Genre {
3534 id: "g2".into(),
3535 name: "Jazz".into(),
3536 album_count: Some(17),
3537 },
3538 Genre {
3539 id: "g3".into(),
3540 name: "Ambient".into(),
3541 album_count: None,
3542 },
3543 ];
3544 let saved = repo
3545 .save_genres_to_cache(Some("music-lib"), &server_genres)
3546 .await
3547 .unwrap();
3548 assert_eq!(saved, 3);
3549
3550 let mut offline_genres = repo.get_genres(Some("music-lib")).await.unwrap();
3552 offline_genres.sort_by(|a, b| a.name.cmp(&b.name));
3553 let names: Vec<&str> = offline_genres.iter().map(|g| g.name.as_str()).collect();
3554 assert_eq!(names, vec!["Ambient", "Jazz", "Rock"]);
3555 let rock = offline_genres.iter().find(|g| g.name == "Rock").unwrap();
3556 assert_eq!(rock.album_count, Some(42));
3557
3558 assert!(repo.get_genres(Some("other-lib")).await.unwrap().is_empty());
3560
3561 let updated = vec![Genre {
3563 id: "g1".into(),
3564 name: "Rock".into(),
3565 album_count: Some(50),
3566 }];
3567 repo.save_genres_to_cache(Some("music-lib"), &updated)
3568 .await
3569 .unwrap();
3570 let after = repo.get_genres(Some("music-lib")).await.unwrap();
3571 assert_eq!(after.len(), 1, "stale genres removed on refresh");
3572 assert_eq!(after[0].album_count, Some(50), "counts updated on refresh");
3573 }
3574
3575 async fn seed_items(repo: &OfflineRepository, ids: &[&str]) {
3579 let items: Vec<MediaItem> = ids
3580 .iter()
3581 .map(|id| create_test_item(id, &format!("Track {}", id), Some("library-1")))
3582 .collect();
3583 repo.save_to_cache("library-1", &items).await.unwrap();
3584 }
3585
3586 async fn insert_item(
3589 db: &Arc<RusqliteService>,
3590 id: &str,
3591 item_type: &str,
3592 album_id: Option<&str>,
3593 series_id: Option<&str>,
3594 season_id: Option<&str>,
3595 ) {
3596 db.execute(Query::with_params(
3597 "INSERT INTO items (id, server_id, name, item_type, album_id, series_id, season_id, synced_at)
3598 VALUES (?1, 'test-server', ?2, ?3, ?4, ?5, ?6, '2024-01-01')",
3599 vec![
3600 QueryParam::String(id.to_string()),
3601 QueryParam::String(format!("Name {id}")),
3602 QueryParam::String(item_type.to_string()),
3603 album_id.map(|s| QueryParam::String(s.to_string())).unwrap_or(QueryParam::Null),
3604 series_id.map(|s| QueryParam::String(s.to_string())).unwrap_or(QueryParam::Null),
3605 season_id.map(|s| QueryParam::String(s.to_string())).unwrap_or(QueryParam::Null),
3606 ],
3607 ))
3608 .await
3609 .unwrap();
3610 }
3611
3612 async fn insert_library_item(
3615 db: &Arc<RusqliteService>,
3616 id: &str,
3617 item_type: &str,
3618 library_id: &str,
3619 album_id: Option<&str>,
3620 ) {
3621 db.execute(Query::with_params(
3622 "INSERT INTO items (id, server_id, library_id, name, item_type, album_id, synced_at)
3623 VALUES (?1, 'test-server', ?2, ?3, ?4, ?5, '2024-01-01')",
3624 vec![
3625 QueryParam::String(id.to_string()),
3626 QueryParam::String(library_id.to_string()),
3627 QueryParam::String(format!("Name {id}")),
3628 QueryParam::String(item_type.to_string()),
3629 album_id
3630 .map(|s| QueryParam::String(s.to_string()))
3631 .unwrap_or(QueryParam::Null),
3632 ],
3633 ))
3634 .await
3635 .unwrap();
3636 }
3637
3638 async fn seed_completed_download(db: &Arc<RusqliteService>, item_id: &str, file_size: i64) {
3639 db.execute(Query::with_params(
3640 "INSERT INTO downloads (item_id, status, file_size) VALUES (?1, 'completed', ?2)",
3641 vec![
3642 QueryParam::String(item_id.to_string()),
3643 QueryParam::Int64(file_size),
3644 ],
3645 ))
3646 .await
3647 .unwrap();
3648 }
3649
3650 async fn seed_library(db: &Arc<RusqliteService>, id: &str, collection_type: &str) {
3651 db.execute(Query::with_params(
3652 "INSERT INTO libraries (id, server_id, name, collection_type, sort_order)
3653 VALUES (?1, 'test-server', ?2, ?3, 0)",
3654 vec![
3655 QueryParam::String(id.to_string()),
3656 QueryParam::String(format!("Lib {id}")),
3657 QueryParam::String(collection_type.to_string()),
3658 ],
3659 ))
3660 .await
3661 .unwrap();
3662 }
3663
3664 fn make_repo(db: &Arc<RusqliteService>) -> OfflineRepository {
3665 OfflineRepository::new(
3666 db.clone(),
3667 "test-server".to_string(),
3668 "test-user".to_string(),
3669 )
3670 }
3671
3672 #[tokio::test]
3679 async fn test_get_latest_items_collapses_tracks_into_their_album() {
3680 let db = create_test_db();
3681 insert_library_item(&db, "album-1", "MusicAlbum", "lib-1", None).await;
3682 for track in ["track-1", "track-2", "track-3"] {
3683 insert_library_item(&db, track, "Audio", "lib-1", Some("album-1")).await;
3684 seed_completed_download(&db, track, 1000).await;
3685 }
3686 insert_library_item(&db, "movie-1", "Movie", "lib-1", None).await;
3688 seed_completed_download(&db, "movie-1", 2000).await;
3689
3690 let repo = make_repo(&db);
3691 let latest = repo.get_latest_items("lib-1", Some(16)).await.unwrap();
3692 let ids: Vec<&str> = latest.iter().map(|i| i.id.as_str()).collect();
3693
3694 assert!(
3695 !ids.iter().any(|id| id.starts_with("track-")),
3696 "individual tracks must collapse into their album, got: {ids:?}"
3697 );
3698 assert!(ids.contains(&"album-1"), "the album itself is listed");
3699 assert!(ids.contains(&"movie-1"), "containerless items still listed");
3700 }
3701
3702 #[tokio::test]
3707 async fn test_get_downloaded_items_returns_leaf_and_container() {
3708 let db = create_test_db();
3709 insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
3710 insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
3711 insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
3712 seed_completed_download(&db, "track-1", 1000).await;
3714
3715 let repo = make_repo(&db);
3716
3717 let in_album = repo.get_downloaded_items("album-1", None).await.unwrap();
3719 let ids: Vec<&str> = in_album.items.iter().map(|i| i.id.as_str()).collect();
3720 assert_eq!(ids, vec!["track-1"], "only the downloaded track is listed");
3721 }
3722
3723 #[tokio::test]
3729 async fn test_get_downloaded_items_library_lists_albums_not_tracks() {
3730 let db = create_test_db();
3731 seed_library(&db, "music-lib", "music").await;
3732 insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
3733 insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
3735 insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
3736 seed_completed_download(&db, "track-1", 1000).await;
3737 seed_completed_download(&db, "track-2", 1000).await;
3738
3739 let repo = make_repo(&db);
3740
3741 let at_library = repo.get_downloaded_items("music-lib", None).await.unwrap();
3743 let ids: Vec<&str> = at_library.items.iter().map(|i| i.id.as_str()).collect();
3744 assert_eq!(
3745 ids,
3746 vec!["album-1"],
3747 "library browse lists the album container, not its tracks"
3748 );
3749
3750 let in_album = repo.get_downloaded_items("album-1", None).await.unwrap();
3752 let mut track_ids: Vec<&str> = in_album.items.iter().map(|i| i.id.as_str()).collect();
3753 track_ids.sort();
3754 assert_eq!(track_ids, vec!["track-1", "track-2"]);
3755 }
3756
3757 #[tokio::test]
3769 async fn test_get_downloaded_items_library_does_not_mix_media_types() {
3770 let db = create_test_db();
3771 seed_library(&db, "music-lib", "music").await;
3772 seed_library(&db, "movie-lib", "movies").await;
3773 seed_library(&db, "tv-lib", "tvshows").await;
3774
3775 insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
3776 insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
3777 insert_item(&db, "movie-1", "Movie", None, None, None).await;
3778 insert_item(&db, "series-1", "Series", None, None, None).await;
3779 insert_item(&db, "episode-1", "Episode", None, Some("series-1"), None).await;
3780
3781 seed_completed_download(&db, "track-1", 1000).await;
3782 seed_completed_download(&db, "movie-1", 2000).await;
3783 seed_completed_download(&db, "episode-1", 3000).await;
3784
3785 let repo = make_repo(&db);
3786
3787 let music: Vec<String> = repo
3788 .get_downloaded_items("music-lib", None)
3789 .await
3790 .unwrap()
3791 .items
3792 .iter()
3793 .map(|i| i.id.clone())
3794 .collect();
3795 assert_eq!(
3796 music,
3797 vec!["album-1"],
3798 "the music library must not list films or series; got {:?}",
3799 music
3800 );
3801
3802 let movies: Vec<String> = repo
3803 .get_downloaded_items("movie-lib", None)
3804 .await
3805 .unwrap()
3806 .items
3807 .iter()
3808 .map(|i| i.id.clone())
3809 .collect();
3810 assert_eq!(
3811 movies,
3812 vec!["movie-1"],
3813 "the movie library must not list albums or series; got {:?}",
3814 movies
3815 );
3816
3817 let tv: Vec<String> = repo
3818 .get_downloaded_items("tv-lib", None)
3819 .await
3820 .unwrap()
3821 .items
3822 .iter()
3823 .map(|i| i.id.clone())
3824 .collect();
3825 assert_eq!(
3826 tv,
3827 vec!["series-1"],
3828 "the TV library must not list albums or films; got {:?}",
3829 tv
3830 );
3831 }
3832
3833 #[tokio::test]
3839 async fn test_get_downloaded_items_library_lists_series_not_episodes() {
3840 let db = create_test_db();
3841 seed_library(&db, "tv-lib", "tvshows").await;
3842 insert_item(&db, "series-1", "Series", None, None, None).await;
3843 insert_item(&db, "season-1", "Season", None, Some("series-1"), None).await;
3845 insert_item(
3846 &db,
3847 "ep-1",
3848 "Episode",
3849 None,
3850 Some("series-1"),
3851 Some("season-1"),
3852 )
3853 .await;
3854 seed_completed_download(&db, "ep-1", 4000).await;
3855
3856 let repo = make_repo(&db);
3857
3858 let at_library = repo.get_downloaded_items("tv-lib", None).await.unwrap();
3860 let ids: Vec<&str> = at_library.items.iter().map(|i| i.id.as_str()).collect();
3861 assert_eq!(
3862 ids,
3863 vec!["series-1"],
3864 "TV library browse lists the series, not seasons/episodes"
3865 );
3866
3867 let in_series = repo.get_downloaded_items("series-1", None).await.unwrap();
3869 assert!(
3870 in_series.items.iter().any(|i| i.id == "season-1"),
3871 "series drill returns the season"
3872 );
3873 let in_season = repo.get_downloaded_items("season-1", None).await.unwrap();
3874 assert!(
3875 in_season.items.iter().any(|i| i.id == "ep-1"),
3876 "season drill returns the episode"
3877 );
3878 }
3879
3880 #[tokio::test]
3885 async fn test_get_downloaded_items_library_keeps_orphan_leaves() {
3886 let db = create_test_db();
3887 seed_library(&db, "movie-lib", "movies").await;
3888 insert_item(&db, "movie-1", "Movie", None, None, None).await;
3889 seed_completed_download(&db, "movie-1", 5000).await;
3890
3891 let repo = make_repo(&db);
3892 let at_library = repo.get_downloaded_items("movie-lib", None).await.unwrap();
3893 let ids: Vec<&str> = at_library.items.iter().map(|i| i.id.as_str()).collect();
3894 assert_eq!(
3895 ids,
3896 vec!["movie-1"],
3897 "a downloaded movie with no container shows"
3898 );
3899 }
3900
3901 #[tokio::test]
3906 async fn test_get_downloaded_items_empty_is_authoritative() {
3907 let db = create_test_db();
3908 insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
3909 insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
3910 set_include_catalog_browse(true);
3912 let repo = make_repo(&db);
3913
3914 let result = repo.get_downloaded_items("album-1", None).await.unwrap();
3915 assert!(
3916 result.items.is_empty(),
3917 "empty downloaded browse returns no items even with catalog-browse on"
3918 );
3919 }
3920
3921 #[tokio::test]
3925 async fn test_get_downloaded_libraries_omits_empty() {
3926 let db = create_test_db();
3927 seed_library(&db, "music-lib", "music").await;
3928 seed_library(&db, "movie-lib", "movies").await;
3929 insert_item(&db, "track-1", "Audio", None, None, None).await;
3930 seed_completed_download(&db, "track-1", 500).await;
3931
3932 let repo = make_repo(&db);
3933 let libs = repo.get_downloaded_libraries().await.unwrap();
3934 let ids: Vec<&str> = libs.iter().map(|l| l.id.as_str()).collect();
3935 assert_eq!(
3936 ids,
3937 vec!["music-lib"],
3938 "movie library with no downloads omitted"
3939 );
3940 }
3941
3942 #[tokio::test]
3947 async fn test_download_disk_usage_aggregates_containers() {
3948 let db = create_test_db();
3949 insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
3950 insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
3951 insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
3952 seed_completed_download(&db, "track-1", 1000).await;
3953 seed_completed_download(&db, "track-2", 2000).await;
3954
3955 let repo = make_repo(&db);
3956 let usage = repo.get_download_disk_usage().await.unwrap();
3957
3958 assert_eq!(usage.item_count, 2, "two leaf downloads");
3959 assert_eq!(
3960 usage.device_total_bytes, 3000,
3961 "device total is the leaf sum"
3962 );
3963 assert_eq!(usage.sizes.get("track-1"), Some(&1000));
3964 assert_eq!(
3965 usage.sizes.get("album-1"),
3966 Some(&3000),
3967 "container = sum of children"
3968 );
3969 assert_eq!(
3971 usage.partial_containers.get("album-1"),
3972 None,
3973 "fully downloaded album is not partial"
3974 );
3975 let leaf_sum: i64 = ["track-1", "track-2"]
3977 .iter()
3978 .map(|id| usage.sizes[*id])
3979 .sum();
3980 assert_eq!(leaf_sum, usage.device_total_bytes);
3981 }
3982
3983 #[tokio::test]
3986 async fn test_download_disk_usage_flags_partial_container() {
3987 let db = create_test_db();
3988 insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
3989 insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
3990 insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
3991 seed_completed_download(&db, "track-1", 1000).await;
3993
3994 let repo = make_repo(&db);
3995 let usage = repo.get_download_disk_usage().await.unwrap();
3996 assert_eq!(
3997 usage.partial_containers.get("album-1"),
3998 Some(&true),
3999 "album with a missing child is partial"
4000 );
4001 }
4002
4003 #[tokio::test]
4004 async fn test_playlist_create_empty() {
4005 let db_service = create_test_db();
4006 let repo = OfflineRepository::new(
4007 db_service.clone(),
4008 "test-server".to_string(),
4009 "test-user".to_string(),
4010 );
4011
4012 let result = repo.create_playlist("My Playlist", &[]).await;
4013 assert!(result.is_ok());
4014 let created = result.unwrap();
4015 assert!(
4016 !created.id.is_empty(),
4017 "Should return a non-empty playlist ID"
4018 );
4019
4020 let name: String = db_service
4022 .query_one(
4023 Query::with_params(
4024 "SELECT name FROM playlists WHERE id = ?",
4025 vec![QueryParam::String(created.id.clone())],
4026 ),
4027 |row| row.get(0),
4028 )
4029 .await
4030 .unwrap();
4031 assert_eq!(name, "My Playlist");
4032 }
4033
4034 #[tokio::test]
4035 async fn test_playlist_create_with_items() {
4036 let db_service = create_test_db();
4037 let repo = OfflineRepository::new(
4038 db_service.clone(),
4039 "test-server".to_string(),
4040 "test-user".to_string(),
4041 );
4042 seed_items(&repo, &["t1", "t2", "t3"]).await;
4043
4044 let created = repo
4045 .create_playlist("With Tracks", &["t1".into(), "t2".into(), "t3".into()])
4046 .await
4047 .unwrap();
4048
4049 let items = repo.get_playlist_items(&created.id).await.unwrap();
4050 assert_eq!(items.len(), 3);
4051 assert_eq!(items[0].item.id, "t1");
4052 assert_eq!(items[1].item.id, "t2");
4053 assert_eq!(items[2].item.id, "t3");
4054 }
4055
4056 #[tokio::test]
4057 async fn test_playlist_delete() {
4058 let db_service = create_test_db();
4059 let repo = OfflineRepository::new(
4060 db_service.clone(),
4061 "test-server".to_string(),
4062 "test-user".to_string(),
4063 );
4064 seed_items(&repo, &["t1"]).await;
4065
4066 let created = repo
4067 .create_playlist("To Delete", &["t1".into()])
4068 .await
4069 .unwrap();
4070
4071 repo.delete_playlist(&created.id).await.unwrap();
4073
4074 let count: i32 = db_service
4076 .query_one(
4077 Query::with_params(
4078 "SELECT COUNT(*) FROM playlists WHERE id = ?",
4079 vec![QueryParam::String(created.id.clone())],
4080 ),
4081 |row| row.get(0),
4082 )
4083 .await
4084 .unwrap();
4085 assert_eq!(count, 0);
4086
4087 let item_count: i32 = db_service
4089 .query_one(
4090 Query::with_params(
4091 "SELECT COUNT(*) FROM playlist_items WHERE playlist_id = ?",
4092 vec![QueryParam::String(created.id)],
4093 ),
4094 |row| row.get(0),
4095 )
4096 .await
4097 .unwrap();
4098 assert_eq!(item_count, 0);
4099 }
4100
4101 #[tokio::test]
4102 async fn test_playlist_rename() {
4103 let db_service = create_test_db();
4104 let repo = OfflineRepository::new(
4105 db_service.clone(),
4106 "test-server".to_string(),
4107 "test-user".to_string(),
4108 );
4109
4110 let created = repo.create_playlist("Original Name", &[]).await.unwrap();
4111 repo.rename_playlist(&created.id, "New Name").await.unwrap();
4112
4113 let name: String = db_service
4114 .query_one(
4115 Query::with_params(
4116 "SELECT name FROM playlists WHERE id = ?",
4117 vec![QueryParam::String(created.id)],
4118 ),
4119 |row| row.get(0),
4120 )
4121 .await
4122 .unwrap();
4123 assert_eq!(name, "New Name");
4124 }
4125
4126 #[tokio::test]
4127 async fn test_playlist_get_items_preserves_order() {
4128 let db_service = create_test_db();
4129 let repo = OfflineRepository::new(
4130 db_service.clone(),
4131 "test-server".to_string(),
4132 "test-user".to_string(),
4133 );
4134 seed_items(&repo, &["a", "b", "c"]).await;
4135
4136 let created = repo
4137 .create_playlist("Ordered", &["c".into(), "a".into(), "b".into()])
4138 .await
4139 .unwrap();
4140 let items = repo.get_playlist_items(&created.id).await.unwrap();
4141
4142 assert_eq!(items.len(), 3);
4143 assert_eq!(items[0].item.id, "c");
4145 assert_eq!(items[1].item.id, "a");
4146 assert_eq!(items[2].item.id, "b");
4147 assert_ne!(items[0].playlist_item_id, items[1].playlist_item_id);
4149 assert_ne!(items[1].playlist_item_id, items[2].playlist_item_id);
4150 }
4151
4152 #[tokio::test]
4153 async fn test_playlist_get_items_empty_playlist() {
4154 let db_service = create_test_db();
4155 let repo = OfflineRepository::new(
4156 db_service.clone(),
4157 "test-server".to_string(),
4158 "test-user".to_string(),
4159 );
4160
4161 let created = repo.create_playlist("Empty", &[]).await.unwrap();
4162 let items = repo.get_playlist_items(&created.id).await.unwrap();
4163 assert!(items.is_empty());
4164 }
4165
4166 #[tokio::test]
4167 async fn test_playlist_add_items() {
4168 let db_service = create_test_db();
4169 let repo = OfflineRepository::new(
4170 db_service.clone(),
4171 "test-server".to_string(),
4172 "test-user".to_string(),
4173 );
4174 seed_items(&repo, &["t1", "t2", "t3"]).await;
4175
4176 let created = repo
4177 .create_playlist("Addable", &["t1".into()])
4178 .await
4179 .unwrap();
4180
4181 repo.add_to_playlist(&created.id, &["t2".into(), "t3".into()])
4183 .await
4184 .unwrap();
4185
4186 let items = repo.get_playlist_items(&created.id).await.unwrap();
4187 assert_eq!(items.len(), 3);
4188 assert_eq!(items[0].item.id, "t1");
4189 assert_eq!(items[1].item.id, "t2");
4190 assert_eq!(items[2].item.id, "t3");
4191 }
4192
4193 #[tokio::test]
4194 async fn test_playlist_add_duplicate_items_ignored() {
4195 let db_service = create_test_db();
4196 let repo = OfflineRepository::new(
4197 db_service.clone(),
4198 "test-server".to_string(),
4199 "test-user".to_string(),
4200 );
4201 seed_items(&repo, &["t1"]).await;
4202
4203 let created = repo.create_playlist("Dupes", &["t1".into()]).await.unwrap();
4204
4205 repo.add_to_playlist(&created.id, &["t1".into()])
4207 .await
4208 .unwrap();
4209
4210 let items = repo.get_playlist_items(&created.id).await.unwrap();
4211 assert_eq!(
4212 items.len(),
4213 1,
4214 "Duplicate should be ignored (UNIQUE constraint)"
4215 );
4216 }
4217
4218 #[tokio::test]
4219 async fn test_playlist_remove_items() {
4220 let db_service = create_test_db();
4221 let repo = OfflineRepository::new(
4222 db_service.clone(),
4223 "test-server".to_string(),
4224 "test-user".to_string(),
4225 );
4226 seed_items(&repo, &["t1", "t2", "t3"]).await;
4227
4228 let created = repo
4229 .create_playlist("Removable", &["t1".into(), "t2".into(), "t3".into()])
4230 .await
4231 .unwrap();
4232 let items = repo.get_playlist_items(&created.id).await.unwrap();
4233 assert_eq!(items.len(), 3);
4234
4235 let entry_id_to_remove = items[1].playlist_item_id.clone();
4237 repo.remove_from_playlist(&created.id, &[entry_id_to_remove])
4238 .await
4239 .unwrap();
4240
4241 let items_after = repo.get_playlist_items(&created.id).await.unwrap();
4242 assert_eq!(items_after.len(), 2);
4243 assert_eq!(items_after[0].item.id, "t1");
4244 assert_eq!(items_after[1].item.id, "t3");
4245 }
4246
4247 #[tokio::test]
4248 async fn test_playlist_move_item_forward() {
4249 let db_service = create_test_db();
4250 let repo = OfflineRepository::new(
4251 db_service.clone(),
4252 "test-server".to_string(),
4253 "test-user".to_string(),
4254 );
4255 seed_items(&repo, &["a", "b", "c", "d"]).await;
4256
4257 let created = repo
4258 .create_playlist("Reorder", &["a".into(), "b".into(), "c".into(), "d".into()])
4259 .await
4260 .unwrap();
4261
4262 repo.move_playlist_item(&created.id, "a", 2).await.unwrap();
4264
4265 let items = repo.get_playlist_items(&created.id).await.unwrap();
4266 let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
4267 assert_eq!(ids, vec!["b", "c", "a", "d"]);
4268 }
4269
4270 #[tokio::test]
4271 async fn test_playlist_move_item_backward() {
4272 let db_service = create_test_db();
4273 let repo = OfflineRepository::new(
4274 db_service.clone(),
4275 "test-server".to_string(),
4276 "test-user".to_string(),
4277 );
4278 seed_items(&repo, &["a", "b", "c", "d"]).await;
4279
4280 let created = repo
4281 .create_playlist(
4282 "Reorder2",
4283 &["a".into(), "b".into(), "c".into(), "d".into()],
4284 )
4285 .await
4286 .unwrap();
4287
4288 repo.move_playlist_item(&created.id, "d", 0).await.unwrap();
4290
4291 let items = repo.get_playlist_items(&created.id).await.unwrap();
4292 let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
4293 assert_eq!(ids, vec!["d", "a", "b", "c"]);
4294 }
4295
4296 #[tokio::test]
4297 async fn test_playlist_move_item_to_end() {
4298 let db_service = create_test_db();
4299 let repo = OfflineRepository::new(
4300 db_service.clone(),
4301 "test-server".to_string(),
4302 "test-user".to_string(),
4303 );
4304 seed_items(&repo, &["a", "b", "c"]).await;
4305
4306 let created = repo
4307 .create_playlist("MoveEnd", &["a".into(), "b".into(), "c".into()])
4308 .await
4309 .unwrap();
4310
4311 repo.move_playlist_item(&created.id, "a", 99).await.unwrap();
4313
4314 let items = repo.get_playlist_items(&created.id).await.unwrap();
4315 let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
4316 assert_eq!(ids, vec!["b", "c", "a"]);
4317 }
4318
4319 #[tokio::test]
4320 async fn test_playlist_move_nonexistent_item_is_noop() {
4321 let db_service = create_test_db();
4322 let repo = OfflineRepository::new(
4323 db_service.clone(),
4324 "test-server".to_string(),
4325 "test-user".to_string(),
4326 );
4327 seed_items(&repo, &["a", "b"]).await;
4328
4329 let created = repo
4330 .create_playlist("NoOp", &["a".into(), "b".into()])
4331 .await
4332 .unwrap();
4333
4334 repo.move_playlist_item(&created.id, "nonexistent", 0)
4336 .await
4337 .unwrap();
4338
4339 let items = repo.get_playlist_items(&created.id).await.unwrap();
4340 let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
4341 assert_eq!(ids, vec!["a", "b"]);
4342 }
4343
4344 async fn seed_favorites(db_service: &Arc<RusqliteService>) {
4347 use crate::storage::db_service::DatabaseService;
4348 for sql in [
4349 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
4350 VALUES ('movie-fav', 'test-server', 'Favourite Movie', 'Movie', 'lib-1', '2026-01-01', 'Favourite Movie')",
4351 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
4352 VALUES ('movie-plain', 'test-server', 'Ordinary Movie', 'Movie', 'lib-1', '2026-01-01', 'Ordinary Movie')",
4353 "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
4354 VALUES ('album-fav', 'test-server', 'Favourite Album', 'MusicAlbum', 'lib-2', '2026-01-01', 'Favourite Album')",
4355 "INSERT INTO libraries (id, server_id, name) VALUES ('lib-1', 'test-server', 'Movies')",
4356 "INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'movie-fav', 1)",
4357 "INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'album-fav', 1)",
4358 "INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'movie-plain', 0)",
4360 "INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('other-user', 'movie-plain', 1)",
4362 ] {
4363 db_service.execute(Query::new(sql)).await.unwrap();
4364 }
4365 }
4366
4367 #[tokio::test]
4372 async fn test_get_favorites_returns_only_scoped_favorites() {
4373 let _guard = lock_catalog_browse();
4374 set_include_catalog_browse(true);
4375
4376 let db_service = create_test_db();
4377 seed_favorites(&db_service).await;
4378 let repo = OfflineRepository::new(
4379 db_service,
4380 "test-server".to_string(),
4381 "test-user".to_string(),
4382 );
4383
4384 let all = repo.get_favorites(SearchScope::All, None).await.unwrap();
4385 let mut ids: Vec<&str> = all.items.iter().map(|i| i.id.as_str()).collect();
4386 ids.sort();
4387 assert_eq!(
4388 ids,
4389 vec!["album-fav", "movie-fav"],
4390 "All scope should return every favourite and nothing else"
4391 );
4392
4393 let movies = repo.get_favorites(SearchScope::Movies, None).await.unwrap();
4394 let ids: Vec<&str> = movies.items.iter().map(|i| i.id.as_str()).collect();
4395 assert_eq!(ids, vec!["movie-fav"]);
4396
4397 let music = repo.get_favorites(SearchScope::Music, None).await.unwrap();
4398 let ids: Vec<&str> = music.items.iter().map(|i| i.id.as_str()).collect();
4399 assert_eq!(ids, vec!["album-fav"]);
4400 }
4401
4402 #[tokio::test]
4407 async fn test_get_favorites_respects_catalog_browse_gate() {
4408 use crate::storage::db_service::DatabaseService;
4409 let _guard = lock_catalog_browse();
4410
4411 let db_service = create_test_db();
4412 seed_favorites(&db_service).await;
4413 db_service
4415 .execute(Query::new(
4416 "INSERT INTO items (id, server_id, name, item_type, album_id, synced_at) \
4417 VALUES ('track-1', 'test-server', 'Track', 'Audio', 'album-fav', '2026-01-01')",
4418 ))
4419 .await
4420 .unwrap();
4421 db_service
4422 .execute(Query::new(
4423 "INSERT INTO downloads (item_id, status) VALUES ('track-1', 'completed')",
4424 ))
4425 .await
4426 .unwrap();
4427
4428 let repo = OfflineRepository::new(
4429 db_service,
4430 "test-server".to_string(),
4431 "test-user".to_string(),
4432 );
4433
4434 set_include_catalog_browse(false);
4435 let offline_only = repo.get_favorites(SearchScope::All, None).await.unwrap();
4436 let ids: Vec<&str> = offline_only.items.iter().map(|i| i.id.as_str()).collect();
4437 assert_eq!(
4438 ids,
4439 vec!["album-fav"],
4440 "with the gate off, only favourites on the device are listed"
4441 );
4442
4443 set_include_catalog_browse(true);
4444 let with_catalog = repo.get_favorites(SearchScope::All, None).await.unwrap();
4445 assert_eq!(with_catalog.items.len(), 2);
4446 }
4447
4448 #[tokio::test]
4452 async fn test_get_items_favorites_only_filters_listing() {
4453 let _guard = lock_catalog_browse();
4454 set_include_catalog_browse(true);
4455
4456 let db_service = create_test_db();
4457 seed_favorites(&db_service).await;
4458 let repo = OfflineRepository::new(
4459 db_service,
4460 "test-server".to_string(),
4461 "test-user".to_string(),
4462 );
4463
4464 let unfiltered = repo
4465 .get_items(
4466 "lib-1",
4467 Some(GetItemsOptions {
4468 include_item_types: Some(vec!["Movie".to_string()]),
4469 ..Default::default()
4470 }),
4471 )
4472 .await
4473 .unwrap();
4474 assert_eq!(unfiltered.items.len(), 2, "both movies without the filter");
4475
4476 let favourites = repo
4477 .get_items(
4478 "lib-1",
4479 Some(GetItemsOptions {
4480 include_item_types: Some(vec!["Movie".to_string()]),
4481 favorites_only: Some(true),
4482 ..Default::default()
4483 }),
4484 )
4485 .await
4486 .unwrap();
4487 let ids: Vec<&str> = favourites.items.iter().map(|i| i.id.as_str()).collect();
4488 assert_eq!(ids, vec!["movie-fav"]);
4489 }
4490
4491 #[tokio::test]
4502 async fn test_get_items_type_filter_is_bound_not_interpolated() {
4503 let _guard = lock_catalog_browse();
4504 set_include_catalog_browse(true);
4505
4506 let db_service = create_test_db();
4507 seed_favorites(&db_service).await;
4508 let repo = OfflineRepository::new(
4509 db_service,
4510 "test-server".to_string(),
4511 "test-user".to_string(),
4512 );
4513
4514 let injected = repo
4515 .get_items(
4516 "lib-1",
4517 Some(GetItemsOptions {
4518 include_item_types: Some(vec!["Movie') OR 1=1 --".to_string()]),
4519 ..Default::default()
4520 }),
4521 )
4522 .await
4523 .expect("a hostile type name must be data, not a broken query");
4524 assert!(
4525 injected.items.is_empty(),
4526 "no cached item has that type, so nothing may come back; got {:?}",
4527 injected
4528 .items
4529 .iter()
4530 .map(|i| i.id.as_str())
4531 .collect::<Vec<_>>()
4532 );
4533
4534 let quoted = repo
4536 .get_items(
4537 "lib-1",
4538 Some(GetItemsOptions {
4539 include_item_types: Some(vec!["Mo'vie".to_string()]),
4540 ..Default::default()
4541 }),
4542 )
4543 .await
4544 .expect("an embedded quote must not break the query");
4545 assert!(quoted.items.is_empty());
4546 }
4547
4548 #[tokio::test]
4555 async fn test_get_items_binds_multiple_types_in_parameter_order() {
4556 let _guard = lock_catalog_browse();
4557 set_include_catalog_browse(true);
4558
4559 let db_service = create_test_db();
4560 seed_favorites(&db_service).await;
4561 let repo = OfflineRepository::new(
4562 db_service,
4563 "test-server".to_string(),
4564 "test-user".to_string(),
4565 );
4566
4567 let both = repo
4568 .get_items(
4569 "lib-1",
4570 Some(GetItemsOptions {
4571 include_item_types: Some(vec!["Movie".to_string(), "MusicAlbum".to_string()]),
4572 ..Default::default()
4573 }),
4574 )
4575 .await
4576 .unwrap();
4577 let mut ids: Vec<&str> = both.items.iter().map(|i| i.id.as_str()).collect();
4578 ids.sort();
4579 assert_eq!(ids, vec!["album-fav", "movie-fav", "movie-plain"]);
4580
4581 let favourites = repo
4583 .get_items(
4584 "lib-1",
4585 Some(GetItemsOptions {
4586 include_item_types: Some(vec!["Movie".to_string(), "MusicAlbum".to_string()]),
4587 favorites_only: Some(true),
4588 ..Default::default()
4589 }),
4590 )
4591 .await
4592 .unwrap();
4593 let mut ids: Vec<&str> = favourites.items.iter().map(|i| i.id.as_str()).collect();
4594 ids.sort();
4595 assert_eq!(ids, vec!["album-fav", "movie-fav"]);
4596 }
4597
4598 #[tokio::test]
4607 async fn test_save_to_cache_mirrors_favorites_without_clobbering_pending() {
4608 use crate::storage::db_service::DatabaseService;
4609 let db_service = create_test_db();
4610 let repo = OfflineRepository::new(
4611 db_service.clone(),
4612 "test-server".to_string(),
4613 "test-user".to_string(),
4614 );
4615
4616 let favourite_flag = |id: &'static str| {
4617 let db = db_service.clone();
4618 async move {
4619 db.query_optional(
4620 Query::with_params(
4621 "SELECT is_favorite, pending_sync FROM user_data \
4622 WHERE user_id = ? AND item_id = ?",
4623 vec![
4624 QueryParam::String("test-user".to_string()),
4625 QueryParam::String(id.to_string()),
4626 ],
4627 ),
4628 |row| Ok((row.get::<_, Option<i32>>(0)?, row.get::<_, Option<i32>>(1)?)),
4629 )
4630 .await
4631 .unwrap()
4632 }
4633 };
4634
4635 let mut favourited = create_test_item("fav-1", "Favourited Elsewhere", None);
4637 favourited.user_data = Some(UserData {
4638 is_favorite: Some(true),
4639 ..Default::default()
4640 });
4641 let untouched = create_test_item("plain-1", "No User Data", None);
4643
4644 repo.save_to_cache("parent-1", &[favourited.clone(), untouched])
4645 .await
4646 .unwrap();
4647
4648 assert_eq!(
4649 favourite_flag("fav-1").await,
4650 Some((Some(1), Some(0))),
4651 "server favourite should be mirrored as synced"
4652 );
4653 assert_eq!(
4654 favourite_flag("plain-1").await,
4655 None,
4656 "an item without UserData should not get an invented user_data row"
4657 );
4658
4659 db_service
4661 .execute(Query::with_params(
4662 "UPDATE user_data SET is_favorite = 0, pending_sync = 1 \
4663 WHERE user_id = ? AND item_id = ?",
4664 vec![
4665 QueryParam::String("test-user".to_string()),
4666 QueryParam::String("fav-1".to_string()),
4667 ],
4668 ))
4669 .await
4670 .unwrap();
4671
4672 repo.save_to_cache("parent-1", &[favourited]).await.unwrap();
4674
4675 assert_eq!(
4676 favourite_flag("fav-1").await,
4677 Some((Some(0), Some(1))),
4678 "an unsynced local toggle must survive a cache write"
4679 );
4680 }
4681
4682 #[tokio::test]
4694 async fn test_save_to_cache_mirrors_playback_position_without_clobbering_pending() {
4695 use crate::storage::db_service::DatabaseService;
4696 let db_service = create_test_db();
4697 let repo = OfflineRepository::new(
4698 db_service.clone(),
4699 "test-server".to_string(),
4700 "test-user".to_string(),
4701 );
4702
4703 let position = |id: &'static str| {
4704 let db = db_service.clone();
4705 async move {
4706 db.query_optional(
4707 Query::with_params(
4708 "SELECT playback_position_ticks, pending_sync FROM user_data \
4709 WHERE user_id = ? AND item_id = ?",
4710 vec![
4711 QueryParam::String("test-user".to_string()),
4712 QueryParam::String(id.to_string()),
4713 ],
4714 ),
4715 |row| Ok((row.get::<_, Option<i64>>(0)?, row.get::<_, Option<i32>>(1)?)),
4716 )
4717 .await
4718 .unwrap()
4719 }
4720 };
4721
4722 let mut watched = create_test_item("ep-1", "Watched Elsewhere", None);
4724 watched.user_data = Some(UserData {
4725 playback_position_ticks: Some(12_000_000_000),
4726 ..Default::default()
4727 });
4728 let untouched = create_test_item("ep-2", "No User Data", None);
4730
4731 repo.save_to_cache("parent-1", &[watched.clone(), untouched])
4732 .await
4733 .unwrap();
4734
4735 assert_eq!(
4736 position("ep-1").await,
4737 Some((Some(12_000_000_000), Some(0))),
4738 "the server's position should be mirrored as synced"
4739 );
4740 assert_eq!(
4741 position("ep-2").await,
4742 None,
4743 "an item without UserData should not get an invented position"
4744 );
4745
4746 db_service
4748 .execute(Query::with_params(
4749 "UPDATE user_data SET playback_position_ticks = ?, pending_sync = 1 \
4750 WHERE user_id = ? AND item_id = ?",
4751 vec![
4752 QueryParam::Int64(30_000_000_000),
4753 QueryParam::String("test-user".to_string()),
4754 QueryParam::String("ep-1".to_string()),
4755 ],
4756 ))
4757 .await
4758 .unwrap();
4759
4760 repo.save_to_cache("parent-1", &[watched]).await.unwrap();
4762
4763 assert_eq!(
4764 position("ep-1").await,
4765 Some((Some(30_000_000_000), Some(1))),
4766 "an unsynced local position must not be pulled backwards"
4767 );
4768 }
4769
4770 #[tokio::test]
4780 async fn test_position_is_mirrored_even_when_no_favourite_flag_is_present() {
4781 use crate::storage::db_service::DatabaseService;
4782 let db_service = create_test_db();
4783 let repo = OfflineRepository::new(
4784 db_service.clone(),
4785 "test-server".to_string(),
4786 "test-user".to_string(),
4787 );
4788
4789 let mut watched = create_test_item("ep-3", "Position Only", None);
4790 watched.user_data = Some(UserData {
4791 is_favorite: None,
4792 playback_position_ticks: Some(9_000_000_000),
4793 ..Default::default()
4794 });
4795
4796 repo.save_to_cache("parent-1", &[watched]).await.unwrap();
4797
4798 let stored = db_service
4799 .query_optional(
4800 Query::with_params(
4801 "SELECT playback_position_ticks FROM user_data \
4802 WHERE user_id = ? AND item_id = ?",
4803 vec![
4804 QueryParam::String("test-user".to_string()),
4805 QueryParam::String("ep-3".to_string()),
4806 ],
4807 ),
4808 |row| row.get::<_, Option<i64>>(0),
4809 )
4810 .await
4811 .unwrap();
4812
4813 assert_eq!(
4814 stored,
4815 Some(Some(9_000_000_000)),
4816 "a position with no favourite flag must still be mirrored"
4817 );
4818 }
4819}