Files
jellytau/src-tauri/src/repository/offline.rs
T
dtourolle 1b70926c36
🏗️ Build and Test JellyTau / Run Tests (push) Successful in 20m34s
Publish Documentation / Build & publish docs to gitea-pages (push) Successful in 6m6s
Traceability Validation / Check Requirement Traces (push) Successful in 18s
Build & Release / Run Tests (push) Successful in 20m26s
🏗️ Build and Test JellyTau / Android Compile Check (push) Successful in 10m3s
Build & Release / Build Linux (push) Successful in 37m59s
Build & Release / Build Windows (push) Successful in 23m0s
Build & Release / Build Android (push) Successful in 40m26s
Build & Release / Create Release (push) Successful in 1m20s
feat(offline): play downloaded video, and drain the offline sync queue (0.4.6)
Bundles this session's work plus the concurrent search/offline/player changes.
Every gate passes on the combined tree: 885 frontend tests, 610 Rust tests,
clippy clean, boundary clean, trace coverage 86%.

Offline video playback — four separate defects, each of which alone stopped it:

  DR-133  A completed download's file_path is already absolute (the worker
          rewrites it on completion), but the player rooted it a second time and
          handed the webview /data/user/0/app//data/user/0/app/videos/x.mp4.
  DR-134  The asset protocol was never enabled: no protocol-asset feature and no
          assetProtocol config, so convertFileSrc produced URLs nothing answered.
          Also silently defeated the cached-thumbnail path, which fails soft to
          the server copy and hid it whenever the server was reachable.
  DR-137  Tauri's asset protocol answers a range-less request by reading the
          whole file into memory, and only advertises Accept-Ranges from inside
          its range branch, so the first request never learns ranges exist.
          Chromium gave up with PIPELINE_ERROR_READ after ~31s. Local media is
          now served by a loopback HTTP server: bounded 4 MiB chunks streamed
          from the file handle, every response length-delimited, and a range-less
          request answered with one chunk rather than the file. Confined by a
          per-session token and to the app data directory, because loopback is
          shared between apps on Android.
  DR-138  Release builds set usesCleartextTraffic=false, so Android rejected the
          request to that server before any I/O. A network-security-config
          exempts 127.0.0.1 only; a remote server must still be HTTPS.

Downloads:

  DR-135  download_item never records media_type and the reconnect resolver read
          that NULL as 'audio', so a movie queued from a media card had its URL
          resolved by get_audio_stream_url and completed as an audio-only
          transcode. The item's own type now decides.
  DR-136  Rows already downloaded that way are requeued on reconnect, since
          prevention alone leaves them reading "downloaded" and still unplayable.

Known limitation: a download taken at `original` quality is a byte copy of the
source, so it can be any container. One such file is an AVI holding XVID, which
the webview cannot play in any case — the media server serves it correctly and
Chromium refuses it. That needs either a transcoded download preset or the
native ExoPlayer surface work, and is not addressed here.

Also fixes two ID collisions between concurrent work: DR-143 defined twice
(search vs offline gate) and UT-131 defined twice (Episode Focus hero vs channel
cap). The search requirement is now DR-147 and the channel-cap test UT-141, with
their code references and matrix rows updated.
2026-08-09 16:38:07 +02:00

4378 lines
175 KiB
Rust

