// Hybrid repository - parallel racing between cache and server // // @req: UR-002 - Access media when online or offline // @req: IR-013 - SQLite integration for local database // @req: DR-012 - Local database for media metadata cache // @req: DR-013 - Repository pattern for online/offline data access // // TRACES: UR-002, UR-052 | IR-013 | DR-012, DR-013, DR-080 #[cfg(test)] use crate::utils::lock::MutexSafe; use std::sync::Arc; use async_trait::async_trait; use log::{debug, warn}; use tokio::time::{timeout, Duration}; use super::exclusions::ExcludeHidden; use super::{types::*, MediaRepository, OfflineRepository, OnlineRepository}; /// Hybrid repository combining online and offline data sources /// /// Uses cache-first parallel racing strategy: /// - Runs SQLite cache and HTTP server queries in parallel /// - Cache has 100ms timeout for fast feedback /// - Returns cache result if it has meaningful content /// - Falls back to server result if cache is empty/stale /// /// @req: UR-002 - Access media when online or offline /// @req: DR-012 - Local database for media metadata cache /// @req: DR-013 - Repository pattern for online/offline data access pub struct HybridRepository { online: Arc, offline: Arc, } impl HybridRepository { pub fn new(online: OnlineRepository, offline: OfflineRepository) -> Self { Self { online: Arc::new(online), offline: Arc::new(offline), } } /// The signed-in user this repository acts for. /// /// TRACES: UR-069 | DR-120 pub fn user_id(&self) -> &str { self.online.user_id() } /// Download raw bytes from a URL using the shared authenticated HTTP client. /// Delegates to online repository for connection reuse and proper auth. pub async fn download_bytes(&self, url: &str) -> Result, String> { self.online.download_bytes(url).await } /// Remove catalog entries the server no longer has. Cache-only, so it goes /// straight to the offline repository. Callers must only invoke this after a /// crawl in which every library succeeded — see /// `OfflineRepository::prune_stale_catalog` for why a partial crawl must not /// sweep. /// /// TRACES: UR-065 | DR-110 pub async fn prune_stale_catalog( &self, cutoff: &str, item_types: &[String], ) -> Result { self.offline.prune_stale_catalog(cutoff, item_types).await } /// Query the JRay plugin for actors on screen at time `t`. Online-only /// (the plugin lives on the Jellyfin server); empty when JRay isn't present. pub async fn get_jray_actors( &self, item_id: &str, t: f64, ) -> Result, RepoError> { self.online.get_jray_actors(item_id, t).await } /// Get video stream URL. This method is online-only since offline playback /// uses local file paths. /// /// Takes no start position: the URL is an HLS playlist spanning the whole /// item, and a position on it would 400 every segment — see /// `OnlineRepository::get_video_stream_url`. Resume by seeking after load. pub async fn get_video_stream_url( &self, item_id: &str, media_source_id: Option<&str>, audio_stream_index: Option, ) -> Result { self.online .get_video_stream_url(item_id, media_source_id, audio_stream_index) .await } /// Get an audio-only stream URL for a video item (background-audio handoff). /// Online-only, like `get_video_stream_url`. /// /// TRACES: UR-040 | JA-032 pub async fn get_audio_only_stream_url_for_video( &self, item_id: &str, media_source_id: Option<&str>, start_time_seconds: Option, audio_stream_index: Option, ) -> Result { self.online .get_audio_only_stream_url_for_video( item_id, media_source_id, start_time_seconds, audio_stream_index, ) .await } /// Every track of an album, asked of the **server** rather than the cache. /// /// Deliberately not `get_items`, which is cache-first: it answers from SQLite /// the moment the cache has any content. That is right for browsing and wrong /// for deciding what to download, because a partial or unlinked cache then /// decides how much of the album gets queued while the user is told the whole /// album is downloading. Downloading is the one operation that must know the /// album's *complete* contents. /// /// Errors when the server cannot answer (offline); the caller falls back to /// the local catalog and the rows are queued either way, resolving on /// reconnect. Server results are written back to the cache, so browsing /// benefits from the round trip too. /// /// TRACES: UR-018, UR-055 | DR-173 pub async fn get_album_tracks(&self, album_id: &str) -> Result, RepoError> { let options = Some(GetItemsOptions { include_item_types: Some(vec!["Audio".to_string()]), sort_by: Some("ParentIndexNumber,IndexNumber,SortName".to_string()), limit: Some(1000), ..Default::default() }); let result = self.online.get_items(album_id, options).await?; if !result.items.is_empty() { if let Err(e) = self.offline.save_to_cache(album_id, &result.items).await { warn!("[HybridRepo] Failed to cache album tracks: {:?}", e); } } Ok(result.items) } /// Immediate children of a container with the user's browsing exclusions /// **not** applied. /// /// Exists for the exclusion picker in settings. Everything else in this /// repository hides what the user has hidden, which would make the setting /// one-way: a folder already excluded would vanish from the list of folders /// to exclude and could never be un-hidden. Server-first so the picker sees /// the real library, falling back to the cache when unreachable. /// /// TRACES: UR-076 | DR-209 pub async fn get_items_unfiltered( &self, parent_id: &str, options: Option, ) -> Result { match self.online.get_items(parent_id, options.clone()).await { Ok(result) => Ok(result), Err(e) => self.offline.get_items(parent_id, options).await.or(Err(e)), } } /// Search only the local SQLite cache (downloaded content). /// /// Fast (100ms timeout) — used to render instant results before the server /// responds. Returns an empty result rather than erroring on timeout so the /// caller can still fall through to the server. pub async fn search_cache_only( &self, query: &str, options: Option, ) -> Result { let offline = Arc::clone(&self.offline); let query = query.to_string(); self.cache_with_timeout(async move { offline.search(&query, options).await }) .await .map(ExcludeHidden::without_excluded) } /// Favourites held locally, without touching the server. Backs the instant /// leg of the two-phase favourites read in the command layer. /// /// TRACES: UR-067 | DR-115 pub async fn get_favorites_cache_only( &self, scope: SearchScope, options: Option, ) -> Result { let offline = Arc::clone(&self.offline); self.cache_with_timeout(async move { offline.get_favorites(scope, options).await }) .await .map(ExcludeHidden::without_excluded) } /// Favourites straight from the server, persisted to the cache on the way /// through — which is also what mirrors their favourite flags into /// `user_data` (DR-114), so the next offline read agrees with the server. /// /// TRACES: UR-067 | DR-115 pub async fn get_favorites_server_only( &self, scope: SearchScope, options: Option, ) -> Result { let result = self.online.get_favorites(scope, options).await?; if !result.items.is_empty() { // Favourites span libraries, so there is no single parent to file // them under; the parent id is only used for stub rows. if let Err(e) = self.offline.save_to_cache("favorites", &result.items).await { debug!("[HybridRepo] Failed to cache favourites: {:?}", e); } } Ok(result.without_excluded()) } /// Fetch a folder's items from the live server and persist them to the /// offline cache synchronously (unlike `get_items`, which saves in a /// fire-and-forget background task after a 100ms cache race). /// /// Used by the full-catalog pre-sync (`sync_full_catalog`) to deterministically /// walk every library while online so the whole catalog is browsable — greyed /// out — offline. Returns the items fetched so the caller can recurse into /// containers. Server-only: errors if unreachable. pub async fn cache_items_from_server( &self, parent_id: &str, options: Option, ) -> Result, RepoError> { let result = self.online.get_items(parent_id, options).await?; if !result.items.is_empty() { self.offline.save_to_cache(parent_id, &result.items).await?; } Ok(result.items) } /// Browse downloaded content only — the dedicated Downloads surface. /// /// Bypasses the cache/server merge entirely and reads the offline repository /// directly, so an empty result is authoritative ("nothing downloaded here") /// and never falls through to the server (DR-080). Available online too — a /// user who is reachable still wants to browse what's on the device. /// /// TRACES: UR-055 | DR-082, DR-083 pub async fn get_downloaded_items( &self, parent_id: &str, options: Option, ) -> Result { // Deliberately *not* filtered by the user's hidden folders: this surface // manages what is on the device, and hiding a download would leave the // user unable to delete a file they can still see the disk usage of. // TRACES: UR-076 | DR-209 self.offline.get_downloaded_items(parent_id, options).await } /// Libraries that contain downloaded content (offline-only, authoritative). /// /// TRACES: UR-055 | DR-082 pub async fn get_downloaded_libraries(&self) -> Result, RepoError> { self.offline.get_downloaded_libraries().await } /// On-disk usage of downloaded content, for the disk-usage display. /// /// TRACES: UR-056 | DR-085 pub async fn get_download_disk_usage(&self) -> Result { self.offline.get_download_disk_usage().await } /// Search only the live Jellyfin server (full library). pub async fn search_server_only( &self, query: &str, options: Option, ) -> Result { self.online .search(query, options) .await .map(ExcludeHidden::without_excluded) } /// Merge cache and server search results into a single de-duplicated list. /// /// Ordering: local (cached/downloaded) items first, then server-only items /// appended. On a duplicate `id`, the server's item wins (fresher, more /// complete metadata) but keeps the local item's earlier position. pub fn merge_search_results(cache: SearchResult, server: SearchResult) -> SearchResult { use std::collections::HashMap; // Index server items by id so we can (a) override duplicates with the // server's metadata and (b) know which server items are brand new. let mut server_by_id: HashMap = HashMap::new(); let mut server_order: Vec = Vec::with_capacity(server.items.len()); for item in server.items { if !server_by_id.contains_key(&item.id) { server_order.push(item.id.clone()); } server_by_id.insert(item.id.clone(), item); } let mut items: Vec = Vec::new(); let mut seen: std::collections::HashSet = std::collections::HashSet::new(); // Local items first, in their original order. If the server also // returned this item, take the server's copy (newer metadata). for local in cache.items { if !seen.insert(local.id.clone()) { continue; } match server_by_id.remove(&local.id) { Some(server_item) => items.push(server_item), None => items.push(local), } } // Then append server-only items, preserving the server's order. for id in server_order { if let Some(server_item) = server_by_id.remove(&id) { if seen.insert(id) { items.push(server_item); } } } let total_record_count = items.len(); SearchResult { items, total_record_count, } } /// Cache-first query: try cache, fall back to server on miss. /// /// 1. Check cache (100ms timeout applied by caller via cache_with_timeout) /// 2. If cache has meaningful content → return immediately (fast path) /// 3. If cache is empty/stale → query server (fresh data) /// 4. If server fails → return cache even if empty (offline fallback) /// /// Both legs are passed through [`ExcludeHidden`] before the "does the cache /// have content?" question is asked. This is the single place the cache and /// server results of a cache-first query converge, so applying the user's /// browsing exclusions here covers every query built on it at once — and /// filtering *before* the content check is what makes a cache page holding /// nothing but hidden items fall through to the server instead of being /// served as an empty listing. /// /// @req: UR-002 - Access media when online or offline /// @req: DR-013 - Repository pattern for online/offline data access /// /// TRACES: UR-002, UR-076 | DR-013, DR-209 async fn parallel_race( &self, cache_future: F1, server_future: F2, ) -> Result where T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static, F1: std::future::Future> + Send, F2: std::future::Future> + Send, { // Try cache first (100ms timeout already applied by callers) let cache_result = cache_future.await.map(ExcludeHidden::without_excluded); if let Ok(data) = &cache_result { if data.has_content() { debug!("[HybridRepo] Cache hit, returning immediately"); return Ok(data.clone()); } } // Cache miss — fall back to server debug!("[HybridRepo] Cache miss, querying server"); match server_future.await { Ok(data) => Ok(data.without_excluded()), Err(e) => { // Server failed, try to return cache even if empty cache_result.or(Err(e)) } } } /// [`Self::parallel_race`], plus a callback fired on the fast path so the /// caller can refresh the cache in the background. /// /// A plain cache hit answers from data that may be arbitrarily old, which /// is right for the *response* and wrong for what it leaves behind: per-user /// state (watch positions, favourites) only reaches the local tables when a /// server result is cached, so a surface that always hits cache never learns /// what another device did. `get_items` had a bespoke version of this; this /// is the same idea, reusable. /// /// The callback runs only on a cache hit — on a miss the server result is /// already being fetched and cached by the normal path. /// /// TRACES: UR-002, UR-025, UR-076 | DR-155, DR-209 async fn race_with_refresh( &self, cache_future: F1, server_future: F2, on_cache_hit: R, ) -> Result where T: MeaningfulContent + ExcludeHidden + Clone + Send + 'static, F1: std::future::Future> + Send, F2: std::future::Future> + Send, R: FnOnce(), { let cache_result = cache_future.await.map(ExcludeHidden::without_excluded); if let Ok(data) = &cache_result { if data.has_content() { debug!("[HybridRepo] Cache hit, returning immediately (refreshing in background)"); on_cache_hit(); return Ok(data.clone()); } } debug!("[HybridRepo] Cache miss, querying server"); match server_future.await { Ok(data) => Ok(data.without_excluded()), Err(e) => cache_result.or(Err(e)), } } /// Simple timeout wrapper for cache queries (100ms timeout) /// /// @req: DR-013 - Repository pattern (cache-first with timeout) async fn cache_with_timeout( &self, future: impl std::future::Future> + Send, ) -> Result { timeout(Duration::from_millis(100), future) .await .unwrap_or_else(|_| { Err(RepoError::Database { message: "Cache query timeout".to_string(), }) }) } } #[async_trait] impl MediaRepository for HybridRepository { async fn get_libraries(&self) -> Result, RepoError> { // Cache-first (100ms). On a cache hit, refresh the cache from the server // in the background. On a miss, fetch from the server and persist so the // list is available on the next (possibly offline) startup. let cache_result = self.cache_with_timeout(self.offline.get_libraries()).await; if let Ok(libs) = &cache_result { if libs.has_content() { debug!("[HybridRepo] Cache hit for libraries, returning immediately"); let online = Arc::clone(&self.online); let offline = Arc::clone(&self.offline); tokio::spawn(async move { if let Ok(server_libs) = online.get_libraries().await { if !server_libs.is_empty() { if let Err(e) = offline.save_libraries_to_cache(&server_libs).await { warn!( "[HybridRepo] Background library cache update failed: {:?}", e ); } } } }); return cache_result; } } // Cache miss — fetch from server and persist for offline use. match self.online.get_libraries().await { Ok(server_libs) => { if !server_libs.is_empty() { if let Err(e) = self.offline.save_libraries_to_cache(&server_libs).await { warn!( "[HybridRepo] Failed to cache {} libraries: {:?}", server_libs.len(), e ); } } Ok(server_libs) } Err(e) => cache_result.or(Err(e)), } } async fn get_items( &self, parent_id: &str, options: Option, ) -> Result { let offline = Arc::clone(&self.offline); let offline_for_save = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let parent_id = parent_id.to_string(); let parent_id_clone = parent_id.clone(); let parent_id_for_save = parent_id.clone(); let opts_clone = options.clone(); // Start server request in background (non-blocking) let server_handle = tokio::spawn(async move { online.get_items(&parent_id_clone, options).await }); // Check cache first (fast, 100ms timeout). // // Exclusions are applied here rather than at each return below so the // "has content" decisions further down are made about what the user will // actually see. `get_items` is the one query that does not go through // `parallel_race` — it interleaves the downloads-only gate and a // background cache write — so it applies the filter itself. // TRACES: UR-076 | DR-209 let cache_result = self .cache_with_timeout(async move { offline.get_items(&parent_id, opts_clone).await }) .await .map(ExcludeHidden::without_excluded); // Downloads-only gate: when the "Show all server media" toggle is off // (offline), an empty offline result is authoritative — the user asked // for downloaded media only and this library has none. Return it as-is // rather than falling through to the server, which would re-pad the page // with the full catalog and re-defeat the filter (DR-080). When the flag // is on (the default, and always so while reachable) behaviour below is // unchanged, including the background cache refresh on a hit. if !crate::repository::offline::include_catalog_browse() { if let Ok(data) = &cache_result { debug!( "[HybridRepo] Downloads-only gate: returning offline result ({} items) as authoritative for parent {}", data.items.len(), &parent_id_for_save[..8.min(parent_id_for_save.len())] ); // Abort the in-flight server request; we won't use it. server_handle.abort(); return Ok(data.clone()); } } // Cache hit: return immediately, update cache in background if let Ok(data) = &cache_result { if data.has_content() { debug!( "[HybridRepo] Cache hit for get_items, returning immediately for parent {}", &parent_id_for_save[..8.min(parent_id_for_save.len())] ); // Background: save server result to cache when it arrives tokio::spawn(async move { match server_handle.await { Ok(Ok(server_data)) if !server_data.items.is_empty() => { if let Err(e) = offline_for_save .save_to_cache(&parent_id_for_save, &server_data.items) .await { warn!("[HybridRepo] Background cache update failed: {:?}", e); } else { debug!( "[HybridRepo] Background updated {} cached items for parent {}", server_data.items.len(), &parent_id_for_save[..8.min(parent_id_for_save.len())] ); } } _ => {} // Server failed or returned empty — keep existing cache } }); return Ok(data.clone()); } } // Cache miss — wait for server result match server_handle.await { Ok(Ok(server_data)) => { if !server_data.items.is_empty() { let items_clone = server_data.items.clone(); tokio::spawn(async move { if let Err(e) = offline_for_save .save_to_cache(&parent_id_for_save, &items_clone) .await { warn!( "[HybridRepo] Failed to save {} items to cache: {:?}", items_clone.len(), e ); } else { debug!( "[HybridRepo] Saved {} items to cache for parent {}", items_clone.len(), &parent_id_for_save[..8.min(parent_id_for_save.len())] ); } }); } // The cache keeps the server's full page (above) — an exclusion // is a view preference and can be undone, so hiding items from // the *cache* would make un-hiding them require a re-crawl. Only // what is handed back is filtered. // TRACES: UR-076 | DR-209 Ok(server_data.without_excluded()) } Ok(Err(e)) => cache_result.or(Err(e)), Err(join_err) => cache_result.or(Err(RepoError::Network { message: format!("Server task failed: {}", join_err), })), } } /// A single item, cache-first — and, on a cache hit, refreshed in the /// background so the stored copy keeps up with the server. /// /// The background refresh is what carries per-user state home: caching an /// item runs `mirror_user_data`, which is the only path by which a watch /// position set on another device reaches the local `user_data` row the /// resume check reads. Without it a cache hit returned this device's own /// stale position forever and cross-device resume silently did nothing — /// `get_items` already refreshes this way, so browsing a season worked /// while opening the episode directly did not. /// /// The refreshed value lands for the *next* read rather than this one: the /// point of the cache-first race is to answer immediately. /// /// TRACES: UR-025, UR-002 | DR-155 | UT-152 async fn get_item(&self, item_id: &str) -> Result { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let item_id = item_id.to_string(); let item_id_clone = item_id.clone(); let cache_future = self.cache_with_timeout(async move { offline.get_item(&item_id).await }); let online_for_refresh = Arc::clone(&self.online); let offline_for_save = Arc::clone(&self.offline); let refresh_id = item_id_clone.clone(); let on_cache_hit = move || { tokio::spawn(async move { match online_for_refresh.get_item(&refresh_id).await { Ok(fresh) => { // `save_to_cache` files the row under a parent; the item's // own parent keeps it where a later listing expects it. let parent = fresh .parent_id .clone() .unwrap_or_else(|| "item".to_string()); if let Err(e) = offline_for_save.save_to_cache(&parent, &[fresh]).await { debug!("[HybridRepo] Background item refresh failed: {:?}", e); } } Err(e) => debug!("[HybridRepo] Background item refresh unavailable: {:?}", e), } }); }; let server_future = async move { online.get_item(&item_id_clone).await }; self.race_with_refresh(cache_future, server_future, on_cache_hit) .await } async fn get_latest_items( &self, parent_id: &str, limit: Option, ) -> Result, RepoError> { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let parent_id = parent_id.to_string(); let parent_id_clone = parent_id.clone(); let limit_clone = limit; let cache_future = self .cache_with_timeout(async move { offline.get_latest_items(&parent_id, limit).await }); let server_future = async move { online.get_latest_items(&parent_id_clone, limit_clone).await }; self.parallel_race(cache_future, server_future).await } async fn get_resume_items( &self, parent_id: Option<&str>, limit: Option, ) -> Result, RepoError> { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let parent_id_str = parent_id.map(|s| s.to_string()); let parent_id_clone = parent_id_str.clone(); let limit_clone = limit; let cache_future = self.cache_with_timeout(async move { offline .get_resume_items(parent_id_str.as_deref(), limit) .await }); let server_future = async move { online .get_resume_items(parent_id_clone.as_deref(), limit_clone) .await }; self.parallel_race(cache_future, server_future).await } async fn get_next_up_episodes( &self, series_id: Option<&str>, limit: Option, ) -> Result, RepoError> { // Next up is dynamic, always fetch from server self.online .get_next_up_episodes(series_id, limit) .await .map(ExcludeHidden::without_excluded) } async fn get_recently_played_audio( &self, limit: Option, ) -> Result, RepoError> { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let limit_clone = limit; let cache_future = self.cache_with_timeout(async move { offline.get_recently_played_audio(limit).await }); let server_future = async move { online.get_recently_played_audio(limit_clone).await }; self.parallel_race(cache_future, server_future).await } async fn get_resume_movies(&self, limit: Option) -> Result, RepoError> { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let limit_clone = limit; let cache_future = self.cache_with_timeout(async move { offline.get_resume_movies(limit).await }); let server_future = async move { online.get_resume_movies(limit_clone).await }; self.parallel_race(cache_future, server_future).await } async fn get_rediscover_albums( &self, parent_id: Option<&str>, limit: Option, ) -> Result, RepoError> { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let parent_id_owned = parent_id.map(|s| s.to_string()); let parent_id_clone = parent_id_owned.clone(); let cache_future = self.cache_with_timeout(async move { offline .get_rediscover_albums(parent_id_owned.as_deref(), limit) .await }); let server_future = async move { online .get_rediscover_albums(parent_id_clone.as_deref(), limit) .await }; self.parallel_race(cache_future, server_future).await } async fn get_genres(&self, parent_id: Option<&str>) -> Result, RepoError> { // Cache-first (100ms). On a cache hit, refresh the cached genre catalog // from the server in the background. On a miss, fetch from the server and // persist so the full genre list is available offline. Mirrors // get_libraries — NOT parallel_race, whose "any non-empty cache wins" // rule would pin genres to whatever sparse set the local albums yield. let parent_id_str = parent_id.map(|s| s.to_string()); let cache_offline = Arc::clone(&self.offline); let cache_pid = parent_id_str.clone(); let cache_result = self .cache_with_timeout(async move { cache_offline.get_genres(cache_pid.as_deref()).await }) .await; if let Ok(genres) = &cache_result { if genres.has_content() { debug!("[HybridRepo] Cache hit for genres, returning immediately"); let online = Arc::clone(&self.online); let offline = Arc::clone(&self.offline); let pid = parent_id_str.clone(); tokio::spawn(async move { if let Ok(server_genres) = online.get_genres(pid.as_deref()).await { if !server_genres.is_empty() { if let Err(e) = offline .save_genres_to_cache(pid.as_deref(), &server_genres) .await { warn!("[HybridRepo] Background genre cache update failed: {:?}", e); } } } }); return cache_result; } } // Cache miss — fetch from server and persist for offline use. match self.online.get_genres(parent_id_str.as_deref()).await { Ok(server_genres) => { if !server_genres.is_empty() { if let Err(e) = self .offline .save_genres_to_cache(parent_id_str.as_deref(), &server_genres) .await { warn!( "[HybridRepo] Failed to cache {} genres: {:?}", server_genres.len(), e ); } } Ok(server_genres) } Err(e) => cache_result.or(Err(e)), } } async fn search( &self, query: &str, options: Option, ) -> Result { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let query = query.to_string(); let query_clone = query.clone(); let opts_clone = options.clone(); let cache_future = self.cache_with_timeout(async move { offline.search(&query, opts_clone).await }); let server_future = async move { online.search(&query_clone, options).await }; self.parallel_race(cache_future, server_future).await } async fn get_playback_info(&self, item_id: &str) -> Result { // Playback info requires server communication for transcoding decisions self.online.get_playback_info(item_id).await } async fn get_audio_stream_url(&self, item_id: &str) -> Result { // Stream URLs require server communication - delegate to online repository self.online.get_audio_stream_url(item_id).await } async fn get_audio_only_stream_url_for_video( &self, item_id: &str, media_source_id: Option<&str>, start_time_seconds: Option, audio_stream_index: Option, ) -> Result { // Audio-only transcode of a video requires the server - delegate to online. self.online .build_audio_only_stream_url_for_video( item_id, media_source_id, start_time_seconds, audio_stream_index, ) .await } async fn get_live_tv_channels(&self) -> Result, RepoError> { // Live TV requires server communication - delegate to online repository self.online .get_live_tv_channels() .await .map(ExcludeHidden::without_excluded) } async fn get_channels(&self) -> Result { // Plugin channels require server communication - delegate to online repository self.online .get_channels() .await .map(ExcludeHidden::without_excluded) } async fn open_live_stream(&self, item_id: &str) -> Result { // Opening a live stream requires server communication - delegate to online self.online.open_live_stream(item_id).await } async fn report_playback_start( &self, item_id: &str, position_ticks: i64, ) -> Result<(), RepoError> { // Playback reporting goes directly to server self.online .report_playback_start(item_id, position_ticks) .await } async fn report_playback_progress( &self, item_id: &str, position_ticks: i64, ) -> Result<(), RepoError> { // Playback reporting goes directly to server self.online .report_playback_progress(item_id, position_ticks) .await } async fn report_playback_stopped( &self, item_id: &str, position_ticks: i64, ) -> Result<(), RepoError> { // Playback reporting goes directly to server self.online .report_playback_stopped(item_id, position_ticks) .await } fn get_image_url( &self, item_id: &str, image_type: ImageType, options: Option, ) -> String { // Always use online URL for images (thumbnail cache handles offline) self.online.get_image_url(item_id, image_type, options) } fn get_subtitle_url( &self, item_id: &str, media_source_id: &str, stream_index: i32, format: &str, ) -> String { // Always use online URL for subtitles self.online .get_subtitle_url(item_id, media_source_id, stream_index, format) } fn get_video_download_url( &self, item_id: &str, quality: &str, media_source_id: Option<&str>, source_audio_codec: Option<&str>, ) -> String { // Always use online URL for downloads self.online .get_video_download_url(item_id, quality, media_source_id, source_audio_codec) } async fn mark_favorite(&self, item_id: &str) -> Result<(), RepoError> { // Write operations go directly to server self.online.mark_favorite(item_id).await } async fn unmark_favorite(&self, item_id: &str) -> Result<(), RepoError> { // Write operations go directly to server self.online.unmark_favorite(item_id).await } async fn clear_watch_history(&self, item_id: &str) -> Result<(), RepoError> { // Write operations go directly to server self.online.clear_watch_history(item_id).await } async fn mark_played(&self, item_id: &str) -> Result<(), RepoError> { // Write operations go directly to server self.online.mark_played(item_id).await } async fn get_person(&self, person_id: &str) -> Result { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let person_id = person_id.to_string(); let person_id_clone = person_id.clone(); let cache_future = self.cache_with_timeout(async move { offline.get_person(&person_id).await }); let server_future = async move { online.get_person(&person_id_clone).await }; self.parallel_race(cache_future, server_future).await } async fn get_items_by_person( &self, person_id: &str, options: Option, ) -> Result { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let person_id = person_id.to_string(); let person_id_clone = person_id.clone(); let opts_clone = options.clone(); let cache_future = self.cache_with_timeout(async move { offline.get_items_by_person(&person_id, opts_clone).await }); let server_future = async move { online.get_items_by_person(&person_id_clone, options).await }; self.parallel_race(cache_future, server_future).await } /// TRACES: UR-067 | DR-115 async fn get_favorites( &self, scope: SearchScope, options: Option, ) -> Result { let cache_result = self.get_favorites_cache_only(scope, options.clone()).await; // Downloads-only gate: with "Show all server media" off, an empty local // result means "nothing favourited is on this device" and is // authoritative. Falling through to the server here would re-pad the // page with the full favourited catalog and defeat the filter (DR-080). if !crate::repository::offline::include_catalog_browse() { if let Ok(data) = &cache_result { return Ok(data.clone()); } } if let Ok(data) = &cache_result { if data.has_content() { return Ok(data.clone()); } } // Cache miss — answer from the server, *saving through* on the way back. // Every other read path persists what it fetches; skipping it here would // mean the favourites page re-queries the server on every visit and the // offline mirror (DR-114) never learns about favourites marked // elsewhere, since this path is what fills it on a fresh install. match self.get_favorites_server_only(scope, options).await { Ok(data) => Ok(data), Err(e) => cache_result.or(Err(e)), } } async fn get_similar_items( &self, item_id: &str, limit: Option, ) -> Result { let offline = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let item_id = item_id.to_string(); let item_id_clone = item_id.clone(); let cache_future = self .cache_with_timeout(async move { offline.get_similar_items(&item_id, limit).await }); let server_future = async move { online.get_similar_items(&item_id_clone, limit).await }; self.parallel_race(cache_future, server_future).await } // ===== Playlist Methods ===== async fn create_playlist( &self, name: &str, item_ids: &[String], ) -> Result { // Write operation - delegate directly to server self.online.create_playlist(name, item_ids).await } async fn delete_playlist(&self, playlist_id: &str) -> Result<(), RepoError> { // Write operation - delegate directly to server self.online.delete_playlist(playlist_id).await } async fn rename_playlist(&self, playlist_id: &str, name: &str) -> Result<(), RepoError> { // Write operation - delegate directly to server self.online.rename_playlist(playlist_id, name).await } async fn get_playlist_items(&self, playlist_id: &str) -> Result, RepoError> { let offline = Arc::clone(&self.offline); let offline_for_save = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let playlist_id = playlist_id.to_string(); let playlist_id_clone = playlist_id.clone(); let playlist_id_for_save = playlist_id.clone(); // Start server request in background (non-blocking) let server_handle = tokio::spawn(async move { online.get_playlist_items(&playlist_id_clone).await }); // Check cache first (fast, 100ms timeout) let cache_result = self .cache_with_timeout(async move { offline.get_playlist_items(&playlist_id).await }) .await; // Cache hit: return immediately, update cache in background if let Ok(data) = &cache_result { if data.has_content() { debug!("[HybridRepo] Cache hit for playlist items, returning immediately"); tokio::spawn(async move { if let Ok(Ok(server_entries)) = server_handle.await { if let Err(e) = offline_for_save .save_playlist_items_to_cache(&playlist_id_for_save, &server_entries) .await { warn!("[HybridRepo] Failed to update playlist cache: {:?}", e); } } }); return cache_result; } } // Cache miss — wait for server result match server_handle.await { Ok(Ok(entries)) => { let entries_clone = entries.clone(); tokio::spawn(async move { if let Err(e) = offline_for_save .save_playlist_items_to_cache(&playlist_id_for_save, &entries_clone) .await { warn!( "[HybridRepo] Failed to save playlist items to cache: {:?}", e ); } }); Ok(entries) } Ok(Err(e)) => cache_result.or(Err(e)), Err(join_err) => cache_result.or(Err(RepoError::Network { message: format!("Server task failed: {}", join_err), })), } } async fn add_to_playlist( &self, playlist_id: &str, item_ids: &[String], ) -> Result<(), RepoError> { // Write operation - delegate directly to server self.online.add_to_playlist(playlist_id, item_ids).await } async fn remove_from_playlist( &self, playlist_id: &str, entry_ids: &[String], ) -> Result<(), RepoError> { // Write operation - delegate directly to server self.online .remove_from_playlist(playlist_id, entry_ids) .await } async fn move_playlist_item( &self, playlist_id: &str, item_id: &str, new_index: u32, ) -> Result<(), RepoError> { // Write operation - delegate directly to server self.online .move_playlist_item(playlist_id, item_id, new_index) .await } } #[cfg(test)] mod tests { // `GATE_TEST_LOCK` below serialises the tests that flip the process-global // `INCLUDE_CATALOG_BROWSE` flag, so its guard is deliberately held across // the `.await` of the repository call under test — that await *is* the // critical section. This is not the production deadlock hazard the lint // targets: the lock is test-only, uncontended outside these tests, and each // `#[tokio::test]` runs on its own single-threaded runtime, so a held guard // cannot block another task on the same worker. Restructuring around it // would reintroduce the flag race the lock exists to prevent. #![allow(clippy::await_holding_lock)] use super::*; use std::sync::Mutex; /// Mock offline repository that tracks queries and saves struct MockOfflineRepo { items: Arc>>, query_count: Arc>, save_count: Arc>, } impl MockOfflineRepo { fn new() -> Self { Self { items: Arc::new(Mutex::new(Vec::new())), query_count: Arc::new(Mutex::new(0)), save_count: Arc::new(Mutex::new(0)), } } fn get_query_count(&self) -> usize { *self.query_count.lock_safe() } fn get_save_count(&self) -> usize { *self.save_count.lock_safe() } async fn save_to_cache( &self, _parent_id: &str, items: &[MediaItem], ) -> Result { *self.save_count.lock_safe() += 1; *self.items.lock_safe() = items.to_vec(); Ok(items.len()) } } #[async_trait] impl MediaRepository for MockOfflineRepo { async fn get_libraries(&self) -> Result, RepoError> { unimplemented!() } async fn get_items( &self, _parent_id: &str, _options: Option, ) -> Result { *self.query_count.lock_safe() += 1; let items = self.items.lock_safe().clone(); let count = items.len(); Ok(SearchResult { items, total_record_count: count, }) } async fn get_item(&self, _item_id: &str) -> Result { unimplemented!() } async fn get_latest_items( &self, _parent_id: &str, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_resume_items( &self, _parent_id: Option<&str>, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_next_up_episodes( &self, _series_id: Option<&str>, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_recently_played_audio( &self, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_resume_movies( &self, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_rediscover_albums( &self, _parent_id: Option<&str>, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_genres(&self, _parent_id: Option<&str>) -> Result, RepoError> { unimplemented!() } async fn search( &self, _query: &str, _options: Option, ) -> Result { unimplemented!() } async fn get_playback_info(&self, _item_id: &str) -> Result { unimplemented!() } async fn get_audio_stream_url(&self, _item_id: &str) -> Result { unimplemented!() } async fn get_audio_only_stream_url_for_video( &self, _item_id: &str, _media_source_id: Option<&str>, _start_time_seconds: Option, _audio_stream_index: Option, ) -> Result { unimplemented!() } async fn get_live_tv_channels(&self) -> Result, RepoError> { unimplemented!() } async fn get_channels(&self) -> Result { unimplemented!() } async fn open_live_stream(&self, _item_id: &str) -> Result { unimplemented!() } async fn report_playback_start( &self, _item_id: &str, _position_ticks: i64, ) -> Result<(), RepoError> { unimplemented!() } async fn report_playback_progress( &self, _item_id: &str, _position_ticks: i64, ) -> Result<(), RepoError> { unimplemented!() } async fn report_playback_stopped( &self, _item_id: &str, _position_ticks: i64, ) -> Result<(), RepoError> { unimplemented!() } fn get_image_url( &self, _item_id: &str, _image_type: ImageType, _options: Option, ) -> String { unimplemented!() } fn get_subtitle_url( &self, _item_id: &str, _media_source_id: &str, _stream_index: i32, _format: &str, ) -> String { unimplemented!() } fn get_video_download_url( &self, _item_id: &str, _quality: &str, _media_source_id: Option<&str>, _source_audio_codec: Option<&str>, ) -> String { unimplemented!() } async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn get_favorites( &self, _scope: SearchScope, _options: Option, ) -> Result { unimplemented!() } async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn get_person(&self, _person_id: &str) -> Result { unimplemented!() } async fn get_items_by_person( &self, _person_id: &str, _options: Option, ) -> Result { unimplemented!() } async fn get_similar_items( &self, _item_id: &str, _limit: Option, ) -> Result { unimplemented!() } async fn create_playlist( &self, _name: &str, _item_ids: &[String], ) -> Result { unimplemented!() } async fn delete_playlist(&self, _playlist_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn rename_playlist(&self, _playlist_id: &str, _name: &str) -> Result<(), RepoError> { unimplemented!() } async fn get_playlist_items( &self, _playlist_id: &str, ) -> Result, RepoError> { unimplemented!() } async fn add_to_playlist( &self, _playlist_id: &str, _item_ids: &[String], ) -> Result<(), RepoError> { unimplemented!() } async fn remove_from_playlist( &self, _playlist_id: &str, _entry_ids: &[String], ) -> Result<(), RepoError> { unimplemented!() } async fn move_playlist_item( &self, _playlist_id: &str, _item_id: &str, _new_index: u32, ) -> Result<(), RepoError> { unimplemented!() } } /// Mock online repository that returns predefined items struct MockOnlineRepo { items: Vec, query_count: Arc>, } impl MockOnlineRepo { fn new(items: Vec) -> Self { Self { items, query_count: Arc::new(Mutex::new(0)), } } fn get_query_count(&self) -> usize { *self.query_count.lock_safe() } } #[async_trait] impl MediaRepository for MockOnlineRepo { async fn get_libraries(&self) -> Result, RepoError> { unimplemented!() } async fn get_items( &self, _parent_id: &str, _options: Option, ) -> Result { *self.query_count.lock_safe() += 1; Ok(SearchResult { items: self.items.clone(), total_record_count: self.items.len(), }) } async fn get_item(&self, _item_id: &str) -> Result { unimplemented!() } async fn get_latest_items( &self, _parent_id: &str, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_resume_items( &self, _parent_id: Option<&str>, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_next_up_episodes( &self, _series_id: Option<&str>, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_recently_played_audio( &self, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_resume_movies( &self, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_rediscover_albums( &self, _parent_id: Option<&str>, _limit: Option, ) -> Result, RepoError> { unimplemented!() } async fn get_genres(&self, _parent_id: Option<&str>) -> Result, RepoError> { unimplemented!() } async fn search( &self, _query: &str, _options: Option, ) -> Result { unimplemented!() } async fn get_playback_info(&self, _item_id: &str) -> Result { unimplemented!() } async fn get_audio_stream_url(&self, _item_id: &str) -> Result { unimplemented!() } async fn get_audio_only_stream_url_for_video( &self, _item_id: &str, _media_source_id: Option<&str>, _start_time_seconds: Option, _audio_stream_index: Option, ) -> Result { unimplemented!() } async fn get_live_tv_channels(&self) -> Result, RepoError> { unimplemented!() } async fn get_channels(&self) -> Result { unimplemented!() } async fn open_live_stream(&self, _item_id: &str) -> Result { unimplemented!() } async fn report_playback_start( &self, _item_id: &str, _position_ticks: i64, ) -> Result<(), RepoError> { unimplemented!() } async fn report_playback_progress( &self, _item_id: &str, _position_ticks: i64, ) -> Result<(), RepoError> { unimplemented!() } async fn report_playback_stopped( &self, _item_id: &str, _position_ticks: i64, ) -> Result<(), RepoError> { unimplemented!() } fn get_image_url( &self, _item_id: &str, _image_type: ImageType, _options: Option, ) -> String { unimplemented!() } fn get_subtitle_url( &self, _item_id: &str, _media_source_id: &str, _stream_index: i32, _format: &str, ) -> String { unimplemented!() } fn get_video_download_url( &self, _item_id: &str, _quality: &str, _media_source_id: Option<&str>, _source_audio_codec: Option<&str>, ) -> String { unimplemented!() } async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn get_favorites( &self, _scope: SearchScope, _options: Option, ) -> Result { unimplemented!() } async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn get_person(&self, _person_id: &str) -> Result { unimplemented!() } async fn get_items_by_person( &self, _person_id: &str, _options: Option, ) -> Result { unimplemented!() } async fn get_similar_items( &self, _item_id: &str, _limit: Option, ) -> Result { unimplemented!() } async fn create_playlist( &self, _name: &str, _item_ids: &[String], ) -> Result { unimplemented!() } async fn delete_playlist(&self, _playlist_id: &str) -> Result<(), RepoError> { unimplemented!() } async fn rename_playlist(&self, _playlist_id: &str, _name: &str) -> Result<(), RepoError> { unimplemented!() } async fn get_playlist_items( &self, _playlist_id: &str, ) -> Result, RepoError> { unimplemented!() } async fn add_to_playlist( &self, _playlist_id: &str, _item_ids: &[String], ) -> Result<(), RepoError> { unimplemented!() } async fn remove_from_playlist( &self, _playlist_id: &str, _entry_ids: &[String], ) -> Result<(), RepoError> { unimplemented!() } async fn move_playlist_item( &self, _playlist_id: &str, _item_id: &str, _new_index: u32, ) -> Result<(), RepoError> { unimplemented!() } } fn create_test_item(id: &str, name: &str) -> MediaItem { MediaItem { id: id.to_string(), name: name.to_string(), item_type: "Movie".to_string(), kind: crate::domain::MediaKind::Movie, is_folder: false, server_id: "test-server".to_string(), parent_id: Some("parent-123".to_string()), library_id: Some("library-456".to_string()), overview: Some("Test overview".to_string()), genres: Some(vec!["Action".to_string(), "Adventure".to_string()]), runtime_ticks: Some(7200000000), duration_ms: Some(720000), production_year: Some(2024), premiere_date: None, community_rating: Some(8.5), official_rating: Some("PG-13".to_string()), primary_image_tag: Some("image-tag-123".to_string()), image_id: Some("image-tag-123".to_string()), backdrop_image_tags: Some(vec!["backdrop-1".to_string()]), parent_backdrop_image_tags: None, album_id: None, album_name: None, album_artist: None, artists: None, artist_items: None, index_number: None, series_id: None, series_name: None, season_id: None, season_name: None, parent_index_number: None, user_data: None, media_streams: None, media_sources: None, people: None, } } /// Helper to test the caching logic struct TestHybridRepo { offline: Arc, online: Arc, } impl TestHybridRepo { fn new(server_items: Vec) -> Self { let offline = Arc::new(MockOfflineRepo::new()); let online = Arc::new(MockOnlineRepo::new(server_items)); Self { offline, online } } /// Test version of get_items that implements the cache logic async fn get_items(&self, parent_id: &str) -> Result { let offline = Arc::clone(&self.offline); let offline_for_save = Arc::clone(&self.offline); let online = Arc::clone(&self.online); let parent_id = parent_id.to_string(); let parent_id_clone = parent_id.clone(); let parent_id_for_save = parent_id.clone(); // Check cache first let cache_future = async move { offline.get_items(&parent_id, None).await }; let server_future = async move { online.get_items(&parent_id_clone, None).await }; // Wait for both, prefer cache if available let (cache_result, server_result) = tokio::join!(cache_future, server_future); // Check if cache had meaningful content let cache_had_content = cache_result .as_ref() .map(|data| data.has_content()) .unwrap_or(false); // Prefer cache if it has content (mimics hybrid.rs get_items logic) let result = if cache_had_content { cache_result? } else { // Use server result and save to cache for next time let server_data = server_result?; if !server_data.items.is_empty() { let items_clone = server_data.items.clone(); offline_for_save .save_to_cache(&parent_id_for_save, &items_clone) .await?; } server_data }; Ok(result) } /// Test version mirroring the real `HybridRepository::get_items` /// downloads-only gate: when `include_catalog_browse()` is false, the /// offline result is authoritative and the server is NOT queried, even /// when the cache is empty. Otherwise falls through to the normal /// cache-first logic in `get_items`. async fn get_items_gated(&self, parent_id: &str) -> Result { if !crate::repository::offline::include_catalog_browse() { let items = self.offline.get_items(parent_id, None).await?; // Authoritative: return as-is, never touch the server. return Ok(items); } self.get_items(parent_id).await } } /// Serialize tests that mutate the process-global INCLUDE_CATALOG_BROWSE /// flag, and always restore it to the default (true) afterwards. static GATE_TEST_LOCK: Mutex<()> = Mutex::new(()); /// UT-070: with the downloads-only gate off, an empty offline result is /// returned as-is and the server is NOT queried. /// /// @req-test: UR-052 - Offline "downloaded only" filtering /// @req-test: DR-080 - Empty offline result is authoritative when gate off #[tokio::test] async fn test_get_items_gate_off_empty_does_not_query_server() { let _guard = GATE_TEST_LOCK.lock_safe(); crate::repository::offline::set_include_catalog_browse(false); // Server has items, cache is empty. Gate off ⇒ the server must be ignored. let repo = TestHybridRepo::new(vec![ create_test_item("s-1", "Server 1"), create_test_item("s-2", "Server 2"), ]); let result = repo.get_items_gated("parent-123").await.unwrap(); assert_eq!( result.items.len(), 0, "empty offline result is authoritative when the gate is off" ); assert_eq!( repo.online.get_query_count(), 0, "server must NOT be queried when the gate is off" ); crate::repository::offline::set_include_catalog_browse(true); } /// Guard the online path: with the gate ON and an empty cache, get_items /// still falls through to the server (unchanged behaviour). /// /// @req-test: UR-052 - Offline "downloaded only" filtering /// @req-test: DR-080 - Gate on ⇒ empty cache still queries the server #[tokio::test] async fn test_get_items_gate_on_empty_falls_through_to_server() { let _guard = GATE_TEST_LOCK.lock_safe(); crate::repository::offline::set_include_catalog_browse(true); let repo = TestHybridRepo::new(vec![ create_test_item("s-1", "Server 1"), create_test_item("s-2", "Server 2"), ]); let result = repo.get_items_gated("parent-123").await.unwrap(); assert_eq!(result.items.len(), 2, "server result used on empty cache"); assert_eq!( repo.online.get_query_count(), 1, "server IS queried when the gate is on and the cache is empty" ); } /// Test cache miss saves server data to cache for next time /// /// @req-test: UR-002 - Access media when online or offline /// @req-test: DR-013 - Repository pattern for online/offline data access /// @req-test: DR-012 - Local database for media metadata cache #[tokio::test] async fn test_cache_miss_saves_to_cache() { // Setup: Server has 3 items, cache is empty let server_items = vec![ create_test_item("item-1", "Movie 1"), create_test_item("item-2", "Movie 2"), create_test_item("item-3", "Movie 3"), ]; let repo = TestHybridRepo::new(server_items.clone()); // First request - cache miss let result = repo.get_items("parent-123").await.unwrap(); // Should return server items assert_eq!(result.items.len(), 3); assert_eq!(result.items[0].id, "item-1"); // Should have queried both cache and server assert_eq!( repo.offline.get_query_count(), 1, "Cache should be queried once" ); assert_eq!( repo.online.get_query_count(), 1, "Server should be queried once" ); // Should have saved to cache assert_eq!( repo.offline.get_save_count(), 1, "Should save to cache on miss" ); } /// Test cache hit prevents duplicate save to cache /// /// Verifies parallel racing strategy: both cache and server are queried, /// but when cache has content, it's used and no duplicate save occurs. /// /// @req-test: UR-002 - Access media when online or offline /// @req-test: DR-013 - Repository pattern for online/offline data access /// @req-test: DR-012 - Local database cache (avoid duplicate writes) #[tokio::test] async fn test_cache_hit_no_save() { // Setup: Server has 3 items, we'll pre-populate cache let server_items = vec![ create_test_item("item-1", "Movie 1"), create_test_item("item-2", "Movie 2"), create_test_item("item-3", "Movie 3"), ]; let repo = TestHybridRepo::new(server_items.clone()); // Pre-populate cache repo.offline .save_to_cache("parent-123", &server_items) .await .unwrap(); assert_eq!(repo.offline.get_save_count(), 1); // Second request - cache hit let result = repo.get_items("parent-123").await.unwrap(); // Should return cached items assert_eq!(result.items.len(), 3); assert_eq!(result.items[0].id, "item-1"); // Should have queried cache and server (parallel race) assert_eq!(repo.offline.get_query_count(), 1, "Cache should be queried"); assert_eq!( repo.online.get_query_count(), 1, "Server is queried in parallel" ); // Should NOT have saved again (no duplicate save) assert_eq!( repo.offline.get_save_count(), 1, "Should NOT save when using cache" ); } /// Test empty results are not saved to cache /// /// @req-test: DR-013 - Repository pattern (edge case handling) /// @req-test: DR-012 - Local database cache (avoid saving empty data) #[tokio::test] async fn test_empty_cache_returns_empty_result() { // Setup: Server has no items let repo = TestHybridRepo::new(vec![]); // Request with empty server let result = repo.get_items("parent-123").await.unwrap(); // Should return empty result assert_eq!(result.items.len(), 0); // Should NOT save empty results assert_eq!( repo.offline.get_save_count(), 0, "Should not save empty results" ); } /// Test SearchResult::has_content helper method /// /// @req-test: DR-013 - Repository pattern (content detection helper) #[tokio::test] async fn test_has_content_check() { // Test that SearchResult::has_content works correctly let empty_result = SearchResult { items: vec![], total_record_count: 0, }; assert!( !empty_result.has_content(), "Empty result should not have content" ); let result_with_items = SearchResult { items: vec![create_test_item("item-1", "Movie 1")], total_record_count: 1, }; assert!( result_with_items.has_content(), "Result with items should have content" ); } #[test] fn test_merge_search_local_first_then_server_appended() { let cache = SearchResult { items: vec![ create_test_item("a", "Cached A"), create_test_item("b", "Cached B"), ], total_record_count: 2, }; let server = SearchResult { items: vec![ create_test_item("c", "Server C"), create_test_item("d", "Server D"), ], total_record_count: 2, }; let merged = HybridRepository::merge_search_results(cache, server); // Local items first (in order), then server-only items appended. let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect(); assert_eq!(ids, vec!["a", "b", "c", "d"]); assert_eq!(merged.total_record_count, 4); } #[test] fn test_merge_search_dedupes_with_server_winning() { // "b" appears in both. Server metadata should win, but the item keeps // its earlier (local) position and is not duplicated. let cache = SearchResult { items: vec![ create_test_item("a", "Cached A"), create_test_item("b", "Cached B"), ], total_record_count: 2, }; let server = SearchResult { items: vec![ create_test_item("b", "Server B (fresher)"), create_test_item("c", "Server C"), ], total_record_count: 2, }; let merged = HybridRepository::merge_search_results(cache, server); let ids: Vec<&str> = merged.items.iter().map(|i| i.id.as_str()).collect(); assert_eq!( ids, vec!["a", "b", "c"], "no duplicate, local position kept" ); let b = merged.items.iter().find(|i| i.id == "b").unwrap(); assert_eq!( b.name, "Server B (fresher)", "server metadata wins on conflict" ); assert_eq!(merged.total_record_count, 3); } #[test] fn test_merge_search_handles_empty_sides() { let only_server = HybridRepository::merge_search_results( SearchResult { items: vec![], total_record_count: 0, }, SearchResult { items: vec![create_test_item("x", "X")], total_record_count: 1, }, ); assert_eq!(only_server.items.len(), 1); assert_eq!(only_server.items[0].id, "x"); let only_cache = HybridRepository::merge_search_results( SearchResult { items: vec![create_test_item("y", "Y")], total_record_count: 1, }, SearchResult { items: vec![], total_record_count: 0, }, ); assert_eq!(only_cache.items.len(), 1); assert_eq!(only_cache.items[0].id, "y"); } }