dtourolle
c0c6c5023e
🏗️ Build and Test JellyTau / Run Tests (push) Failing after 5m10s
🏗️ Build and Test JellyTau / Android Compile Check (push) Skipped
Publish Documentation / Build & publish docs to gitea-pages (push) Successful in 5m30s
Traceability Validation / Check Requirement Traces (push) Successful in 19s
A resumed transcode played nothing at all: every segment came back 400, hls.js exhausted its retries and gave up, while the same episode from the beginning was fine. Jellyfin builds each segment URI by echoing the master playlist's query string into it, and its segment handler opens by rejecting any request carrying StartTimeTicks > 0 (ArgumentException → 400). So one resume position on the playlist is copied onto every hls1/main/N.ts and 400s all of them — the `> 0` being exactly why starting from the beginning survived. HLS does not need the parameter: a playlist spans the whole item and asking for segment N *is* the seek. It is removed from the URL builder entirely rather than conditionalised — the builder cannot know whether its response will be segmented — and the position becomes a seek issued once the player has loaded. The progressive /Audio/universal builder behind the background-audio handoff has no segments and keeps its StartTimeTicks, which is why audio-only handoffs resumed correctly and video ones did not. Completing that across the boundary, since the URL no longer starts where the caller asked: - reloadSource(url, position) now means "reload and resume AT this absolute position": it seeks the element once the source is playable and clears the transcode offset to zero. It previously set the offset to the position and seeked nothing, which was correct only while the URL itself began there — left in place it would have shown 20:00 on the scrubber while the opening titles played, with no seek ever happening. - The transcoded resume path in the player page collapses into the same "seek after load" branch direct streams already used. - VideoPlayer's background-audio return does the same: no base, seek to the absolute position. - The stale test asserting StartTimeTicks is present is rewritten to keep its other half (an HLS master playlist, never a progressive stream.mp4, carrying the chosen source and audio track). TRACES: UR-004, UR-005, UR-019, UR-021, UR-074 | DR-181 | UT-182, UT-183
2110 lines
73 KiB
Rust
2110 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 {
|
|
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");
|
|
}
|
|
}
|