1#[cfg(test)]
11use crate::utils::lock::MutexSafe;
12use std::sync::Arc;
13
14use async_trait::async_trait;
15use log::{debug, warn};
16use tokio::time::{timeout, Duration};
17
18use super::exclusions::ExcludeHidden;
19use super::{types::*, MediaRepository, OfflineRepository, OnlineRepository};
20
21enum CacheLeg<T> {
30 Ready(Result<T, RepoError>),
32 Slow(tokio::task::JoinHandle<Result<T, RepoError>>),
34}
35
36impl<T> CacheLeg<T> {
37 #[allow(clippy::type_complexity)]
40 fn split(
41 self,
42 ) -> (
43 Option<Result<T, RepoError>>,
44 Option<tokio::task::JoinHandle<Result<T, RepoError>>>,
45 ) {
46 match self {
47 CacheLeg::Ready(result) => (Some(result), None),
48 CacheLeg::Slow(handle) => (None, Some(handle)),
49 }
50 }
51}
52
53fn log_slow_leg(
56 leg: &str,
57 parent_id: &str,
58 started: std::time::Instant,
59 result: &Result<SearchResult, RepoError>,
60) {
61 let ms = started.elapsed().as_millis();
62 let outcome = match result {
63 Ok(data) => format!("{} rows", data.items.len()),
64 Err(e) => format!("error: {e:?}"),
65 };
66 let parent = &parent_id[..8.min(parent_id.len())];
67 if ms >= 250 {
68 log::info!("[HybridRepo] get_items {leg} leg for {parent} took {ms} ms ({outcome})");
69 } else {
70 debug!("[HybridRepo] get_items {leg} leg for {parent} took {ms} ms ({outcome})");
71 }
72}
73
74enum Raced<T> {
77 Cache(T),
79 Server(T),
82 Fallback(T),
85 Failed(RepoError),
87}
88
89pub struct HybridRepository {
101 online: Arc<OnlineRepository>,
102 offline: Arc<OfflineRepository>,
103}
104
105impl HybridRepository {
106 pub fn new(online: OnlineRepository, offline: OfflineRepository) -> Self {
107 Self {
108 online: Arc::new(online),
109 offline: Arc::new(offline),
110 }
111 }
112
113 pub fn user_id(&self) -> &str {
117 self.online.user_id()
118 }
119
120 pub async fn download_bytes(&self, url: &str) -> Result<Vec<u8>, String> {
123 self.online.download_bytes(url).await
124 }
125
126 pub async fn prune_stale_catalog(
134 &self,
135 cutoff: &str,
136 item_types: &[String],
137 ) -> Result<usize, RepoError> {
138 self.offline.prune_stale_catalog(cutoff, item_types).await
139 }
140
141 pub async fn get_jray_actors(
144 &self,
145 item_id: &str,
146 t: f64,
147 ) -> Result<Vec<super::JRayActor>, RepoError> {
148 self.online.get_jray_actors(item_id, t).await
149 }
150
151 pub async fn get_video_stream_url(
158 &self,
159 item_id: &str,
160 media_source_id: Option<&str>,
161 audio_stream_index: Option<i32>,
162 ) -> Result<String, RepoError> {
163 self.online
164 .get_video_stream_url(item_id, media_source_id, audio_stream_index)
165 .await
166 }
167
168 pub async fn get_stream_selection(
176 &self,
177 item_id: &str,
178 media_source_id: Option<&str>,
179 audio_stream_index: Option<i32>,
180 ) -> Result<super::StreamSelection, RepoError> {
181 self.online
182 .get_stream_selection(item_id, media_source_id, audio_stream_index)
183 .await
184 }
185
186 pub async fn get_audio_only_stream_url_for_video(
191 &self,
192 item_id: &str,
193 media_source_id: Option<&str>,
194 start_time_seconds: Option<f64>,
195 audio_stream_index: Option<i32>,
196 ) -> Result<String, RepoError> {
197 self.online
198 .get_audio_only_stream_url_for_video(
199 item_id,
200 media_source_id,
201 start_time_seconds,
202 audio_stream_index,
203 )
204 .await
205 }
206
207 pub async fn get_album_tracks(&self, album_id: &str) -> Result<Vec<MediaItem>, RepoError> {
223 let options = Some(GetItemsOptions {
224 include_item_types: Some(vec!["Audio".to_string()]),
225 sort_by: Some("ParentIndexNumber,IndexNumber,SortName".to_string()),
226 limit: Some(1000),
227 ..Default::default()
228 });
229
230 let result = self.online.get_items(album_id, options).await?;
231
232 if !result.items.is_empty() {
233 if let Err(e) = self.offline.save_to_cache(album_id, &result.items).await {
234 warn!("[HybridRepo] Failed to cache album tracks: {:?}", e);
235 }
236 }
237
238 Ok(result.items)
239 }
240
241 pub async fn get_items_unfiltered(
252 &self,
253 parent_id: &str,
254 options: Option<GetItemsOptions>,
255 ) -> Result<SearchResult, RepoError> {
256 match self.online.get_items(parent_id, options.clone()).await {
257 Ok(result) => Ok(result),
258 Err(e) => self.offline.get_items(parent_id, options).await.or(Err(e)),
259 }
260 }
261
262 pub async fn search_cache_only(
268 &self,
269 query: &str,
270 options: Option<SearchOptions>,
271 ) -> Result<SearchResult, RepoError> {
272 let offline = Arc::clone(&self.offline);
273 let query = query.to_string();
274 offline
277 .search(&query, options)
278 .await
279 .map(ExcludeHidden::without_excluded)
280 }
281
282 pub async fn get_favorites_cache_only(
287 &self,
288 scope: SearchScope,
289 options: Option<GetItemsOptions>,
290 ) -> Result<SearchResult, RepoError> {
291 let offline = Arc::clone(&self.offline);
292 offline
294 .get_favorites(scope, options)
295 .await
296 .map(ExcludeHidden::without_excluded)
297 }
298
299 pub async fn get_favorites_server_only(
305 &self,
306 scope: SearchScope,
307 options: Option<GetItemsOptions>,
308 ) -> Result<SearchResult, RepoError> {
309 let result = self.online.get_favorites(scope, options).await?;
310 if !result.items.is_empty() {
311 if let Err(e) = self.offline.save_to_cache("favorites", &result.items).await {
314 debug!("[HybridRepo] Failed to cache favourites: {:?}", e);
315 }
316 }
317 Ok(result.without_excluded())
318 }
319
320 pub async fn cache_items_from_server(
329 &self,
330 parent_id: &str,
331 options: Option<GetItemsOptions>,
332 ) -> Result<Vec<MediaItem>, RepoError> {
333 let result = self.online.get_items(parent_id, options).await?;
334 if !result.items.is_empty() {
335 self.offline.save_to_cache(parent_id, &result.items).await?;
336 }
337 Ok(result.items)
338 }
339
340 pub async fn get_downloaded_items(
349 &self,
350 parent_id: &str,
351 options: Option<GetItemsOptions>,
352 ) -> Result<SearchResult, RepoError> {
353 self.offline.get_downloaded_items(parent_id, options).await
358 }
359
360 pub async fn get_downloaded_libraries(&self) -> Result<Vec<Library>, RepoError> {
364 self.offline.get_downloaded_libraries().await
365 }
366
367 pub async fn get_download_disk_usage(&self) -> Result<DownloadDiskUsage, RepoError> {
371 self.offline.get_download_disk_usage().await
372 }
373
374 pub async fn search_server_only(
376 &self,
377 query: &str,
378 options: Option<SearchOptions>,
379 ) -> Result<SearchResult, RepoError> {
380 self.online
381 .search(query, options)
382 .await
383 .map(ExcludeHidden::without_excluded)
384 }
385
386 pub fn merge_search_results(cache: SearchResult, server: SearchResult) -> SearchResult {
392 use std::collections::HashMap;
393
394 let mut server_by_id: HashMap<String, MediaItem> = HashMap::new();
397 let mut server_order: Vec<String> = Vec::with_capacity(server.items.len());
398 for item in server.items {
399 if !server_by_id.contains_key(&item.id) {
400 server_order.push(item.id.clone());
401 }
402 server_by_id.insert(item.id.clone(), item);
403 }
404
405 let mut items: Vec<MediaItem> = Vec::new();
406 let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
407
408 for local in cache.items {
411 if !seen.insert(local.id.clone()) {
412 continue;
413 }
414 match server_by_id.remove(&local.id) {
415 Some(server_item) => items.push(server_item),
416 None => items.push(local),
417 }
418 }
419
420 for id in server_order {
422 if let Some(server_item) = server_by_id.remove(&id) {
423 if seen.insert(id) {
424 items.push(server_item);
425 }
426 }
427 }
428
429 let total_record_count = items.len();
430 SearchResult {
431 items,
432 total_record_count,
433 }
434 }
435
436 async fn race_slow_cache<T, F>(
449 mut slow: tokio::task::JoinHandle<Result<T, RepoError>>,
450 server: F,
451 ) -> Raced<T>
452 where
453 T: MeaningfulContent + ExcludeHidden,
454 F: std::future::Future<Output = Result<T, RepoError>>,
455 {
456 tokio::pin!(server);
457 tokio::select! {
460 biased;
461 server_result = &mut server => match server_result {
462 Ok(data) => Raced::Server(data),
463 Err(e) => {
464 debug!("[HybridRepo] Server failed; waiting for the slow cache query");
465 match slow.await {
466 Ok(Ok(data)) => Raced::Fallback(data.without_excluded()),
467 _ => Raced::Failed(e),
468 }
469 }
470 },
471 cache_result = &mut slow => {
472 let cache = cache_result
473 .unwrap_or_else(|join| Err(RepoError::Database {
474 message: format!("Cache query failed: {join}"),
475 }))
476 .map(ExcludeHidden::without_excluded);
477 match cache {
478 Ok(data) if data.has_content() => {
479 debug!("[HybridRepo] Slow cache answered before the server");
480 Raced::Cache(data)
481 }
482 other => match server.await {
483 Ok(data) => Raced::Server(data),
484 Err(e) => match other {
485 Ok(data) => Raced::Fallback(data),
486 Err(_) => Raced::Failed(e),
487 },
488 },
489 }
490 }
491 }
492 }
493
494 async fn parallel_race<T, F2>(cache: CacheLeg<T>, server_future: F2) -> Result<T, RepoError>
514 where
515 T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static,
516 F2: std::future::Future<Output = Result<T, RepoError>> + Send,
517 {
518 Self::race_with_refresh(cache, server_future, || {}).await
519 }
520
521 async fn race_with_refresh<T, F2, R>(
536 cache: CacheLeg<T>,
537 server_future: F2,
538 on_cache_hit: R,
539 ) -> Result<T, RepoError>
540 where
541 T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static,
542 F2: std::future::Future<Output = Result<T, RepoError>> + Send,
543 R: FnOnce(),
544 {
545 let (fast, slow) = cache.split();
546 let fast = fast.map(|r| r.map(ExcludeHidden::without_excluded));
547
548 if let Some(Ok(data)) = &fast {
549 if data.has_content() {
550 debug!("[HybridRepo] Cache hit, returning immediately (refreshing in background)");
551 on_cache_hit();
552 return Ok(data.clone());
553 }
554 }
555
556 if let Some(handle) = slow {
561 return match Self::race_slow_cache(handle, server_future).await {
562 Raced::Cache(data) => {
563 on_cache_hit();
564 Ok(data)
565 }
566 Raced::Server(data) => Ok(data.without_excluded()),
567 Raced::Fallback(data) => Ok(data),
568 Raced::Failed(e) => Err(e),
569 };
570 }
571
572 debug!("[HybridRepo] Cache miss, querying server");
573 match server_future.await {
574 Ok(data) => Ok(data.without_excluded()),
575 Err(e) => fast.unwrap_or(Err(e)),
578 }
579 }
580
581 const CACHE_FAST_PATH: Duration = Duration::from_millis(100);
585
586 async fn cache_leg<T>(
600 future: impl std::future::Future<Output = Result<T, RepoError>> + Send + 'static,
601 ) -> CacheLeg<T>
602 where
603 T: Send + 'static,
604 {
605 let mut handle = tokio::spawn(future);
608 match timeout(Self::CACHE_FAST_PATH, &mut handle).await {
609 Ok(Ok(result)) => CacheLeg::Ready(result),
610 Ok(Err(join)) => CacheLeg::Ready(Err(RepoError::Database {
611 message: format!("Cache query failed: {join}"),
612 })),
613 Err(_) => {
614 debug!("[HybridRepo] Cache missed the fast path; leaving it running");
615 CacheLeg::Slow(handle)
616 }
617 }
618 }
619
620 async fn cache_try<T>(
632 future: impl std::future::Future<Output = Result<T, RepoError>> + Send + 'static,
633 ) -> (
634 Result<T, RepoError>,
635 Option<tokio::task::JoinHandle<Result<T, RepoError>>>,
636 )
637 where
638 T: Send + 'static,
639 {
640 let (fast, slow) = Self::cache_leg(future).await.split();
641 let fast = fast.unwrap_or_else(|| {
642 Err(RepoError::Database {
643 message: "Cache query still running".to_string(),
644 })
645 });
646 (fast, slow)
647 }
648
649 async fn settle<T>(
654 fast: Result<T, RepoError>,
655 slow: Option<tokio::task::JoinHandle<Result<T, RepoError>>>,
656 server_err: RepoError,
657 ) -> Result<T, RepoError> {
658 if let Some(handle) = slow {
659 debug!("[HybridRepo] Server failed; waiting for the slow cache read");
660 return match handle.await {
661 Ok(Ok(data)) => Ok(data),
662 _ => Err(server_err),
663 };
664 }
665 fast.or(Err(server_err))
666 }
667
668 fn save_when_server_answers(
672 server: tokio::task::JoinHandle<Result<SearchResult, RepoError>>,
673 offline: Arc<OfflineRepository>,
674 parent_id: String,
675 ) {
676 tokio::spawn(async move {
677 if let Ok(Ok(server_data)) = server.await {
678 Self::save_in_background(offline, parent_id, &server_data);
679 }
680 });
681 }
682
683 fn save_in_background(
686 offline: Arc<OfflineRepository>,
687 parent_id: String,
688 server_data: &SearchResult,
689 ) {
690 if server_data.items.is_empty() {
691 return;
692 }
693 let items = server_data.items.clone();
694 tokio::spawn(async move {
695 match offline.save_to_cache(&parent_id, &items).await {
696 Ok(_) => debug!(
697 "[HybridRepo] Cached {} items for parent {}",
698 items.len(),
699 &parent_id[..8.min(parent_id.len())]
700 ),
701 Err(e) => warn!(
702 "[HybridRepo] Failed to cache {} items: {:?}",
703 items.len(),
704 e
705 ),
706 }
707 });
708 }
709
710 async fn cache_or(
713 fast: Result<SearchResult, RepoError>,
714 slow: Option<tokio::task::JoinHandle<Result<SearchResult, RepoError>>>,
715 server_err: RepoError,
716 ) -> Result<SearchResult, RepoError> {
717 Self::settle(fast, slow, server_err)
718 .await
719 .map(ExcludeHidden::without_excluded)
720 }
721}
722
723#[async_trait]
724impl MediaRepository for HybridRepository {
725 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
726 let offline_read = Arc::clone(&self.offline);
730 let (cache_result, slow_cache) =
731 Self::cache_try(async move { offline_read.get_libraries().await }).await;
732
733 if let Ok(libs) = &cache_result {
734 if libs.has_content() {
735 debug!("[HybridRepo] Cache hit for libraries, returning immediately");
736 let online = Arc::clone(&self.online);
737 let offline = Arc::clone(&self.offline);
738 tokio::spawn(async move {
739 if let Ok(server_libs) = online.get_libraries().await {
740 if !server_libs.is_empty() {
741 if let Err(e) = offline.save_libraries_to_cache(&server_libs).await {
742 warn!(
743 "[HybridRepo] Background library cache update failed: {:?}",
744 e
745 );
746 }
747 }
748 }
749 });
750 return cache_result;
751 }
752 }
753
754 match self.online.get_libraries().await {
756 Ok(server_libs) => {
757 if !server_libs.is_empty() {
758 if let Err(e) = self.offline.save_libraries_to_cache(&server_libs).await {
759 warn!(
760 "[HybridRepo] Failed to cache {} libraries: {:?}",
761 server_libs.len(),
762 e
763 );
764 }
765 }
766 Ok(server_libs)
767 }
768 Err(e) => Self::settle(cache_result, slow_cache, e).await,
770 }
771 }
772
773 async fn get_items(
774 &self,
775 parent_id: &str,
776 options: Option<GetItemsOptions>,
777 ) -> Result<SearchResult, RepoError> {
778 let offline = Arc::clone(&self.offline);
779 let offline_for_save = Arc::clone(&self.offline);
780 let online = Arc::clone(&self.online);
781 let parent_id = parent_id.to_string();
782 let parent_id_clone = parent_id.clone();
783 let parent_id_for_save = parent_id.clone();
784 let opts_clone = options.clone();
785
786 let mut server_handle = tokio::spawn(async move {
788 let started = std::time::Instant::now();
789 let result = online.get_items(&parent_id_clone, options).await;
790 log_slow_leg("server", &parent_id_clone, started, &result);
791 result
792 });
793
794 let (cache_result, slow_cache) = Self::cache_try(async move {
810 let started = std::time::Instant::now();
811 let result = offline.get_items(&parent_id, opts_clone).await;
812 log_slow_leg("cache", &parent_id, started, &result);
813 result
814 })
815 .await;
816 let cache_result = cache_result.map(ExcludeHidden::without_excluded);
817
818 if !crate::repository::offline::include_catalog_browse() {
826 if let Ok(data) = &cache_result {
827 debug!(
828 "[HybridRepo] Downloads-only gate: returning offline result ({} items) as authoritative for parent {}",
829 data.items.len(),
830 &parent_id_for_save[..8.min(parent_id_for_save.len())]
831 );
832 server_handle.abort();
834 return Ok(data.clone());
835 }
836 }
837
838 if let Ok(data) = &cache_result {
840 if data.has_content() {
841 debug!(
842 "[HybridRepo] Cache hit for get_items, returning immediately for parent {}",
843 &parent_id_for_save[..8.min(parent_id_for_save.len())]
844 );
845 Self::save_when_server_answers(server_handle, offline_for_save, parent_id_for_save);
846 return Ok(data.clone());
847 }
848 }
849
850 if let Some(slow) = slow_cache {
853 let server = async {
854 (&mut server_handle).await.unwrap_or_else(|join| {
855 Err(RepoError::Network {
856 message: format!("Server task failed: {}", join),
857 })
858 })
859 };
860 return match Self::race_slow_cache(slow, server).await {
861 Raced::Cache(data) => {
862 Self::save_when_server_answers(
865 server_handle,
866 offline_for_save,
867 parent_id_for_save,
868 );
869 Ok(data)
870 }
871 Raced::Server(server_data) => {
872 Self::save_in_background(offline_for_save, parent_id_for_save, &server_data);
873 Ok(server_data.without_excluded())
874 }
875 Raced::Fallback(data) => Ok(data),
876 Raced::Failed(e) => Err(e),
877 };
878 }
879
880 match server_handle.await {
882 Ok(Ok(server_data)) => {
883 Self::save_in_background(offline_for_save, parent_id_for_save, &server_data);
884 Ok(server_data.without_excluded())
890 }
891 Ok(Err(e)) => Self::cache_or(cache_result, None, e).await,
892 Err(join_err) => {
893 let e = RepoError::Network {
894 message: format!("Server task failed: {}", join_err),
895 };
896 Self::cache_or(cache_result, None, e).await
897 }
898 }
899 }
900
901 async fn get_item(&self, item_id: &str) -> Result<MediaItem, RepoError> {
917 let offline = Arc::clone(&self.offline);
918 let online = Arc::clone(&self.online);
919 let item_id = item_id.to_string();
920 let item_id_clone = item_id.clone();
921
922 let cache_future = Self::cache_leg(async move { offline.get_item(&item_id).await }).await;
923
924 let online_for_refresh = Arc::clone(&self.online);
925 let offline_for_save = Arc::clone(&self.offline);
926 let refresh_id = item_id_clone.clone();
927 let on_cache_hit = move || {
928 tokio::spawn(async move {
929 match online_for_refresh.get_item(&refresh_id).await {
930 Ok(fresh) => {
931 let parent = fresh
934 .parent_id
935 .clone()
936 .unwrap_or_else(|| "item".to_string());
937 if let Err(e) = offline_for_save.save_to_cache(&parent, &[fresh]).await {
938 debug!("[HybridRepo] Background item refresh failed: {:?}", e);
939 }
940 }
941 Err(e) => debug!("[HybridRepo] Background item refresh unavailable: {:?}", e),
942 }
943 });
944 };
945
946 let server_future = async move { online.get_item(&item_id_clone).await };
947
948 Self::race_with_refresh(cache_future, server_future, on_cache_hit).await
949 }
950
951 async fn get_latest_items(
952 &self,
953 parent_id: &str,
954 limit: Option<usize>,
955 ) -> Result<Vec<MediaItem>, RepoError> {
956 let offline = Arc::clone(&self.offline);
957 let online = Arc::clone(&self.online);
958 let parent_id = parent_id.to_string();
959 let parent_id_clone = parent_id.clone();
960 let limit_clone = limit;
961
962 let cache_future =
963 Self::cache_leg(async move { offline.get_latest_items(&parent_id, limit).await }).await;
964
965 let server_future =
966 async move { online.get_latest_items(&parent_id_clone, limit_clone).await };
967
968 Self::parallel_race(cache_future, server_future).await
969 }
970
971 async fn get_resume_items(
972 &self,
973 parent_id: Option<&str>,
974 limit: Option<usize>,
975 ) -> Result<Vec<MediaItem>, RepoError> {
976 let offline = Arc::clone(&self.offline);
977 let online = Arc::clone(&self.online);
978 let parent_id_str = parent_id.map(|s| s.to_string());
979 let parent_id_clone = parent_id_str.clone();
980 let limit_clone = limit;
981
982 let cache_future = Self::cache_leg(async move {
983 offline
984 .get_resume_items(parent_id_str.as_deref(), limit)
985 .await
986 })
987 .await;
988
989 let server_future = async move {
990 online
991 .get_resume_items(parent_id_clone.as_deref(), limit_clone)
992 .await
993 };
994
995 Self::parallel_race(cache_future, server_future).await
996 }
997
998 async fn get_next_up_episodes(
999 &self,
1000 series_id: Option<&str>,
1001 limit: Option<usize>,
1002 ) -> Result<Vec<MediaItem>, RepoError> {
1003 let offline = Arc::clone(&self.offline);
1013 let online = Arc::clone(&self.online);
1014 let series = series_id.map(str::to_string);
1015 let series_for_server = series.clone();
1016
1017 let cache_future =
1018 Self::cache_leg(
1019 async move { offline.get_next_up_episodes(series.as_deref(), limit).await },
1020 )
1021 .await;
1022 let server_future = async move {
1023 online
1024 .get_next_up_episodes(series_for_server.as_deref(), limit)
1025 .await
1026 };
1027
1028 Self::parallel_race(cache_future, server_future).await
1029 }
1030
1031 async fn get_recently_played_audio(
1032 &self,
1033 limit: Option<usize>,
1034 ) -> Result<Vec<MediaItem>, RepoError> {
1035 let offline = Arc::clone(&self.offline);
1036 let online = Arc::clone(&self.online);
1037 let limit_clone = limit;
1038
1039 let cache_future =
1040 Self::cache_leg(async move { offline.get_recently_played_audio(limit).await }).await;
1041
1042 let server_future = async move { online.get_recently_played_audio(limit_clone).await };
1043
1044 Self::parallel_race(cache_future, server_future).await
1045 }
1046
1047 async fn get_resume_movies(&self, limit: Option<usize>) -> Result<Vec<MediaItem>, RepoError> {
1048 let offline = Arc::clone(&self.offline);
1049 let online = Arc::clone(&self.online);
1050 let limit_clone = limit;
1051
1052 let cache_future =
1053 Self::cache_leg(async move { offline.get_resume_movies(limit).await }).await;
1054
1055 let server_future = async move { online.get_resume_movies(limit_clone).await };
1056
1057 Self::parallel_race(cache_future, server_future).await
1058 }
1059
1060 async fn get_rediscover_albums(
1061 &self,
1062 parent_id: Option<&str>,
1063 limit: Option<usize>,
1064 ) -> Result<Vec<MediaItem>, RepoError> {
1065 let offline = Arc::clone(&self.offline);
1066 let online = Arc::clone(&self.online);
1067 let parent_id_owned = parent_id.map(|s| s.to_string());
1068 let parent_id_clone = parent_id_owned.clone();
1069
1070 let cache_future = Self::cache_leg(async move {
1071 offline
1072 .get_rediscover_albums(parent_id_owned.as_deref(), limit)
1073 .await
1074 })
1075 .await;
1076
1077 let server_future = async move {
1078 online
1079 .get_rediscover_albums(parent_id_clone.as_deref(), limit)
1080 .await
1081 };
1082
1083 Self::parallel_race(cache_future, server_future).await
1084 }
1085
1086 async fn get_genres(&self, parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
1087 let parent_id_str = parent_id.map(|s| s.to_string());
1093
1094 let cache_offline = Arc::clone(&self.offline);
1095 let cache_pid = parent_id_str.clone();
1096 let (cache_result, slow_cache) =
1097 Self::cache_try(async move { cache_offline.get_genres(cache_pid.as_deref()).await })
1098 .await;
1099
1100 if let Ok(genres) = &cache_result {
1101 if genres.has_content() {
1102 debug!("[HybridRepo] Cache hit for genres, returning immediately");
1103 let online = Arc::clone(&self.online);
1104 let offline = Arc::clone(&self.offline);
1105 let pid = parent_id_str.clone();
1106 tokio::spawn(async move {
1107 if let Ok(server_genres) = online.get_genres(pid.as_deref()).await {
1108 if !server_genres.is_empty() {
1109 if let Err(e) = offline
1110 .save_genres_to_cache(pid.as_deref(), &server_genres)
1111 .await
1112 {
1113 warn!("[HybridRepo] Background genre cache update failed: {:?}", e);
1114 }
1115 }
1116 }
1117 });
1118 return cache_result;
1119 }
1120 }
1121
1122 match self.online.get_genres(parent_id_str.as_deref()).await {
1124 Ok(server_genres) => {
1125 if !server_genres.is_empty() {
1126 if let Err(e) = self
1127 .offline
1128 .save_genres_to_cache(parent_id_str.as_deref(), &server_genres)
1129 .await
1130 {
1131 warn!(
1132 "[HybridRepo] Failed to cache {} genres: {:?}",
1133 server_genres.len(),
1134 e
1135 );
1136 }
1137 }
1138 Ok(server_genres)
1139 }
1140 Err(e) => Self::settle(cache_result, slow_cache, e).await,
1141 }
1142 }
1143
1144 async fn search(
1145 &self,
1146 query: &str,
1147 options: Option<SearchOptions>,
1148 ) -> Result<SearchResult, RepoError> {
1149 let offline = Arc::clone(&self.offline);
1150 let online = Arc::clone(&self.online);
1151 let query = query.to_string();
1152 let query_clone = query.clone();
1153 let opts_clone = options.clone();
1154
1155 let cache_future =
1156 Self::cache_leg(async move { offline.search(&query, opts_clone).await }).await;
1157
1158 let server_future = async move { online.search(&query_clone, options).await };
1159
1160 Self::parallel_race(cache_future, server_future).await
1161 }
1162
1163 async fn get_playback_info(&self, item_id: &str) -> Result<PlaybackInfo, RepoError> {
1164 if let Some(local) = self.offline.local_playback_info(item_id).await? {
1171 return Ok(local);
1172 }
1173 self.online.get_playback_info(item_id).await
1174 }
1175
1176 async fn get_audio_stream_url(&self, item_id: &str) -> Result<String, RepoError> {
1177 self.online.get_audio_stream_url(item_id).await
1179 }
1180
1181 async fn get_audio_only_stream_url_for_video(
1182 &self,
1183 item_id: &str,
1184 media_source_id: Option<&str>,
1185 start_time_seconds: Option<f64>,
1186 audio_stream_index: Option<i32>,
1187 ) -> Result<String, RepoError> {
1188 self.online
1190 .build_audio_only_stream_url_for_video(
1191 item_id,
1192 media_source_id,
1193 start_time_seconds,
1194 audio_stream_index,
1195 )
1196 .await
1197 }
1198
1199 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
1200 self.online
1202 .get_live_tv_channels()
1203 .await
1204 .map(ExcludeHidden::without_excluded)
1205 }
1206
1207 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
1208 self.online
1210 .get_channels()
1211 .await
1212 .map(ExcludeHidden::without_excluded)
1213 }
1214
1215 async fn open_live_stream(&self, item_id: &str) -> Result<LiveStreamInfo, RepoError> {
1216 self.online.open_live_stream(item_id).await
1218 }
1219
1220 async fn report_playback_start(
1221 &self,
1222 item_id: &str,
1223 position_ticks: i64,
1224 ) -> Result<(), RepoError> {
1225 self.online
1227 .report_playback_start(item_id, position_ticks)
1228 .await
1229 }
1230
1231 async fn report_playback_progress(
1232 &self,
1233 item_id: &str,
1234 position_ticks: i64,
1235 ) -> Result<(), RepoError> {
1236 self.online
1238 .report_playback_progress(item_id, position_ticks)
1239 .await
1240 }
1241
1242 async fn report_playback_stopped(
1243 &self,
1244 item_id: &str,
1245 position_ticks: i64,
1246 ) -> Result<(), RepoError> {
1247 self.online
1249 .report_playback_stopped(item_id, position_ticks)
1250 .await
1251 }
1252
1253 fn get_image_url(
1254 &self,
1255 item_id: &str,
1256 image_type: ImageType,
1257 options: Option<ImageOptions>,
1258 ) -> String {
1259 self.online.get_image_url(item_id, image_type, options)
1261 }
1262
1263 fn get_subtitle_url(
1264 &self,
1265 item_id: &str,
1266 media_source_id: &str,
1267 stream_index: i32,
1268 format: &str,
1269 ) -> String {
1270 self.online
1272 .get_subtitle_url(item_id, media_source_id, stream_index, format)
1273 }
1274
1275 fn get_video_download_url(
1276 &self,
1277 item_id: &str,
1278 quality: &str,
1279 media_source_id: Option<&str>,
1280 source_audio_codec: Option<&str>,
1281 ) -> String {
1282 self.online
1284 .get_video_download_url(item_id, quality, media_source_id, source_audio_codec)
1285 }
1286
1287 async fn mark_favorite(&self, item_id: &str) -> Result<(), RepoError> {
1288 self.online.mark_favorite(item_id).await
1290 }
1291
1292 async fn unmark_favorite(&self, item_id: &str) -> Result<(), RepoError> {
1293 self.online.unmark_favorite(item_id).await
1295 }
1296
1297 async fn clear_watch_history(&self, item_id: &str) -> Result<(), RepoError> {
1298 self.online.clear_watch_history(item_id).await
1300 }
1301
1302 async fn mark_played(&self, item_id: &str) -> Result<(), RepoError> {
1303 self.online.mark_played(item_id).await
1305 }
1306
1307 async fn get_person(&self, person_id: &str) -> Result<MediaItem, RepoError> {
1308 let offline = Arc::clone(&self.offline);
1309 let online = Arc::clone(&self.online);
1310 let person_id = person_id.to_string();
1311 let person_id_clone = person_id.clone();
1312
1313 let cache_future =
1314 Self::cache_leg(async move { offline.get_person(&person_id).await }).await;
1315
1316 let server_future = async move { online.get_person(&person_id_clone).await };
1317
1318 Self::parallel_race(cache_future, server_future).await
1319 }
1320
1321 async fn get_items_by_person(
1322 &self,
1323 person_id: &str,
1324 options: Option<GetItemsOptions>,
1325 ) -> Result<SearchResult, RepoError> {
1326 let offline = Arc::clone(&self.offline);
1327 let online = Arc::clone(&self.online);
1328 let person_id = person_id.to_string();
1329 let person_id_clone = person_id.clone();
1330 let opts_clone = options.clone();
1331
1332 let cache_future =
1333 Self::cache_leg(
1334 async move { offline.get_items_by_person(&person_id, opts_clone).await },
1335 )
1336 .await;
1337
1338 let server_future =
1339 async move { online.get_items_by_person(&person_id_clone, options).await };
1340
1341 Self::parallel_race(cache_future, server_future).await
1342 }
1343
1344 async fn get_favorites(
1346 &self,
1347 scope: SearchScope,
1348 options: Option<GetItemsOptions>,
1349 ) -> Result<SearchResult, RepoError> {
1350 let cache_result = self.get_favorites_cache_only(scope, options.clone()).await;
1351
1352 if !crate::repository::offline::include_catalog_browse() {
1357 if let Ok(data) = &cache_result {
1358 return Ok(data.clone());
1359 }
1360 }
1361
1362 if let Ok(data) = &cache_result {
1363 if data.has_content() {
1364 return Ok(data.clone());
1365 }
1366 }
1367
1368 match self.get_favorites_server_only(scope, options).await {
1374 Ok(data) => Ok(data),
1375 Err(e) => cache_result.or(Err(e)),
1376 }
1377 }
1378
1379 async fn get_similar_items(
1380 &self,
1381 item_id: &str,
1382 limit: Option<usize>,
1383 ) -> Result<SearchResult, RepoError> {
1384 let offline = Arc::clone(&self.offline);
1385 let online = Arc::clone(&self.online);
1386 let item_id = item_id.to_string();
1387 let item_id_clone = item_id.clone();
1388
1389 let cache_future =
1390 Self::cache_leg(async move { offline.get_similar_items(&item_id, limit).await }).await;
1391
1392 let server_future = async move { online.get_similar_items(&item_id_clone, limit).await };
1393
1394 Self::parallel_race(cache_future, server_future).await
1395 }
1396
1397 async fn create_playlist(
1400 &self,
1401 name: &str,
1402 item_ids: &[String],
1403 ) -> Result<PlaylistCreatedResult, RepoError> {
1404 self.online.create_playlist(name, item_ids).await
1406 }
1407
1408 async fn delete_playlist(&self, playlist_id: &str) -> Result<(), RepoError> {
1409 self.online.delete_playlist(playlist_id).await
1411 }
1412
1413 async fn rename_playlist(&self, playlist_id: &str, name: &str) -> Result<(), RepoError> {
1414 self.online.rename_playlist(playlist_id, name).await
1416 }
1417
1418 async fn get_playlist_items(&self, playlist_id: &str) -> Result<Vec<PlaylistEntry>, RepoError> {
1419 let offline = Arc::clone(&self.offline);
1420 let offline_for_save = Arc::clone(&self.offline);
1421 let online = Arc::clone(&self.online);
1422 let playlist_id = playlist_id.to_string();
1423 let playlist_id_clone = playlist_id.clone();
1424 let playlist_id_for_save = playlist_id.clone();
1425
1426 let server_handle =
1428 tokio::spawn(async move { online.get_playlist_items(&playlist_id_clone).await });
1429
1430 let (cache_result, slow_cache) =
1432 Self::cache_try(async move { offline.get_playlist_items(&playlist_id).await }).await;
1433
1434 if let Ok(data) = &cache_result {
1436 if data.has_content() {
1437 debug!("[HybridRepo] Cache hit for playlist items, returning immediately");
1438 tokio::spawn(async move {
1439 if let Ok(Ok(server_entries)) = server_handle.await {
1440 if let Err(e) = offline_for_save
1441 .save_playlist_items_to_cache(&playlist_id_for_save, &server_entries)
1442 .await
1443 {
1444 warn!("[HybridRepo] Failed to update playlist cache: {:?}", e);
1445 }
1446 }
1447 });
1448 return cache_result;
1449 }
1450 }
1451
1452 match server_handle.await {
1454 Ok(Ok(entries)) => {
1455 let entries_clone = entries.clone();
1456 tokio::spawn(async move {
1457 if let Err(e) = offline_for_save
1458 .save_playlist_items_to_cache(&playlist_id_for_save, &entries_clone)
1459 .await
1460 {
1461 warn!(
1462 "[HybridRepo] Failed to save playlist items to cache: {:?}",
1463 e
1464 );
1465 }
1466 });
1467 Ok(entries)
1468 }
1469 Ok(Err(e)) => Self::settle(cache_result, slow_cache, e).await,
1470 Err(join_err) => {
1471 let e = RepoError::Network {
1472 message: format!("Server task failed: {}", join_err),
1473 };
1474 Self::settle(cache_result, slow_cache, e).await
1475 }
1476 }
1477 }
1478
1479 async fn add_to_playlist(
1480 &self,
1481 playlist_id: &str,
1482 item_ids: &[String],
1483 ) -> Result<(), RepoError> {
1484 self.online.add_to_playlist(playlist_id, item_ids).await
1486 }
1487
1488 async fn remove_from_playlist(
1489 &self,
1490 playlist_id: &str,
1491 entry_ids: &[String],
1492 ) -> Result<(), RepoError> {
1493 self.online
1495 .remove_from_playlist(playlist_id, entry_ids)
1496 .await
1497 }
1498
1499 async fn move_playlist_item(
1500 &self,
1501 playlist_id: &str,
1502 item_id: &str,
1503 new_index: u32,
1504 ) -> Result<(), RepoError> {
1505 self.online
1507 .move_playlist_item(playlist_id, item_id, new_index)
1508 .await
1509 }
1510}
1511
1512#[cfg(test)]
1513mod tests {
1514 #![allow(clippy::await_holding_lock)]
1523
1524 use super::*;
1525 use std::sync::Mutex;
1526
1527 #[tokio::test]
1537 async fn a_slow_cache_still_answers_when_the_server_is_gone() {
1538 let cache = HybridRepository::cache_leg(async {
1539 tokio::time::sleep(Duration::from_millis(250)).await;
1540 Ok(vec![MediaItem {
1541 id: "cached-item".to_string(),
1542 ..Default::default()
1543 }])
1544 })
1545 .await;
1546
1547 let server = async {
1548 Err(RepoError::Network {
1549 message: "offline".to_string(),
1550 })
1551 };
1552
1553 let got = HybridRepository::parallel_race(cache, server)
1554 .await
1555 .expect("a slow cache read must still be delivered when the server is gone");
1556 assert_eq!(got.len(), 1);
1557 assert_eq!(got[0].id, "cached-item");
1558 }
1559
1560 #[tokio::test]
1564 async fn a_fast_cache_hit_never_reaches_the_server() {
1565 let cache = HybridRepository::cache_leg(async {
1566 Ok(vec![MediaItem {
1567 id: "fast".to_string(),
1568 ..Default::default()
1569 }])
1570 })
1571 .await;
1572
1573 let server = async {
1574 panic!("the server leg must not run on a cache hit");
1575 };
1576
1577 let got = HybridRepository::parallel_race(cache, server)
1578 .await
1579 .unwrap();
1580 assert_eq!(got[0].id, "fast");
1581 }
1582
1583 #[tokio::test]
1587 async fn a_failing_slow_cache_reports_the_server_error() {
1588 let cache: CacheLeg<Vec<MediaItem>> = HybridRepository::cache_leg(async {
1589 tokio::time::sleep(Duration::from_millis(250)).await;
1590 Err(RepoError::Database {
1591 message: "disk gone".to_string(),
1592 })
1593 })
1594 .await;
1595
1596 let server = async {
1597 Err(RepoError::Network {
1598 message: "offline".to_string(),
1599 })
1600 };
1601
1602 let err = HybridRepository::parallel_race(cache, server)
1603 .await
1604 .unwrap_err();
1605 assert!(matches!(err, RepoError::Network { .. }), "got {err:?}");
1606 }
1607
1608 #[tokio::test]
1618 async fn a_slow_cache_that_beats_the_server_wins() {
1619 let cache = HybridRepository::cache_leg(async {
1620 tokio::time::sleep(Duration::from_millis(150)).await;
1621 Ok(vec![MediaItem {
1622 id: "slow-cache".to_string(),
1623 ..Default::default()
1624 }])
1625 })
1626 .await;
1627
1628 let server = async {
1629 tokio::time::sleep(Duration::from_millis(1500)).await;
1630 Ok(vec![MediaItem {
1631 id: "server".to_string(),
1632 ..Default::default()
1633 }])
1634 };
1635
1636 let started = std::time::Instant::now();
1637 let got = HybridRepository::parallel_race(cache, server)
1638 .await
1639 .unwrap();
1640 assert_eq!(
1641 got[0].id, "slow-cache",
1642 "waited for the server over a cache answer"
1643 );
1644 assert!(started.elapsed() < Duration::from_millis(1000));
1645 }
1646
1647 #[tokio::test]
1651 async fn a_server_that_beats_a_slow_cache_wins() {
1652 let cache = HybridRepository::cache_leg(async {
1653 tokio::time::sleep(Duration::from_millis(1500)).await;
1654 Ok(vec![MediaItem {
1655 id: "slow-cache".to_string(),
1656 ..Default::default()
1657 }])
1658 })
1659 .await;
1660
1661 let server = async {
1662 tokio::time::sleep(Duration::from_millis(150)).await;
1663 Ok(vec![MediaItem {
1664 id: "server".to_string(),
1665 ..Default::default()
1666 }])
1667 };
1668
1669 let started = std::time::Instant::now();
1670 let got = HybridRepository::parallel_race(cache, server)
1671 .await
1672 .unwrap();
1673 assert_eq!(got[0].id, "server");
1674 assert!(started.elapsed() < Duration::from_millis(1000));
1675 }
1676
1677 #[tokio::test]
1682 async fn an_empty_slow_cache_defers_to_the_server() {
1683 let cache = HybridRepository::cache_leg(async {
1684 tokio::time::sleep(Duration::from_millis(150)).await;
1685 Ok(Vec::<MediaItem>::new())
1686 })
1687 .await;
1688
1689 let server = async {
1690 tokio::time::sleep(Duration::from_millis(400)).await;
1691 Ok(vec![MediaItem {
1692 id: "server".to_string(),
1693 ..Default::default()
1694 }])
1695 };
1696
1697 let got = HybridRepository::parallel_race(cache, server)
1698 .await
1699 .unwrap();
1700 assert_eq!(got[0].id, "server");
1701 }
1702
1703 struct MockOfflineRepo {
1705 items: Arc<Mutex<Vec<MediaItem>>>,
1706 query_count: Arc<Mutex<usize>>,
1707 save_count: Arc<Mutex<usize>>,
1708 }
1709
1710 impl MockOfflineRepo {
1711 fn new() -> Self {
1712 Self {
1713 items: Arc::new(Mutex::new(Vec::new())),
1714 query_count: Arc::new(Mutex::new(0)),
1715 save_count: Arc::new(Mutex::new(0)),
1716 }
1717 }
1718
1719 fn get_query_count(&self) -> usize {
1720 *self.query_count.lock_safe()
1721 }
1722
1723 fn get_save_count(&self) -> usize {
1724 *self.save_count.lock_safe()
1725 }
1726
1727 async fn save_to_cache(
1728 &self,
1729 _parent_id: &str,
1730 items: &[MediaItem],
1731 ) -> Result<usize, RepoError> {
1732 *self.save_count.lock_safe() += 1;
1733 *self.items.lock_safe() = items.to_vec();
1734 Ok(items.len())
1735 }
1736 }
1737
1738 #[async_trait]
1739 impl MediaRepository for MockOfflineRepo {
1740 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
1741 unimplemented!()
1742 }
1743
1744 async fn get_items(
1745 &self,
1746 _parent_id: &str,
1747 _options: Option<GetItemsOptions>,
1748 ) -> Result<SearchResult, RepoError> {
1749 *self.query_count.lock_safe() += 1;
1750 let items = self.items.lock_safe().clone();
1751 let count = items.len();
1752 Ok(SearchResult {
1753 items,
1754 total_record_count: count,
1755 })
1756 }
1757
1758 async fn get_item(&self, _item_id: &str) -> Result<MediaItem, RepoError> {
1759 unimplemented!()
1760 }
1761
1762 async fn get_latest_items(
1763 &self,
1764 _parent_id: &str,
1765 _limit: Option<usize>,
1766 ) -> Result<Vec<MediaItem>, RepoError> {
1767 unimplemented!()
1768 }
1769
1770 async fn get_resume_items(
1771 &self,
1772 _parent_id: Option<&str>,
1773 _limit: Option<usize>,
1774 ) -> Result<Vec<MediaItem>, RepoError> {
1775 unimplemented!()
1776 }
1777
1778 async fn get_next_up_episodes(
1779 &self,
1780 _series_id: Option<&str>,
1781 _limit: Option<usize>,
1782 ) -> Result<Vec<MediaItem>, RepoError> {
1783 unimplemented!()
1784 }
1785
1786 async fn get_recently_played_audio(
1787 &self,
1788 _limit: Option<usize>,
1789 ) -> Result<Vec<MediaItem>, RepoError> {
1790 unimplemented!()
1791 }
1792
1793 async fn get_resume_movies(
1794 &self,
1795 _limit: Option<usize>,
1796 ) -> Result<Vec<MediaItem>, RepoError> {
1797 unimplemented!()
1798 }
1799
1800 async fn get_rediscover_albums(
1801 &self,
1802 _parent_id: Option<&str>,
1803 _limit: Option<usize>,
1804 ) -> Result<Vec<MediaItem>, RepoError> {
1805 unimplemented!()
1806 }
1807
1808 async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
1809 unimplemented!()
1810 }
1811
1812 async fn search(
1813 &self,
1814 _query: &str,
1815 _options: Option<SearchOptions>,
1816 ) -> Result<SearchResult, RepoError> {
1817 unimplemented!()
1818 }
1819
1820 async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
1821 unimplemented!()
1822 }
1823
1824 async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
1825 unimplemented!()
1826 }
1827
1828 async fn get_audio_only_stream_url_for_video(
1829 &self,
1830 _item_id: &str,
1831 _media_source_id: Option<&str>,
1832 _start_time_seconds: Option<f64>,
1833 _audio_stream_index: Option<i32>,
1834 ) -> Result<String, RepoError> {
1835 unimplemented!()
1836 }
1837
1838 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
1839 unimplemented!()
1840 }
1841
1842 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
1843 unimplemented!()
1844 }
1845
1846 async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
1847 unimplemented!()
1848 }
1849
1850 async fn report_playback_start(
1851 &self,
1852 _item_id: &str,
1853 _position_ticks: i64,
1854 ) -> Result<(), RepoError> {
1855 unimplemented!()
1856 }
1857
1858 async fn report_playback_progress(
1859 &self,
1860 _item_id: &str,
1861 _position_ticks: i64,
1862 ) -> Result<(), RepoError> {
1863 unimplemented!()
1864 }
1865
1866 async fn report_playback_stopped(
1867 &self,
1868 _item_id: &str,
1869 _position_ticks: i64,
1870 ) -> Result<(), RepoError> {
1871 unimplemented!()
1872 }
1873
1874 fn get_image_url(
1875 &self,
1876 _item_id: &str,
1877 _image_type: ImageType,
1878 _options: Option<ImageOptions>,
1879 ) -> String {
1880 unimplemented!()
1881 }
1882
1883 fn get_subtitle_url(
1884 &self,
1885 _item_id: &str,
1886 _media_source_id: &str,
1887 _stream_index: i32,
1888 _format: &str,
1889 ) -> String {
1890 unimplemented!()
1891 }
1892
1893 fn get_video_download_url(
1894 &self,
1895 _item_id: &str,
1896 _quality: &str,
1897 _media_source_id: Option<&str>,
1898 _source_audio_codec: Option<&str>,
1899 ) -> String {
1900 unimplemented!()
1901 }
1902
1903 async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1904 unimplemented!()
1905 }
1906
1907 async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1908 unimplemented!()
1909 }
1910
1911 async fn get_favorites(
1912 &self,
1913 _scope: SearchScope,
1914 _options: Option<GetItemsOptions>,
1915 ) -> Result<SearchResult, RepoError> {
1916 unimplemented!()
1917 }
1918
1919 async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
1920 unimplemented!()
1921 }
1922
1923 async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> {
1924 unimplemented!()
1925 }
1926
1927 async fn get_person(&self, _person_id: &str) -> Result<MediaItem, RepoError> {
1928 unimplemented!()
1929 }
1930
1931 async fn get_items_by_person(
1932 &self,
1933 _person_id: &str,
1934 _options: Option<GetItemsOptions>,
1935 ) -> Result<SearchResult, RepoError> {
1936 unimplemented!()
1937 }
1938
1939 async fn get_similar_items(
1940 &self,
1941 _item_id: &str,
1942 _limit: Option<usize>,
1943 ) -> Result<SearchResult, RepoError> {
1944 unimplemented!()
1945 }
1946
1947 async fn create_playlist(
1948 &self,
1949 _name: &str,
1950 _item_ids: &[String],
1951 ) -> Result<PlaylistCreatedResult, RepoError> {
1952 unimplemented!()
1953 }
1954
1955 async fn delete_playlist(&self, _playlist_id: &str) -> Result<(), RepoError> {
1956 unimplemented!()
1957 }
1958
1959 async fn rename_playlist(&self, _playlist_id: &str, _name: &str) -> Result<(), RepoError> {
1960 unimplemented!()
1961 }
1962
1963 async fn get_playlist_items(
1964 &self,
1965 _playlist_id: &str,
1966 ) -> Result<Vec<PlaylistEntry>, RepoError> {
1967 unimplemented!()
1968 }
1969
1970 async fn add_to_playlist(
1971 &self,
1972 _playlist_id: &str,
1973 _item_ids: &[String],
1974 ) -> Result<(), RepoError> {
1975 unimplemented!()
1976 }
1977
1978 async fn remove_from_playlist(
1979 &self,
1980 _playlist_id: &str,
1981 _entry_ids: &[String],
1982 ) -> Result<(), RepoError> {
1983 unimplemented!()
1984 }
1985
1986 async fn move_playlist_item(
1987 &self,
1988 _playlist_id: &str,
1989 _item_id: &str,
1990 _new_index: u32,
1991 ) -> Result<(), RepoError> {
1992 unimplemented!()
1993 }
1994 }
1995
1996 struct MockOnlineRepo {
1998 items: Vec<MediaItem>,
1999 query_count: Arc<Mutex<usize>>,
2000 }
2001
2002 impl MockOnlineRepo {
2003 fn new(items: Vec<MediaItem>) -> Self {
2004 Self {
2005 items,
2006 query_count: Arc::new(Mutex::new(0)),
2007 }
2008 }
2009
2010 fn get_query_count(&self) -> usize {
2011 *self.query_count.lock_safe()
2012 }
2013 }
2014
2015 #[async_trait]
2016 impl MediaRepository for MockOnlineRepo {
2017 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
2018 unimplemented!()
2019 }
2020
2021 async fn get_items(
2022 &self,
2023 _parent_id: &str,
2024 _options: Option<GetItemsOptions>,
2025 ) -> Result<SearchResult, RepoError> {
2026 *self.query_count.lock_safe() += 1;
2027 Ok(SearchResult {
2028 items: self.items.clone(),
2029 total_record_count: self.items.len(),
2030 })
2031 }
2032
2033 async fn get_item(&self, _item_id: &str) -> Result<MediaItem, RepoError> {
2034 unimplemented!()
2035 }
2036
2037 async fn get_latest_items(
2038 &self,
2039 _parent_id: &str,
2040 _limit: Option<usize>,
2041 ) -> Result<Vec<MediaItem>, RepoError> {
2042 unimplemented!()
2043 }
2044
2045 async fn get_resume_items(
2046 &self,
2047 _parent_id: Option<&str>,
2048 _limit: Option<usize>,
2049 ) -> Result<Vec<MediaItem>, RepoError> {
2050 unimplemented!()
2051 }
2052
2053 async fn get_next_up_episodes(
2054 &self,
2055 _series_id: Option<&str>,
2056 _limit: Option<usize>,
2057 ) -> Result<Vec<MediaItem>, RepoError> {
2058 unimplemented!()
2059 }
2060
2061 async fn get_recently_played_audio(
2062 &self,
2063 _limit: Option<usize>,
2064 ) -> Result<Vec<MediaItem>, RepoError> {
2065 unimplemented!()
2066 }
2067
2068 async fn get_resume_movies(
2069 &self,
2070 _limit: Option<usize>,
2071 ) -> Result<Vec<MediaItem>, RepoError> {
2072 unimplemented!()
2073 }
2074
2075 async fn get_rediscover_albums(
2076 &self,
2077 _parent_id: Option<&str>,
2078 _limit: Option<usize>,
2079 ) -> Result<Vec<MediaItem>, RepoError> {
2080 unimplemented!()
2081 }
2082
2083 async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
2084 unimplemented!()
2085 }
2086
2087 async fn search(
2088 &self,
2089 _query: &str,
2090 _options: Option<SearchOptions>,
2091 ) -> Result<SearchResult, RepoError> {
2092 unimplemented!()
2093 }
2094
2095 async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
2096 unimplemented!()
2097 }
2098
2099 async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
2100 unimplemented!()
2101 }
2102
2103 async fn get_audio_only_stream_url_for_video(
2104 &self,
2105 _item_id: &str,
2106 _media_source_id: Option<&str>,
2107 _start_time_seconds: Option<f64>,
2108 _audio_stream_index: Option<i32>,
2109 ) -> Result<String, RepoError> {
2110 unimplemented!()
2111 }
2112
2113 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
2114 unimplemented!()
2115 }
2116
2117 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
2118 unimplemented!()
2119 }
2120
2121 async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
2122 unimplemented!()
2123 }
2124
2125 async fn report_playback_start(
2126 &self,
2127 _item_id: &str,
2128 _position_ticks: i64,
2129 ) -> Result<(), RepoError> {
2130 unimplemented!()
2131 }
2132
2133 async fn report_playback_progress(
2134 &self,
2135 _item_id: &str,
2136 _position_ticks: i64,
2137 ) -> Result<(), RepoError> {
2138 unimplemented!()
2139 }
2140
2141 async fn report_playback_stopped(
2142 &self,
2143 _item_id: &str,
2144 _position_ticks: i64,
2145 ) -> Result<(), RepoError> {
2146 unimplemented!()
2147 }
2148
2149 fn get_image_url(
2150 &self,
2151 _item_id: &str,
2152 _image_type: ImageType,
2153 _options: Option<ImageOptions>,
2154 ) -> String {
2155 unimplemented!()
2156 }
2157
2158 fn get_subtitle_url(
2159 &self,
2160 _item_id: &str,
2161 _media_source_id: &str,
2162 _stream_index: i32,
2163 _format: &str,
2164 ) -> String {
2165 unimplemented!()
2166 }
2167
2168 fn get_video_download_url(
2169 &self,
2170 _item_id: &str,
2171 _quality: &str,
2172 _media_source_id: Option<&str>,
2173 _source_audio_codec: Option<&str>,
2174 ) -> String {
2175 unimplemented!()
2176 }
2177
2178 async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
2179 unimplemented!()
2180 }
2181
2182 async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
2183 unimplemented!()
2184 }
2185
2186 async fn get_favorites(
2187 &self,
2188 _scope: SearchScope,
2189 _options: Option<GetItemsOptions>,
2190 ) -> Result<SearchResult, RepoError> {
2191 unimplemented!()
2192 }
2193
2194 async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
2195 unimplemented!()
2196 }
2197
2198 async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> {
2199 unimplemented!()
2200 }
2201
2202 async fn get_person(&self, _person_id: &str) -> Result<MediaItem, RepoError> {
2203 unimplemented!()
2204 }
2205
2206 async fn get_items_by_person(
2207 &self,
2208 _person_id: &str,
2209 _options: Option<GetItemsOptions>,
2210 ) -> Result<SearchResult, RepoError> {
2211 unimplemented!()
2212 }
2213
2214 async fn get_similar_items(
2215 &self,
2216 _item_id: &str,
2217 _limit: Option<usize>,
2218 ) -> Result<SearchResult, RepoError> {
2219 unimplemented!()
2220 }
2221
2222 async fn create_playlist(
2223 &self,
2224 _name: &str,
2225 _item_ids: &[String],
2226 ) -> Result<PlaylistCreatedResult, RepoError> {
2227 unimplemented!()
2228 }
2229
2230 async fn delete_playlist(&self, _playlist_id: &str) -> Result<(), RepoError> {
2231 unimplemented!()
2232 }
2233
2234 async fn rename_playlist(&self, _playlist_id: &str, _name: &str) -> Result<(), RepoError> {
2235 unimplemented!()
2236 }
2237
2238 async fn get_playlist_items(
2239 &self,
2240 _playlist_id: &str,
2241 ) -> Result<Vec<PlaylistEntry>, RepoError> {
2242 unimplemented!()
2243 }
2244
2245 async fn add_to_playlist(
2246 &self,
2247 _playlist_id: &str,
2248 _item_ids: &[String],
2249 ) -> Result<(), RepoError> {
2250 unimplemented!()
2251 }
2252
2253 async fn remove_from_playlist(
2254 &self,
2255 _playlist_id: &str,
2256 _entry_ids: &[String],
2257 ) -> Result<(), RepoError> {
2258 unimplemented!()
2259 }
2260
2261 async fn move_playlist_item(
2262 &self,
2263 _playlist_id: &str,
2264 _item_id: &str,
2265 _new_index: u32,
2266 ) -> Result<(), RepoError> {
2267 unimplemented!()
2268 }
2269 }
2270
2271 fn create_test_item(id: &str, name: &str) -> MediaItem {
2272 MediaItem {
2273 id: id.to_string(),
2274 name: name.to_string(),
2275 item_type: "Movie".to_string(),
2276 kind: crate::domain::MediaKind::Movie,
2277 is_folder: false,
2278 server_id: "test-server".to_string(),
2279 parent_id: Some("parent-123".to_string()),
2280 library_id: Some("library-456".to_string()),
2281 overview: Some("Test overview".to_string()),
2282 genres: Some(vec!["Action".to_string(), "Adventure".to_string()]),
2283 runtime_ticks: Some(7200000000),
2284 duration_ms: Some(720000),
2285 production_year: Some(2024),
2286 premiere_date: None,
2287 community_rating: Some(8.5),
2288 official_rating: Some("PG-13".to_string()),
2289 primary_image_tag: Some("image-tag-123".to_string()),
2290 image_id: Some("image-tag-123".to_string()),
2291 backdrop_image_tags: Some(vec!["backdrop-1".to_string()]),
2292 parent_backdrop_image_tags: None,
2293 album_id: None,
2294 album_name: None,
2295 album_artist: None,
2296 artists: None,
2297 artist_items: None,
2298 index_number: None,
2299 series_id: None,
2300 series_name: None,
2301 season_id: None,
2302 season_name: None,
2303 parent_index_number: None,
2304 user_data: None,
2305 media_streams: None,
2306 media_sources: None,
2307 people: None,
2308 }
2309 }
2310
2311 struct TestHybridRepo {
2313 offline: Arc<MockOfflineRepo>,
2314 online: Arc<MockOnlineRepo>,
2315 }
2316
2317 impl TestHybridRepo {
2318 fn new(server_items: Vec<MediaItem>) -> Self {
2319 let offline = Arc::new(MockOfflineRepo::new());
2320 let online = Arc::new(MockOnlineRepo::new(server_items));
2321 Self { offline, online }
2322 }
2323
2324 async fn get_items(&self, parent_id: &str) -> Result<SearchResult, RepoError> {
2326 let offline = Arc::clone(&self.offline);
2327 let offline_for_save = Arc::clone(&self.offline);
2328 let online = Arc::clone(&self.online);
2329 let parent_id = parent_id.to_string();
2330 let parent_id_clone = parent_id.clone();
2331 let parent_id_for_save = parent_id.clone();
2332
2333 let cache_future = async move { offline.get_items(&parent_id, None).await };
2335
2336 let server_future = async move { online.get_items(&parent_id_clone, None).await };
2337
2338 let (cache_result, server_result) = tokio::join!(cache_future, server_future);
2340
2341 let cache_had_content = cache_result
2343 .as_ref()
2344 .map(|data| data.has_content())
2345 .unwrap_or(false);
2346
2347 let result = if cache_had_content {
2349 cache_result?
2350 } else {
2351 let server_data = server_result?;
2353
2354 if !server_data.items.is_empty() {
2355 let items_clone = server_data.items.clone();
2356 offline_for_save
2357 .save_to_cache(&parent_id_for_save, &items_clone)
2358 .await?;
2359 }
2360
2361 server_data
2362 };
2363
2364 Ok(result)
2365 }
2366
2367 async fn get_items_gated(&self, parent_id: &str) -> Result<SearchResult, RepoError> {
2373 if !crate::repository::offline::include_catalog_browse() {
2374 let items = self.offline.get_items(parent_id, None).await?;
2375 return Ok(items);
2377 }
2378 self.get_items(parent_id).await
2379 }
2380 }
2381
2382 static GATE_TEST_LOCK: Mutex<()> = Mutex::new(());
2385
2386 #[tokio::test]
2392 async fn test_get_items_gate_off_empty_does_not_query_server() {
2393 let _guard = GATE_TEST_LOCK.lock_safe();
2394 crate::repository::offline::set_include_catalog_browse(false);
2395
2396 let repo = TestHybridRepo::new(vec![
2398 create_test_item("s-1", "Server 1"),
2399 create_test_item("s-2", "Server 2"),
2400 ]);
2401
2402 let result = repo.get_items_gated("parent-123").await.unwrap();
2403
2404 assert_eq!(
2405 result.items.len(),
2406 0,
2407 "empty offline result is authoritative when the gate is off"
2408 );
2409 assert_eq!(
2410 repo.online.get_query_count(),
2411 0,
2412 "server must NOT be queried when the gate is off"
2413 );
2414
2415 crate::repository::offline::set_include_catalog_browse(true);
2416 }
2417
2418 #[tokio::test]
2424 async fn test_get_items_gate_on_empty_falls_through_to_server() {
2425 let _guard = GATE_TEST_LOCK.lock_safe();
2426 crate::repository::offline::set_include_catalog_browse(true);
2427
2428 let repo = TestHybridRepo::new(vec![
2429 create_test_item("s-1", "Server 1"),
2430 create_test_item("s-2", "Server 2"),
2431 ]);
2432
2433 let result = repo.get_items_gated("parent-123").await.unwrap();
2434
2435 assert_eq!(result.items.len(), 2, "server result used on empty cache");
2436 assert_eq!(
2437 repo.online.get_query_count(),
2438 1,
2439 "server IS queried when the gate is on and the cache is empty"
2440 );
2441 }
2442
2443 #[tokio::test]
2449 async fn test_cache_miss_saves_to_cache() {
2450 let server_items = vec![
2452 create_test_item("item-1", "Movie 1"),
2453 create_test_item("item-2", "Movie 2"),
2454 create_test_item("item-3", "Movie 3"),
2455 ];
2456
2457 let repo = TestHybridRepo::new(server_items.clone());
2458
2459 let result = repo.get_items("parent-123").await.unwrap();
2461
2462 assert_eq!(result.items.len(), 3);
2464 assert_eq!(result.items[0].id, "item-1");
2465
2466 assert_eq!(
2468 repo.offline.get_query_count(),
2469 1,
2470 "Cache should be queried once"
2471 );
2472 assert_eq!(
2473 repo.online.get_query_count(),
2474 1,
2475 "Server should be queried once"
2476 );
2477
2478 assert_eq!(
2480 repo.offline.get_save_count(),
2481 1,
2482 "Should save to cache on miss"
2483 );
2484 }
2485
2486 #[tokio::test]
2495 async fn test_cache_hit_no_save() {
2496 let server_items = vec![
2498 create_test_item("item-1", "Movie 1"),
2499 create_test_item("item-2", "Movie 2"),
2500 create_test_item("item-3", "Movie 3"),
2501 ];
2502
2503 let repo = TestHybridRepo::new(server_items.clone());
2504
2505 repo.offline
2507 .save_to_cache("parent-123", &server_items)
2508 .await
2509 .unwrap();
2510 assert_eq!(repo.offline.get_save_count(), 1);
2511
2512 let result = repo.get_items("parent-123").await.unwrap();
2514
2515 assert_eq!(result.items.len(), 3);
2517 assert_eq!(result.items[0].id, "item-1");
2518
2519 assert_eq!(repo.offline.get_query_count(), 1, "Cache should be queried");
2521 assert_eq!(
2522 repo.online.get_query_count(),
2523 1,
2524 "Server is queried in parallel"
2525 );
2526
2527 assert_eq!(
2529 repo.offline.get_save_count(),
2530 1,
2531 "Should NOT save when using cache"
2532 );
2533 }
2534
2535 #[tokio::test]
2540 async fn test_empty_cache_returns_empty_result() {
2541 let repo = TestHybridRepo::new(vec![]);
2543
2544 let result = repo.get_items("parent-123").await.unwrap();
2546
2547 assert_eq!(result.items.len(), 0);
2549
2550 assert_eq!(
2552 repo.offline.get_save_count(),
2553 0,
2554 "Should not save empty results"
2555 );
2556 }
2557
2558 #[tokio::test]
2562 async fn test_has_content_check() {
2563 let empty_result = SearchResult {
2565 items: vec![],
2566 total_record_count: 0,
2567 };
2568 assert!(
2569 !empty_result.has_content(),
2570 "Empty result should not have content"
2571 );
2572
2573 let result_with_items = SearchResult {
2574 items: vec![create_test_item("item-1", "Movie 1")],
2575 total_record_count: 1,
2576 };
2577 assert!(
2578 result_with_items.has_content(),
2579 "Result with items should have content"
2580 );
2581 }
2582
2583 #[test]
2584 fn test_merge_search_local_first_then_server_appended() {
2585 let cache = SearchResult {
2586 items: vec![
2587 create_test_item("a", "Cached A"),
2588 create_test_item("b", "Cached B"),
2589 ],
2590 total_record_count: 2,
2591 };
2592 let server = SearchResult {
2593 items: vec![
2594 create_test_item("c", "Server C"),
2595 create_test_item("d", "Server D"),
2596 ],
2597 total_record_count: 2,
2598 };
2599
2600 let merged = HybridRepository::merge_search_results(cache, server);
2601
2602 let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect();
2604 assert_eq!(ids, vec!["a", "b", "c", "d"]);
2605 assert_eq!(merged.total_record_count, 4);
2606 }
2607
2608 #[test]
2609 fn test_merge_search_dedupes_with_server_winning() {
2610 let cache = SearchResult {
2613 items: vec![
2614 create_test_item("a", "Cached A"),
2615 create_test_item("b", "Cached B"),
2616 ],
2617 total_record_count: 2,
2618 };
2619 let server = SearchResult {
2620 items: vec![
2621 create_test_item("b", "Server B (fresher)"),
2622 create_test_item("c", "Server C"),
2623 ],
2624 total_record_count: 2,
2625 };
2626
2627 let merged = HybridRepository::merge_search_results(cache, server);
2628
2629 let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect();
2630 assert_eq!(
2631 ids,
2632 vec!["a", "b", "c"],
2633 "no duplicate, local position kept"
2634 );
2635
2636 let b = merged.items.iter().find(|i| i.id == "b").unwrap();
2637 assert_eq!(
2638 b.name, "Server B (fresher)",
2639 "server metadata wins on conflict"
2640 );
2641 assert_eq!(merged.total_record_count, 3);
2642 }
2643
2644 #[test]
2645 fn test_merge_search_handles_empty_sides() {
2646 let only_server = HybridRepository::merge_search_results(
2647 SearchResult {
2648 items: vec![],
2649 total_record_count: 0,
2650 },
2651 SearchResult {
2652 items: vec![create_test_item("x", "X")],
2653 total_record_count: 1,
2654 },
2655 );
2656 assert_eq!(only_server.items.len(), 1);
2657 assert_eq!(only_server.items[0].id, "x");
2658
2659 let only_cache = HybridRepository::merge_search_results(
2660 SearchResult {
2661 items: vec![create_test_item("y", "Y")],
2662 total_record_count: 1,
2663 },
2664 SearchResult {
2665 items: vec![],
2666 total_record_count: 0,
2667 },
2668 );
2669 assert_eq!(only_cache.items.len(), 1);
2670 assert_eq!(only_cache.items[0].id, "y");
2671 }
2672}