// Offline repository - queries SQLite database for cached data
//
// TRACES: UR-002, UR-052 | DR-012, DR-013, DR-078
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use async_trait::async_trait;
use log::debug;
use super::{types::*, MediaRepository};
use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};
/// Whether offline library queries may include catalog items that are merely
/// *browsed/synced* but not downloaded (the greyed-out "browse the whole server"
/// view). Defaults to `true` so online browsing (which reads this same cache as
/// a fast path) still sees the full catalog.
///
/// While offline, the frontend drives this from the "Show all server media"
/// toggle: OFF means library pages show only downloaded/local media, ON reveals
/// the full greyed-out catalog. See `set_include_catalog_browse` and the
/// `showServerCatalog` UI flag. Fixes the bug where offline library pages showed
/// every server item regardless of the toggle.
///
/// TRACES: UR-052 | DR-078
static INCLUDE_CATALOG_BROWSE: AtomicBool = AtomicBool::new(true);
/// Set whether offline `get_items` includes non-downloaded (synced-only) catalog
/// items. Called from the frontend: `true` when online or when the offline
/// "Show all server media" toggle is on; `false` when offline with the toggle
/// off (show downloaded/local media only).
pub fn set_include_catalog_browse(include: bool) {
INCLUDE_CATALOG_BROWSE.store(include, Ordering::Relaxed);
}
/// Whether offline `get_items` currently includes the synced-but-not-downloaded
/// catalog (the greyed-out browse view). Mirrors `set_include_catalog_browse`.
///
/// Exposed so the hybrid repo can tell "cache is cold, ask the server" from
/// "user asked for downloads only and there are none here": when this is false,
/// an empty offline `get_items` is authoritative and must not fall through to
/// the server. See hybrid.rs `get_items`.
///
/// TRACES: UR-052 | DR-078, DR-080
pub fn include_catalog_browse() -> bool {
INCLUDE_CATALOG_BROWSE.load(Ordering::Relaxed)
}
/// Build a safe FTS5 prefix query from raw user input.
///
/// Every whitespace-separated token is emitted as a *quoted phrase*, so
/// punctuation the user types (apostrophes in `Bob's Burgers`, hyphens in
/// `Spider-Man`, the `/` in `AC/DC`) is treated as data rather than FTS5
/// operator syntax — unquoted, those characters make `MATCH` raise a syntax
/// error and the whole search fails. The final token carries the `*` prefix
/// operator so results appear while the user is still typing; earlier tokens are
/// implicitly ANDed, which preserves the pre-existing matching semantics.
///
/// Returns `None` when the input holds nothing searchable (empty, or pure
/// punctuation), so callers skip the query rather than handing FTS5 a string it
/// will reject — `search("")` is a real call site, used to list all playlists.
///
/// TRACES: UR-065 | DR-108 | UT-111
fn build_fts_prefix_query(query: &str) -> Option<String> {
let tokens: Vec<String> = query
.split_whitespace()
.filter(|token| token.chars().any(char::is_alphanumeric))
.map(|token| format!("\"{}\"", token.replace('"', "\"\"")))
.collect();
let last = tokens.len().checked_sub(1)?;
Some(
tokens
.iter()
.enumerate()
.map(|(i, token)| {
if i == last {
format!("{}*", token)
} else {
token.clone()
}
})
.collect::<Vec<_>>()
.join(" "),
)
}
pub struct OfflineRepository {
db_service: Arc<RusqliteService>,
server_id: String,
user_id: String,
}
impl OfflineRepository {
pub fn new(db_service: Arc<RusqliteService>, server_id: String, user_id: String) -> Self {
Self {
db_service,
server_id,
user_id,
}
}
/// Helper to convert CachedItem from storage to MediaItem
fn cached_item_to_media_item(item: CachedItem, user_data: Option<UserData>) -> MediaItem {
let artists_vec = item
.artists
.as_ref()
.and_then(|s| serde_json::from_str::<Vec<String>>(s).ok())
.unwrap_or_default();
let kind = crate::domain::kind_from_jellyfin(&item.item_type, item.is_folder);
MediaItem {
id: item.id.clone(),
name: item.name,
item_type: item.item_type,
kind,
is_folder: item.is_folder,
server_id: item.server_id,
parent_id: item.parent_id,
library_id: item.library_id,
overview: item.overview,
genres: item
.genres
.as_ref()
.and_then(|s| serde_json::from_str::<Vec<String>>(s).ok()),
runtime_ticks: item.runtime_ticks,
duration_ms: item.runtime_ticks.map(crate::domain::ticks_to_ms),
production_year: item.production_year,
premiere_date: item.premiere_date,
community_rating: item.community_rating,
official_rating: item.official_rating,
primary_image_tag: item.primary_image_tag.clone(),
image_id: item.primary_image_tag,
backdrop_image_tags: item.backdrop_image_tags,
parent_backdrop_image_tags: item.parent_backdrop_image_tags,
album_id: item.album_id,
album_name: item.album_name,
album_artist: item.album_artist,
artists: Some(artists_vec),
artist_items: None, // Not stored in cache yet - TODO: add to database schema
index_number: item.index_number,
series_id: item.series_id,
series_name: item.series_name,
season_id: item.season_id,
season_name: item.season_name,
parent_index_number: item.parent_index_number,
user_data,
media_streams: None, // Not cached offline
media_sources: None, // Not cached offline
people: None, // Not cached offline - TODO: add to database schema
}
}
/// Get user data for an item (playback position, favorite, etc.)
async fn get_user_data(&self, item_id: &str) -> Option<UserData> {
let query = Query::with_params(
"SELECT playback_position_ticks, is_played, is_favorite, play_count, last_played_at, playback_context_type, playback_context_id
FROM user_data WHERE user_id = ? AND item_id = ?",
vec![
QueryParam::String(self.user_id.clone()),
QueryParam::String(item_id.to_string()),
],
);
self.db_service
.query_optional(query, |row| {
let playback_position_ticks: Option<i64> = row.get(0).ok();
Ok(UserData {
playback_position_ticks,
playback_position_ms: playback_position_ticks.map(crate::domain::ticks_to_ms),
is_played: row.get::<_, Option<i32>>(1).ok().flatten().map(|v| v != 0),
is_favorite: row.get::<_, Option<i32>>(2).ok().flatten().map(|v| v != 0),
play_count: row.get(3).ok(),
last_played_date: row.get(4).ok(),
playback_context_type: row.get(5).ok(),
playback_context_id: row.get(6).ok(),
})
})
.await
.ok()
.flatten()
}
}
// Helper struct matching storage.rs CachedItem structure
#[derive(Debug)]
struct CachedItem {
id: String,
name: String,
item_type: String,
is_folder: bool,
server_id: String,
parent_id: Option<String>,
library_id: Option<String>,
overview: Option<String>,
genres: Option<String>,
runtime_ticks: Option<i64>,
production_year: Option<i32>,
premiere_date: Option<String>,
community_rating: Option<f64>,
official_rating: Option<String>,
primary_image_tag: Option<String>,
backdrop_image_tags: Option<Vec<String>>,
parent_backdrop_image_tags: Option<Vec<String>>,
album_id: Option<String>,
album_name: Option<String>,
album_artist: Option<String>,
artists: Option<String>,
index_number: Option<i32>,
series_id: Option<String>,
series_name: Option<String>,
season_id: Option<String>,
season_name: Option<String>,
parent_index_number: Option<i32>,
}
fn row_to_cached_item(row: &rusqlite::Row) -> rusqlite::Result<CachedItem> {
Ok(CachedItem {
id: row.get(0)?,
name: row.get(1)?,
item_type: row.get(2)?,
server_id: row.get(3)?,
parent_id: row.get(4)?,
library_id: row.get(5)?,
overview: row.get(6)?,
genres: row.get(7)?,
runtime_ticks: row.get(8)?,
production_year: row.get(9)?,
community_rating: row.get(10)?,
official_rating: row.get(11)?,
primary_image_tag: row.get(12)?,
backdrop_image_tags: None, // TODO: Add to DB schema
parent_backdrop_image_tags: None, // TODO: Add to DB schema
album_id: row.get(13)?,
album_name: row.get(14)?,
album_artist: row.get(15)?,
artists: row.get(16)?,
index_number: row.get(17)?,
series_id: row.get(18)?,
series_name: row.get(19)?,
season_id: row.get(20)?,
season_name: row.get(21)?,
parent_index_number: row.get(22)?,
// Appended as the final columns in every SELECT that maps through this fn.
is_folder: row.get::<_, Option<i64>>(23)?.unwrap_or(0) != 0,
premiere_date: row.get(24)?,
})
}
impl OfflineRepository {
/// Remove catalog entries the server no longer has (mark-and-sweep).
///
/// `save_to_cache` stamps every row it writes with a fresh `synced_at`, so
/// after a complete crawl anything still on the server carries a timestamp
/// newer than `cutoff` (taken before the crawl began) and anything deleted
/// server-side kept its older one. Sweeping by timestamp avoids binding the
/// crawl's entire id set, which would blow past SQLite's variable limit on a
/// large library.
///
/// Three exclusions, each load-bearing:
///
/// * **Only `item_types` the crawl actually requested.** The crawl asks for
/// `CATALOG_ITEM_TYPES`; rows of any other type (artists, playlists, the
/// `Folder` parent stubs `save_to_cache` inserts) are never refreshed by
/// it, so sweeping by age alone would delete every one of them.
/// * **Anything a completed download depends on** — the downloaded item
/// itself, and any container with a downloaded child. The user has those
/// bytes on disk; dropping the row would orphan the file.
/// * Callers must only invoke this after a crawl in which *every* library
/// succeeded. `sync_full_catalog` is best-effort per library, and
/// `items.parent_id` is `ON DELETE CASCADE`, so sweeping after a partial
/// crawl could cascade an entire series away because one request timed out.
///
/// Returns the number of rows removed.
///
/// TRACES: UR-065 | DR-110 | UT-113
pub async fn prune_stale_catalog(
&self,
cutoff: &str,
item_types: &[String],
) -> Result<usize, RepoError> {
if item_types.is_empty() {
return Ok(0);
}
let placeholders = vec!["?"; item_types.len()].join(",");
let sql = format!(
"DELETE FROM items
WHERE server_id = ?
AND synced_at IS NOT NULL
AND synced_at < ?
AND item_type IN ({})
AND id NOT IN (
-- Playable items with completed downloads
SELECT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
UNION
-- Containers with downloaded children
SELECT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
)",
placeholders
);
let mut params = vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(cutoff.to_string()),
];
params.extend(item_types.iter().cloned().map(QueryParam::String));
let removed = self
.db_service
.execute(Query::with_params(sql, params))
.await
.map_err(|e| RepoError::Database { message: e })?;
Ok(removed)
}
/// Full-text search over cached people (cast and crew).
///
/// People are stored in their own table rather than in `items`, so search
/// has to look them up separately and adapt them to `MediaItem`. Availability
/// gating deliberately does not apply: a person is metadata, never a
/// download, so there is nothing to be offline about.
///
/// TRACES: UR-065, UR-060 | DR-111 | UT-114
async fn search_people(
&self,
fts_query: &str,
limit: usize,
) -> Result<Vec<MediaItem>, RepoError> {
let sql = format!(
"SELECT p.id, p.name, p.overview, p.primary_image_tag, p.premiere_date
FROM people p
JOIN people_fts fts ON fts.rowid = p.rowid
WHERE p.server_id = ? AND people_fts MATCH ?
ORDER BY rank
LIMIT {}",
limit
);
let rows = self
.db_service
.query_many(
Query::with_params(
sql,
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(fts_query.to_string()),
],
),
|row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, Option<String>>(3)?,
row.get::<_, Option<String>>(4)?,
))
},
)
.await
.map_err(|e| RepoError::Database { message: e })?;
Ok(rows
.into_iter()
.map(
|(id, name, overview, primary_image_tag, premiere_date)| MediaItem {
id,
name,
item_type: "Person".to_string(),
kind: crate::domain::MediaKind::Person,
is_folder: false,
server_id: self.server_id.clone(),
overview,
primary_image_tag,
premiere_date,
..Default::default()
},
)
.collect())
}
/// Save browsed items to cache for faster subsequent loading
///
/// This persists metadata for items that were browsed (not necessarily downloaded).
/// Items are marked with current timestamp for freshness tracking.
pub async fn save_to_cache(
&self,
parent_id: &str,
items: &[MediaItem],
) -> Result<usize, RepoError> {
if items.is_empty() {
return Ok(0);
}
let now = chrono::Utc::now().to_rfc3339();
// Temporarily disable foreign key constraints to avoid CASCADE DELETE issues
// when replacing stub parent items with their actual data
self.db_service
.execute(Query::new("PRAGMA foreign_keys = OFF"))
.await
.map_err(|e| RepoError::Database { message: e })?;
// Ensure we re-enable foreign keys even if an error occurs
let result = self.save_to_cache_impl(parent_id, items, &now).await;
// Re-enable foreign key constraints
let _ = self
.db_service
.execute(Query::new("PRAGMA foreign_keys = ON"))
.await;
result
}
async fn save_to_cache_impl(
&self,
parent_id: &str,
items: &[MediaItem],
now: &str,
) -> Result<usize, RepoError> {
// Collect all unique parent IDs referenced by items being saved
let mut parent_ids = std::collections::HashSet::new();
parent_ids.insert(parent_id.to_string());
for item in items {
if let Some(pid) = &item.parent_id {
parent_ids.insert(pid.clone());
}
}
// Insert stub entries for all parent IDs to satisfy FK constraints
#[cfg(test)]
println!("Creating stub parents for: {:?}", parent_ids);
for pid in parent_ids {
let parent_query = Query::with_params(
"INSERT OR IGNORE INTO items (id, server_id, name, item_type, synced_at)
VALUES (?1, ?2, ?3, ?4, ?5)",
vec![
QueryParam::String(pid.clone()),
QueryParam::String(self.server_id.clone()),
QueryParam::String("Parent".to_string()),
QueryParam::String("Folder".to_string()),
QueryParam::String(now.to_string()),
],
);
let _stub_rows = self
.db_service
.execute(parent_query)
.await
.map_err(|e| RepoError::Database { message: e })?;
#[cfg(test)]
println!(
" Created stub parent {} (rows affected: {})",
pid, _stub_rows
);
}
let mut count = 0;
for item in items {
// Convert Option<Vec<String>> to JSON strings for storage
let genres_json = item
.genres
.as_ref()
.map(|g| serde_json::to_string(g).unwrap_or_else(|_| "[]".to_string()));
let artists_json = item
.artists
.as_ref()
.map(|a| serde_json::to_string(a).unwrap_or_else(|_| "[]".to_string()));
let backdrop_tags_json = item
.backdrop_image_tags
.as_ref()
.map(|b| serde_json::to_string(b).unwrap_or_else(|_| "[]".to_string()));
// A real UPSERT, not INSERT OR REPLACE. REPLACE deletes the existing
// row and inserts a new one, which (a) fires no AFTER DELETE trigger
// unless `recursive_triggers` is on — it is not, so `items_ad` never
// ran and the old `items_fts` row was orphaned — and (b) assigns a
// *fresh rowid*, because `items.id` is a TEXT PRIMARY KEY, so
// `items_ai` then appended a second index entry. The result was one
// duplicate FTS index per catalog pass. ON CONFLICT keeps the rowid
// that `items_fts.content_rowid` refers to and fires `items_au`,
// which correctly replaces the entry.
//
// It also preserves columns absent from this statement (etag,
// sort_name, tagline, …) instead of resetting them to defaults the
// way REPLACE did.
//
// TRACES: UR-065 | DR-110 | UT-112
let query = Query::with_params(
"INSERT INTO items (
id, server_id, library_id, parent_id,
name, item_type, is_folder, overview,
genres, series_id, series_name,
season_id, season_name, index_number, parent_index_number,
album_id, album_name, album_artist, artists,
production_year, premiere_date, runtime_ticks,
primary_image_tag, backdrop_image_tags,
community_rating, official_rating,
synced_at
) VALUES (
?1, ?2, ?3, ?4,
?5, ?6, ?7, ?8,
?9, ?10, ?11,
?12, ?13, ?14, ?15,
?16, ?17, ?18, ?19,
?20, ?21, ?22,
?23, ?24,
?25, ?26,
?27
)
ON CONFLICT(id) DO UPDATE SET
server_id = excluded.server_id,
-- This call site never supplies library_id (it is always
-- bound NULL), so keep whatever another path recorded rather
-- than clearing it the way REPLACE did.
library_id = COALESCE(excluded.library_id, items.library_id),
parent_id = excluded.parent_id,
name = excluded.name,
item_type = excluded.item_type,
is_folder = excluded.is_folder,
overview = excluded.overview,
genres = excluded.genres,
series_id = excluded.series_id,
series_name = excluded.series_name,
season_id = excluded.season_id,
season_name = excluded.season_name,
index_number = excluded.index_number,
parent_index_number = excluded.parent_index_number,
album_id = excluded.album_id,
album_name = excluded.album_name,
album_artist = excluded.album_artist,
artists = excluded.artists,
production_year = excluded.production_year,
premiere_date = excluded.premiere_date,
runtime_ticks = excluded.runtime_ticks,
primary_image_tag = excluded.primary_image_tag,
backdrop_image_tags = excluded.backdrop_image_tags,
community_rating = excluded.community_rating,
official_rating = excluded.official_rating,
synced_at = excluded.synced_at",
vec![
QueryParam::String(item.id.clone()),
QueryParam::String(self.server_id.clone()),
// Library is NULL for cached items (may not be synced yet)
QueryParam::Null, // library_id
// Use the item's actual parent_id, not the function parameter
match &item.parent_id {
Some(pid) => QueryParam::String(pid.clone()),
None => QueryParam::Null,
},
QueryParam::String(item.name.clone()),
QueryParam::String(item.item_type.clone()),
QueryParam::Int(if item.is_folder { 1 } else { 0 }),
match &item.overview {
Some(o) => QueryParam::String(o.clone()),
None => QueryParam::Null,
},
match genres_json {
Some(g) => QueryParam::String(g),
None => QueryParam::Null,
},
match &item.series_id {
Some(s) => QueryParam::String(s.clone()),
None => QueryParam::Null,
},
match &item.series_name {
Some(s) => QueryParam::String(s.clone()),
None => QueryParam::Null,
},
match &item.season_id {
Some(s) => QueryParam::String(s.clone()),
None => QueryParam::Null,
},
match &item.season_name {
Some(s) => QueryParam::String(s.clone()),
None => QueryParam::Null,
},
match item.index_number {
Some(i) => QueryParam::Int(i),
None => QueryParam::Null,
},
match item.parent_index_number {
Some(i) => QueryParam::Int(i),
None => QueryParam::Null,
},
match &item.album_id {
Some(a) => QueryParam::String(a.clone()),
None => QueryParam::Null,
},
match &item.album_name {
Some(a) => QueryParam::String(a.clone()),
None => QueryParam::Null,
},
match &item.album_artist {
Some(a) => QueryParam::String(a.clone()),
None => QueryParam::Null,
},
match artists_json {
Some(a) => QueryParam::String(a),
None => QueryParam::Null,
},
match item.production_year {
Some(y) => QueryParam::Int(y),
None => QueryParam::Null,
},
match &item.premiere_date {
Some(d) => QueryParam::String(d.clone()),
None => QueryParam::Null,
},
match item.runtime_ticks {
Some(r) => QueryParam::Int64(r),
None => QueryParam::Null,
},
match &item.primary_image_tag {
Some(t) => QueryParam::String(t.clone()),
None => QueryParam::Null,
},
match backdrop_tags_json {
Some(b) => QueryParam::String(b),
None => QueryParam::Null,
},
match item.community_rating {
Some(r) => QueryParam::Float(r),
None => QueryParam::Null,
},
match &item.official_rating {
Some(r) => QueryParam::String(r.clone()),
None => QueryParam::Null,
},
QueryParam::String(now.to_string()),
],
);
let _rows_affected =
self.db_service
.execute(query)
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to insert item {}: {}", item.id, e),
})?;
#[cfg(test)]
println!(
" [save_to_cache] Saved item {} (rows affected: {})",
item.id, _rows_affected
);
self.mirror_user_data(item, now).await?;
count += 1;
}
Ok(count)
}
/// Mirror the server's per-user state for an item into the local
/// `user_data` table, so favourites marked on any other client are visible
/// here — including offline, where the local table is the only source.
///
/// The `WHERE user_data.pending_sync = 0` on the conflict clause is the
/// conflict rule: a toggle made while the server was unreachable is still
/// waiting to be pushed, and must not be clobbered by the stale value the
/// server is still reporting. Rows carrying no favourite state are skipped
/// entirely rather than written as `0`, which would fabricate an
/// "unfavourited" record from an endpoint that simply omits `UserData`.
///
/// TRACES: UR-069 | DR-114 | UT-102
async fn mirror_user_data(&self, item: &MediaItem, now: &str) -> Result<(), RepoError> {
let Some(is_favorite) = item.user_data.as_ref().and_then(|ud| ud.is_favorite) else {
return Ok(());
};
let query = Query::with_params(
"INSERT INTO user_data (user_id, item_id, is_favorite, synced_at, pending_sync)
VALUES (?1, ?2, ?3, ?4, 0)
ON CONFLICT(user_id, item_id) DO UPDATE SET
is_favorite = excluded.is_favorite,
synced_at = excluded.synced_at
WHERE user_data.pending_sync = 0",
vec![
QueryParam::String(self.user_id.clone()),
QueryParam::String(item.id.clone()),
QueryParam::Int(if is_favorite { 1 } else { 0 }),
QueryParam::String(now.to_string()),
],
);
// A missing item row (FK) is not fatal here — the favourite mirror is
// best-effort metadata, and failing the whole cache write over it would
// break browsing.
if let Err(e) = self.db_service.execute(query).await {
debug!(
"[OfflineRepo] user_data mirror skipped for {}: {}",
item.id, e
);
}
Ok(())
}
/// Cache the library (view) list from the server into the local database.
/// Called by HybridRepository after a successful online fetch so the list is
/// available offline. Without this, the `libraries` table stays empty and
/// offline startup shows no libraries at all.
pub async fn save_libraries_to_cache(&self, libraries: &[Library]) -> Result<usize, RepoError> {
if libraries.is_empty() {
return Ok(0);
}
let mut count = 0;
for (idx, lib) in libraries.iter().enumerate() {
let query = Query::with_params(
"INSERT OR REPLACE INTO libraries (id, server_id, name, collection_type, image_tag, sort_order, synced_at)
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)",
vec![
QueryParam::String(lib.id.clone()),
QueryParam::String(self.server_id.clone()),
QueryParam::String(lib.name.clone()),
QueryParam::String(lib.collection_type.clone()),
lib.image_tag.clone().map(QueryParam::String).unwrap_or(QueryParam::Null),
QueryParam::Int(idx as i32),
],
);
self.db_service
.execute(query)
.await
.map_err(|e| RepoError::Database { message: e })?;
count += 1;
}
Ok(count)
}
/// Cache the full server genre catalog for a library, so offline (and the
/// hybrid cache-first race) can return the complete list instead of only the
/// genres derivable from locally-cached albums. Replaces the scope's rows
/// wholesale so genres removed on the server don't linger.
pub async fn save_genres_to_cache(
&self,
parent_id: Option<&str>,
genres: &[Genre],
) -> Result<usize, RepoError> {
if genres.is_empty() {
return Ok(0);
}
// library_id is part of the primary key; NULL keys don't de-dupe in
// SQLite, so store the "no library" scope as an empty string.
let library_id = parent_id.unwrap_or("").to_string();
let server_id = self.server_id.clone();
let genres: Vec<(String, String, Option<u32>)> = genres
.iter()
.map(|g| (g.id.clone(), g.name.clone(), g.album_count))
.collect();
let saved = genres.len();
self.db_service
.transaction(move |tx| {
use crate::storage::db_service::{Query, QueryParam};
// Clear the scope's existing genres, then re-insert the fresh set.
tx.execute(Query::with_params(
"DELETE FROM genres WHERE server_id = ? AND library_id = ?",
vec![
QueryParam::String(server_id.clone()),
QueryParam::String(library_id.clone()),
],
))?;
for (id, name, album_count) in &genres {
tx.execute(Query::with_params(
"INSERT OR REPLACE INTO genres (id, server_id, library_id, name, album_count, synced_at)
VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP)",
vec![
QueryParam::String(id.clone()),
QueryParam::String(server_id.clone()),
QueryParam::String(library_id.clone()),
QueryParam::String(name.clone()),
album_count.map(|c| QueryParam::Int(c as i32)).unwrap_or(QueryParam::Null),
],
))?;
}
Ok(())
})
.await
.map_err(|e| RepoError::Database { message: e })?;
Ok(saved)
}
/// SQL fragment: the set of item ids that are "on the device" — playable
/// items with a completed download, plus containers (album/series/season)
/// that have at least one downloaded child. This is the `get_items` CTE with
/// the synced-but-not-downloaded catalog branch deliberately excluded, so it
/// is authoritative regardless of the process-wide catalog-browse flag.
///
/// TRACES: UR-055 | DR-082, DR-083
const DOWNLOADED_ITEMS_CTE: &'static str = "
WITH downloaded_items AS (
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
)";
/// Downloaded-only browse: items under `parent_id` that are on the device.
///
/// Unlike [`MediaRepository::get_items`], this never includes the
/// synced-but-not-downloaded catalog and never consults the process-wide
/// `INCLUDE_CATALOG_BROWSE` flag — it is the dedicated Downloads surface.
/// An empty result is authoritative ("nothing downloaded here"), so the
/// hybrid repo must call this directly rather than racing the server.
///
/// TRACES: UR-055 | DR-082, DR-083
pub async fn get_downloaded_items(
&self,
parent_id: &str,
options: Option<GetItemsOptions>,
) -> Result<SearchResult, RepoError> {
let opts = options.unwrap_or_default();
let limit = opts.limit.unwrap_or(10000);
let start_index = opts.start_index.unwrap_or(0);
let type_filter = if let Some(include_item_types) = &opts.include_item_types {
if !include_item_types.is_empty() {
let types = include_item_types
.iter()
.map(|t| format!("'{}'", t.replace('\'', "''")))
.collect::<Vec<_>>()
.join(",");
format!(" AND i.item_type IN ({})", types)
} else {
String::new()
}
} else {
String::new()
};
// When the parent is a LIBRARY, cached items carry no link back to it
// (library_id/parent_id are NULL), so the `libraries` EXISTS clause below
// matches every downloaded item on the server — both containers
// (MusicAlbum/Series/…) AND their leaves (Audio/Episode). Listing the
// leaves alongside the containers is the "I see individual songs, not
// albums" bug: a library landing page must show only *top-level* items.
// So at the library level we exclude any leaf whose own container
// (album/season/series/parent) is itself present in `downloaded_items` —
// that container represents it in the grid. Items with no downloaded
// container (e.g. a downloaded Movie, or a stray track whose album isn't
// cached) still surface. This mirrors the online music library, which
// routes to a dedicated albums view. See [[offline-libraries-never-cached]].
let sql = format!(
"{cte}
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
INNER JOIN downloaded_items di ON i.id = di.id
WHERE i.server_id = ?
AND (
i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ?
OR (
EXISTS (
SELECT 1 FROM libraries l
WHERE l.id = ? AND l.server_id = i.server_id
)
-- Top-level only: hide leaves whose container is downloaded.
AND NOT EXISTS (
SELECT 1 FROM downloaded_items parent
WHERE parent.id = i.album_id
OR parent.id = i.season_id
OR parent.id = i.series_id
OR parent.id = i.parent_id
)
)
){type_filter}
ORDER BY i.sort_name ASC, i.name ASC
LIMIT {limit} OFFSET {start_index}",
cte = Self::DOWNLOADED_ITEMS_CTE,
);
let query = Query::with_params(
sql,
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(parent_id.to_string()),
QueryParam::String(parent_id.to_string()),
QueryParam::String(parent_id.to_string()),
QueryParam::String(parent_id.to_string()),
QueryParam::String(parent_id.to_string()),
],
);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
let total_record_count = items.len();
Ok(SearchResult {
items,
total_record_count,
})
}
/// Libraries that contain at least one downloaded item. Libraries with
/// nothing on the device are omitted, so the Downloaded surface only lists
/// libraries the user actually has offline content in.
///
/// TRACES: UR-055 | DR-082
pub async fn get_downloaded_libraries(&self) -> Result<Vec<Library>, RepoError> {
// A downloaded item links back to its library only indirectly (the
// cache leaves library_id NULL — see [[offline-libraries-never-cached]]).
// We match a library by collection_type ↔ item_type instead: any
// completed download of a given media kind qualifies that library.
let query = Query::with_params(
&format!(
"{cte}
SELECT l.id, l.name, l.collection_type, l.image_tag
FROM libraries l
WHERE l.server_id = ?
AND EXISTS (
SELECT 1 FROM items i
INNER JOIN downloaded_items di ON i.id = di.id
WHERE i.server_id = l.server_id
AND (
(l.collection_type = 'music' AND i.item_type IN ('MusicAlbum', 'MusicArtist', 'Audio'))
OR (l.collection_type = 'movies' AND i.item_type = 'Movie')
OR (l.collection_type = 'tvshows' AND i.item_type IN ('Series', 'Season', 'Episode'))
OR (l.collection_type NOT IN ('music', 'movies', 'tvshows'))
)
)
ORDER BY l.sort_order ASC, l.name ASC",
cte = Self::DOWNLOADED_ITEMS_CTE,
),
vec![QueryParam::String(self.server_id.clone())],
);
self.db_service
.query_many(query, |row| {
Ok(Library {
id: row.get(0)?,
name: row.get(1)?,
collection_type: row
.get::<_, Option<String>>(2)?
.unwrap_or_else(|| "unknown".to_string()),
image_tag: row.get(3)?,
})
})
.await
.map_err(|e| RepoError::Database { message: e })
}
/// On-disk bytes for downloaded content, for the disk-usage display.
///
/// Returns one entry per *container or leaf* that appears in the Downloaded
/// browse: a leaf's own `file_size`, a container's summed downloaded
/// descendants — plus the device total and item (leaf) count. This is pure
/// aggregation over `downloads.file_size`, not new tracking.
///
/// TRACES: UR-056 | DR-085
pub async fn get_download_disk_usage(&self) -> Result<DownloadDiskUsage, RepoError> {
// Per-leaf sizes (completed playable downloads only).
let leaf_query = Query::with_params(
"SELECT d.item_id, COALESCE(d.file_size, 0)
FROM downloads d
INNER JOIN items i ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.server_id = ?
AND i.item_type IN ('Audio', 'Movie', 'Episode')",
vec![QueryParam::String(self.server_id.clone())],
);
let leaves: Vec<(String, i64)> = self
.db_service
.query_many(leaf_query, |row| Ok((row.get(0)?, row.get(1)?)))
.await
.map_err(|e| RepoError::Database { message: e })?;
// Container subtotals: sum each container's downloaded descendants.
let container_query = Query::with_params(
"SELECT c.id, COALESCE(SUM(d.file_size), 0)
FROM items c
INNER JOIN items children
ON (children.parent_id = c.id OR children.album_id = c.id OR children.season_id = c.id OR children.series_id = c.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND c.server_id = ?
AND c.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
GROUP BY c.id",
vec![QueryParam::String(self.server_id.clone())],
);
let containers: Vec<(String, i64)> = self
.db_service
.query_many(container_query, |row| Ok((row.get(0)?, row.get(1)?)))
.await
.map_err(|e| RepoError::Database { message: e })?;
// Partiality per container: a container is "partial" when it has cached
// descendants that are NOT downloaded. We compare downloaded-descendant
// count against total-cached-descendant count (the offline cache holds
// the synced full catalog, so this is meaningful).
//
// Perf: restrict `c` to containers that actually have a completed
// download *first* (the CTE), so the OR-based self-join runs over that
// handful of rows instead of the entire synced catalog. Without this the
// join is an unindexable O(items²) scan and the Downloaded page hangs on
// a large library ("Loading your downloads…" forever).
let partial_query = Query::with_params(
"WITH downloaded_containers AS (
SELECT DISTINCT c.id
FROM items c
INNER JOIN items children
ON (children.parent_id = c.id OR children.album_id = c.id OR children.season_id = c.id OR children.series_id = c.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND c.server_id = ?
AND c.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
)
SELECT c.id,
COUNT(children.id) AS total_children,
SUM(CASE WHEN d.status = 'completed' THEN 1 ELSE 0 END) AS downloaded_children
FROM items c
INNER JOIN downloaded_containers dc ON dc.id = c.id
INNER JOIN items children
ON (children.parent_id = c.id OR children.album_id = c.id OR children.season_id = c.id OR children.series_id = c.id)
LEFT JOIN downloads d ON children.id = d.item_id AND d.status = 'completed'
WHERE children.item_type IN ('Audio', 'Movie', 'Episode', 'Season')
GROUP BY c.id",
vec![QueryParam::String(self.server_id.clone())],
);
let partial_rows: Vec<(String, i64, i64)> = self
.db_service
.query_many(partial_query, |row| {
Ok((
row.get(0)?,
row.get(1)?,
row.get::<_, Option<i64>>(2)?.unwrap_or(0),
))
})
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut partial_containers = std::collections::HashMap::new();
for (id, total, downloaded) in partial_rows {
// Only record containers that actually have a download (they appear
// in the browse); mark partial when some cached child is missing.
if downloaded > 0 && downloaded < total {
partial_containers.insert(id, true);
}
}
let item_count = leaves.len() as u32;
let device_total_bytes: i64 = leaves.iter().map(|(_, b)| *b).sum();
let mut sizes = std::collections::HashMap::new();
for (id, bytes) in leaves.into_iter().chain(containers.into_iter()) {
// A container id can never collide with a leaf id, so a plain insert
// is fine; use entry to be defensive against duplicate rows.
*sizes.entry(id).or_insert(0) += bytes;
}
Ok(DownloadDiskUsage {
sizes,
partial_containers,
device_total_bytes,
item_count,
})
}
/// Cache playlist items from server into local database
/// Called by HybridRepository after fetching from online
pub async fn save_playlist_items_to_cache(
&self,
playlist_id: &str,
entries: &[PlaylistEntry],
) -> Result<(), RepoError> {
let playlist_id = playlist_id.to_string();
let user_id = self.user_id.clone();
let entries: Vec<(String, String, usize)> = entries
.iter()
.enumerate()
.map(|(i, e)| (e.playlist_item_id.clone(), e.item.id.clone(), i))
.collect();
self.db_service
.transaction(move |tx| {
use crate::storage::db_service::{Query, QueryParam};
// Ensure playlist record exists
tx.execute(Query::with_params(
"INSERT OR IGNORE INTO playlists (id, user_id, name, is_local) VALUES (?1, ?2, '', 0)",
vec![QueryParam::String(playlist_id.clone()), QueryParam::String(user_id)],
))?;
// Clear existing entries and re-insert
tx.execute(Query::with_params(
"DELETE FROM playlist_items WHERE playlist_id = ?",
vec![QueryParam::String(playlist_id.clone())],
))?;
for (_, item_id, sort_order) in &entries {
tx.execute(Query::with_params(
"INSERT OR IGNORE INTO playlist_items (playlist_id, item_id, sort_order) VALUES (?1, ?2, ?3)",
vec![
QueryParam::String(playlist_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::Int(*sort_order as i32),
],
))?;
}
Ok(())
})
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to cache playlist items: {}", e),
})
}
}
#[async_trait]
impl MediaRepository for OfflineRepository {
async fn get_libraries(&self) -> Result<Vec<Library>, RepoError> {
// Return every cached library for this server. We deliberately do NOT
// gate on `items.library_id` here: that column is not populated in the
// cache (the Jellyfin client doesn't parse it), so the old
// `INNER JOIN items i ON i.library_id = l.id` matched nothing and left
// offline startup with zero libraries. Navigating into a library still
// filters to downloaded content via get_items, so listing all cached
// libraries is correct — it's the "local first" list the UI browses.
let query = Query::with_params(
"SELECT l.id, l.name, l.collection_type, l.image_tag
FROM libraries l
WHERE l.server_id = ?
ORDER BY l.sort_order ASC, l.name ASC",
vec![QueryParam::String(self.server_id.clone())],
);
self.db_service
.query_many(query, |row| {
Ok(Library {
id: row.get(0)?,
name: row.get(1)?,
collection_type: row
.get::<_, Option<String>>(2)?
.unwrap_or_else(|| "unknown".to_string()),
image_tag: row.get(3)?,
})
})
.await
.map_err(|e| RepoError::Database { message: e })
}
async fn get_items(
&self,
parent_id: &str,
options: Option<GetItemsOptions>,
) -> Result<SearchResult, RepoError> {
debug!(
"[OfflineRepo] get_items called for parent_id: {}",
&parent_id[..8.min(parent_id.len())]
);
let opts = options.unwrap_or_default();
let limit = opts.limit.unwrap_or(10000); // Match frontend limit for full library loading
let start_index = opts.start_index.unwrap_or(0);
// SortBy=Random is the only sort the landing pages rely on offline (the
// hero "surprise" pool); everything else keeps the stable name order.
let order_by = match opts.sort_by.as_deref() {
Some("Random") => "RANDOM()",
_ => "i.sort_name ASC, i.name ASC",
};
// Build type filter for optional filtering
let type_filter = if let Some(include_item_types) = &opts.include_item_types {
if !include_item_types.is_empty() {
let types = include_item_types
.iter()
.map(|t| format!("'{}'", t))
.collect::<Vec<_>>()
.join(",");
format!(" AND i.item_type IN ({})", types)
} else {
String::new()
}
} else {
String::new()
};
// Favourites narrowing for a normal library listing. Bound rather than
// interpolated, and appended after the parent-matching group so its
// parameter is simply the last one in the vec below.
// TRACES: UR-067 | DR-116 | UT-104
let favorites_filter = if opts.favorites_only == Some(true) {
" AND EXISTS (
SELECT 1 FROM user_data ud
WHERE ud.item_id = i.id AND ud.user_id = ? AND ud.is_favorite = 1
)"
} else {
""
};
// Use CTE to find items that are either:
// 1. Playable items (Audio, Movie, Episode) with completed downloads (offline mode)
// 2. Container items (MusicAlbum, Series, Season) with at least one downloaded child (offline mode)
// 3. Cached items with recent synced_at timestamp (fast online browsing, or the
// offline "Show all server media" catalog view) — only when the catalog-browse
// flag is set. When offline with the toggle off, this branch is omitted so the
// page shows downloaded/local media only. See `set_include_catalog_browse`.
let catalog_branch = if include_catalog_browse() {
"UNION
-- Cached items for fast browsing (online) or the offline catalog view
SELECT DISTINCT i.id
FROM items i
WHERE i.synced_at IS NOT NULL"
} else {
""
};
let sql = format!(
"WITH available_items AS (
-- Playable items with completed downloads
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
-- Containers with downloaded children
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
{catalog_branch}
)
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
INNER JOIN available_items ai ON i.id = ai.id
WHERE i.server_id = ?
AND (
i.parent_id = ? OR i.album_id = ? OR i.season_id = ? OR i.series_id = ?
-- When the requested parent is a LIBRARY, there is no per-item
-- link back to it (library_id/parent_id are NULL in the cache),
-- so match every item on the server and let the type filter
-- (e.g. MusicAlbum / Movie / Series) narrow it. This is what
-- makes library landing pages show albums/movies/shows offline.
OR EXISTS (
SELECT 1 FROM libraries l
WHERE l.id = ? AND l.server_id = i.server_id
)
){}{}
ORDER BY {}
LIMIT {} OFFSET {}",
type_filter, favorites_filter, order_by, limit, start_index
);
// The requested id is compared against every hierarchy-linkage column
// because `parent_id` is not populated for cached items — music tracks
// link to their album via `album_id`, episodes to their season/series
// via `season_id`/`series_id`, and a library parent matches via the
// `libraries` EXISTS clause. See [[offline-libraries-never-cached]].
let mut params = vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(parent_id.to_string()), // i.parent_id = ?
QueryParam::String(parent_id.to_string()), // i.album_id = ?
QueryParam::String(parent_id.to_string()), // i.season_id = ?
QueryParam::String(parent_id.to_string()), // i.series_id = ?
QueryParam::String(parent_id.to_string()), // libraries.id = ?
];
if !favorites_filter.is_empty() {
params.push(QueryParam::String(self.user_id.clone())); // ud.user_id = ?
}
let query = Query::with_params(sql, params);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
debug!(
"[OfflineRepo] Found {} cached items for parent {}",
cached_items.len(),
&parent_id[..8.min(parent_id.len())]
);
// Fetch user data for each item
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
let total_record_count = items.len();
debug!(
"[OfflineRepo] Returning {} items for parent {}",
total_record_count,
&parent_id[..8.min(parent_id.len())]
);
Ok(SearchResult {
items,
total_record_count,
})
}
async fn get_item(&self, item_id: &str) -> Result<MediaItem, RepoError> {
// Check if item is available offline (either downloaded itself or has downloaded children)
let query = Query::with_params(
"WITH downloaded_items AS (
-- Playable items with completed downloads
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
-- Containers with downloaded children
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
)
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
INNER JOIN downloaded_items di ON i.id = di.id
WHERE i.id = ?",
vec![QueryParam::String(item_id.to_string())],
);
let cached = self
.db_service
.query_optional(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?
.ok_or_else(|| RepoError::NotFound {
message: format!(
"Item {} not found in offline cache or not downloaded",
item_id
),
})?;
let user_data = self.get_user_data(item_id).await;
Ok(Self::cached_item_to_media_item(cached, user_data))
}
async fn get_latest_items(
&self,
parent_id: &str,
limit: Option<usize>,
) -> Result<Vec<MediaItem>, RepoError> {
let limit_val = limit.unwrap_or(16);
let query = Query::with_params(
format!(
"WITH downloaded_items AS (
-- Playable items with completed downloads
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
-- Containers with downloaded children
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
)
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
INNER JOIN downloaded_items di ON i.id = di.id
WHERE i.server_id = ? AND i.library_id = ?
ORDER BY i.synced_at DESC
LIMIT {}", limit_val
),
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(parent_id.to_string()),
],
);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
Ok(items)
}
async fn get_resume_items(
&self,
parent_id: Option<&str>,
limit: Option<usize>,
) -> Result<Vec<MediaItem>, RepoError> {
let limit_val = limit.unwrap_or(12);
// Resume items are video-only (Movie, Episode) - audio is handled by get_recently_played_audio
let (sql, params) = if let Some(pid) = parent_id {
(
format!(
"SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
i.overview, i.genres, i.runtime_ticks, i.production_year,
i.community_rating, i.official_rating, i.primary_image_tag,
i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id,
i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
JOIN user_data ud ON i.id = ud.item_id
INNER JOIN downloads d ON i.id = d.item_id
WHERE i.server_id = ? AND ud.user_id = ? AND i.library_id = ?
AND ud.playback_position_ticks > 0 AND ud.is_played = 0
AND d.status = 'completed'
AND i.item_type IN ('Movie', 'Episode')
ORDER BY ud.last_played_at DESC
LIMIT {}",
limit_val
),
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(self.user_id.clone()),
QueryParam::String(pid.to_string()),
],
)
} else {
(
format!(
"SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
i.overview, i.genres, i.runtime_ticks, i.production_year,
i.community_rating, i.official_rating, i.primary_image_tag,
i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id,
i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
JOIN user_data ud ON i.id = ud.item_id
INNER JOIN downloads d ON i.id = d.item_id
WHERE i.server_id = ? AND ud.user_id = ?
AND ud.playback_position_ticks > 0 AND ud.is_played = 0
AND d.status = 'completed'
AND i.item_type IN ('Movie', 'Episode')
ORDER BY ud.last_played_at DESC
LIMIT {}",
limit_val
),
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(self.user_id.clone()),
],
)
};
let query = Query::with_params(sql, params);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
Ok(items)
}
async fn get_next_up_episodes(
&self,
_series_id: Option<&str>,
_limit: Option<usize>,
) -> Result<Vec<MediaItem>, RepoError> {
// Next up is complex - would need to track watched episodes and find the next unwatched
// For now, return empty for offline mode
Ok(Vec::new())
}
async fn get_recently_played_audio(
&self,
limit: Option<usize>,
) -> Result<Vec<MediaItem>, RepoError> {
let limit_val = limit.unwrap_or(12);
// Use CTE to intelligently group by playback context and filter by downloads
// Shows containers (albums) when context_type='container', individual tracks when context_type='single'
// Falls back to album grouping for legacy data (NULL context)
// Only shows items that are downloaded or have downloaded children
let query = Query::with_params(
format!(
"WITH downloaded_items AS (
-- Playable items with completed downloads (Audio tracks)
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type = 'Audio'
UNION
-- Containers with downloaded children (Albums)
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type = 'MusicAlbum'
),
ranked_plays AS (
SELECT
CASE
WHEN ud.playback_context_type = 'container' THEN ud.playback_context_id
WHEN ud.playback_context_type = 'single' THEN ud.item_id
ELSE COALESCE(i.album_id, ud.item_id)
END AS display_id,
MAX(ud.last_played_at) AS most_recent_play
FROM user_data ud
JOIN items i ON ud.item_id = i.id
WHERE ud.user_id = ? AND i.server_id = ?
AND i.item_type = 'Audio'
AND ud.last_played_at IS NOT NULL
GROUP BY display_id
ORDER BY most_recent_play DESC
LIMIT {}
)
SELECT DISTINCT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
i.overview, i.genres, i.runtime_ticks, i.production_year,
i.community_rating, i.official_rating, i.primary_image_tag,
i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id,
i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
FROM ranked_plays rp
JOIN items i ON rp.display_id = i.id
INNER JOIN downloaded_items di ON i.id = di.id
ORDER BY rp.most_recent_play DESC",
limit_val
),
vec![
QueryParam::String(self.user_id.clone()),
QueryParam::String(self.server_id.clone()),
],
);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
Ok(items)
}
async fn get_rediscover_albums(
&self,
_parent_id: Option<&str>,
_limit: Option<usize>,
) -> Result<Vec<MediaItem>, RepoError> {
// "Rediscover" is a discovery feature over the full server library.
// Offline only holds downloaded items, so there is nothing meaningful
// to surface here; the hybrid repo serves this from the server instead.
Ok(Vec::new())
}
async fn get_resume_movies(&self, limit: Option<usize>) -> Result<Vec<MediaItem>, RepoError> {
let limit_val = limit.unwrap_or(12);
// Resume movies are playable items, so simple JOIN with downloads
let query = Query::with_params(
format!(
"SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
i.overview, i.genres, i.runtime_ticks, i.production_year,
i.community_rating, i.official_rating, i.primary_image_tag,
i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id,
i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
JOIN user_data ud ON i.id = ud.item_id
INNER JOIN downloads d ON i.id = d.item_id
WHERE i.server_id = ? AND ud.user_id = ? AND i.item_type = 'Movie'
AND ud.playback_position_ticks > 0 AND ud.is_played = 0
AND d.status = 'completed'
ORDER BY ud.last_played_at DESC
LIMIT {}",
limit_val
),
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(self.user_id.clone()),
],
);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
Ok(items)
}
async fn get_genres(&self, parent_id: Option<&str>) -> Result<Vec<Genre>, RepoError> {
// Read the cached server genre catalog (populated by the hybrid repo via
// save_genres_to_cache). This is the FULL genre list for the library, not
// just the genres derivable from locally-cached albums — so offline keeps
// the same variety the server has. library_id NULL is stored as ''.
let library_id = parent_id.unwrap_or("").to_string();
let query = Query::with_params(
"SELECT id, name, album_count FROM genres WHERE server_id = ? AND library_id = ?",
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(library_id),
],
);
let genres: Vec<Genre> = self
.db_service
.query_many(query, |row| {
Ok(Genre {
id: row.get(0)?,
name: row.get(1)?,
album_count: row.get::<_, Option<i64>>(2)?.map(|c| c as u32),
})
})
.await
.map_err(|e| RepoError::Database { message: e })?;
Ok(genres)
}
async fn search(
&self,
query: &str,
options: Option<SearchOptions>,
) -> Result<SearchResult, RepoError> {
let opts = options.unwrap_or_default();
let limit = opts.limit.unwrap_or(20);
// Nothing searchable (empty query, or pure punctuation): return an empty
// result rather than letting FTS5 reject the string and fail the search.
let Some(fts_query) = build_fts_prefix_query(query) else {
return Ok(SearchResult {
items: Vec::new(),
total_record_count: 0,
});
};
// Bind the type filter rather than interpolating it: `include_item_types`
// is settable straight from the frontend (GenericMediaListPage passes it),
// so a quote in a type must be data, not syntax.
let type_values: &[String] = opts
.include_item_types
.as_deref()
.filter(|types| !types.is_empty())
.unwrap_or(&[]);
let type_filter = if type_values.is_empty() {
String::new()
} else {
let placeholders = vec!["?"; type_values.len()].join(",");
format!(" AND i.item_type IN ({})", placeholders)
};
// Availability CTE — deliberately identical to the one `get_items` uses,
// including the `include_catalog_browse()` gate, so search and browse can
// never disagree about what is visible. Before DR-108 this leg was
// downloads-only, which meant a user with no downloads got nothing from
// the local index and every keystroke fell through to the server.
let catalog_branch = if include_catalog_browse() {
"UNION
-- Synced catalog: fast online search, or the offline
-- 'Show all server media' view. See set_include_catalog_browse.
SELECT DISTINCT i.id
FROM items i
WHERE i.synced_at IS NOT NULL"
} else {
""
};
let sql = format!(
"WITH available_items AS (
-- Playable items with completed downloads
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
-- Containers with downloaded children
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
{}
)
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
i.overview, i.genres, i.runtime_ticks, i.production_year,
i.community_rating, i.official_rating, i.primary_image_tag,
i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id,
i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
JOIN items_fts fts ON fts.rowid = i.rowid
INNER JOIN available_items ai ON i.id = ai.id
WHERE i.server_id = ? AND items_fts MATCH ?{}
ORDER BY rank
LIMIT {}",
catalog_branch, type_filter, limit
);
let mut params = vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(fts_query.clone()),
];
params.extend(type_values.iter().cloned().map(QueryParam::String));
let db_query = Query::with_params(sql, params);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(db_query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
// People are not rows in `items` — they live in their own table — so
// they need a second lookup. Only when the scope admits them: an
// explicit type filter (Music/Movies/TV) never includes People, whereas
// `SearchScope::All` expands to *no* filter (DR-063), which is exactly
// the case where the People group should be populated.
if type_values.is_empty() {
items.extend(self.search_people(&fts_query, limit).await?);
}
let total_record_count = items.len();
Ok(SearchResult {
items,
total_record_count,
})
}
async fn get_playback_info(&self, _item_id: &str) -> Result<PlaybackInfo, RepoError> {
// Playback info requires server communication for transcoding decisions
Err(RepoError::Offline)
}
async fn get_audio_stream_url(&self, _item_id: &str) -> Result<String, RepoError> {
// Cannot get stream URLs while offline - offline tracks use local paths
Err(RepoError::Offline)
}
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 requires the server; offline downloads play locally.
Err(RepoError::Offline)
}
async fn get_live_tv_channels(&self) -> Result<Vec<MediaItem>, RepoError> {
// Live TV is inherently online-only.
Err(RepoError::Offline)
}
async fn get_channels(&self) -> Result<SearchResult, RepoError> {
// Plugin channels are inherently online-only.
Err(RepoError::Offline)
}
async fn open_live_stream(&self, _item_id: &str) -> Result<LiveStreamInfo, RepoError> {
// Live streams cannot be opened offline.
Err(RepoError::Offline)
}
async fn report_playback_start(
&self,
_item_id: &str,
_position_ticks: i64,
) -> Result<(), RepoError> {
// Cannot report to server while offline
Err(RepoError::Offline)
}
async fn report_playback_progress(
&self,
_item_id: &str,
_position_ticks: i64,
) -> Result<(), RepoError> {
// Cannot report to server while offline
Err(RepoError::Offline)
}
async fn report_playback_stopped(
&self,
_item_id: &str,
_position_ticks: i64,
) -> Result<(), RepoError> {
// Cannot report to server while offline
Err(RepoError::Offline)
}
fn get_image_url(
&self,
item_id: &str,
image_type: ImageType,
options: Option<ImageOptions>,
) -> String {
// Return a placeholder path for offline image retrieval
// The actual image should be in thumbnail cache
let type_str = match image_type {
ImageType::Primary => "Primary",
ImageType::Backdrop => "Backdrop",
ImageType::Logo => "Logo",
ImageType::Thumb => "Thumb",
ImageType::Banner => "Banner",
};
if let Some(opts) = options {
if let Some(tag) = opts.tag {
return format!("offline://{}/{}/{}", item_id, type_str, tag);
}
}
format!("offline://{}/{}", item_id, type_str)
}
fn get_subtitle_url(
&self,
_item_id: &str,
_media_source_id: &str,
_stream_index: i32,
_format: &str,
) -> String {
// Subtitles not available offline
String::new()
}
fn get_video_download_url(
&self,
_item_id: &str,
_quality: &str,
_media_source_id: Option<&str>,
) -> String {
// Cannot download while offline
String::new()
}
async fn mark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
// Cannot update server while offline
Err(RepoError::Offline)
}
async fn unmark_favorite(&self, _item_id: &str) -> Result<(), RepoError> {
// Cannot update server while offline
Err(RepoError::Offline)
}
/// Favourites held locally — the ones mirrored from the server by
/// `save_to_cache` plus anything favourited on this device.
///
/// Gated by the same `available_items` rules as browsing, so with "Show all
/// server media" off this returns favourites that are actually on the
/// device rather than the whole favourited catalog (DR-080).
///
/// TRACES: UR-067 | DR-115 | UT-101
async fn get_favorites(
&self,
scope: SearchScope,
options: Option<GetItemsOptions>,
) -> Result<SearchResult, RepoError> {
let opts = options.unwrap_or_default();
let limit = opts.limit.unwrap_or(10000);
let start_index = opts.start_index.unwrap_or(0);
// Scope → item types is expanded in Rust (DR-063); `All` yields no
// filter at all rather than a union.
let type_filter = match scope.item_types() {
Some(types) if !types.is_empty() => {
let placeholders = vec!["?"; types.len()].join(",");
format!(" AND i.item_type IN ({})", placeholders)
}
_ => String::new(),
};
let catalog_branch = if include_catalog_browse() {
"UNION
SELECT DISTINCT i.id
FROM items i
WHERE i.synced_at IS NOT NULL"
} else {
""
};
let sql = format!(
"WITH available_items AS (
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
{catalog_branch}
)
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres,
i.runtime_ticks, i.production_year, i.community_rating, i.official_rating,
i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id, i.season_name,
i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
INNER JOIN available_items ai ON i.id = ai.id
INNER JOIN user_data ud ON ud.item_id = i.id
WHERE i.server_id = ?
AND ud.user_id = ?
AND ud.is_favorite = 1{}
ORDER BY i.sort_name ASC, i.name ASC
LIMIT {} OFFSET {}",
type_filter, limit, start_index
);
let mut params = vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(self.user_id.clone()),
];
if let Some(types) = scope.item_types() {
params.extend(types.into_iter().map(QueryParam::String));
}
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(Query::with_params(sql, params), row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
let total_record_count = items.len();
debug!("[OfflineRepo] Returning {} favourites", total_record_count);
Ok(SearchResult {
items,
total_record_count,
})
}
async fn clear_watch_history(&self, _item_id: &str) -> Result<(), RepoError> {
// Erasing history has to reach the server to be meaningful — clearing
// it only locally would be silently undone by the next sync.
Err(RepoError::Offline)
}
async fn mark_played(&self, _item_id: &str) -> Result<(), RepoError> {
// Offline the local flag is written by `storage_mark_played` and the
// server half is queued in `sync_queue`; this path has no server.
Err(RepoError::Offline)
}
async fn get_person(&self, person_id: &str) -> Result<MediaItem, RepoError> {
let query = Query::with_params(
"SELECT id, name, overview, primary_image_tag
FROM people WHERE id = ?",
vec![QueryParam::String(person_id.to_string())],
);
let person_data = self
.db_service
.query_optional(query, |row| {
Ok((
row.get::<_, String>(0)?,
row.get::<_, String>(1)?,
row.get::<_, Option<String>>(2)?,
row.get::<_, Option<String>>(3)?,
))
})
.await
.map_err(|e| RepoError::Database { message: e })?
.ok_or_else(|| RepoError::NotFound {
message: format!("Person {} not found in cache", person_id),
})?;
Ok(MediaItem {
id: person_data.0,
name: person_data.1,
item_type: "Person".to_string(),
kind: crate::domain::MediaKind::Person,
is_folder: false,
server_id: self.server_id.clone(),
parent_id: None,
library_id: None,
overview: person_data.2,
genres: None,
runtime_ticks: None,
duration_ms: None,
production_year: None,
premiere_date: None,
community_rating: None,
official_rating: None,
primary_image_tag: person_data.3.clone(),
image_id: person_data.3,
backdrop_image_tags: None,
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,
})
}
async fn get_items_by_person(
&self,
person_id: &str,
options: Option<GetItemsOptions>,
) -> Result<SearchResult, RepoError> {
let opts = options.unwrap_or_default();
let limit = opts.limit.unwrap_or(10000); // Match frontend limit
// Filter by downloads using CTE
let query = Query::with_params(
format!(
"WITH downloaded_items AS (
-- Playable items with completed downloads
SELECT DISTINCT i.id
FROM items i
INNER JOIN downloads d ON i.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('Audio', 'Movie', 'Episode')
UNION
-- Containers with downloaded children
SELECT DISTINCT i.id
FROM items i
INNER JOIN items children ON (children.parent_id = i.id OR children.album_id = i.id OR children.season_id = i.id OR children.series_id = i.id)
INNER JOIN downloads d ON children.id = d.item_id
WHERE d.status = 'completed'
AND i.item_type IN ('MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder')
)
SELECT i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id,
i.overview, i.genres, i.runtime_ticks, i.production_year,
i.community_rating, i.official_rating, i.primary_image_tag,
i.album_id, i.album_name, i.album_artist, i.artists,
i.index_number, i.series_id, i.series_name, i.season_id,
i.season_name, i.parent_index_number, i.is_folder, i.premiere_date
FROM items i
JOIN item_people ip ON i.id = ip.item_id
INNER JOIN downloaded_items di ON i.id = di.id
WHERE i.server_id = ? AND ip.person_id = ?
ORDER BY i.production_year DESC, i.sort_name ASC
LIMIT {}", limit
),
vec![
QueryParam::String(self.server_id.clone()),
QueryParam::String(person_id.to_string()),
],
);
let cached_items: Vec<CachedItem> = self
.db_service
.query_many(query, row_to_cached_item)
.await
.map_err(|e| RepoError::Database { message: e })?;
let mut items = Vec::new();
for cached in cached_items {
let user_data = self.get_user_data(&cached.id).await;
items.push(Self::cached_item_to_media_item(cached, user_data));
}
let total_record_count = items.len();
Ok(SearchResult {
items,
total_record_count,
})
}
async fn get_similar_items(
&self,
_item_id: &str,
_limit: Option<usize>,
) -> Result<SearchResult, RepoError> {
// Similar items require server-side computation and are not available offline
Err(RepoError::Offline)
}
// ===== Playlist Methods =====
async fn create_playlist(
&self,
name: &str,
item_ids: &[String],
) -> Result<PlaylistCreatedResult, RepoError> {
let playlist_id = uuid::Uuid::new_v4().to_string();
let user_id = self.user_id.clone();
let name = name.to_string();
let item_ids = item_ids.to_vec();
let pid = playlist_id.clone();
self.db_service
.transaction(move |tx| {
use crate::storage::db_service::{Query, QueryParam};
tx.execute(Query::with_params(
"INSERT INTO playlists (id, user_id, name, is_local) VALUES (?1, ?2, ?3, 1)",
vec![QueryParam::String(pid.clone()), QueryParam::String(user_id), QueryParam::String(name)],
))?;
for (i, item_id) in item_ids.iter().enumerate() {
tx.execute(Query::with_params(
"INSERT OR IGNORE INTO playlist_items (playlist_id, item_id, sort_order) VALUES (?1, ?2, ?3)",
vec![QueryParam::String(pid.clone()), QueryParam::String(item_id.clone()), QueryParam::Int(i as i32)],
))?;
}
Ok(())
})
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to create playlist: {}", e),
})?;
Ok(PlaylistCreatedResult { id: playlist_id })
}
async fn delete_playlist(&self, playlist_id: &str) -> Result<(), RepoError> {
let query = Query::with_params(
"DELETE FROM playlists WHERE id = ?",
vec![QueryParam::String(playlist_id.to_string())],
);
self.db_service
.execute(query)
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to delete playlist: {}", e),
})?;
Ok(())
}
async fn rename_playlist(&self, playlist_id: &str, name: &str) -> Result<(), RepoError> {
let query = Query::with_params(
"UPDATE playlists SET name = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
vec![
QueryParam::String(name.to_string()),
QueryParam::String(playlist_id.to_string()),
],
);
self.db_service
.execute(query)
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to rename playlist: {}", e),
})?;
Ok(())
}
async fn get_playlist_items(&self, playlist_id: &str) -> Result<Vec<PlaylistEntry>, RepoError> {
let query = Query::with_params(
"SELECT pi.id, \
i.id, i.name, i.item_type, i.server_id, i.parent_id, i.library_id, i.overview, i.genres, \
i.runtime_ticks, i.production_year, i.community_rating, i.official_rating, \
i.primary_image_tag, i.album_id, i.album_name, i.album_artist, i.artists, \
i.index_number, i.series_id, i.series_name, i.season_id, i.season_name, \
i.parent_index_number, i.is_folder, i.premiere_date \
FROM playlist_items pi \
JOIN items i ON pi.item_id = i.id \
WHERE pi.playlist_id = ? \
ORDER BY pi.sort_order ASC",
vec![QueryParam::String(playlist_id.to_string())],
);
let items = self
.db_service
.query_many(query, |row| {
let entry_id: i64 = row.get(0)?;
// Columns offset by 1 because first column is pi.id
let cached = CachedItem {
id: row.get(1)?,
name: row.get(2)?,
item_type: row.get(3)?,
server_id: row.get(4)?,
parent_id: row.get(5)?,
library_id: row.get(6)?,
overview: row.get(7)?,
genres: row.get(8)?,
runtime_ticks: row.get(9)?,
production_year: row.get(10)?,
community_rating: row.get(11)?,
official_rating: row.get(12)?,
primary_image_tag: row.get(13)?,
backdrop_image_tags: None,
parent_backdrop_image_tags: None,
album_id: row.get(14)?,
album_name: row.get(15)?,
album_artist: row.get(16)?,
artists: row.get(17)?,
index_number: row.get(18)?,
series_id: row.get(19)?,
series_name: row.get(20)?,
season_id: row.get(21)?,
season_name: row.get(22)?,
parent_index_number: row.get(23)?,
is_folder: row.get::<_, Option<i64>>(24)?.unwrap_or(0) != 0,
premiere_date: row.get(25)?,
};
Ok((entry_id.to_string(), cached))
})
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to get playlist items: {}", e),
})?;
Ok(items
.into_iter()
.map(|(entry_id, cached)| PlaylistEntry {
playlist_item_id: entry_id,
item: Self::cached_item_to_media_item(cached, None),
})
.collect())
}
async fn add_to_playlist(
&self,
playlist_id: &str,
item_ids: &[String],
) -> Result<(), RepoError> {
// Get current max sort_order
let max_query = Query::with_params(
"SELECT COALESCE(MAX(sort_order), -1) FROM playlist_items WHERE playlist_id = ?",
vec![QueryParam::String(playlist_id.to_string())],
);
let max_order: i32 = self
.db_service
.query_one(max_query, |row| row.get(0))
.await
.unwrap_or(-1);
let playlist_id = playlist_id.to_string();
let item_ids = item_ids.to_vec();
self.db_service
.transaction(move |tx| {
use crate::storage::db_service::{Query, QueryParam};
for (i, item_id) in item_ids.iter().enumerate() {
tx.execute(Query::with_params(
"INSERT OR IGNORE INTO playlist_items (playlist_id, item_id, sort_order) VALUES (?1, ?2, ?3)",
vec![
QueryParam::String(playlist_id.clone()),
QueryParam::String(item_id.clone()),
QueryParam::Int(max_order + 1 + i as i32),
],
))?;
}
Ok(())
})
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to add items to playlist: {}", e),
})?;
Ok(())
}
async fn remove_from_playlist(
&self,
playlist_id: &str,
entry_ids: &[String],
) -> Result<(), RepoError> {
let playlist_id = playlist_id.to_string();
let entry_ids = entry_ids.to_vec();
self.db_service
.transaction(move |tx| {
use crate::storage::db_service::{Query, QueryParam};
for entry_id in &entry_ids {
tx.execute(Query::with_params(
"DELETE FROM playlist_items WHERE playlist_id = ? AND id = ?",
vec![
QueryParam::String(playlist_id.clone()),
QueryParam::String(entry_id.clone()),
],
))?;
}
Ok(())
})
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to remove items from playlist: {}", e),
})?;
Ok(())
}
async fn move_playlist_item(
&self,
playlist_id: &str,
item_id: &str,
new_index: u32,
) -> Result<(), RepoError> {
let playlist_id = playlist_id.to_string();
let item_id = item_id.to_string();
self.db_service
.transaction(move |tx| {
use crate::storage::db_service::{Query, QueryParam};
// Get all items ordered by sort_order
let items: Vec<(i64, String)> = tx.query_many(
Query::with_params(
"SELECT id, item_id FROM playlist_items WHERE playlist_id = ? ORDER BY sort_order",
vec![QueryParam::String(playlist_id)],
),
|row| Ok((row.get(0)?, row.get(1)?)),
)?;
// Find the item to move
let old_idx = items.iter().position(|(_, iid)| iid == &item_id);
if let Some(old_pos) = old_idx {
let mut ids = items;
let entry = ids.remove(old_pos);
let insert_at = (new_index as usize).min(ids.len());
ids.insert(insert_at, entry);
// Renumber all sort_orders
for (i, (entry_id, _)) in ids.iter().enumerate() {
tx.execute(Query::with_params(
"UPDATE playlist_items SET sort_order = ? WHERE id = ?",
vec![QueryParam::Int(i as i32), QueryParam::Int64(*entry_id)],
))?;
}
}
Ok(())
})
.await
.map_err(|e| RepoError::Database {
message: format!("Failed to move playlist item: {}", e),
})?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::db_service::RusqliteService;
use rusqlite::Connection;
use std::sync::{Arc, Mutex};
/// Helper to create a test database with the necessary schema
/// `INCLUDE_CATALOG_BROWSE` is a process-global `AtomicBool`, so every test
/// that flips it must hold this lock — cargo runs tests in parallel threads
/// and would otherwise let them race, producing intermittent failures that
/// look like query bugs. Poisoning is recovered rather than propagated: a
/// panic in one such test should not cascade into unrelated ones.
static CATALOG_BROWSE_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn lock_catalog_browse() -> std::sync::MutexGuard<'static, ()> {
CATALOG_BROWSE_LOCK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
/// TRACES: UR-065 | DR-108 | UT-111
#[test]
fn test_build_fts_prefix_query() {
// Plain input: quoted phrase plus the prefix operator.
assert_eq!(build_fts_prefix_query("Arr").as_deref(), Some("\"Arr\"*"));
// Multiple tokens stay implicitly ANDed; only the last is a prefix, so
// results still narrow while the user is typing.
assert_eq!(
build_fts_prefix_query("parks rec").as_deref(),
Some("\"parks\" \"rec\"*")
);
// Punctuation is data, not FTS5 syntax. Unquoted, each of these makes
// MATCH raise a syntax error and the whole search fail.
for query in ["Bob's Burgers", "Spider-Man", "AC/DC", "Wall-E", "9-1-1"] {
let built = build_fts_prefix_query(query).expect("should build");
assert!(
built.starts_with('"') && built.ends_with("*"),
"{query:?} produced {built:?}"
);
}
// An embedded double quote is escaped by doubling, not left to close
// the phrase early.
assert_eq!(
build_fts_prefix_query("say \"hi\"").as_deref(),
Some("\"say\" \"\"\"hi\"\"\"*")
);
// Nothing searchable => None, so callers skip the query instead of
// handing FTS5 a string it rejects. `search("")` is a real call site.
assert_eq!(build_fts_prefix_query(""), None);
assert_eq!(build_fts_prefix_query(" "), None);
assert_eq!(build_fts_prefix_query("-"), None);
}
/// An empty query must yield an empty result, not a database error — the
/// "list all playlists" call site passes one.
///
/// TRACES: UR-065 | DR-108 | UT-111
#[tokio::test]
async fn test_search_empty_query_returns_empty_not_error() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service,
"test-server".to_string(),
"test-user".to_string(),
);
let result = repo.search("", None).await;
assert!(
result.is_ok(),
"empty query must not error: {:?}",
result.err()
);
assert!(result.unwrap().items.is_empty());
}
fn create_test_db() -> Arc<RusqliteService> {
let conn = Connection::open_in_memory().unwrap();
// Enable foreign key constraints (they're disabled by default in SQLite)
conn.execute("PRAGMA foreign_keys = ON", []).unwrap();
// Create minimal schema for testing
conn.execute_batch(
r#"
CREATE TABLE servers (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
url TEXT NOT NULL UNIQUE
);
CREATE TABLE items (
id TEXT PRIMARY KEY,
server_id TEXT NOT NULL REFERENCES servers(id) ON DELETE CASCADE,
library_id TEXT,
parent_id TEXT REFERENCES items(id) ON DELETE CASCADE,
name TEXT NOT NULL,
item_type TEXT NOT NULL,
is_folder INTEGER DEFAULT 0,
overview TEXT,
genres TEXT,
runtime_ticks INTEGER,
production_year INTEGER,
premiere_date TEXT,
community_rating REAL,
official_rating TEXT,
primary_image_tag TEXT,
backdrop_image_tags TEXT,
album_id TEXT,
album_name TEXT,
album_artist TEXT,
artists TEXT,
index_number INTEGER,
series_id TEXT,
series_name TEXT,
season_id TEXT,
season_name TEXT,
parent_index_number INTEGER,
synced_at TEXT,
sort_name TEXT
);
-- Mirrors the real FTS5 index and its triggers (schema.rs migration
-- 001) so search can be exercised in tests at all.
CREATE VIRTUAL TABLE items_fts USING fts5(
name, overview, album_name, album_artist, artists, series_name,
content='items', content_rowid='rowid'
);
CREATE TRIGGER items_ai AFTER INSERT ON items BEGIN
INSERT INTO items_fts(rowid, name, overview, album_name, album_artist, artists, series_name)
VALUES (new.rowid, new.name, new.overview, new.album_name, new.album_artist, new.artists, new.series_name);
END;
CREATE TRIGGER items_ad AFTER DELETE ON items BEGIN
INSERT INTO items_fts(items_fts, rowid, name, overview, album_name, album_artist, artists, series_name)
VALUES('delete', old.rowid, old.name, old.overview, old.album_name, old.album_artist, old.artists, old.series_name);
END;
CREATE TRIGGER items_au AFTER UPDATE ON items BEGIN
INSERT INTO items_fts(items_fts, rowid, name, overview, album_name, album_artist, artists, series_name)
VALUES('delete', old.rowid, old.name, old.overview, old.album_name, old.album_artist, old.artists, old.series_name);
INSERT INTO items_fts(rowid, name, overview, album_name, album_artist, artists, series_name)
VALUES (new.rowid, new.name, new.overview, new.album_name, new.album_artist, new.artists, new.series_name);
END;
CREATE TABLE user_data (
user_id TEXT NOT NULL,
item_id TEXT NOT NULL,
playback_position_ticks INTEGER,
is_played INTEGER,
is_favorite INTEGER,
play_count INTEGER,
last_played_at TEXT,
playback_context_type TEXT,
playback_context_id TEXT,
synced_at TEXT,
pending_sync INTEGER DEFAULT 0,
PRIMARY KEY (user_id, item_id)
);
CREATE TABLE playlists (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
name TEXT NOT NULL,
is_local INTEGER DEFAULT 0,
jellyfin_id TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT
);
CREATE TABLE playlist_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
playlist_id TEXT NOT NULL REFERENCES playlists(id) ON DELETE CASCADE,
item_id TEXT NOT NULL REFERENCES items(id) ON DELETE CASCADE,
sort_order INTEGER NOT NULL,
added_at TEXT DEFAULT CURRENT_TIMESTAMP,
UNIQUE(playlist_id, item_id)
);
CREATE INDEX idx_playlist_items_playlist ON playlist_items(playlist_id, sort_order);
CREATE TABLE downloads (
id INTEGER PRIMARY KEY AUTOINCREMENT,
item_id TEXT NOT NULL,
status TEXT NOT NULL,
file_size INTEGER
);
CREATE TABLE libraries (
id TEXT PRIMARY KEY,
server_id TEXT NOT NULL,
name TEXT NOT NULL,
collection_type TEXT,
image_tag TEXT,
sort_order INTEGER DEFAULT 0,
synced_at TEXT
);
-- Mirrors migration 009 + the migration 022 FTS index.
CREATE TABLE people (
id TEXT PRIMARY KEY,
server_id TEXT NOT NULL,
name TEXT NOT NULL,
overview TEXT,
primary_image_tag TEXT,
premiere_date TEXT,
end_date TEXT,
synced_at TEXT DEFAULT CURRENT_TIMESTAMP
);
CREATE VIRTUAL TABLE people_fts USING fts5(
name, overview, content='people', content_rowid='rowid'
);
CREATE TRIGGER people_ai AFTER INSERT ON people BEGIN
INSERT INTO people_fts(rowid, name, overview)
VALUES (new.rowid, new.name, new.overview);
END;
CREATE TRIGGER people_ad AFTER DELETE ON people BEGIN
INSERT INTO people_fts(people_fts, rowid, name, overview)
VALUES('delete', old.rowid, old.name, old.overview);
END;
CREATE TRIGGER people_au AFTER UPDATE ON people BEGIN
INSERT INTO people_fts(people_fts, rowid, name, overview)
VALUES('delete', old.rowid, old.name, old.overview);
INSERT INTO people_fts(rowid, name, overview)
VALUES (new.rowid, new.name, new.overview);
END;
CREATE TABLE genres (
id TEXT NOT NULL,
server_id TEXT NOT NULL,
library_id TEXT,
name TEXT NOT NULL,
album_count INTEGER,
synced_at TEXT DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (server_id, library_id, name)
);
"#,
)
.unwrap();
// Insert a test server
conn.execute(
"INSERT INTO servers (id, name, url) VALUES ('test-server', 'Test Server', 'http://test')",
[],
).unwrap();
Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
}
/// Helper to create a test MediaItem
fn create_test_item(id: &str, name: &str, parent_id: Option<&str>) -> MediaItem {
MediaItem {
id: id.to_string(),
name: name.to_string(),
item_type: "Audio".to_string(),
kind: crate::domain::MediaKind::Track,
is_folder: false,
server_id: "test-server".to_string(),
parent_id: parent_id.map(|s| s.to_string()),
library_id: None,
overview: None,
genres: None,
runtime_ticks: None,
duration_ms: None,
production_year: None,
premiere_date: None,
community_rating: None,
official_rating: None,
primary_image_tag: None,
image_id: None,
backdrop_image_tags: None,
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,
}
}
#[tokio::test]
async fn test_save_to_cache_with_missing_parent_fk() {
let db_service = create_test_db();
// Verify FK constraints are actually enabled
let fk_enabled: i32 = db_service
.query_one(Query::new("PRAGMA foreign_keys"), |row| row.get(0))
.await
.unwrap();
println!("Foreign keys enabled: {}", fk_enabled);
assert_eq!(fk_enabled, 1, "Foreign keys should be enabled");
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
// Create items where CHILDREN come BEFORE their PARENTS in the list
// This simulates the real-world scenario where Jellyfin's API
// returns items in an arbitrary order (e.g., alphabetically)
let items = vec![
// Tracks from Album 1 (album-1 doesn't exist yet!)
create_test_item("track-1", "Track 1", Some("album-1")),
create_test_item("track-2", "Track 2", Some("album-1")),
// Tracks from Album 2 (album-2 doesn't exist yet!)
create_test_item("track-3", "Track 3", Some("album-2")),
create_test_item("track-4", "Track 4", Some("album-2")),
// Albums come later in the list
create_test_item("album-1", "Album One", Some("library-123")),
create_test_item("album-2", "Album Two", Some("library-123")),
];
println!("Attempting to save {} items...", items.len());
for (i, item) in items.iter().enumerate() {
println!(" Item {}: {} (parent: {:?})", i, item.id, item.parent_id);
}
// This should fail with the current implementation because:
// 1. It only creates a stub for "library-123" (the parent_id parameter)
// 2. When it tries to insert track-1 with parent_id = "album-1",
// album-1 doesn't exist yet, causing FK constraint failure
let result = repo.save_to_cache("library-123", &items).await;
// After the fix, this should succeed and preserve parent relationships
match &result {
Ok(count) => {
println!("✓ Saved {} items", count);
assert_eq!(*count, 6);
// Debug: Check all items in the database
let all_items: Vec<(String, Option<String>)> = db_service
.query_many(
Query::new("SELECT id, parent_id FROM items ORDER BY id"),
|row| Ok((row.get(0)?, row.get(1)?)),
)
.await
.unwrap();
println!("\nAll items in database:");
for (id, parent) in &all_items {
println!(" {} -> parent: {:?}", id, parent);
}
// Verify the ACTUAL parent_ids in the database are preserved correctly
let track1_parent: Option<String> = db_service
.query_optional(
Query::with_params(
"SELECT parent_id FROM items WHERE id = ?",
vec![QueryParam::String("track-1".to_string())],
),
|row| row.get(0),
)
.await
.unwrap()
.flatten();
println!("\ntrack-1 parent_id in DB: {:?}", track1_parent);
println!("track-1 expected parent_id: Some(\"album-1\")");
// Verify parent relationships are preserved
assert_eq!(
track1_parent,
Some("album-1".to_string()),
"track-1 should have parent_id='album-1'"
);
// Verify album-1 has the correct parent too
let album1_parent: Option<String> = db_service
.query_optional(
Query::with_params(
"SELECT parent_id FROM items WHERE id = ?",
vec![QueryParam::String("album-1".to_string())],
),
|row| row.get(0),
)
.await
.unwrap()
.flatten();
assert_eq!(
album1_parent,
Some("library-123".to_string()),
"album-1 should have parent_id='library-123'"
);
}
Err(e) => panic!("Unexpected error: {:?}", e),
}
}
#[tokio::test]
async fn test_save_to_cache_simple_case() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
// Simple case: all items have the same parent_id as the parameter
let items = vec![
create_test_item("item-1", "Item 1", Some("parent-123")),
create_test_item("item-2", "Item 2", Some("parent-123")),
create_test_item("item-3", "Item 3", Some("parent-123")),
];
let result = repo.save_to_cache("parent-123", &items).await;
assert!(result.is_ok(), "Simple case should work: {:?}", result);
assert_eq!(result.unwrap(), 3);
}
/// Regression: a MusicAlbum whose tracks link via `album_id` (and have a
/// NULL `parent_id`, which is how the Jellyfin cache actually stores them)
/// must be recognized as available offline when a track is downloaded.
///
/// Before the fix, `get_item(album_id)` only matched children by
/// `children.parent_id = i.id`, so a fully-downloaded album returned
/// NotFound offline and playback fell through to the (unreachable) server.
#[tokio::test]
async fn test_get_item_album_available_via_album_id_link() {
use crate::storage::db_service::DatabaseService;
let db_service = create_test_db();
for sql in [
// Album container (no children by parent_id).
"INSERT INTO items (id, server_id, name, item_type, album_id, parent_id) \
VALUES ('album-1', 'test-server', 'Hadestown', 'MusicAlbum', NULL, NULL)",
// Track linked to the album ONLY via album_id, parent_id NULL.
"INSERT INTO items (id, server_id, name, item_type, album_id, parent_id) \
VALUES ('track-1', 'test-server', 'Wait For Me', 'Audio', 'album-1', NULL)",
"INSERT INTO downloads (item_id, status) VALUES ('track-1', 'completed')",
] {
db_service.execute(Query::new(sql)).await.unwrap();
}
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
// The downloaded track itself resolves offline.
assert!(
repo.get_item("track-1").await.is_ok(),
"downloaded track should be available offline"
);
// The album must also resolve offline because it has a downloaded child
// linked by album_id (not parent_id).
let album = repo.get_item("album-1").await;
assert!(
album.is_ok(),
"album with an album_id-linked downloaded track should be available offline, got {:?}",
album.err()
);
assert_eq!(album.unwrap().id, "album-1");
// Browsing into the album (get_items) must return its tracks even though
// they link by album_id and have a NULL parent_id. This is the call
// play_album_track makes to build the queue.
let tracks = repo.get_items("album-1", None).await.unwrap();
assert_eq!(
tracks.items.len(),
1,
"get_items(album_id) should return the track"
);
assert_eq!(tracks.items[0].id, "track-1");
}
/// Regression: offline library pages must honor the "Show all server media"
/// toggle. With `include_catalog_browse` off, `get_items` returns only
/// downloaded media — not the whole synced catalog. With it on, the full
/// (synced-but-not-downloaded) catalog is revealed. Fixes the bug where
/// offline library pages showed every server item regardless of the toggle.
///
/// This is also the backend half of the end-to-end offline-listing scenario
/// IT-016: toggle off ⇒ downloaded media only; toggle on ⇒ the cached server
/// catalog is additionally revealed (greyed-out in the UI, distinguished by
/// the absence of a `downloads` row — see `MediaCard.isServerOnly`).
///
/// TRACES: UR-052 | DR-078 | UT-067, IT-016
#[tokio::test]
async fn test_get_items_toggle_gates_synced_catalog() {
use crate::storage::db_service::DatabaseService;
let _guard = lock_catalog_browse();
let db_service = create_test_db();
for sql in [
// Two movies in a library, both merely synced (browsed) — no download.
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
VALUES ('movie-dl', 'test-server', 'Downloaded', 'Movie', 'lib-1', '2026-01-01')",
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
VALUES ('movie-cat', 'test-server', 'CatalogOnly', 'Movie', 'lib-1', '2026-01-01')",
// Only the first movie is actually downloaded.
"INSERT INTO downloads (item_id, status) VALUES ('movie-dl', 'completed')",
// A library row so the library-parent EXISTS clause matches.
"INSERT INTO libraries (id, server_id, name) VALUES ('lib-1', 'test-server', 'Movies')",
] {
db_service.execute(Query::new(sql)).await.unwrap();
}
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let opts = Some(GetItemsOptions {
include_item_types: Some(vec!["Movie".to_string()]),
..Default::default()
});
// Toggle OFF: only the downloaded movie is returned.
set_include_catalog_browse(false);
let local_only = repo.get_items("lib-1", opts.clone()).await.unwrap();
let ids: Vec<&str> = local_only.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(
ids,
vec!["movie-dl"],
"toggle off should show downloaded media only"
);
// Toggle ON: both the downloaded and the catalog-only movie are returned.
set_include_catalog_browse(true);
let full_catalog = repo.get_items("lib-1", opts).await.unwrap();
let mut ids: Vec<&str> = full_catalog.items.iter().map(|i| i.id.as_str()).collect();
ids.sort();
assert_eq!(
ids,
vec!["movie-cat", "movie-dl"],
"toggle on should reveal the full catalog"
);
// Restore default for other tests sharing this process-global flag.
set_include_catalog_browse(true);
}
/// Searching an actor's name must reach them from the local index, and only
/// under a scope that admits People — `SearchScope::All`, which expands to
/// no type filter (DR-063). Before DR-111, `people` had no FTS index at all,
/// so the People group UR-060 requires could only be filled by the server.
///
/// TRACES: UR-065, UR-060 | DR-111 | UT-114
#[tokio::test]
async fn test_search_includes_cached_people() {
use crate::storage::db_service::DatabaseService;
let _guard = lock_catalog_browse();
let db_service = create_test_db();
for sql in [
"INSERT INTO people (id, server_id, name, overview) \
VALUES ('p1', 'test-server', 'Tilda Swinton', 'Actor')",
// A movie that also matches, to prove people are added to — not
// substituted for — item results.
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('m1', 'test-server', 'Tilda the Movie', 'Movie', '2026-01-01')",
] {
db_service.execute(Query::new(sql)).await.unwrap();
}
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
set_include_catalog_browse(true);
// Unscoped search (SearchScope::All => no type filter) reaches people.
let all = repo.search("Tilda", None).await.unwrap();
let mut ids: Vec<&str> = all.items.iter().map(|i| i.id.as_str()).collect();
ids.sort();
assert_eq!(ids, vec!["m1", "p1"], "unscoped search must include people");
let person = all.items.iter().find(|i| i.id == "p1").unwrap();
assert_eq!(person.item_type, "Person");
assert_eq!(person.kind, crate::domain::MediaKind::Person);
// A scoped search names item types, and People are never among them.
let scoped = repo
.search(
"Tilda",
Some(SearchOptions {
include_item_types: Some(vec!["Movie".to_string()]),
..Default::default()
}),
)
.await
.unwrap();
let ids: Vec<&str> = scoped.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(ids, vec!["m1"], "a scoped search must not leak people in");
set_include_catalog_browse(true);
}
/// The post-crawl sweep must remove what the server dropped, and nothing
/// else. Before DR-110 there was no `DELETE FROM items` anywhere in the
/// codebase, so the local catalog was append-only and media removed
/// server-side stayed searchable forever.
///
/// TRACES: UR-065 | DR-110 | UT-113
#[tokio::test]
async fn test_prune_stale_catalog() {
use crate::storage::db_service::DatabaseService;
let db_service = create_test_db();
// "old" predates the crawl; "new" is what this crawl just wrote.
let old = "2026-01-01T00:00:00+00:00";
let new = "2026-06-01T00:00:00+00:00";
let cutoff = "2026-03-01T00:00:00+00:00";
for sql in [
// A second server, so the sweep can be shown to stay scoped to one.
"INSERT INTO servers (id, name, url) \
VALUES ('other-server', 'Other', 'http://other')"
.to_string(),
// Refreshed by the crawl => still on the server => keep.
format!(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('keep-fresh', 'test-server', 'Fresh', 'Movie', '{new}')"
),
// Not refreshed => gone from the server => sweep.
format!(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('gone', 'test-server', 'Vanished', 'Movie', '{old}')"
),
// Not refreshed, but downloaded => the file is on disk => keep.
format!(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('keep-dl', 'test-server', 'Downloaded', 'Movie', '{old}')"
),
"INSERT INTO downloads (item_id, status) VALUES ('keep-dl', 'completed')".to_string(),
// Not refreshed, but a container whose child is downloaded => keep both.
format!(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('keep-album', 'test-server', 'Album', 'MusicAlbum', '{old}')"
),
format!(
"INSERT INTO items (id, server_id, name, item_type, album_id, synced_at) \
VALUES ('keep-track', 'test-server', 'Track', 'Audio', 'keep-album', '{old}')"
),
"INSERT INTO downloads (item_id, status) VALUES ('keep-track', 'completed')"
.to_string(),
// A type the crawl never requests => it is never refreshed, so age
// says nothing about it => keep.
format!(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('keep-artist', 'test-server', 'Artist', 'MusicArtist', '{old}')"
),
// Another server's row must be untouched.
format!(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('keep-other', 'other-server', 'Elsewhere', 'Movie', '{old}')"
),
] {
db_service.execute(Query::new(&sql)).await.unwrap();
}
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let crawled_types = vec![
"Movie".to_string(),
"MusicAlbum".to_string(),
"Audio".to_string(),
];
let removed = repo
.prune_stale_catalog(cutoff, &crawled_types)
.await
.unwrap();
assert_eq!(removed, 1, "only the vanished movie should be swept");
let mut surviving: Vec<String> = db_service
.query_many(Query::new("SELECT id FROM items"), |row| row.get(0))
.await
.unwrap();
surviving.sort();
assert_eq!(
surviving,
vec![
"keep-album",
"keep-artist",
"keep-dl",
"keep-fresh",
"keep-other",
"keep-track",
]
);
// An empty type list is a no-op, not a "delete everything".
assert_eq!(repo.prune_stale_catalog(cutoff, &[]).await.unwrap(), 0);
}
/// Re-caching an item must not append a second FTS entry for it.
///
/// `INSERT OR REPLACE` fires no `AFTER DELETE` trigger unless
/// `recursive_triggers` is on (it is not — storage/mod.rs sets only
/// `foreign_keys` and `journal_mode`), so `items_ad` never ran and the old
/// index row was orphaned; and because `items.id` is a `TEXT PRIMARY KEY`,
/// the replacement row also took a *fresh rowid* and inserted a second
/// entry. Every catalog pass therefore appended a duplicate index. Results
/// stayed correct (the rowid join hides orphans) but `MATCH` degraded
/// permanently — and once the DR-110 deletion sweep starts freeing rowids,
/// a reused rowid would collide with a stale orphan and produce a genuine
/// false positive.
///
/// TRACES: UR-065 | DR-110 | UT-112
#[tokio::test]
async fn test_repeated_cache_does_not_duplicate_fts_entries() {
use crate::storage::db_service::DatabaseService;
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let items = vec![create_test_item("track-1", "Wait For Me", None)];
// Three catalog passes over unchanged content, as happens on every app
// start.
for _ in 0..3 {
repo.save_to_cache("parent-1", &items).await.unwrap();
}
// (`save_to_cache` also inserts a stub row for the parent, so scope this
// to the item itself.)
let item_rows: i64 = db_service
.query_one(
Query::new("SELECT COUNT(*) FROM items WHERE id = 'track-1'"),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(item_rows, 1, "three passes must leave one item row");
let fts_hits: i64 = db_service
.query_one(
Query::with_params(
"SELECT COUNT(*) FROM items_fts WHERE items_fts MATCH ?",
vec![QueryParam::String("\"Wait\"*".to_string())],
),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(
fts_hits, 1,
"the FTS index must hold one entry per item, not one per sync pass"
);
}
/// Search must read the *synced catalog*, not only downloaded media —
/// mirroring `get_items` and gated on the same `include_catalog_browse`
/// flag. Before DR-108 the cache leg wrapped its FTS query in a
/// `downloaded_items` CTE requiring `d.status = 'completed'`, so a user with
/// no downloads got an empty instant result on every keystroke and every
/// query fell through to a full `Recursive=true` server request.
///
/// TRACES: UR-065 | DR-108 | UT-109
#[tokio::test]
async fn test_search_toggle_gates_synced_catalog() {
use crate::storage::db_service::DatabaseService;
let _guard = lock_catalog_browse();
let db_service = create_test_db();
for sql in [
// Two movies, both merely synced (indexed by the catalog crawl).
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
VALUES ('movie-dl', 'test-server', 'Arrival', 'Movie', 'lib-1', '2026-01-01')",
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at) \
VALUES ('movie-cat', 'test-server', 'Arrakis', 'Movie', 'lib-1', '2026-01-01')",
// Only the first is actually downloaded.
"INSERT INTO downloads (item_id, status) VALUES ('movie-dl', 'completed')",
] {
db_service.execute(Query::new(sql)).await.unwrap();
}
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
// Toggle ON (the online case, and the offline "Show all server media"
// case): the whole synced catalog is searchable.
set_include_catalog_browse(true);
let full = repo.search("Arr", None).await.unwrap();
let mut ids: Vec<&str> = full.items.iter().map(|i| i.id.as_str()).collect();
ids.sort();
assert_eq!(
ids,
vec!["movie-cat", "movie-dl"],
"with catalog browse on, search must cover synced-but-not-downloaded items"
);
// Toggle OFF (offline, downloads-only): unchanged from today.
set_include_catalog_browse(false);
let local_only = repo.search("Arr", None).await.unwrap();
let ids: Vec<&str> = local_only.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(
ids,
vec!["movie-dl"],
"with catalog browse off, search stays downloads-only"
);
// Restore default for other tests sharing this process-global flag.
set_include_catalog_browse(true);
}
/// The scope/type filter must be *bound*, not interpolated into the SQL
/// string. `SearchOptions.include_item_types` is settable from the frontend
/// (GenericMediaListPage passes it directly), so a value containing a quote
/// must not be able to alter the query.
///
/// TRACES: UR-065 | DR-108 | UT-110
#[tokio::test]
async fn test_search_type_filter_is_parameterised() {
use crate::storage::db_service::DatabaseService;
let _guard = lock_catalog_browse();
let db_service = create_test_db();
db_service
.execute(Query::new(
"INSERT INTO items (id, server_id, name, item_type, synced_at) \
VALUES ('m1', 'test-server', 'Arrival', 'Movie', '2026-01-01')",
))
.await
.unwrap();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
set_include_catalog_browse(true);
// A quote-bearing type must be treated as data: no SQL error, no match.
let opts = SearchOptions {
include_item_types: Some(vec!["Movie') OR 1=1 --".to_string()]),
..Default::default()
};
let result = repo.search("Arr", Some(opts)).await;
assert!(
result.is_ok(),
"a quote in an item type must not break the query: {:?}",
result.err()
);
assert!(
result.unwrap().items.is_empty(),
"an injected type filter must not widen the result set"
);
// The legitimate filter still works.
let opts = SearchOptions {
include_item_types: Some(vec!["Movie".to_string()]),
..Default::default()
};
assert_eq!(repo.search("Arr", Some(opts)).await.unwrap().items.len(), 1);
}
/// Regression: TV episodes link to their season/series via `season_id` /
/// `series_id` (NOT `parent_id`, which is NULL in the cache). A downloaded
/// episode must make both its Season and Series available offline, and
/// browsing either container must return the episode.
#[tokio::test]
async fn test_get_item_tv_available_via_season_series_link() {
use crate::storage::db_service::DatabaseService;
let db_service = create_test_db();
for sql in [
"INSERT INTO items (id, server_id, name, item_type, parent_id) \
VALUES ('series-1', 'test-server', 'Gilmore Girls', 'Series', NULL)",
"INSERT INTO items (id, server_id, name, item_type, series_id, parent_id) \
VALUES ('season-1', 'test-server', 'Season 1', 'Season', 'series-1', NULL)",
// Episode links to both season and series; parent_id NULL.
"INSERT INTO items (id, server_id, name, item_type, season_id, series_id, parent_id) \
VALUES ('ep-1', 'test-server', 'Pilot', 'Episode', 'season-1', 'series-1', NULL)",
"INSERT INTO downloads (item_id, status) VALUES ('ep-1', 'completed')",
] {
db_service.execute(Query::new(sql)).await.unwrap();
}
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
assert!(
repo.get_item("ep-1").await.is_ok(),
"downloaded episode available offline"
);
assert!(
repo.get_item("season-1").await.is_ok(),
"season with a season_id-linked downloaded episode should be available offline"
);
assert!(
repo.get_item("series-1").await.is_ok(),
"series with a series_id-linked downloaded episode should be available offline"
);
// Browsing the season returns the episode.
let season_items = repo.get_items("season-1", None).await.unwrap();
assert!(
season_items.items.iter().any(|i| i.id == "ep-1"),
"get_items(season_id) should return the episode"
);
// Browsing the series returns the episode (via series_id link).
let series_items = repo.get_items("series-1", None).await.unwrap();
assert!(
series_items.items.iter().any(|i| i.id == "ep-1"),
"get_items(series_id) should surface the downloaded episode"
);
}
/// Regression: offline startup must list cached libraries. Previously
/// `get_libraries` joined on `items.library_id` (always NULL in the cache),
/// so it returned nothing offline and the app showed no libraries at all.
/// The list must round-trip through `save_libraries_to_cache`.
#[tokio::test]
async fn test_libraries_cache_roundtrip_available_offline() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
// Empty cache → no libraries (this is the state that fell through to the
// server and hung offline).
assert!(repo.get_libraries().await.unwrap().is_empty());
// Simulate the online path persisting the server's library list.
let server_libs = vec![
Library {
id: "music".into(),
name: "Music".into(),
collection_type: "music".into(),
image_tag: None,
},
Library {
id: "movies".into(),
name: "Movies".into(),
collection_type: "movies".into(),
image_tag: Some("tag".into()),
},
];
let saved = repo.save_libraries_to_cache(&server_libs).await.unwrap();
assert_eq!(saved, 2);
// Now offline get_libraries returns them without touching the server.
let offline_libs = repo.get_libraries().await.unwrap();
let names: Vec<&str> = offline_libs.iter().map(|l| l.name.as_str()).collect();
assert_eq!(
names,
vec!["Music", "Movies"],
"cached libraries available offline in sort order"
);
// Re-saving is idempotent (INSERT OR REPLACE), not duplicating rows.
repo.save_libraries_to_cache(&server_libs).await.unwrap();
assert_eq!(repo.get_libraries().await.unwrap().len(), 2);
}
/// Regression: the music/TV/movie landing pages lost genre variety because
/// offline `get_genres` derived genres from cached albums only — so the
/// hybrid cache-first race pinned the list to whatever sparse set the local
/// albums yielded instead of the server's full catalog. The full genre list
/// must round-trip through `save_genres_to_cache` and come back scoped by
/// library.
#[tokio::test]
async fn test_genres_cache_roundtrip_scoped_by_library() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
// Empty cache → no genres.
assert!(repo.get_genres(Some("music-lib")).await.unwrap().is_empty());
// Simulate the online path persisting the server's full genre catalog.
let server_genres = vec![
Genre {
id: "g1".into(),
name: "Rock".into(),
album_count: Some(42),
},
Genre {
id: "g2".into(),
name: "Jazz".into(),
album_count: Some(17),
},
Genre {
id: "g3".into(),
name: "Ambient".into(),
album_count: None,
},
];
let saved = repo
.save_genres_to_cache(Some("music-lib"), &server_genres)
.await
.unwrap();
assert_eq!(saved, 3);
// Offline get_genres returns the full set for that library, counts intact.
let mut offline_genres = repo.get_genres(Some("music-lib")).await.unwrap();
offline_genres.sort_by(|a, b| a.name.cmp(&b.name));
let names: Vec<&str> = offline_genres.iter().map(|g| g.name.as_str()).collect();
assert_eq!(names, vec!["Ambient", "Jazz", "Rock"]);
let rock = offline_genres.iter().find(|g| g.name == "Rock").unwrap();
assert_eq!(rock.album_count, Some(42));
// Genres are scoped: a different library sees nothing.
assert!(repo.get_genres(Some("other-lib")).await.unwrap().is_empty());
// Re-saving replaces the scope's rows (server removed "Jazz").
let updated = vec![Genre {
id: "g1".into(),
name: "Rock".into(),
album_count: Some(50),
}];
repo.save_genres_to_cache(Some("music-lib"), &updated)
.await
.unwrap();
let after = repo.get_genres(Some("music-lib")).await.unwrap();
assert_eq!(after.len(), 1, "stale genres removed on refresh");
assert_eq!(after[0].album_count, Some(50), "counts updated on refresh");
}
// ===== Playlist Tests =====
/// Helper to seed items into the DB for playlist tests
async fn seed_items(repo: &OfflineRepository, ids: &[&str]) {
let items: Vec<MediaItem> = ids
.iter()
.map(|id| create_test_item(id, &format!("Track {}", id), Some("library-1")))
.collect();
repo.save_to_cache("library-1", &items).await.unwrap();
}
/// Insert a fully-formed item row of a given type (bypasses save_to_cache's
/// stub-parent machinery so containers/leaves can be linked precisely).
async fn insert_item(
db: &Arc<RusqliteService>,
id: &str,
item_type: &str,
album_id: Option<&str>,
series_id: Option<&str>,
season_id: Option<&str>,
) {
db.execute(Query::with_params(
"INSERT INTO items (id, server_id, name, item_type, album_id, series_id, season_id, synced_at)
VALUES (?1, 'test-server', ?2, ?3, ?4, ?5, ?6, '2024-01-01')",
vec![
QueryParam::String(id.to_string()),
QueryParam::String(format!("Name {id}")),
QueryParam::String(item_type.to_string()),
album_id.map(|s| QueryParam::String(s.to_string())).unwrap_or(QueryParam::Null),
series_id.map(|s| QueryParam::String(s.to_string())).unwrap_or(QueryParam::Null),
season_id.map(|s| QueryParam::String(s.to_string())).unwrap_or(QueryParam::Null),
],
))
.await
.unwrap();
}
async fn seed_completed_download(db: &Arc<RusqliteService>, item_id: &str, file_size: i64) {
db.execute(Query::with_params(
"INSERT INTO downloads (item_id, status, file_size) VALUES (?1, 'completed', ?2)",
vec![
QueryParam::String(item_id.to_string()),
QueryParam::Int64(file_size),
],
))
.await
.unwrap();
}
async fn seed_library(db: &Arc<RusqliteService>, id: &str, collection_type: &str) {
db.execute(Query::with_params(
"INSERT INTO libraries (id, server_id, name, collection_type, sort_order)
VALUES (?1, 'test-server', ?2, ?3, 0)",
vec![
QueryParam::String(id.to_string()),
QueryParam::String(format!("Lib {id}")),
QueryParam::String(collection_type.to_string()),
],
))
.await
.unwrap();
}
fn make_repo(db: &Arc<RusqliteService>) -> OfflineRepository {
OfflineRepository::new(
db.clone(),
"test-server".to_string(),
"test-user".to_string(),
)
}
/// UT: downloaded-only browse returns a downloaded leaf AND its container,
/// filtered to the requested album parent. A non-downloaded sibling is omitted.
///
/// TRACES: UR-055 | DR-082, DR-083 | UT-072
#[tokio::test]
async fn test_get_downloaded_items_returns_leaf_and_container() {
let db = create_test_db();
insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
// track-1 downloaded; track-2 is NOT downloaded.
seed_completed_download(&db, "track-1", 1000).await;
let repo = make_repo(&db);
// Browsing the album shows only the downloaded track.
let in_album = repo.get_downloaded_items("album-1", None).await.unwrap();
let ids: Vec<&str> = in_album.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(ids, vec!["track-1"], "only the downloaded track is listed");
}
/// Regression: browsing a downloaded *library* (top level) lists containers,
/// not their leaves — a music library shows the album, not the individual
/// downloaded songs. The leaf is still reachable by drilling into the album.
///
/// TRACES: UR-055 | DR-082, DR-083 | UT-076
#[tokio::test]
async fn test_get_downloaded_items_library_lists_albums_not_tracks() {
let db = create_test_db();
seed_library(&db, "music-lib", "music").await;
insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
// Tracks link to the album via album_id (parent_id NULL in the cache).
insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
seed_completed_download(&db, "track-1", 1000).await;
seed_completed_download(&db, "track-2", 1000).await;
let repo = make_repo(&db);
// Library level: only the album shows, not the two tracks.
let at_library = repo.get_downloaded_items("music-lib", None).await.unwrap();
let ids: Vec<&str> = at_library.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(
ids,
vec!["album-1"],
"library browse lists the album container, not its tracks"
);
// Drilling into the album still returns the downloaded tracks.
let in_album = repo.get_downloaded_items("album-1", None).await.unwrap();
let mut track_ids: Vec<&str> = in_album.items.iter().map(|i| i.id.as_str()).collect();
track_ids.sort();
assert_eq!(track_ids, vec!["track-1", "track-2"]);
}
/// Regression: a downloaded TV library lists the Series, not its Seasons or
/// Episodes — the same "individual songs" bug seen for music, for TV. The
/// season and episode are still reachable by drilling into the series.
///
/// TRACES: UR-055 | DR-082, DR-083 | UT-077
#[tokio::test]
async fn test_get_downloaded_items_library_lists_series_not_episodes() {
let db = create_test_db();
seed_library(&db, "tv-lib", "tvshows").await;
insert_item(&db, "series-1", "Series", None, None, None).await;
// Season links to its series; episode links to both season and series.
insert_item(&db, "season-1", "Season", None, Some("series-1"), None).await;
insert_item(
&db,
"ep-1",
"Episode",
None,
Some("series-1"),
Some("season-1"),
)
.await;
seed_completed_download(&db, "ep-1", 4000).await;
let repo = make_repo(&db);
// Library level: only the series shows.
let at_library = repo.get_downloaded_items("tv-lib", None).await.unwrap();
let ids: Vec<&str> = at_library.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(
ids,
vec!["series-1"],
"TV library browse lists the series, not seasons/episodes"
);
// Drilling into the series returns its season; into the season, the episode.
let in_series = repo.get_downloaded_items("series-1", None).await.unwrap();
assert!(
in_series.items.iter().any(|i| i.id == "season-1"),
"series drill returns the season"
);
let in_season = repo.get_downloaded_items("season-1", None).await.unwrap();
assert!(
in_season.items.iter().any(|i| i.id == "ep-1"),
"season drill returns the episode"
);
}
/// A downloaded leaf with no cached container (e.g. a Movie, or a track whose
/// album isn't in the cache) still surfaces at the library level.
///
/// TRACES: UR-055 | DR-082, DR-083 | UT-078
#[tokio::test]
async fn test_get_downloaded_items_library_keeps_orphan_leaves() {
let db = create_test_db();
seed_library(&db, "movie-lib", "movies").await;
insert_item(&db, "movie-1", "Movie", None, None, None).await;
seed_completed_download(&db, "movie-1", 5000).await;
let repo = make_repo(&db);
let at_library = repo.get_downloaded_items("movie-lib", None).await.unwrap();
let ids: Vec<&str> = at_library.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(
ids,
vec!["movie-1"],
"a downloaded movie with no container shows"
);
}
/// UT: an empty downloaded-only browse is authoritative — no rows, no error,
/// regardless of the catalog-browse flag (which the DR-080 fallthrough uses).
///
/// TRACES: UR-055 | DR-082 | UT-073
#[tokio::test]
async fn test_get_downloaded_items_empty_is_authoritative() {
let db = create_test_db();
insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
// Nothing downloaded, and the catalog-browse flag is ON (online default).
set_include_catalog_browse(true);
let repo = make_repo(&db);
let result = repo.get_downloaded_items("album-1", None).await.unwrap();
assert!(
result.items.is_empty(),
"empty downloaded browse returns no items even with catalog-browse on"
);
}
/// UT: only libraries with downloaded content are listed; an empty one is omitted.
///
/// TRACES: UR-055 | DR-082 | UT-074
#[tokio::test]
async fn test_get_downloaded_libraries_omits_empty() {
let db = create_test_db();
seed_library(&db, "music-lib", "music").await;
seed_library(&db, "movie-lib", "movies").await;
insert_item(&db, "track-1", "Audio", None, None, None).await;
seed_completed_download(&db, "track-1", 500).await;
let repo = make_repo(&db);
let libs = repo.get_downloaded_libraries().await.unwrap();
let ids: Vec<&str> = libs.iter().map(|l| l.id.as_str()).collect();
assert_eq!(
ids,
vec!["music-lib"],
"movie library with no downloads omitted"
);
}
/// UT: disk usage reports a leaf's own size, a container's summed descendants,
/// and reconciles the device total with the sum of leaves.
///
/// TRACES: UR-056 | DR-085 | UT-075
#[tokio::test]
async fn test_download_disk_usage_aggregates_containers() {
let db = create_test_db();
insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
seed_completed_download(&db, "track-1", 1000).await;
seed_completed_download(&db, "track-2", 2000).await;
let repo = make_repo(&db);
let usage = repo.get_download_disk_usage().await.unwrap();
assert_eq!(usage.item_count, 2, "two leaf downloads");
assert_eq!(
usage.device_total_bytes, 3000,
"device total is the leaf sum"
);
assert_eq!(usage.sizes.get("track-1"), Some(&1000));
assert_eq!(
usage.sizes.get("album-1"),
Some(&3000),
"container = sum of children"
);
// Both children downloaded ⇒ album is NOT partial.
assert_eq!(
usage.partial_containers.get("album-1"),
None,
"fully downloaded album is not partial"
);
// Device total reconciles with the sum of the listed leaves.
let leaf_sum: i64 = ["track-1", "track-2"]
.iter()
.map(|id| usage.sizes[*id])
.sum();
assert_eq!(leaf_sum, usage.device_total_bytes);
}
/// UT: a container with a downloaded child AND a non-downloaded cached child
/// is flagged partial. TRACES: UR-055 | DR-083 | UT-051
#[tokio::test]
async fn test_download_disk_usage_flags_partial_container() {
let db = create_test_db();
insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
insert_item(&db, "track-1", "Audio", Some("album-1"), None, None).await;
insert_item(&db, "track-2", "Audio", Some("album-1"), None, None).await;
// Only track-1 downloaded; track-2 is cached but not downloaded.
seed_completed_download(&db, "track-1", 1000).await;
let repo = make_repo(&db);
let usage = repo.get_download_disk_usage().await.unwrap();
assert_eq!(
usage.partial_containers.get("album-1"),
Some(&true),
"album with a missing child is partial"
);
}
#[tokio::test]
async fn test_playlist_create_empty() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let result = repo.create_playlist("My Playlist", &[]).await;
assert!(result.is_ok());
let created = result.unwrap();
assert!(
!created.id.is_empty(),
"Should return a non-empty playlist ID"
);
// Verify playlist exists in DB
let name: String = db_service
.query_one(
Query::with_params(
"SELECT name FROM playlists WHERE id = ?",
vec![QueryParam::String(created.id.clone())],
),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(name, "My Playlist");
}
#[tokio::test]
async fn test_playlist_create_with_items() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["t1", "t2", "t3"]).await;
let created = repo
.create_playlist("With Tracks", &["t1".into(), "t2".into(), "t3".into()])
.await
.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
assert_eq!(items.len(), 3);
assert_eq!(items[0].item.id, "t1");
assert_eq!(items[1].item.id, "t2");
assert_eq!(items[2].item.id, "t3");
}
#[tokio::test]
async fn test_playlist_delete() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["t1"]).await;
let created = repo
.create_playlist("To Delete", &["t1".into()])
.await
.unwrap();
// Delete it
repo.delete_playlist(&created.id).await.unwrap();
// Verify playlist is gone
let count: i32 = db_service
.query_one(
Query::with_params(
"SELECT COUNT(*) FROM playlists WHERE id = ?",
vec![QueryParam::String(created.id.clone())],
),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(count, 0);
// Verify cascade deleted playlist_items
let item_count: i32 = db_service
.query_one(
Query::with_params(
"SELECT COUNT(*) FROM playlist_items WHERE playlist_id = ?",
vec![QueryParam::String(created.id)],
),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(item_count, 0);
}
#[tokio::test]
async fn test_playlist_rename() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let created = repo.create_playlist("Original Name", &[]).await.unwrap();
repo.rename_playlist(&created.id, "New Name").await.unwrap();
let name: String = db_service
.query_one(
Query::with_params(
"SELECT name FROM playlists WHERE id = ?",
vec![QueryParam::String(created.id)],
),
|row| row.get(0),
)
.await
.unwrap();
assert_eq!(name, "New Name");
}
#[tokio::test]
async fn test_playlist_get_items_preserves_order() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["a", "b", "c"]).await;
let created = repo
.create_playlist("Ordered", &["c".into(), "a".into(), "b".into()])
.await
.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
assert_eq!(items.len(), 3);
// Order should match insertion order: c, a, b
assert_eq!(items[0].item.id, "c");
assert_eq!(items[1].item.id, "a");
assert_eq!(items[2].item.id, "b");
// Each entry should have a unique playlist_item_id
assert_ne!(items[0].playlist_item_id, items[1].playlist_item_id);
assert_ne!(items[1].playlist_item_id, items[2].playlist_item_id);
}
#[tokio::test]
async fn test_playlist_get_items_empty_playlist() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let created = repo.create_playlist("Empty", &[]).await.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
assert!(items.is_empty());
}
#[tokio::test]
async fn test_playlist_add_items() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["t1", "t2", "t3"]).await;
let created = repo
.create_playlist("Addable", &["t1".into()])
.await
.unwrap();
// Add two more tracks
repo.add_to_playlist(&created.id, &["t2".into(), "t3".into()])
.await
.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
assert_eq!(items.len(), 3);
assert_eq!(items[0].item.id, "t1");
assert_eq!(items[1].item.id, "t2");
assert_eq!(items[2].item.id, "t3");
}
#[tokio::test]
async fn test_playlist_add_duplicate_items_ignored() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["t1"]).await;
let created = repo.create_playlist("Dupes", &["t1".into()]).await.unwrap();
// Try to add the same item again
repo.add_to_playlist(&created.id, &["t1".into()])
.await
.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
assert_eq!(
items.len(),
1,
"Duplicate should be ignored (UNIQUE constraint)"
);
}
#[tokio::test]
async fn test_playlist_remove_items() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["t1", "t2", "t3"]).await;
let created = repo
.create_playlist("Removable", &["t1".into(), "t2".into(), "t3".into()])
.await
.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
assert_eq!(items.len(), 3);
// Remove the middle track by its entry ID
let entry_id_to_remove = items[1].playlist_item_id.clone();
repo.remove_from_playlist(&created.id, &[entry_id_to_remove])
.await
.unwrap();
let items_after = repo.get_playlist_items(&created.id).await.unwrap();
assert_eq!(items_after.len(), 2);
assert_eq!(items_after[0].item.id, "t1");
assert_eq!(items_after[1].item.id, "t3");
}
#[tokio::test]
async fn test_playlist_move_item_forward() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["a", "b", "c", "d"]).await;
let created = repo
.create_playlist("Reorder", &["a".into(), "b".into(), "c".into(), "d".into()])
.await
.unwrap();
// Move 'a' (index 0) to index 2: expect b, c, a, d
repo.move_playlist_item(&created.id, "a", 2).await.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
assert_eq!(ids, vec!["b", "c", "a", "d"]);
}
#[tokio::test]
async fn test_playlist_move_item_backward() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["a", "b", "c", "d"]).await;
let created = repo
.create_playlist(
"Reorder2",
&["a".into(), "b".into(), "c".into(), "d".into()],
)
.await
.unwrap();
// Move 'd' (index 3) to index 0: expect d, a, b, c
repo.move_playlist_item(&created.id, "d", 0).await.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
assert_eq!(ids, vec!["d", "a", "b", "c"]);
}
#[tokio::test]
async fn test_playlist_move_item_to_end() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["a", "b", "c"]).await;
let created = repo
.create_playlist("MoveEnd", &["a".into(), "b".into(), "c".into()])
.await
.unwrap();
// Move 'a' to index 99 (beyond end, should clamp): expect b, c, a
repo.move_playlist_item(&created.id, "a", 99).await.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
assert_eq!(ids, vec!["b", "c", "a"]);
}
#[tokio::test]
async fn test_playlist_move_nonexistent_item_is_noop() {
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
seed_items(&repo, &["a", "b"]).await;
let created = repo
.create_playlist("NoOp", &["a".into(), "b".into()])
.await
.unwrap();
// Move a nonexistent item - should not error, just no-op
repo.move_playlist_item(&created.id, "nonexistent", 0)
.await
.unwrap();
let items = repo.get_playlist_items(&created.id).await.unwrap();
let ids: Vec<&str> = items.iter().map(|e| e.item.id.as_str()).collect();
assert_eq!(ids, vec!["a", "b"]);
}
/// Seed two movies and one album, favouriting a subset, for the favourites
/// query tests below.
async fn seed_favorites(db_service: &Arc<RusqliteService>) {
use crate::storage::db_service::DatabaseService;
for sql in [
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
VALUES ('movie-fav', 'test-server', 'Favourite Movie', 'Movie', 'lib-1', '2026-01-01', 'Favourite Movie')",
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
VALUES ('movie-plain', 'test-server', 'Ordinary Movie', 'Movie', 'lib-1', '2026-01-01', 'Ordinary Movie')",
"INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
VALUES ('album-fav', 'test-server', 'Favourite Album', 'MusicAlbum', 'lib-2', '2026-01-01', 'Favourite Album')",
"INSERT INTO libraries (id, server_id, name) VALUES ('lib-1', 'test-server', 'Movies')",
"INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'movie-fav', 1)",
"INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'album-fav', 1)",
// Explicitly not a favourite — must never show up.
"INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'movie-plain', 0)",
// Another user's favourite must not leak into this user's list.
"INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('other-user', 'movie-plain', 1)",
] {
db_service.execute(Query::new(sql)).await.unwrap();
}
}
/// UT-101 — the offline favourites query returns only favourites, scoped,
/// and only for the current user.
///
/// TRACES: UR-067 | DR-115 | UT-101
#[tokio::test]
async fn test_get_favorites_returns_only_scoped_favorites() {
let _guard = lock_catalog_browse();
set_include_catalog_browse(true);
let db_service = create_test_db();
seed_favorites(&db_service).await;
let repo = OfflineRepository::new(
db_service,
"test-server".to_string(),
"test-user".to_string(),
);
let all = repo.get_favorites(SearchScope::All, None).await.unwrap();
let mut ids: Vec<&str> = all.items.iter().map(|i| i.id.as_str()).collect();
ids.sort();
assert_eq!(
ids,
vec!["album-fav", "movie-fav"],
"All scope should return every favourite and nothing else"
);
let movies = repo.get_favorites(SearchScope::Movies, None).await.unwrap();
let ids: Vec<&str> = movies.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(ids, vec!["movie-fav"]);
let music = repo.get_favorites(SearchScope::Music, None).await.unwrap();
let ids: Vec<&str> = music.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(ids, vec!["album-fav"]);
}
/// With "Show all server media" off, favourites narrow to what is actually
/// on the device — the same gate browsing obeys (DR-080).
///
/// TRACES: UR-067 | DR-115 | UT-101
#[tokio::test]
async fn test_get_favorites_respects_catalog_browse_gate() {
use crate::storage::db_service::DatabaseService;
let _guard = lock_catalog_browse();
let db_service = create_test_db();
seed_favorites(&db_service).await;
// Only the album is downloaded... via a child track, the container case.
db_service
.execute(Query::new(
"INSERT INTO items (id, server_id, name, item_type, album_id, synced_at) \
VALUES ('track-1', 'test-server', 'Track', 'Audio', 'album-fav', '2026-01-01')",
))
.await
.unwrap();
db_service
.execute(Query::new(
"INSERT INTO downloads (item_id, status) VALUES ('track-1', 'completed')",
))
.await
.unwrap();
let repo = OfflineRepository::new(
db_service,
"test-server".to_string(),
"test-user".to_string(),
);
set_include_catalog_browse(false);
let offline_only = repo.get_favorites(SearchScope::All, None).await.unwrap();
let ids: Vec<&str> = offline_only.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(
ids,
vec!["album-fav"],
"with the gate off, only favourites on the device are listed"
);
set_include_catalog_browse(true);
let with_catalog = repo.get_favorites(SearchScope::All, None).await.unwrap();
assert_eq!(with_catalog.items.len(), 2);
}
/// UT-104 — the per-library favourites toggle narrows a normal listing.
///
/// TRACES: UR-067 | DR-116 | UT-104
#[tokio::test]
async fn test_get_items_favorites_only_filters_listing() {
let _guard = lock_catalog_browse();
set_include_catalog_browse(true);
let db_service = create_test_db();
seed_favorites(&db_service).await;
let repo = OfflineRepository::new(
db_service,
"test-server".to_string(),
"test-user".to_string(),
);
let unfiltered = repo
.get_items(
"lib-1",
Some(GetItemsOptions {
include_item_types: Some(vec!["Movie".to_string()]),
..Default::default()
}),
)
.await
.unwrap();
assert_eq!(unfiltered.items.len(), 2, "both movies without the filter");
let favourites = repo
.get_items(
"lib-1",
Some(GetItemsOptions {
include_item_types: Some(vec!["Movie".to_string()]),
favorites_only: Some(true),
..Default::default()
}),
)
.await
.unwrap();
let ids: Vec<&str> = favourites.items.iter().map(|i| i.id.as_str()).collect();
assert_eq!(ids, vec!["movie-fav"]);
}
/// UT-102 — caching a server result mirrors its favourite state locally,
/// but never over a row still waiting to be pushed.
///
/// The second half is the one that matters: favourite something with the
/// server unreachable, and the next successful browse would otherwise
/// overwrite it with the server's stale `false` before the drain ever ran.
///
/// TRACES: UR-069 | DR-114 | UT-102
#[tokio::test]
async fn test_save_to_cache_mirrors_favorites_without_clobbering_pending() {
use crate::storage::db_service::DatabaseService;
let db_service = create_test_db();
let repo = OfflineRepository::new(
db_service.clone(),
"test-server".to_string(),
"test-user".to_string(),
);
let favourite_flag = |id: &'static str| {
let db = db_service.clone();
async move {
db.query_optional(
Query::with_params(
"SELECT is_favorite, pending_sync FROM user_data \
WHERE user_id = ? AND item_id = ?",
vec![
QueryParam::String("test-user".to_string()),
QueryParam::String(id.to_string()),
],
),
|row| Ok((row.get::<_, Option<i32>>(0)?, row.get::<_, Option<i32>>(1)?)),
)
.await
.unwrap()
}
};
// A server item the user favourited elsewhere.
let mut favourited = create_test_item("fav-1", "Favourited Elsewhere", None);
favourited.user_data = Some(UserData {
is_favorite: Some(true),
..Default::default()
});
// A server item with no user data at all — must not fabricate a row.
let untouched = create_test_item("plain-1", "No User Data", None);
repo.save_to_cache("parent-1", &[favourited.clone(), untouched])
.await
.unwrap();
assert_eq!(
favourite_flag("fav-1").await,
Some((Some(1), Some(0))),
"server favourite should be mirrored as synced"
);
assert_eq!(
favourite_flag("plain-1").await,
None,
"an item without UserData should not get an invented user_data row"
);
// The user un-favourites it while offline: local write, pending_sync = 1.
db_service
.execute(Query::with_params(
"UPDATE user_data SET is_favorite = 0, pending_sync = 1 \
WHERE user_id = ? AND item_id = ?",
vec![
QueryParam::String("test-user".to_string()),
QueryParam::String("fav-1".to_string()),
],
))
.await
.unwrap();
// The server still reports it as a favourite; caching must not win.
repo.save_to_cache("parent-1", &[favourited]).await.unwrap();
assert_eq!(
favourite_flag("fav-1").await,
Some((Some(0), Some(1))),
"an unsynced local toggle must survive a cache write"
);
}
}