`cargo clippy --all-targets` went from 51 warnings (23 in the lib) to zero. Most were mechanical — needless borrows, `assert_eq!` against a bool literal, `vec!` where an array does, `or_insert_with(Vec::new)`, a loop index used only to index — and were applied with `clippy --fix`, then reviewed line by line. That review caught one auto-fix that was *not* semantically neutral: dropping the redundant `use hostname;` left its `#[cfg(target_os = "linux")]` orphaned directly above `SERVICE_NAME`, which would have silently cfg'd the constant out of every non-Linux build. Removed the stray attribute with the import. Where a lint asked for a risky change rather than a better one, it is suppressed with a comment saying why: - `too_many_arguments` on five `#[tauri::command]` handlers and `ThumbnailCache::save_thumbnail` — most of the arity is `State<'_, _>` injection, and a parameter struct would change the IPC contract and the generated TypeScript for no readability gain. - `large_enum_variant` on `PlayerStatusEvent` and `AutoplayDecision` — both are serde + specta wire types emitted a handful of times a second, never bulk allocated; boxing would have to stay invisible to the generated bindings while every match arm gained a deref. - `await_holding_lock` on the `hybrid`/`offline` test modules — the guard is a test-only serialisation lock for the process-global `INCLUDE_CATALOG_BROWSE` flag, and the await it spans *is* the critical section. Each `#[tokio::test]` gets its own single-threaded runtime, so this is not the production deadlock class the lint targets; restructuring would reintroduce the flag race. Real fixes elsewhere: `JellyfinItem::to_media_item` takes `self` by value, so it is now `into_media_item`; the five-tuple episode row in the download commands has a named `EpisodeRow` alias; the mpv `PropertyChange` arm matches `name: "pause"` instead of guarding on it. Also converted the last 27 raw `.lock().unwrap()` call sites to `lock_safe()`, completing the `MutexSafe`/`RwLockSafe` convention. All of them turned out to be in test modules — production code was already clean — so this is consistency rather than a fix. The two raw locks in `utils/lock.rs` stay raw on purpose: those tests deliberately poison a mutex to prove the helpers recover from it. Pure refactoring: all 698 tests still pass.
2120 lines
73 KiB
Rust
2120 lines
73 KiB
Rust
// 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::{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<OnlineRepository>,
|
|
offline: Arc<OfflineRepository>,
|
|
}
|
|
|
|
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<Vec<u8>, 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<usize, RepoError> {
|
|
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<Vec<super::JRayActor>, 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<i32>,
|
|
) -> Result<String, RepoError> {
|
|
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<f64>,
|
|
audio_stream_index: Option<i32>,
|
|
) -> Result<String, RepoError> {
|
|
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<Vec<MediaItem>, 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)
|
|
}
|
|
|
|
/// 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<SearchOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
let offline = Arc::clone(&self.offline);
|
|
let query = query.to_string();
|
|
self.cache_with_timeout(async move { offline.search(&query, options).await })
|
|
.await
|
|
}
|
|
|
|
/// 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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
let offline = Arc::clone(&self.offline);
|
|
self.cache_with_timeout(async move { offline.get_favorites(scope, options).await })
|
|
.await
|
|
}
|
|
|
|
/// 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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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)
|
|
}
|
|
|
|
/// 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<GetItemsOptions>,
|
|
) -> Result<Vec<MediaItem>, 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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<Vec<Library>, 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<DownloadDiskUsage, RepoError> {
|
|
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<SearchOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
self.online.search(query, options).await
|
|
}
|
|
|
|
/// 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<String, MediaItem> = HashMap::new();
|
|
let mut server_order: Vec<String> = 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<MediaItem> = Vec::new();
|
|
let mut seen: std::collections::HashSet<String> = 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)
|
|
///
|
|
/// @req: UR-002 - Access media when online or offline
|
|
/// @req: DR-013 - Repository pattern for online/offline data access
|
|
async fn parallel_race<T, F1, F2>(
|
|
&self,
|
|
cache_future: F1,
|
|
server_future: F2,
|
|
) -> Result<T, RepoError>
|
|
where
|
|
T: MeaningfulContent + Clone + Send + 'static,
|
|
F1: std::future::Future<Output = Result<T, RepoError>> + Send,
|
|
F2: std::future::Future<Output = Result<T, RepoError>> + Send,
|
|
{
|
|
// Try cache first (100ms timeout already applied by callers)
|
|
let cache_result = cache_future.await;
|
|
|
|
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),
|
|
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 | DR-155
|
|
async fn race_with_refresh<T, F1, F2, R>(
|
|
&self,
|
|
cache_future: F1,
|
|
server_future: F2,
|
|
on_cache_hit: R,
|
|
) -> Result<T, RepoError>
|
|
where
|
|
T: MeaningfulContent + Clone + Send + 'static,
|
|
F1: std::future::Future<Output = Result<T, RepoError>> + Send,
|
|
F2: std::future::Future<Output = Result<T, RepoError>> + Send,
|
|
R: FnOnce(),
|
|
{
|
|
let cache_result = cache_future.await;
|
|
|
|
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),
|
|
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<T>(
|
|
&self,
|
|
future: impl std::future::Future<Output = Result<T, RepoError>> + Send,
|
|
) -> Result<T, RepoError> {
|
|
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<Vec<Library>, 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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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)
|
|
let cache_result = self
|
|
.cache_with_timeout(async move { offline.get_items(&parent_id, opts_clone).await })
|
|
.await;
|
|
|
|
// 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())]
|
|
);
|
|
}
|
|
});
|
|
}
|
|
Ok(server_data)
|
|
}
|
|
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<MediaItem, RepoError> {
|
|
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<usize>,
|
|
) -> Result<Vec<MediaItem>, 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<usize>,
|
|
) -> Result<Vec<MediaItem>, 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<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
// Next up is dynamic, always fetch from server
|
|
self.online.get_next_up_episodes(series_id, limit).await
|
|
}
|
|
|
|
async fn get_recently_played_audio(
|
|
&self,
|
|
limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, 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<usize>) -> Result<Vec<MediaItem>, 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<usize>,
|
|
) -> Result<Vec<MediaItem>, 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<Vec<Genre>, 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<SearchOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<PlaybackInfo, RepoError> {
|
|
// 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<String, RepoError> {
|
|
// 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<f64>,
|
|
audio_stream_index: Option<i32>,
|
|
) -> Result<String, RepoError> {
|
|
// 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<Vec<MediaItem>, RepoError> {
|
|
// Live TV requires server communication - delegate to online repository
|
|
self.online.get_live_tv_channels().await
|
|
}
|
|
|
|
async fn get_channels(&self) -> Result<SearchResult, RepoError> {
|
|
// Plugin channels require server communication - delegate to online repository
|
|
self.online.get_channels().await
|
|
}
|
|
|
|
async fn open_live_stream(&self, item_id: &str) -> Result<LiveStreamInfo, RepoError> {
|
|
// 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<ImageOptions>,
|
|
) -> 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<MediaItem, RepoError> {
|
|
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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<usize>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<PlaylistCreatedResult, RepoError> {
|
|
// 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<Vec<PlaylistEntry>, 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<Mutex<Vec<MediaItem>>>,
|
|
query_count: Arc<Mutex<usize>>,
|
|
save_count: Arc<Mutex<usize>>,
|
|
}
|
|
|
|
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<usize, RepoError> {
|
|
*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<Vec<Library>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_items(
|
|
&self,
|
|
_parent_id: &str,
|
|
_options: Option<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
*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<MediaItem, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_latest_items(
|
|
&self,
|
|
_parent_id: &str,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_resume_items(
|
|
&self,
|
|
_parent_id: Option<&str>,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_next_up_episodes(
|
|
&self,
|
|
_series_id: Option<&str>,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_recently_played_audio(
|
|
&self,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_resume_movies(
|
|
&self,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_rediscover_albums(
|
|
&self,
|
|
_parent_id: Option<&str>,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn search(
|
|
&self,
|
|
_query: &str,
|
|
_options: Option<SearchOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_audio_only_stream_url_for_video(
|
|
&self,
|
|
_item_id: &str,
|
|
_media_source_id: Option<&str>,
|
|
_start_time_seconds: Option<f64>,
|
|
_audio_stream_index: Option<i32>,
|
|
) -> Result<String, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_channels(&self) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
|
|
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<ImageOptions>,
|
|
) -> 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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<MediaItem, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_items_by_person(
|
|
&self,
|
|
_person_id: &str,
|
|
_options: Option<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_similar_items(
|
|
&self,
|
|
_item_id: &str,
|
|
_limit: Option<usize>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn create_playlist(
|
|
&self,
|
|
_name: &str,
|
|
_item_ids: &[String],
|
|
) -> Result<PlaylistCreatedResult, RepoError> {
|
|
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<Vec<PlaylistEntry>, 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<MediaItem>,
|
|
query_count: Arc<Mutex<usize>>,
|
|
}
|
|
|
|
impl MockOnlineRepo {
|
|
fn new(items: Vec<MediaItem>) -> 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<Vec<Library>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_items(
|
|
&self,
|
|
_parent_id: &str,
|
|
_options: Option<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
*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<MediaItem, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_latest_items(
|
|
&self,
|
|
_parent_id: &str,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_resume_items(
|
|
&self,
|
|
_parent_id: Option<&str>,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_next_up_episodes(
|
|
&self,
|
|
_series_id: Option<&str>,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_recently_played_audio(
|
|
&self,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_resume_movies(
|
|
&self,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_rediscover_albums(
|
|
&self,
|
|
_parent_id: Option<&str>,
|
|
_limit: Option<usize>,
|
|
) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_genres(&self, _parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn search(
|
|
&self,
|
|
_query: &str,
|
|
_options: Option<SearchOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_audio_only_stream_url_for_video(
|
|
&self,
|
|
_item_id: &str,
|
|
_media_source_id: Option<&str>,
|
|
_start_time_seconds: Option<f64>,
|
|
_audio_stream_index: Option<i32>,
|
|
) -> Result<String, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_channels(&self) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
|
|
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<ImageOptions>,
|
|
) -> 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<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
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<MediaItem, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_items_by_person(
|
|
&self,
|
|
_person_id: &str,
|
|
_options: Option<GetItemsOptions>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn get_similar_items(
|
|
&self,
|
|
_item_id: &str,
|
|
_limit: Option<usize>,
|
|
) -> Result<SearchResult, RepoError> {
|
|
unimplemented!()
|
|
}
|
|
|
|
async fn create_playlist(
|
|
&self,
|
|
_name: &str,
|
|
_item_ids: &[String],
|
|
) -> Result<PlaylistCreatedResult, RepoError> {
|
|
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<Vec<PlaylistEntry>, 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<MockOfflineRepo>,
|
|
online: Arc<MockOnlineRepo>,
|
|
}
|
|
|
|
impl TestHybridRepo {
|
|
fn new(server_items: Vec<MediaItem>) -> 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<SearchResult, RepoError> {
|
|
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<SearchResult, RepoError> {
|
|
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");
|
|
}
|
|
}
|