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
21pub struct HybridRepository {
33 online: Arc<OnlineRepository>,
34 offline: Arc<OfflineRepository>,
35}
36
37impl HybridRepository {
38 pub fn new(online: OnlineRepository, offline: OfflineRepository) -> Self {
39 Self {
40 online: Arc::new(online),
41 offline: Arc::new(offline),
42 }
43 }
44
45 pub fn user_id(&self) -> &str {
49 self.online.user_id()
50 }
51
52 pub async fn download_bytes(&self, url: &str) -> Result<Vec<u8>, String> {
55 self.online.download_bytes(url).await
56 }
57
58 pub async fn prune_stale_catalog(
66 &self,
67 cutoff: &str,
68 item_types: &[String],
69 ) -> Result<usize, RepoError> {
70 self.offline.prune_stale_catalog(cutoff, item_types).await
71 }
72
73 pub async fn get_jray_actors(
76 &self,
77 item_id: &str,
78 t: f64,
79 ) -> Result<Vec<super::JRayActor>, RepoError> {
80 self.online.get_jray_actors(item_id, t).await
81 }
82
83 pub async fn get_video_stream_url(
90 &self,
91 item_id: &str,
92 media_source_id: Option<&str>,
93 audio_stream_index: Option<i32>,
94 ) -> Result<String, RepoError> {
95 self.online
96 .get_video_stream_url(item_id, media_source_id, audio_stream_index)
97 .await
98 }
99
100 pub async fn get_audio_only_stream_url_for_video(
105 &self,
106 item_id: &str,
107 media_source_id: Option<&str>,
108 start_time_seconds: Option<f64>,
109 audio_stream_index: Option<i32>,
110 ) -> Result<String, RepoError> {
111 self.online
112 .get_audio_only_stream_url_for_video(
113 item_id,
114 media_source_id,
115 start_time_seconds,
116 audio_stream_index,
117 )
118 .await
119 }
120
121 pub async fn get_album_tracks(&self, album_id: &str) -> Result<Vec<MediaItem>, RepoError> {
137 let options = Some(GetItemsOptions {
138 include_item_types: Some(vec!["Audio".to_string()]),
139 sort_by: Some("ParentIndexNumber,IndexNumber,SortName".to_string()),
140 limit: Some(1000),
141 ..Default::default()
142 });
143
144 let result = self.online.get_items(album_id, options).await?;
145
146 if !result.items.is_empty() {
147 if let Err(e) = self.offline.save_to_cache(album_id, &result.items).await {
148 warn!("[HybridRepo] Failed to cache album tracks: {:?}", e);
149 }
150 }
151
152 Ok(result.items)
153 }
154
155 pub async fn get_items_unfiltered(
166 &self,
167 parent_id: &str,
168 options: Option<GetItemsOptions>,
169 ) -> Result<SearchResult, RepoError> {
170 match self.online.get_items(parent_id, options.clone()).await {
171 Ok(result) => Ok(result),
172 Err(e) => self.offline.get_items(parent_id, options).await.or(Err(e)),
173 }
174 }
175
176 pub async fn search_cache_only(
182 &self,
183 query: &str,
184 options: Option<SearchOptions>,
185 ) -> Result<SearchResult, RepoError> {
186 let offline = Arc::clone(&self.offline);
187 let query = query.to_string();
188 self.cache_with_timeout(async move { offline.search(&query, options).await })
189 .await
190 .map(ExcludeHidden::without_excluded)
191 }
192
193 pub async fn get_favorites_cache_only(
198 &self,
199 scope: SearchScope,
200 options: Option<GetItemsOptions>,
201 ) -> Result<SearchResult, RepoError> {
202 let offline = Arc::clone(&self.offline);
203 self.cache_with_timeout(async move { offline.get_favorites(scope, options).await })
204 .await
205 .map(ExcludeHidden::without_excluded)
206 }
207
208 pub async fn get_favorites_server_only(
214 &self,
215 scope: SearchScope,
216 options: Option<GetItemsOptions>,
217 ) -> Result<SearchResult, RepoError> {
218 let result = self.online.get_favorites(scope, options).await?;
219 if !result.items.is_empty() {
220 if let Err(e) = self.offline.save_to_cache("favorites", &result.items).await {
223 debug!("[HybridRepo] Failed to cache favourites: {:?}", e);
224 }
225 }
226 Ok(result.without_excluded())
227 }
228
229 pub async fn cache_items_from_server(
238 &self,
239 parent_id: &str,
240 options: Option<GetItemsOptions>,
241 ) -> Result<Vec<MediaItem>, RepoError> {
242 let result = self.online.get_items(parent_id, options).await?;
243 if !result.items.is_empty() {
244 self.offline.save_to_cache(parent_id, &result.items).await?;
245 }
246 Ok(result.items)
247 }
248
249 pub async fn get_downloaded_items(
258 &self,
259 parent_id: &str,
260 options: Option<GetItemsOptions>,
261 ) -> Result<SearchResult, RepoError> {
262 self.offline.get_downloaded_items(parent_id, options).await
267 }
268
269 pub async fn get_downloaded_libraries(&self) -> Result<Vec<Library>, RepoError> {
273 self.offline.get_downloaded_libraries().await
274 }
275
276 pub async fn get_download_disk_usage(&self) -> Result<DownloadDiskUsage, RepoError> {
280 self.offline.get_download_disk_usage().await
281 }
282
283 pub async fn search_server_only(
285 &self,
286 query: &str,
287 options: Option<SearchOptions>,
288 ) -> Result<SearchResult, RepoError> {
289 self.online
290 .search(query, options)
291 .await
292 .map(ExcludeHidden::without_excluded)
293 }
294
295 pub fn merge_search_results(cache: SearchResult, server: SearchResult) -> SearchResult {
301 use std::collections::HashMap;
302
303 let mut server_by_id: HashMap<String, MediaItem> = HashMap::new();
306 let mut server_order: Vec<String> = Vec::with_capacity(server.items.len());
307 for item in server.items {
308 if !server_by_id.contains_key(&item.id) {
309 server_order.push(item.id.clone());
310 }
311 server_by_id.insert(item.id.clone(), item);
312 }
313
314 let mut items: Vec<MediaItem> = Vec::new();
315 let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
316
317 for local in cache.items {
320 if !seen.insert(local.id.clone()) {
321 continue;
322 }
323 match server_by_id.remove(&local.id) {
324 Some(server_item) => items.push(server_item),
325 None => items.push(local),
326 }
327 }
328
329 for id in server_order {
331 if let Some(server_item) = server_by_id.remove(&id) {
332 if seen.insert(id) {
333 items.push(server_item);
334 }
335 }
336 }
337
338 let total_record_count = items.len();
339 SearchResult {
340 items,
341 total_record_count,
342 }
343 }
344
345 async fn parallel_race<T, F1, F2>(
365 &self,
366 cache_future: F1,
367 server_future: F2,
368 ) -> Result<T, RepoError>
369 where
370 T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static,
371 F1: std::future::Future<Output = Result<T, RepoError>> + Send,
372 F2: std::future::Future<Output = Result<T, RepoError>> + Send,
373 {
374 let cache_result = cache_future.await.map(ExcludeHidden::without_excluded);
376
377 if let Ok(data) = &cache_result {
378 if data.has_content() {
379 debug!("[HybridRepo] Cache hit, returning immediately");
380 return Ok(data.clone());
381 }
382 }
383
384 debug!("[HybridRepo] Cache miss, querying server");
386 match server_future.await {
387 Ok(data) => Ok(data.without_excluded()),
388 Err(e) => {
389 cache_result.or(Err(e))
391 }
392 }
393 }
394
395 async fn race_with_refresh<T, F1, F2, R>(
410 &self,
411 cache_future: F1,
412 server_future: F2,
413 on_cache_hit: R,
414 ) -> Result<T, RepoError>
415 where
416 T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static,
417 F1: std::future::Future<Output = Result<T, RepoError>> + Send,
418 F2: std::future::Future<Output = Result<T, RepoError>> + Send,
419 R: FnOnce(),
420 {
421 let cache_result = cache_future.await.map(ExcludeHidden::without_excluded);
422
423 if let Ok(data) = &cache_result {
424 if data.has_content() {
425 debug!("[HybridRepo] Cache hit, returning immediately (refreshing in background)");
426 on_cache_hit();
427 return Ok(data.clone());
428 }
429 }
430
431 debug!("[HybridRepo] Cache miss, querying server");
432 match server_future.await {
433 Ok(data) => Ok(data.without_excluded()),
434 Err(e) => cache_result.or(Err(e)),
435 }
436 }
437
438 async fn cache_with_timeout<T>(
442 &self,
443 future: impl std::future::Future<Output = Result<T, RepoError>> + Send,
444 ) -> Result<T, RepoError> {
445 timeout(Duration::from_millis(100), future)
446 .await
447 .unwrap_or_else(|_| {
448 Err(RepoError::Database {
449 message: "Cache query timeout".to_string(),
450 })
451 })
452 }
453}
454
455#[async_trait]
456impl MediaRepository for HybridRepository {
457 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
458 let cache_result = self.cache_with_timeout(self.offline.get_libraries()).await;
462
463 if let Ok(libs) = &cache_result {
464 if libs.has_content() {
465 debug!("[HybridRepo] Cache hit for libraries, returning immediately");
466 let online = Arc::clone(&self.online);
467 let offline = Arc::clone(&self.offline);
468 tokio::spawn(async move {
469 if let Ok(server_libs) = online.get_libraries().await {
470 if !server_libs.is_empty() {
471 if let Err(e) = offline.save_libraries_to_cache(&server_libs).await {
472 warn!(
473 "[HybridRepo] Background library cache update failed: {:?}",
474 e
475 );
476 }
477 }
478 }
479 });
480 return cache_result;
481 }
482 }
483
484 match self.online.get_libraries().await {
486 Ok(server_libs) => {
487 if !server_libs.is_empty() {
488 if let Err(e) = self.offline.save_libraries_to_cache(&server_libs).await {
489 warn!(
490 "[HybridRepo] Failed to cache {} libraries: {:?}",
491 server_libs.len(),
492 e
493 );
494 }
495 }
496 Ok(server_libs)
497 }
498 Err(e) => cache_result.or(Err(e)),
499 }
500 }
501
502 async fn get_items(
503 &self,
504 parent_id: &str,
505 options: Option<GetItemsOptions>,
506 ) -> Result<SearchResult, RepoError> {
507 let offline = Arc::clone(&self.offline);
508 let offline_for_save = Arc::clone(&self.offline);
509 let online = Arc::clone(&self.online);
510 let parent_id = parent_id.to_string();
511 let parent_id_clone = parent_id.clone();
512 let parent_id_for_save = parent_id.clone();
513 let opts_clone = options.clone();
514
515 let server_handle =
517 tokio::spawn(async move { online.get_items(&parent_id_clone, options).await });
518
519 let cache_result = self
528 .cache_with_timeout(async move { offline.get_items(&parent_id, opts_clone).await })
529 .await
530 .map(ExcludeHidden::without_excluded);
531
532 if !crate::repository::offline::include_catalog_browse() {
540 if let Ok(data) = &cache_result {
541 debug!(
542 "[HybridRepo] Downloads-only gate: returning offline result ({} items) as authoritative for parent {}",
543 data.items.len(),
544 &parent_id_for_save[..8.min(parent_id_for_save.len())]
545 );
546 server_handle.abort();
548 return Ok(data.clone());
549 }
550 }
551
552 if let Ok(data) = &cache_result {
554 if data.has_content() {
555 debug!(
556 "[HybridRepo] Cache hit for get_items, returning immediately for parent {}",
557 &parent_id_for_save[..8.min(parent_id_for_save.len())]
558 );
559 tokio::spawn(async move {
561 match server_handle.await {
562 Ok(Ok(server_data)) if !server_data.items.is_empty() => {
563 if let Err(e) = offline_for_save
564 .save_to_cache(&parent_id_for_save, &server_data.items)
565 .await
566 {
567 warn!("[HybridRepo] Background cache update failed: {:?}", e);
568 } else {
569 debug!(
570 "[HybridRepo] Background updated {} cached items for parent {}",
571 server_data.items.len(),
572 &parent_id_for_save[..8.min(parent_id_for_save.len())]
573 );
574 }
575 }
576 _ => {} }
578 });
579 return Ok(data.clone());
580 }
581 }
582
583 match server_handle.await {
585 Ok(Ok(server_data)) => {
586 if !server_data.items.is_empty() {
587 let items_clone = server_data.items.clone();
588 tokio::spawn(async move {
589 if let Err(e) = offline_for_save
590 .save_to_cache(&parent_id_for_save, &items_clone)
591 .await
592 {
593 warn!(
594 "[HybridRepo] Failed to save {} items to cache: {:?}",
595 items_clone.len(),
596 e
597 );
598 } else {
599 debug!(
600 "[HybridRepo] Saved {} items to cache for parent {}",
601 items_clone.len(),
602 &parent_id_for_save[..8.min(parent_id_for_save.len())]
603 );
604 }
605 });
606 }
607 Ok(server_data.without_excluded())
613 }
614 Ok(Err(e)) => cache_result.or(Err(e)),
615 Err(join_err) => cache_result.or(Err(RepoError::Network {
616 message: format!("Server task failed: {}", join_err),
617 })),
618 }
619 }
620
621 async fn get_item(&self, item_id: &str) -> Result<MediaItem, RepoError> {
637 let offline = Arc::clone(&self.offline);
638 let online = Arc::clone(&self.online);
639 let item_id = item_id.to_string();
640 let item_id_clone = item_id.clone();
641
642 let cache_future = self.cache_with_timeout(async move { offline.get_item(&item_id).await });
643
644 let online_for_refresh = Arc::clone(&self.online);
645 let offline_for_save = Arc::clone(&self.offline);
646 let refresh_id = item_id_clone.clone();
647 let on_cache_hit = move || {
648 tokio::spawn(async move {
649 match online_for_refresh.get_item(&refresh_id).await {
650 Ok(fresh) => {
651 let parent = fresh
654 .parent_id
655 .clone()
656 .unwrap_or_else(|| "item".to_string());
657 if let Err(e) = offline_for_save.save_to_cache(&parent, &[fresh]).await {
658 debug!("[HybridRepo] Background item refresh failed: {:?}", e);
659 }
660 }
661 Err(e) => debug!("[HybridRepo] Background item refresh unavailable: {:?}", e),
662 }
663 });
664 };
665
666 let server_future = async move { online.get_item(&item_id_clone).await };
667
668 self.race_with_refresh(cache_future, server_future, on_cache_hit)
669 .await
670 }
671
672 async fn get_latest_items(
673 &self,
674 parent_id: &str,
675 limit: Option<usize>,
676 ) -> Result<Vec<MediaItem>, RepoError> {
677 let offline = Arc::clone(&self.offline);
678 let online = Arc::clone(&self.online);
679 let parent_id = parent_id.to_string();
680 let parent_id_clone = parent_id.clone();
681 let limit_clone = limit;
682
683 let cache_future = self
684 .cache_with_timeout(async move { offline.get_latest_items(&parent_id, limit).await });
685
686 let server_future =
687 async move { online.get_latest_items(&parent_id_clone, limit_clone).await };
688
689 self.parallel_race(cache_future, server_future).await
690 }
691
692 async fn get_resume_items(
693 &self,
694 parent_id: Option<&str>,
695 limit: Option<usize>,
696 ) -> Result<Vec<MediaItem>, RepoError> {
697 let offline = Arc::clone(&self.offline);
698 let online = Arc::clone(&self.online);
699 let parent_id_str = parent_id.map(|s| s.to_string());
700 let parent_id_clone = parent_id_str.clone();
701 let limit_clone = limit;
702
703 let cache_future = self.cache_with_timeout(async move {
704 offline
705 .get_resume_items(parent_id_str.as_deref(), limit)
706 .await
707 });
708
709 let server_future = async move {
710 online
711 .get_resume_items(parent_id_clone.as_deref(), limit_clone)
712 .await
713 };
714
715 self.parallel_race(cache_future, server_future).await
716 }
717
718 async fn get_next_up_episodes(
719 &self,
720 series_id: Option<&str>,
721 limit: Option<usize>,
722 ) -> Result<Vec<MediaItem>, RepoError> {
723 self.online
725 .get_next_up_episodes(series_id, limit)
726 .await
727 .map(ExcludeHidden::without_excluded)
728 }
729
730 async fn get_recently_played_audio(
731 &self,
732 limit: Option<usize>,
733 ) -> Result<Vec<MediaItem>, RepoError> {
734 let offline = Arc::clone(&self.offline);
735 let online = Arc::clone(&self.online);
736 let limit_clone = limit;
737
738 let cache_future =
739 self.cache_with_timeout(async move { offline.get_recently_played_audio(limit).await });
740
741 let server_future = async move { online.get_recently_played_audio(limit_clone).await };
742
743 self.parallel_race(cache_future, server_future).await
744 }
745
746 async fn get_resume_movies(&self, limit: Option<usize>) -> Result<Vec<MediaItem>, RepoError> {
747 let offline = Arc::clone(&self.offline);
748 let online = Arc::clone(&self.online);
749 let limit_clone = limit;
750
751 let cache_future =
752 self.cache_with_timeout(async move { offline.get_resume_movies(limit).await });
753
754 let server_future = async move { online.get_resume_movies(limit_clone).await };
755
756 self.parallel_race(cache_future, server_future).await
757 }
758
759 async fn get_rediscover_albums(
760 &self,
761 parent_id: Option<&str>,
762 limit: Option<usize>,
763 ) -> Result<Vec<MediaItem>, RepoError> {
764 let offline = Arc::clone(&self.offline);
765 let online = Arc::clone(&self.online);
766 let parent_id_owned = parent_id.map(|s| s.to_string());
767 let parent_id_clone = parent_id_owned.clone();
768
769 let cache_future = self.cache_with_timeout(async move {
770 offline
771 .get_rediscover_albums(parent_id_owned.as_deref(), limit)
772 .await
773 });
774
775 let server_future = async move {
776 online
777 .get_rediscover_albums(parent_id_clone.as_deref(), limit)
778 .await
779 };
780
781 self.parallel_race(cache_future, server_future).await
782 }
783
784 async fn get_genres(&self, parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
785 let parent_id_str = parent_id.map(|s| s.to_string());
791
792 let cache_offline = Arc::clone(&self.offline);
793 let cache_pid = parent_id_str.clone();
794 let cache_result = self
795 .cache_with_timeout(async move { cache_offline.get_genres(cache_pid.as_deref()).await })
796 .await;
797
798 if let Ok(genres) = &cache_result {
799 if genres.has_content() {
800 debug!("[HybridRepo] Cache hit for genres, returning immediately");
801 let online = Arc::clone(&self.online);
802 let offline = Arc::clone(&self.offline);
803 let pid = parent_id_str.clone();
804 tokio::spawn(async move {
805 if let Ok(server_genres) = online.get_genres(pid.as_deref()).await {
806 if !server_genres.is_empty() {
807 if let Err(e) = offline
808 .save_genres_to_cache(pid.as_deref(), &server_genres)
809 .await
810 {
811 warn!("[HybridRepo] Background genre cache update failed: {:?}", e);
812 }
813 }
814 }
815 });
816 return cache_result;
817 }
818 }
819
820 match self.online.get_genres(parent_id_str.as_deref()).await {
822 Ok(server_genres) => {
823 if !server_genres.is_empty() {
824 if let Err(e) = self
825 .offline
826 .save_genres_to_cache(parent_id_str.as_deref(), &server_genres)
827 .await
828 {
829 warn!(
830 "[HybridRepo] Failed to cache {} genres: {:?}",
831 server_genres.len(),
832 e
833 );
834 }
835 }
836 Ok(server_genres)
837 }
838 Err(e) => cache_result.or(Err(e)),
839 }
840 }
841
842 async fn search(
843 &self,
844 query: &str,
845 options: Option<SearchOptions>,
846 ) -> Result<SearchResult, RepoError> {
847 let offline = Arc::clone(&self.offline);
848 let online = Arc::clone(&self.online);
849 let query = query.to_string();
850 let query_clone = query.clone();
851 let opts_clone = options.clone();
852
853 let cache_future =
854 self.cache_with_timeout(async move { offline.search(&query, opts_clone).await });
855
856 let server_future = async move { online.search(&query_clone, options).await };
857
858 self.parallel_race(cache_future, server_future).await
859 }
860
861 async fn get_playback_info(&self, item_id: &str) -> Result<PlaybackInfo, RepoError> {
862 self.online.get_playback_info(item_id).await
864 }
865
866 async fn get_audio_stream_url(&self, item_id: &str) -> Result<String, RepoError> {
867 self.online.get_audio_stream_url(item_id).await
869 }
870
871 async fn get_audio_only_stream_url_for_video(
872 &self,
873 item_id: &str,
874 media_source_id: Option<&str>,
875 start_time_seconds: Option<f64>,
876 audio_stream_index: Option<i32>,
877 ) -> Result<String, RepoError> {
878 self.online
880 .build_audio_only_stream_url_for_video(
881 item_id,
882 media_source_id,
883 start_time_seconds,
884 audio_stream_index,
885 )
886 .await
887 }
888
889 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
890 self.online
892 .get_live_tv_channels()
893 .await
894 .map(ExcludeHidden::without_excluded)
895 }
896
897 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
898 self.online
900 .get_channels()
901 .await
902 .map(ExcludeHidden::without_excluded)
903 }
904
905 async fn open_live_stream(&self, item_id: &str) -> Result<LiveStreamInfo, RepoError> {
906 self.online.open_live_stream(item_id).await
908 }
909
910 async fn report_playback_start(
911 &self,
912 item_id: &str,
913 position_ticks: i64,
914 ) -> Result<(), RepoError> {
915 self.online
917 .report_playback_start(item_id, position_ticks)
918 .await
919 }
920
921 async fn report_playback_progress(
922 &self,
923 item_id: &str,
924 position_ticks: i64,
925 ) -> Result<(), RepoError> {
926 self.online
928 .report_playback_progress(item_id, position_ticks)
929 .await
930 }
931
932 async fn report_playback_stopped(
933 &self,
934 item_id: &str,
935 position_ticks: i64,
936 ) -> Result<(), RepoError> {
937 self.online
939 .report_playback_stopped(item_id, position_ticks)
940 .await
941 }
942
943 fn get_image_url(
944 &self,
945 item_id: &str,
946 image_type: ImageType,
947 options: Option<ImageOptions>,
948 ) -> String {
949 self.online.get_image_url(item_id, image_type, options)
951 }
952
953 fn get_subtitle_url(
954 &self,
955 item_id: &str,
956 media_source_id: &str,
957 stream_index: i32,
958 format: &str,
959 ) -> String {
960 self.online
962 .get_subtitle_url(item_id, media_source_id, stream_index, format)
963 }
964
965 fn get_video_download_url(
966 &self,
967 item_id: &str,
968 quality: &str,
969 media_source_id: Option<&str>,
970 source_audio_codec: Option<&str>,
971 ) -> String {
972 self.online
974 .get_video_download_url(item_id, quality, media_source_id, source_audio_codec)
975 }
976
977 async fn mark_favorite(&self, item_id: &str) -> Result<(), RepoError> {
978 self.online.mark_favorite(item_id).await
980 }
981
982 async fn unmark_favorite(&self, item_id: &str) -> Result<(), RepoError> {
983 self.online.unmark_favorite(item_id).await
985 }
986
987 async fn clear_watch_history(&self, item_id: &str) -> Result<(), RepoError> {
988 self.online.clear_watch_history(item_id).await
990 }
991
992 async fn mark_played(&self, item_id: &str) -> Result<(), RepoError> {
993 self.online.mark_played(item_id).await
995 }
996
997 async fn get_person(&self, person_id: &str) -> Result<MediaItem, RepoError> {
998 let offline = Arc::clone(&self.offline);
999 let online = Arc::clone(&self.online);
1000 let person_id = person_id.to_string();
1001 let person_id_clone = person_id.clone();
1002
1003 let cache_future =
1004 self.cache_with_timeout(async move { offline.get_person(&person_id).await });
1005
1006 let server_future = async move { online.get_person(&person_id_clone).await };
1007
1008 self.parallel_race(cache_future, server_future).await
1009 }
1010
1011 async fn get_items_by_person(
1012 &self,
1013 person_id: &str,
1014 options: Option<GetItemsOptions>,
1015 ) -> Result<SearchResult, RepoError> {
1016 let offline = Arc::clone(&self.offline);
1017 let online = Arc::clone(&self.online);
1018 let person_id = person_id.to_string();
1019 let person_id_clone = person_id.clone();
1020 let opts_clone = options.clone();
1021
1022 let cache_future = self.cache_with_timeout(async move {
1023 offline.get_items_by_person(&person_id, opts_clone).await
1024 });
1025
1026 let server_future =
1027 async move { online.get_items_by_person(&person_id_clone, options).await };
1028
1029 self.parallel_race(cache_future, server_future).await
1030 }
1031
1032 async fn get_favorites(
1034 &self,
1035 scope: SearchScope,
1036 options: Option<GetItemsOptions>,
1037 ) -> Result<SearchResult, RepoError> {
1038 let cache_result = self.get_favorites_cache_only(scope, options.clone()).await;
1039
1040 if !crate::repository::offline::include_catalog_browse() {
1045 if let Ok(data) = &cache_result {
1046 return Ok(data.clone());
1047 }
1048 }
1049
1050 if let Ok(data) = &cache_result {
1051 if data.has_content() {
1052 return Ok(data.clone());
1053 }
1054 }
1055
1056 match self.get_favorites_server_only(scope, options).await {
1062 Ok(data) => Ok(data),
1063 Err(e) => cache_result.or(Err(e)),
1064 }
1065 }
1066
1067 async fn get_similar_items(
1068 &self,
1069 item_id: &str,
1070 limit: Option<usize>,
1071 ) -> Result<SearchResult, RepoError> {
1072 let offline = Arc::clone(&self.offline);
1073 let online = Arc::clone(&self.online);
1074 let item_id = item_id.to_string();
1075 let item_id_clone = item_id.clone();
1076
1077 let cache_future = self
1078 .cache_with_timeout(async move { offline.get_similar_items(&item_id, limit).await });
1079
1080 let server_future = async move { online.get_similar_items(&item_id_clone, limit).await };
1081
1082 self.parallel_race(cache_future, server_future).await
1083 }
1084
1085 async fn create_playlist(
1088 &self,
1089 name: &str,
1090 item_ids: &[String],
1091 ) -> Result<PlaylistCreatedResult, RepoError> {
1092 self.online.create_playlist(name, item_ids).await
1094 }
1095
1096 async fn delete_playlist(&self, playlist_id: &str) -> Result<(), RepoError> {
1097 self.online.delete_playlist(playlist_id).await
1099 }
1100
1101 async fn rename_playlist(&self, playlist_id: &str, name: &str) -> Result<(), RepoError> {
1102 self.online.rename_playlist(playlist_id, name).await
1104 }
1105
1106 async fn get_playlist_items(&self, playlist_id: &str) -> Result<Vec<PlaylistEntry>, RepoError> {
1107 let offline = Arc::clone(&self.offline);
1108 let offline_for_save = Arc::clone(&self.offline);
1109 let online = Arc::clone(&self.online);
1110 let playlist_id = playlist_id.to_string();
1111 let playlist_id_clone = playlist_id.clone();
1112 let playlist_id_for_save = playlist_id.clone();
1113
1114 let server_handle =
1116 tokio::spawn(async move { online.get_playlist_items(&playlist_id_clone).await });
1117
1118 let cache_result = self
1120 .cache_with_timeout(async move { offline.get_playlist_items(&playlist_id).await })
1121 .await;
1122
1123 if let Ok(data) = &cache_result {
1125 if data.has_content() {
1126 debug!("[HybridRepo] Cache hit for playlist items, returning immediately");
1127 tokio::spawn(async move {
1128 if let Ok(Ok(server_entries)) = server_handle.await {
1129 if let Err(e) = offline_for_save
1130 .save_playlist_items_to_cache(&playlist_id_for_save, &server_entries)
1131 .await
1132 {
1133 warn!("[HybridRepo] Failed to update playlist cache: {:?}", e);
1134 }
1135 }
1136 });
1137 return cache_result;
1138 }
1139 }
1140
1141 match server_handle.await {
1143 Ok(Ok(entries)) => {
1144 let entries_clone = entries.clone();
1145 tokio::spawn(async move {
1146 if let Err(e) = offline_for_save
1147 .save_playlist_items_to_cache(&playlist_id_for_save, &entries_clone)
1148 .await
1149 {
1150 warn!(
1151 "[HybridRepo] Failed to save playlist items to cache: {:?}",
1152 e
1153 );
1154 }
1155 });
1156 Ok(entries)
1157 }
1158 Ok(Err(e)) => cache_result.or(Err(e)),
1159 Err(join_err) => cache_result.or(Err(RepoError::Network {
1160 message: format!("Server task failed: {}", join_err),
1161 })),
1162 }
1163 }
1164
1165 async fn add_to_playlist(
1166 &self,
1167 playlist_id: &str,
1168 item_ids: &[String],
1169 ) -> Result<(), RepoError> {
1170 self.online.add_to_playlist(playlist_id, item_ids).await
1172 }
1173
1174 async fn remove_from_playlist(
1175 &self,
1176 playlist_id: &str,
1177 entry_ids: &[String],
1178 ) -> Result<(), RepoError> {
1179 self.online
1181 .remove_from_playlist(playlist_id, entry_ids)
1182 .await
1183 }
1184
1185 async fn move_playlist_item(
1186 &self,
1187 playlist_id: &str,
1188 item_id: &str,
1189 new_index: u32,
1190 ) -> Result<(), RepoError> {
1191 self.online
1193 .move_playlist_item(playlist_id, item_id, new_index)
1194 .await
1195 }
1196}
1197
1198#[cfg(test)]
1199mod tests {
1200 #![allow(clippy::await_holding_lock)]
1209
1210 use super::*;
1211 use std::sync::Mutex;
1212
1213 struct MockOfflineRepo {
1215 items: Arc<Mutex<Vec<MediaItem>>>,
1216 query_count: Arc<Mutex<usize>>,
1217 save_count: Arc<Mutex<usize>>,
1218 }
1219
1220 impl MockOfflineRepo {
1221 fn new() -> Self {
1222 Self {
1223 items: Arc::new(Mutex::new(Vec::new())),
1224 query_count: Arc::new(Mutex::new(0)),
1225 save_count: Arc::new(Mutex::new(0)),
1226 }
1227 }
1228
1229 fn get_query_count(&self) -> usize {
1230 *self.query_count.lock_safe()
1231 }
1232
1233 fn get_save_count(&self) -> usize {
1234 *self.save_count.lock_safe()
1235 }
1236
1237 async fn save_to_cache(
1238 &self,
1239 _parent_id: &str,
1240 items: &[MediaItem],
1241 ) -> Result<usize, RepoError> {
1242 *self.save_count.lock_safe() += 1;
1243 *self.items.lock_safe() = items.to_vec();
1244 Ok(items.len())
1245 }
1246 }
1247
1248 #[async_trait]
1249 impl MediaRepository for MockOfflineRepo {
1250 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
1251 unimplemented!()
1252 }
1253
1254 async fn get_items(
1255 &self,
1256 _parent_id: &str,
1257 _options: Option<GetItemsOptions>,
1258 ) -> Result<SearchResult, RepoError> {
1259 *self.query_count.lock_safe() += 1;
1260 let items = self.items.lock_safe().clone();
1261 let count = items.len();
1262 Ok(SearchResult {
1263 items,
1264 total_record_count: count,
1265 })
1266 }
1267
1268 async fn get_item(&self, _item_id: &str) -> Result<MediaItem, RepoError> {
1269 unimplemented!()
1270 }
1271
1272 async fn get_latest_items(
1273 &self,
1274 _parent_id: &str,
1275 _limit: Option<usize>,
1276 ) -> Result<Vec<MediaItem>, RepoError> {
1277 unimplemented!()
1278 }
1279
1280 async fn get_resume_items(
1281 &self,
1282 _parent_id: Option<&str>,
1283 _limit: Option<usize>,
1284 ) -> Result<Vec<MediaItem>, RepoError> {
1285 unimplemented!()
1286 }
1287
1288 async fn get_next_up_episodes(
1289 &self,
1290 _series_id: Option<&str>,
1291 _limit: Option<usize>,
1292 ) -> Result<Vec<MediaItem>, RepoError> {
1293 unimplemented!()
1294 }
1295
1296 async fn get_recently_played_audio(
1297 &self,
1298 _limit: Option<usize>,
1299 ) -> Result<Vec<MediaItem>, RepoError> {
1300 unimplemented!()
1301 }
1302
1303 async fn get_resume_movies(
1304 &self,
1305 _limit: Option<usize>,
1306 ) -> Result<Vec<MediaItem>, RepoError> {
1307 unimplemented!()
1308 }
1309
1310 async fn get_rediscover_albums(
1311 &self,
1312 _parent_id: Option<&str>,
1313 _limit: Option<usize>,
1314 ) -> Result<Vec<MediaItem>, RepoError> {
1315 unimplemented!()
1316 }
1317
1318 async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
1319 unimplemented!()
1320 }
1321
1322 async fn search(
1323 &self,
1324 _query: &str,
1325 _options: Option<SearchOptions>,
1326 ) -> Result<SearchResult, RepoError> {
1327 unimplemented!()
1328 }
1329
1330 async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
1331 unimplemented!()
1332 }
1333
1334 async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
1335 unimplemented!()
1336 }
1337
1338 async fn get_audio_only_stream_url_for_video(
1339 &self,
1340 _item_id: &str,
1341 _media_source_id: Option<&str>,
1342 _start_time_seconds: Option<f64>,
1343 _audio_stream_index: Option<i32>,
1344 ) -> Result<String, RepoError> {
1345 unimplemented!()
1346 }
1347
1348 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
1349 unimplemented!()
1350 }
1351
1352 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
1353 unimplemented!()
1354 }
1355
1356 async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
1357 unimplemented!()
1358 }
1359
1360 async fn report_playback_start(
1361 &self,
1362 _item_id: &str,
1363 _position_ticks: i64,
1364 ) -> Result<(), RepoError> {
1365 unimplemented!()
1366 }
1367
1368 async fn report_playback_progress(
1369 &self,
1370 _item_id: &str,
1371 _position_ticks: i64,
1372 ) -> Result<(), RepoError> {
1373 unimplemented!()
1374 }
1375
1376 async fn report_playback_stopped(
1377 &self,
1378 _item_id: &str,
1379 _position_ticks: i64,
1380 ) -> Result<(), RepoError> {
1381 unimplemented!()
1382 }
1383
1384 fn get_image_url(
1385 &self,
1386 _item_id: &str,
1387 _image_type: ImageType,
1388 _options: Option<ImageOptions>,
1389 ) -> String {
1390 unimplemented!()
1391 }
1392
1393 fn get_subtitle_url(
1394 &self,
1395 _item_id: &str,
1396 _media_source_id: &str,
1397 _stream_index: i32,
1398 _format: &str,
1399 ) -> String {
1400 unimplemented!()
1401 }
1402
1403 fn get_video_download_url(
1404 &self,
1405 _item_id: &str,
1406 _quality: &str,
1407 _media_source_id: Option<&str>,
1408 _source_audio_codec: Option<&str>,
1409 ) -> String {
1410 unimplemented!()
1411 }
1412
1413 async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1414 unimplemented!()
1415 }
1416
1417 async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1418 unimplemented!()
1419 }
1420
1421 async fn get_favorites(
1422 &self,
1423 _scope: SearchScope,
1424 _options: Option<GetItemsOptions>,
1425 ) -> Result<SearchResult, RepoError> {
1426 unimplemented!()
1427 }
1428
1429 async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
1430 unimplemented!()
1431 }
1432
1433 async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> {
1434 unimplemented!()
1435 }
1436
1437 async fn get_person(&self, _person_id: &str) -> Result<MediaItem, RepoError> {
1438 unimplemented!()
1439 }
1440
1441 async fn get_items_by_person(
1442 &self,
1443 _person_id: &str,
1444 _options: Option<GetItemsOptions>,
1445 ) -> Result<SearchResult, RepoError> {
1446 unimplemented!()
1447 }
1448
1449 async fn get_similar_items(
1450 &self,
1451 _item_id: &str,
1452 _limit: Option<usize>,
1453 ) -> Result<SearchResult, RepoError> {
1454 unimplemented!()
1455 }
1456
1457 async fn create_playlist(
1458 &self,
1459 _name: &str,
1460 _item_ids: &[String],
1461 ) -> Result<PlaylistCreatedResult, RepoError> {
1462 unimplemented!()
1463 }
1464
1465 async fn delete_playlist(&self, _playlist_id: &str) -> Result<(), RepoError> {
1466 unimplemented!()
1467 }
1468
1469 async fn rename_playlist(&self, _playlist_id: &str, _name: &str) -> Result<(), RepoError> {
1470 unimplemented!()
1471 }
1472
1473 async fn get_playlist_items(
1474 &self,
1475 _playlist_id: &str,
1476 ) -> Result<Vec<PlaylistEntry>, RepoError> {
1477 unimplemented!()
1478 }
1479
1480 async fn add_to_playlist(
1481 &self,
1482 _playlist_id: &str,
1483 _item_ids: &[String],
1484 ) -> Result<(), RepoError> {
1485 unimplemented!()
1486 }
1487
1488 async fn remove_from_playlist(
1489 &self,
1490 _playlist_id: &str,
1491 _entry_ids: &[String],
1492 ) -> Result<(), RepoError> {
1493 unimplemented!()
1494 }
1495
1496 async fn move_playlist_item(
1497 &self,
1498 _playlist_id: &str,
1499 _item_id: &str,
1500 _new_index: u32,
1501 ) -> Result<(), RepoError> {
1502 unimplemented!()
1503 }
1504 }
1505
1506 struct MockOnlineRepo {
1508 items: Vec<MediaItem>,
1509 query_count: Arc<Mutex<usize>>,
1510 }
1511
1512 impl MockOnlineRepo {
1513 fn new(items: Vec<MediaItem>) -> Self {
1514 Self {
1515 items,
1516 query_count: Arc::new(Mutex::new(0)),
1517 }
1518 }
1519
1520 fn get_query_count(&self) -> usize {
1521 *self.query_count.lock_safe()
1522 }
1523 }
1524
1525 #[async_trait]
1526 impl MediaRepository for MockOnlineRepo {
1527 async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
1528 unimplemented!()
1529 }
1530
1531 async fn get_items(
1532 &self,
1533 _parent_id: &str,
1534 _options: Option<GetItemsOptions>,
1535 ) -> Result<SearchResult, RepoError> {
1536 *self.query_count.lock_safe() += 1;
1537 Ok(SearchResult {
1538 items: self.items.clone(),
1539 total_record_count: self.items.len(),
1540 })
1541 }
1542
1543 async fn get_item(&self, _item_id: &str) -> Result<MediaItem, RepoError> {
1544 unimplemented!()
1545 }
1546
1547 async fn get_latest_items(
1548 &self,
1549 _parent_id: &str,
1550 _limit: Option<usize>,
1551 ) -> Result<Vec<MediaItem>, RepoError> {
1552 unimplemented!()
1553 }
1554
1555 async fn get_resume_items(
1556 &self,
1557 _parent_id: Option<&str>,
1558 _limit: Option<usize>,
1559 ) -> Result<Vec<MediaItem>, RepoError> {
1560 unimplemented!()
1561 }
1562
1563 async fn get_next_up_episodes(
1564 &self,
1565 _series_id: Option<&str>,
1566 _limit: Option<usize>,
1567 ) -> Result<Vec<MediaItem>, RepoError> {
1568 unimplemented!()
1569 }
1570
1571 async fn get_recently_played_audio(
1572 &self,
1573 _limit: Option<usize>,
1574 ) -> Result<Vec<MediaItem>, RepoError> {
1575 unimplemented!()
1576 }
1577
1578 async fn get_resume_movies(
1579 &self,
1580 _limit: Option<usize>,
1581 ) -> Result<Vec<MediaItem>, RepoError> {
1582 unimplemented!()
1583 }
1584
1585 async fn get_rediscover_albums(
1586 &self,
1587 _parent_id: Option<&str>,
1588 _limit: Option<usize>,
1589 ) -> Result<Vec<MediaItem>, RepoError> {
1590 unimplemented!()
1591 }
1592
1593 async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
1594 unimplemented!()
1595 }
1596
1597 async fn search(
1598 &self,
1599 _query: &str,
1600 _options: Option<SearchOptions>,
1601 ) -> Result<SearchResult, RepoError> {
1602 unimplemented!()
1603 }
1604
1605 async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
1606 unimplemented!()
1607 }
1608
1609 async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
1610 unimplemented!()
1611 }
1612
1613 async fn get_audio_only_stream_url_for_video(
1614 &self,
1615 _item_id: &str,
1616 _media_source_id: Option<&str>,
1617 _start_time_seconds: Option<f64>,
1618 _audio_stream_index: Option<i32>,
1619 ) -> Result<String, RepoError> {
1620 unimplemented!()
1621 }
1622
1623 async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
1624 unimplemented!()
1625 }
1626
1627 async fn get_channels(&self) -> Result<SearchResult, RepoError> {
1628 unimplemented!()
1629 }
1630
1631 async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
1632 unimplemented!()
1633 }
1634
1635 async fn report_playback_start(
1636 &self,
1637 _item_id: &str,
1638 _position_ticks: i64,
1639 ) -> Result<(), RepoError> {
1640 unimplemented!()
1641 }
1642
1643 async fn report_playback_progress(
1644 &self,
1645 _item_id: &str,
1646 _position_ticks: i64,
1647 ) -> Result<(), RepoError> {
1648 unimplemented!()
1649 }
1650
1651 async fn report_playback_stopped(
1652 &self,
1653 _item_id: &str,
1654 _position_ticks: i64,
1655 ) -> Result<(), RepoError> {
1656 unimplemented!()
1657 }
1658
1659 fn get_image_url(
1660 &self,
1661 _item_id: &str,
1662 _image_type: ImageType,
1663 _options: Option<ImageOptions>,
1664 ) -> String {
1665 unimplemented!()
1666 }
1667
1668 fn get_subtitle_url(
1669 &self,
1670 _item_id: &str,
1671 _media_source_id: &str,
1672 _stream_index: i32,
1673 _format: &str,
1674 ) -> String {
1675 unimplemented!()
1676 }
1677
1678 fn get_video_download_url(
1679 &self,
1680 _item_id: &str,
1681 _quality: &str,
1682 _media_source_id: Option<&str>,
1683 _source_audio_codec: Option<&str>,
1684 ) -> String {
1685 unimplemented!()
1686 }
1687
1688 async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1689 unimplemented!()
1690 }
1691
1692 async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
1693 unimplemented!()
1694 }
1695
1696 async fn get_favorites(
1697 &self,
1698 _scope: SearchScope,
1699 _options: Option<GetItemsOptions>,
1700 ) -> Result<SearchResult, RepoError> {
1701 unimplemented!()
1702 }
1703
1704 async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
1705 unimplemented!()
1706 }
1707
1708 async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> {
1709 unimplemented!()
1710 }
1711
1712 async fn get_person(&self, _person_id: &str) -> Result<MediaItem, RepoError> {
1713 unimplemented!()
1714 }
1715
1716 async fn get_items_by_person(
1717 &self,
1718 _person_id: &str,
1719 _options: Option<GetItemsOptions>,
1720 ) -> Result<SearchResult, RepoError> {
1721 unimplemented!()
1722 }
1723
1724 async fn get_similar_items(
1725 &self,
1726 _item_id: &str,
1727 _limit: Option<usize>,
1728 ) -> Result<SearchResult, RepoError> {
1729 unimplemented!()
1730 }
1731
1732 async fn create_playlist(
1733 &self,
1734 _name: &str,
1735 _item_ids: &[String],
1736 ) -> Result<PlaylistCreatedResult, RepoError> {
1737 unimplemented!()
1738 }
1739
1740 async fn delete_playlist(&self, _playlist_id: &str) -> Result<(), RepoError> {
1741 unimplemented!()
1742 }
1743
1744 async fn rename_playlist(&self, _playlist_id: &str, _name: &str) -> Result<(), RepoError> {
1745 unimplemented!()
1746 }
1747
1748 async fn get_playlist_items(
1749 &self,
1750 _playlist_id: &str,
1751 ) -> Result<Vec<PlaylistEntry>, RepoError> {
1752 unimplemented!()
1753 }
1754
1755 async fn add_to_playlist(
1756 &self,
1757 _playlist_id: &str,
1758 _item_ids: &[String],
1759 ) -> Result<(), RepoError> {
1760 unimplemented!()
1761 }
1762
1763 async fn remove_from_playlist(
1764 &self,
1765 _playlist_id: &str,
1766 _entry_ids: &[String],
1767 ) -> Result<(), RepoError> {
1768 unimplemented!()
1769 }
1770
1771 async fn move_playlist_item(
1772 &self,
1773 _playlist_id: &str,
1774 _item_id: &str,
1775 _new_index: u32,
1776 ) -> Result<(), RepoError> {
1777 unimplemented!()
1778 }
1779 }
1780
1781 fn create_test_item(id: &str, name: &str) -> MediaItem {
1782 MediaItem {
1783 id: id.to_string(),
1784 name: name.to_string(),
1785 item_type: "Movie".to_string(),
1786 kind: crate::domain::MediaKind::Movie,
1787 is_folder: false,
1788 server_id: "test-server".to_string(),
1789 parent_id: Some("parent-123".to_string()),
1790 library_id: Some("library-456".to_string()),
1791 overview: Some("Test overview".to_string()),
1792 genres: Some(vec!["Action".to_string(), "Adventure".to_string()]),
1793 runtime_ticks: Some(7200000000),
1794 duration_ms: Some(720000),
1795 production_year: Some(2024),
1796 premiere_date: None,
1797 community_rating: Some(8.5),
1798 official_rating: Some("PG-13".to_string()),
1799 primary_image_tag: Some("image-tag-123".to_string()),
1800 image_id: Some("image-tag-123".to_string()),
1801 backdrop_image_tags: Some(vec!["backdrop-1".to_string()]),
1802 parent_backdrop_image_tags: None,
1803 album_id: None,
1804 album_name: None,
1805 album_artist: None,
1806 artists: None,
1807 artist_items: None,
1808 index_number: None,
1809 series_id: None,
1810 series_name: None,
1811 season_id: None,
1812 season_name: None,
1813 parent_index_number: None,
1814 user_data: None,
1815 media_streams: None,
1816 media_sources: None,
1817 people: None,
1818 }
1819 }
1820
1821 struct TestHybridRepo {
1823 offline: Arc<MockOfflineRepo>,
1824 online: Arc<MockOnlineRepo>,
1825 }
1826
1827 impl TestHybridRepo {
1828 fn new(server_items: Vec<MediaItem>) -> Self {
1829 let offline = Arc::new(MockOfflineRepo::new());
1830 let online = Arc::new(MockOnlineRepo::new(server_items));
1831 Self { offline, online }
1832 }
1833
1834 async fn get_items(&self, parent_id: &str) -> Result<SearchResult, RepoError> {
1836 let offline = Arc::clone(&self.offline);
1837 let offline_for_save = Arc::clone(&self.offline);
1838 let online = Arc::clone(&self.online);
1839 let parent_id = parent_id.to_string();
1840 let parent_id_clone = parent_id.clone();
1841 let parent_id_for_save = parent_id.clone();
1842
1843 let cache_future = async move { offline.get_items(&parent_id, None).await };
1845
1846 let server_future = async move { online.get_items(&parent_id_clone, None).await };
1847
1848 let (cache_result, server_result) = tokio::join!(cache_future, server_future);
1850
1851 let cache_had_content = cache_result
1853 .as_ref()
1854 .map(|data| data.has_content())
1855 .unwrap_or(false);
1856
1857 let result = if cache_had_content {
1859 cache_result?
1860 } else {
1861 let server_data = server_result?;
1863
1864 if !server_data.items.is_empty() {
1865 let items_clone = server_data.items.clone();
1866 offline_for_save
1867 .save_to_cache(&parent_id_for_save, &items_clone)
1868 .await?;
1869 }
1870
1871 server_data
1872 };
1873
1874 Ok(result)
1875 }
1876
1877 async fn get_items_gated(&self, parent_id: &str) -> Result<SearchResult, RepoError> {
1883 if !crate::repository::offline::include_catalog_browse() {
1884 let items = self.offline.get_items(parent_id, None).await?;
1885 return Ok(items);
1887 }
1888 self.get_items(parent_id).await
1889 }
1890 }
1891
1892 static GATE_TEST_LOCK: Mutex<()> = Mutex::new(());
1895
1896 #[tokio::test]
1902 async fn test_get_items_gate_off_empty_does_not_query_server() {
1903 let _guard = GATE_TEST_LOCK.lock_safe();
1904 crate::repository::offline::set_include_catalog_browse(false);
1905
1906 let repo = TestHybridRepo::new(vec![
1908 create_test_item("s-1", "Server 1"),
1909 create_test_item("s-2", "Server 2"),
1910 ]);
1911
1912 let result = repo.get_items_gated("parent-123").await.unwrap();
1913
1914 assert_eq!(
1915 result.items.len(),
1916 0,
1917 "empty offline result is authoritative when the gate is off"
1918 );
1919 assert_eq!(
1920 repo.online.get_query_count(),
1921 0,
1922 "server must NOT be queried when the gate is off"
1923 );
1924
1925 crate::repository::offline::set_include_catalog_browse(true);
1926 }
1927
1928 #[tokio::test]
1934 async fn test_get_items_gate_on_empty_falls_through_to_server() {
1935 let _guard = GATE_TEST_LOCK.lock_safe();
1936 crate::repository::offline::set_include_catalog_browse(true);
1937
1938 let repo = TestHybridRepo::new(vec![
1939 create_test_item("s-1", "Server 1"),
1940 create_test_item("s-2", "Server 2"),
1941 ]);
1942
1943 let result = repo.get_items_gated("parent-123").await.unwrap();
1944
1945 assert_eq!(result.items.len(), 2, "server result used on empty cache");
1946 assert_eq!(
1947 repo.online.get_query_count(),
1948 1,
1949 "server IS queried when the gate is on and the cache is empty"
1950 );
1951 }
1952
1953 #[tokio::test]
1959 async fn test_cache_miss_saves_to_cache() {
1960 let server_items = vec![
1962 create_test_item("item-1", "Movie 1"),
1963 create_test_item("item-2", "Movie 2"),
1964 create_test_item("item-3", "Movie 3"),
1965 ];
1966
1967 let repo = TestHybridRepo::new(server_items.clone());
1968
1969 let result = repo.get_items("parent-123").await.unwrap();
1971
1972 assert_eq!(result.items.len(), 3);
1974 assert_eq!(result.items[0].id, "item-1");
1975
1976 assert_eq!(
1978 repo.offline.get_query_count(),
1979 1,
1980 "Cache should be queried once"
1981 );
1982 assert_eq!(
1983 repo.online.get_query_count(),
1984 1,
1985 "Server should be queried once"
1986 );
1987
1988 assert_eq!(
1990 repo.offline.get_save_count(),
1991 1,
1992 "Should save to cache on miss"
1993 );
1994 }
1995
1996 #[tokio::test]
2005 async fn test_cache_hit_no_save() {
2006 let server_items = vec![
2008 create_test_item("item-1", "Movie 1"),
2009 create_test_item("item-2", "Movie 2"),
2010 create_test_item("item-3", "Movie 3"),
2011 ];
2012
2013 let repo = TestHybridRepo::new(server_items.clone());
2014
2015 repo.offline
2017 .save_to_cache("parent-123", &server_items)
2018 .await
2019 .unwrap();
2020 assert_eq!(repo.offline.get_save_count(), 1);
2021
2022 let result = repo.get_items("parent-123").await.unwrap();
2024
2025 assert_eq!(result.items.len(), 3);
2027 assert_eq!(result.items[0].id, "item-1");
2028
2029 assert_eq!(repo.offline.get_query_count(), 1, "Cache should be queried");
2031 assert_eq!(
2032 repo.online.get_query_count(),
2033 1,
2034 "Server is queried in parallel"
2035 );
2036
2037 assert_eq!(
2039 repo.offline.get_save_count(),
2040 1,
2041 "Should NOT save when using cache"
2042 );
2043 }
2044
2045 #[tokio::test]
2050 async fn test_empty_cache_returns_empty_result() {
2051 let repo = TestHybridRepo::new(vec![]);
2053
2054 let result = repo.get_items("parent-123").await.unwrap();
2056
2057 assert_eq!(result.items.len(), 0);
2059
2060 assert_eq!(
2062 repo.offline.get_save_count(),
2063 0,
2064 "Should not save empty results"
2065 );
2066 }
2067
2068 #[tokio::test]
2072 async fn test_has_content_check() {
2073 let empty_result = SearchResult {
2075 items: vec![],
2076 total_record_count: 0,
2077 };
2078 assert!(
2079 !empty_result.has_content(),
2080 "Empty result should not have content"
2081 );
2082
2083 let result_with_items = SearchResult {
2084 items: vec![create_test_item("item-1", "Movie 1")],
2085 total_record_count: 1,
2086 };
2087 assert!(
2088 result_with_items.has_content(),
2089 "Result with items should have content"
2090 );
2091 }
2092
2093 #[test]
2094 fn test_merge_search_local_first_then_server_appended() {
2095 let cache = SearchResult {
2096 items: vec![
2097 create_test_item("a", "Cached A"),
2098 create_test_item("b", "Cached B"),
2099 ],
2100 total_record_count: 2,
2101 };
2102 let server = SearchResult {
2103 items: vec![
2104 create_test_item("c", "Server C"),
2105 create_test_item("d", "Server D"),
2106 ],
2107 total_record_count: 2,
2108 };
2109
2110 let merged = HybridRepository::merge_search_results(cache, server);
2111
2112 let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect();
2114 assert_eq!(ids, vec!["a", "b", "c", "d"]);
2115 assert_eq!(merged.total_record_count, 4);
2116 }
2117
2118 #[test]
2119 fn test_merge_search_dedupes_with_server_winning() {
2120 let cache = SearchResult {
2123 items: vec![
2124 create_test_item("a", "Cached A"),
2125 create_test_item("b", "Cached B"),
2126 ],
2127 total_record_count: 2,
2128 };
2129 let server = SearchResult {
2130 items: vec![
2131 create_test_item("b", "Server B (fresher)"),
2132 create_test_item("c", "Server C"),
2133 ],
2134 total_record_count: 2,
2135 };
2136
2137 let merged = HybridRepository::merge_search_results(cache, server);
2138
2139 let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect();
2140 assert_eq!(
2141 ids,
2142 vec!["a", "b", "c"],
2143 "no duplicate, local position kept"
2144 );
2145
2146 let b = merged.items.iter().find(|i| i.id == "b").unwrap();
2147 assert_eq!(
2148 b.name, "Server B (fresher)",
2149 "server metadata wins on conflict"
2150 );
2151 assert_eq!(merged.total_record_count, 3);
2152 }
2153
2154 #[test]
2155 fn test_merge_search_handles_empty_sides() {
2156 let only_server = HybridRepository::merge_search_results(
2157 SearchResult {
2158 items: vec![],
2159 total_record_count: 0,
2160 },
2161 SearchResult {
2162 items: vec![create_test_item("x", "X")],
2163 total_record_count: 1,
2164 },
2165 );
2166 assert_eq!(only_server.items.len(), 1);
2167 assert_eq!(only_server.items[0].id, "x");
2168
2169 let only_cache = HybridRepository::merge_search_results(
2170 SearchResult {
2171 items: vec![create_test_item("y", "Y")],
2172 total_record_count: 1,
2173 },
2174 SearchResult {
2175 items: vec![],
2176 total_record_count: 0,
2177 },
2178 );
2179 assert_eq!(only_cache.items.len(), 1);
2180 assert_eq!(only_cache.items[0].id, "y");
2181 }
2182}