jellytau_lib/repository/

offline.rs

// 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(" "),
    )
}

/// The Jellyfin taxonomy half of "does cached item `i` belong to library `l`":
/// the library's `collection_type` against the item's `item_type`.
///
/// A macro rather than a `const` because both callers need it *inside* a larger
/// SQL string literal, and `concat!` cannot take a const. One definition, so the
/// two sites cannot drift — they did once already, and opening any downloaded
/// library then listed every downloaded item on the server (DR-167).
///
/// Deliberately has **no fall-open arm**. Adding "…or the type is unknown"
/// makes the clause true for every row, which is precisely the defect it exists
/// to prevent; callers that want that behaviour must say so themselves and
/// justify it, as `LIBRARY_HOLDS_ITEM` does.
///
/// TRACES: UR-007, UR-055 | DR-167, DR-277
macro_rules! library_type_matches_item {
    () => {
        "(
               (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'))
           )"
    };
}

pub struct OfflineRepository {
    db_service: Arc<RusqliteService>,
    server_id: String,
    user_id: String,
}

impl OfflineRepository {
    /// Playback info for an item this user has downloaded, built from the
    /// download row — no server involved. `None` when there is no completed
    /// local file, which leaves the question to the server.
    ///
    /// This exists because playing a download asked the server first. The
    /// player needs only the media-source id (subtitle URLs are keyed by it),
    /// and fetching that from `/PlaybackInfo` meant a downloaded film would not
    /// play offline: the call retried for seven seconds and failed, and the
    /// file on disk was never opened.
    ///
    /// The media-source id is the item id. A download never names a source —
    /// the URL carries no `mediaSourceId` — so the server serves its default,
    /// and Jellyfin gives an item's default source the item's own id.
    ///
    /// TRACES: UR-002, UR-071 | DR-294 | UT-260
    pub async fn local_playback_info(
        &self,
        item_id: &str,
    ) -> Result<Option<PlaybackInfo>, RepoError> {
        let rows = self
            .db_service
            .query_many(
                Query::with_params(
                    "SELECT file_path FROM downloads \
                     WHERE item_id = ? AND user_id = ? AND status = 'completed' \
                       AND file_path IS NOT NULL \
                     LIMIT 1",
                    vec![
                        QueryParam::String(item_id.to_string()),
                        QueryParam::String(self.user_id.clone()),
                    ],
                ),
                |row| row.get::<_, String>(0),
            )
            .await
            .map_err(|e| RepoError::Database { message: e })?;

        Ok(rows.into_iter().next().map(|file_path| PlaybackInfo {
            media_source_id: item_id.to_string(),
            // No server session: nothing was negotiated, and a local file has no
            // transcode job for a session id to name.
            play_session_id: String::new(),
            stream_url: file_path,
            direct_play: true,
            needs_transcoding: false,
        }))
    }

    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| Ok(row_to_user_data(row, 0)))
            .await
            .ok()
            .flatten()
    }

    /// Attach each cached row's user data, fetched in batches.
    ///
    /// Listings used to call `get_user_data` once per row — a series page's
    /// ~275 cached rows meant ~275 sequential database round trips, which alone
    /// pushed the read past the cache fast path so the page waited for the
    /// server every time. One `IN (…)` query per chunk instead; the chunk stays
    /// well under SQLite's bound-parameter limit.
    ///
    /// TRACES: UR-002 | DR-013
    async fn with_user_data(&self, cached_items: Vec<CachedItem>) -> Vec<MediaItem> {
        const CHUNK: usize = 500;
        let mut by_id: std::collections::HashMap<String, UserData> =
            std::collections::HashMap::new();

        for chunk in cached_items.chunks(CHUNK) {
            let placeholders = vec!["?"; chunk.len()].join(",");
            let mut params = vec![QueryParam::String(self.user_id.clone())];
            params.extend(chunk.iter().map(|c| QueryParam::String(c.id.clone())));
            let query = Query::with_params(
                format!(
                    "SELECT item_id, 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 IN ({placeholders})"
                ),
                params,
            );
            match self
                .db_service
                .query_many(query, |row| {
                    Ok((row.get::<_, String>(0)?, row_to_user_data(row, 1)))
                })
                .await
            {
                Ok(rows) => by_id.extend(rows),
                // Same as the per-row lookup: missing user data is not fatal.
                Err(e) => debug!("[OfflineRepo] user_data batch lookup failed: {}", e),
            }
        }

        cached_items
            .into_iter()
            .map(|cached| {
                let user_data = by_id.remove(&cached.id);
                Self::cached_item_to_media_item(cached, user_data)
            })
            .collect()
    }
}

/// A container row to create if it was never cached. See `save_to_cache`.
struct ContainerPlaceholder {
    id: String,
    name: String,
    item_type: &'static str,
    series_id: Option<String>,
    series_name: Option<String>,
    album_artist: Option<String>,
}

/// The logical containers `item` lists under, as far as its own fields name
/// them — the same rule as `container_id` in migration 027: an episode's
/// season and series, a season's series, a track's album.
///
/// TRACES: UR-002, UR-007 | DR-013
fn container_placeholders(item: &MediaItem) -> Vec<ContainerPlaceholder> {
    let mut out = Vec::new();
    let series = |item: &MediaItem| {
        item.series_id.clone().map(|id| ContainerPlaceholder {
            id,
            name: item
                .series_name
                .clone()
                .unwrap_or_else(|| "Series".to_string()),
            item_type: "Series",
            series_id: None,
            series_name: None,
            album_artist: None,
        })
    };
    match item.item_type.as_str() {
        "Episode" => {
            if let Some(id) = item.season_id.clone() {
                out.push(ContainerPlaceholder {
                    id,
                    name: item
                        .season_name
                        .clone()
                        .unwrap_or_else(|| "Season".to_string()),
                    item_type: "Season",
                    series_id: item.series_id.clone(),
                    series_name: item.series_name.clone(),
                    album_artist: None,
                });
            }
            out.extend(series(item));
        }
        "Season" => out.extend(series(item)),
        "Audio" => {
            if let Some(id) = item.album_id.clone() {
                out.push(ContainerPlaceholder {
                    id,
                    name: item
                        .album_name
                        .clone()
                        .unwrap_or_else(|| "Album".to_string()),
                    item_type: "MusicAlbum",
                    series_id: None,
                    series_name: None,
                    album_artist: item.album_artist.clone(),
                });
            }
        }
        _ => {}
    }
    out
}

/// A `user_data` row's columns, starting at `offset`, in the order
/// `playback_position_ticks, is_played, is_favorite, play_count,
/// last_played_at, playback_context_type, playback_context_id`.
fn row_to_user_data(row: &rusqlite::Row, offset: usize) -> UserData {
    let playback_position_ticks: Option<i64> = row.get(offset).ok();
    UserData {
        playback_position_ticks,
        playback_position_ms: playback_position_ticks.map(crate::domain::ticks_to_ms),
        is_played: row
            .get::<_, Option<i32>>(offset + 1)
            .ok()
            .flatten()
            .map(|v| v != 0),
        is_favorite: row
            .get::<_, Option<i32>>(offset + 2)
            .ok()
            .flatten()
            .map(|v| v != 0),
        play_count: row.get(offset + 3).ok(),
        last_played_date: row.get(offset + 4).ok(),
        playback_context_type: row.get(offset + 5).ok(),
        playback_context_id: row.get(offset + 6).ok(),
    }
}

/// Item types that can be downloaded themselves.
const PLAYABLE_TYPES: &str = "'Audio', 'Movie', 'Episode'";
/// Item types that are available when something below them is downloaded.
const CONTAINER_TYPES: &str =
    "'MusicAlbum', 'Series', 'Season', 'BoxSet', 'Folder', 'CollectionFolder'";

/// SQL predicate: the row `alias` is on the device — a `playable` item with a
/// completed download, or a `container` with a downloaded descendant.
///
/// Evaluated per row, with `EXISTS`, on whatever rows the query has already
/// narrowed to. Every query used to build the set of *all* downloaded items
/// in a CTE first and join against it, paying for the whole table on every
/// call however few rows it wanted (see 08-database-design.md → "Listing
/// query shape"). Descendants are found through all four link columns on
/// purpose: a series is available through an episode two levels down.
///
/// TRACES: UR-002, UR-055 | DR-013, DR-082
fn downloaded_sql(alias: &str, playable: &str, containers: &str) -> String {
    format!(
        "(({a}.item_type IN ({playable})
             AND EXISTS (SELECT 1 FROM downloads dl
                         WHERE dl.item_id = {a}.id AND dl.status = 'completed'))
          OR ({a}.item_type IN ({containers})
             AND EXISTS (SELECT 1 FROM items dc
                         INNER JOIN downloads dl ON dl.item_id = dc.id AND dl.status = 'completed'
                         WHERE dc.parent_id = {a}.id OR dc.album_id = {a}.id
                            OR dc.season_id = {a}.id OR dc.series_id = {a}.id)))",
        a = alias
    )
}

/// SQL predicate: the row `alias` can be shown — downloaded (see
/// [`downloaded_sql`]) or, with `include_catalog`, cached for browsing.
/// `include_catalog` is the catalog-browse flag; offline with "Show all server
/// media" off it is false and only downloaded media shows
/// (`set_include_catalog_browse`).
fn available_sql(alias: &str, include_catalog: bool) -> String {
    let downloaded = downloaded_sql(alias, PLAYABLE_TYPES, CONTAINER_TYPES);
    if include_catalog {
        format!("({alias}.synced_at IS NOT NULL OR {downloaded})")
    } else {
        downloaded
    }
}

/// The listing query behind `get_items`: the cached children of one parent
/// that are available to show.
///
/// An item is available when it is cached for browsing (`synced_at`, only
/// while the catalog-browse flag is on — offline with the toggle off the page
/// shows downloaded media only; see `set_include_catalog_browse`), when it is
/// a playable item with a completed download, or when it is a container with a
/// downloaded child. That is checked per row with `EXISTS` on the rows the
/// parent selects. It used to be a CTE that materialised the id of *every*
/// available item in the database (a `UNION` over the whole table) before
/// filtering to the parent, which cost ~80 ms on a desktop for a 100k-item
/// cache whatever the parent — several times that on a phone.
///
/// Children are matched on `container_id`, the logical container resolved by
/// migration 027 (an episode's season, a season's series, a track's album,
/// otherwise the parent) — one index range, already in display order. It
/// replaced a four-column `OR` that was both slow and wrong: every episode
/// carries its series id, so a series listed its episodes beside its seasons.
/// The library clause is added only when the parent *is* a library; inside the
/// same `OR` it forces a scan of every row. The `+` on `server_id` keeps the
/// planner off the server index, which every row shares.
///
/// Bind order: server id; the parent id (twice for a library parent); the
/// type-filter values; with the favourites filter, the user id.
///
/// TRACES: UR-002, UR-007 | DR-013, DR-277
#[allow(clippy::too_many_arguments)]
fn items_listing_sql(
    parent_is_library: bool,
    include_catalog: bool,
    type_filter: &str,
    favorites_filter: &str,
    order_by: &str,
    limit: usize,
    start_index: usize,
) -> String {
    let available = available_sql("i", include_catalog);
    let parent_match = if parent_is_library {
        format!(
            "i.server_id = ?
             AND (
               i.container_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.
               --
               -- The type correlation is NOT optional. Without it this
               -- EXISTS never mentions the item, so it is true for every
               -- cached row as soon as the requested parent is any library.
               -- Music/Movies/TV got away with that because their landing
               -- pages pass `include_item_types`, which narrowed the result;
               -- the generic library page passes none, so a Books or Photos
               -- library served the entire cached server (DR-277).
               --
               -- `library_id` wins wherever it survived the cache write:
               -- it is the server's own answer, and it is the only thing
               -- that can scope a library whose type has no mapping (Books,
               -- Photos, Collections) or none at all (a mixed library, where
               -- Jellyfin sends CollectionType null). The taxonomy is the
               -- fallback for rows that predate it being stored.
               --
               -- A library with neither a stored link nor a mapped type now
               -- matches nothing here and falls through to the server, which
               -- does know what is in it. Showing nothing briefly beats
               -- showing somebody else's films with confidence.
               OR EXISTS (
                   SELECT 1 FROM libraries l
                   WHERE l.id = ? AND l.server_id = i.server_id
                     AND (
                          i.library_id = l.id
                       OR (i.library_id IS NULL AND {})
                     )
               )
             )",
            library_type_matches_item!()
        )
    } else {
        "+i.server_id = ? AND i.container_id = ?".to_string()
    };
    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
         WHERE {parent_match}
           AND {available}{type_filter}{favorites_filter}
         ORDER BY {order_by}
         LIMIT {limit} OFFSET {start_index}"
    )
}

// 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();

        // Which library do these items belong to?
        //
        // Resolved once per call, from the parent being browsed. Two cases and
        // nothing else:
        //
        //   * the parent IS a library  -> these are its direct children
        //   * the parent is an item    -> inherit whatever library that item is
        //                                 already known to belong to, so tracks
        //                                 under an album and episodes under a
        //                                 season land in the same library as
        //                                 their container
        //
        // Synthetic parents ("favorites" and friends) match neither and stay
        // NULL, which is correct: they are not a library and their contents
        // span several.
        //
        // Until this existed, `library_id` was bound NULL for every cached row
        // and the only way to associate an item with a library was the
        // `collection_type` ↔ `item_type` taxonomy. That cannot tell two
        // libraries of the *same* type apart — a server with "TV" and "Shows"
        // served both the same contents — and has nothing to say about a
        // library whose type it does not map (DR-278).
        //
        // TRACES: UR-007 | DR-278
        let owning_library = self.resolve_owning_library(parent_id).await;

        // Placeholders for the logical containers these items list under
        // (`container_id`, migration 027) when those were never cached: an
        // episode fetched through Next Up or Latest has a season and series
        // this device may never have browsed, and without their rows it would
        // be unreachable from its series page offline. Named from the fields
        // the child carries; `synced_at` stays NULL so they show only when a
        // download makes them available, and the server's real row replaces
        // them. Inserted before the parent stubs below, so a season that is
        // also a parent gets a Season row rather than a nameless folder.
        let mut placeholder_ids = std::collections::HashSet::new();
        let mut placeholders: Vec<Query> = Vec::new();
        for item in items {
            for p in container_placeholders(item) {
                if !placeholder_ids.insert(p.id.clone()) {
                    continue;
                }
                placeholders.push(Query::with_params(
                    "INSERT OR IGNORE INTO items
                        (id, server_id, library_id, name, item_type, is_folder,
                         series_id, series_name, album_artist)
                     VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?)",
                    vec![
                        QueryParam::String(p.id),
                        QueryParam::String(self.server_id.clone()),
                        owning_library
                            .clone()
                            .map(QueryParam::String)
                            .unwrap_or(QueryParam::Null),
                        QueryParam::String(p.name),
                        QueryParam::String(p.item_type.to_string()),
                        p.series_id
                            .map(QueryParam::String)
                            .unwrap_or(QueryParam::Null),
                        p.series_name
                            .map(QueryParam::String)
                            .unwrap_or(QueryParam::Null),
                        p.album_artist
                            .map(QueryParam::String)
                            .unwrap_or(QueryParam::Null),
                    ],
                ));
            }
        }

        // Stub rows for every parent referenced, so `items.parent_id` resolves
        // whatever order the server returned children and parents in.
        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());
            }
        }
        let stubs: Vec<Query> = parent_ids
            .into_iter()
            .map(|pid| {
                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),
                        QueryParam::String(self.server_id.clone()),
                        QueryParam::String("Parent".to_string()),
                        QueryParam::String("Folder".to_string()),
                        QueryParam::String(now.clone()),
                    ],
                )
            })
            .collect();

        let rows: Vec<(String, Query, Option<Query>)> = items
            .iter()
            .map(|item| {
                (
                    item.id.clone(),
                    self.item_upsert_query(item, &owning_library, &now),
                    self.user_data_mirror_query(item, &now),
                )
            })
            .collect();

        // The whole page is one transaction — one job on the writer. It used to
        // be one commit per stub, item and user_data row: opening a series
        // queued hundreds of them, and every read in the app waited behind the
        // pile.
        //
        // Foreign-key checks are off for this job only. They used to be toggled
        // on the shared connection around the save's many awaits, so they were
        // off for every *other* write that ran in between too.
        //
        // TRACES: UR-002, UR-007 | DR-012
        self.db_service
            .transaction_without_foreign_keys(move |tx| {
                for placeholder in placeholders {
                    tx.execute(placeholder)?;
                }
                for stub in stubs {
                    tx.execute(stub)?;
                }
                let mut count = 0;
                for (id, item_query, user_data_query) in rows {
                    tx.execute(item_query)
                        .map_err(|e| format!("Failed to insert item {}: {}", id, e))?;
                    if let Some(query) = user_data_query {
                        if let Err(e) = tx.execute(query) {
                            debug!("[OfflineRepo] user_data mirror skipped for {}: {}", id, e);
                        }
                    }
                    count += 1;
                }
                Ok(count)
            })
            .await
            .map_err(|e| RepoError::Database { message: e })
    }

    /// Whether `id` is one of this server's libraries. Listings decide this
    /// before building their query: the library clause is only correct — and
    /// only affordable — when it is.
    async fn is_library(&self, id: &str) -> Result<bool, RepoError> {
        self.db_service
            .query_optional(
                Query::with_params(
                    "SELECT 1 FROM libraries WHERE id = ? AND server_id = ?",
                    vec![
                        QueryParam::String(id.to_string()),
                        QueryParam::String(self.server_id.clone()),
                    ],
                ),
                |row| row.get::<_, i64>(0),
            )
            .await
            .map(|row| row.is_some())
            .map_err(|e| RepoError::Database { message: e })
    }

    /// Which library the children of `parent_id` belong to.
    ///
    /// `Some(parent_id)` when the parent is itself a library, otherwise the
    /// library the parent item was already filed under — so the association
    /// propagates down a hierarchy as it is browsed, without needing the server
    /// to repeat it on every item. `None` for a parent that is neither, which
    /// is how synthetic parents like "favorites" avoid being filed anywhere.
    ///
    /// TRACES: UR-007 | DR-278
    async fn resolve_owning_library(&self, parent_id: &str) -> Option<String> {
        let is_library: Option<String> = self
            .db_service
            .query_optional(
                Query::with_params(
                    "SELECT id FROM libraries WHERE id = ? AND server_id = ?",
                    vec![
                        QueryParam::String(parent_id.to_string()),
                        QueryParam::String(self.server_id.clone()),
                    ],
                ),
                |row| row.get(0),
            )
            .await
            .ok()
            .flatten();

        if is_library.is_some() {
            return is_library;
        }

        self.db_service
            .query_optional(
                Query::with_params(
                    "SELECT library_id FROM items WHERE id = ? AND library_id IS NOT NULL",
                    vec![QueryParam::String(parent_id.to_string())],
                ),
                |row| row.get(0),
            )
            .await
            .ok()
            .flatten()
    }

    /// The UPSERT that caches one item under `owning_library`.
    fn item_upsert_query(
        &self,
        item: &MediaItem,
        owning_library: &Option<String>,
        now: &str,
    ) -> Query {
        // 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
        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()),
                // The library this browse belongs to; NULL only for
                // synthetic parents. See `resolve_owning_library`.
                match &owning_library {
                    Some(lib) => QueryParam::String(lib.clone()),
                    None => 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()),
            ],
        )
    }

    /// The write that mirrors the server's per-user state for an item into the local
    /// `user_data` table, so favourites marked — and positions watched — 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 change 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. For a position that means it is never pulled
    /// *backwards* by a server that has not yet heard where we got to.
    ///
    /// Each field is mirrored only when the server actually reported it —
    /// `COALESCE(excluded.x, user_data.x)` keeps the stored value for anything
    /// absent, and a row with neither field is skipped outright rather than
    /// written as zeroes, which would fabricate an "unfavourited, unwatched"
    /// record from an endpoint that simply omits `UserData`.
    ///
    /// The position half is what makes cross-device resume work: the resume
    /// check reads this table alone, so before it was mirrored an item watched
    /// elsewhere resumed from whatever *this* device last saw, or not at all.
    /// The played flag rides along for the same reason: nothing else writes it
    /// but an explicit local toggle, so a cached episode list read every
    /// episode back as unwatched — the list the season view ticks and the one
    /// `pick_current_episode` reads to decide what is up next (DR-264).
    ///
    /// Best-effort: `save_to_cache` logs and skips a failure rather than
    /// failing the whole cache write over it, which would break browsing.
    ///
    /// TRACES: UR-025, UR-062, UR-069 | DR-114, DR-155, DR-264 | UT-102, UT-152, UT-240
    fn user_data_mirror_query(&self, item: &MediaItem, now: &str) -> Option<Query> {
        let user_data = item.user_data.as_ref();
        let is_favorite = user_data.and_then(|ud| ud.is_favorite);
        let position_ticks = user_data.and_then(|ud| ud.playback_position_ticks);
        let is_played = user_data.and_then(|ud| ud.is_played);

        // Nothing the server actually told us about — do not invent a row.
        if is_favorite.is_none() && position_ticks.is_none() && is_played.is_none() {
            return None;
        }

        let query = Query::with_params(
            "INSERT INTO user_data
                (user_id, item_id, is_favorite, playback_position_ticks, is_played,
                 synced_at, pending_sync)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, 0)
             ON CONFLICT(user_id, item_id) DO UPDATE SET
                is_favorite = COALESCE(excluded.is_favorite, user_data.is_favorite),
                playback_position_ticks = COALESCE(
                    excluded.playback_position_ticks, user_data.playback_position_ticks),
                is_played = COALESCE(excluded.is_played, user_data.is_played),
                synced_at = excluded.synced_at
             WHERE user_data.pending_sync = 0",
            vec![
                QueryParam::String(self.user_id.clone()),
                QueryParam::String(item.id.clone()),
                is_favorite
                    .map(|f| QueryParam::Int(if f { 1 } else { 0 }))
                    .unwrap_or(QueryParam::Null),
                position_ticks
                    .map(QueryParam::Int64)
                    .unwrap_or(QueryParam::Null),
                is_played
                    .map(|p| QueryParam::Int(if p { 1 } else { 0 }))
                    .unwrap_or(QueryParam::Null),
                QueryParam::String(now.to_string()),
            ],
        );

        Some(query)
    }

    /// 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.
    ///
    /// Whether cached item `i` belongs to library `l`, decided by media kind.
    ///
    /// The cache leaves `library_id`/`parent_id` NULL on every item
    /// ([[offline-libraries-never-cached]]), so there is no link to follow: a
    /// library's `collection_type` and an item's `item_type` are the only things
    /// that can associate them. This is Jellyfin taxonomy and therefore lives in
    /// Rust, never in the frontend.
    ///
    /// It is a named constant because it is needed in two places that must agree
    /// — which library *appears* in the Downloaded list, and which items appear
    /// *inside* it. They disagreed: the listing query used this mapping while the
    /// browse query only checked that the requested library existed, so opening
    /// any library showed every downloaded top-level item on the server.
    ///
    /// A library of some other (or unknown) type keeps everything, since there is
    /// no mapping to narrow it by and hiding its contents would be worse.
    ///
    /// TRACES: UR-055 | DR-082, DR-167
    const LIBRARY_HOLDS_ITEM: &'static str = concat!(
        "(",
        library_type_matches_item!(),
        "  OR l.collection_type IS NULL
           OR l.collection_type NOT IN ('music', 'movies', 'tvshows')
        )"
    );

    /// 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 parent_is_library = self.is_library(parent_id).await?;
        let downloaded = downloaded_sql("i", PLAYABLE_TYPES, CONTAINER_TYPES);
        let downloaded_parent = downloaded_sql("parent", PLAYABLE_TYPES, CONTAINER_TYPES);
        // A non-library parent lists its children by `container_id` (migration
        // 027) — series → seasons → episodes, one index range. The library
        // clause is only added for a library, where it is the whole point.
        let parent_match = if parent_is_library {
            format!(
                "i.server_id = ?
                 AND (
                   i.container_id = ?
                   OR (
                       EXISTS (
                           SELECT 1 FROM libraries l
                           WHERE l.id = ? AND l.server_id = i.server_id
                             AND {membership}
                       )
                       -- Top-level only: hide leaves whose container is downloaded.
                       AND NOT EXISTS (
                           SELECT 1 FROM items parent
                           WHERE parent.id IN (i.album_id, i.season_id, i.series_id, i.parent_id)
                             AND {downloaded_parent}
                       )
                   )
                 )",
                membership = Self::LIBRARY_HOLDS_ITEM,
            )
        } else {
            "+i.server_id = ? AND i.container_id = ?".to_string()
        };
        let sql = 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
             WHERE {parent_match}
               AND {downloaded}{type_filter}
             ORDER BY i.sort_name ASC, i.name ASC
             LIMIT {limit} OFFSET {start_index}",
        );

        let mut params = vec![
            QueryParam::String(self.server_id.clone()),
            QueryParam::String(parent_id.to_string()),
        ];
        if parent_is_library {
            params.push(QueryParam::String(parent_id.to_string()));
        }
        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 items = self.with_user_data(cached_items).await;

        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!(
                "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
                      WHERE i.server_id = l.server_id
                        AND {membership}
                        AND {downloaded}
                  )
                ORDER BY l.sort_order ASC, l.name ASC",
                membership = Self::LIBRARY_HOLDS_ITEM,
                downloaded = downloaded_sql("i", PLAYABLE_TYPES, CONTAINER_TYPES),
            ),
            vec![QueryParam::String(self.server_id.clone())],
        );

        self.db_service
            .query_many(query, |row| {
                Ok(Library::new(
                    row.get(0)?,
                    row.get(1)?,
                    row.get::<_, Option<String>>(2)?
                        .unwrap_or_else(|| "unknown".to_string()),
                    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) {
            // 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::new(
                    row.get(0)?,
                    row.get(1)?,
                    row.get::<_, Option<String>>(2)?
                        .unwrap_or_else(|| "unknown".to_string()),
                    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); PremiereDate is what a channel folder's
        // children are listed by (DR-257), so the cached leg of the race agrees
        // with the server's order instead of flashing a name-sorted list first.
        // Everything else keeps the stable name order.
        //
        // Rows with no premiere date sort last rather than leading the list.
        let default_sort = default_listing_sort(opts.parent_kind);
        let sort_field = opts
            .sort_by
            .as_deref()
            .or(default_sort.map(|(field, _)| field));
        let descending = opts
            .sort_order
            .as_deref()
            .or(default_sort.map(|(_, order)| order))
            == Some("Descending");
        let order_by = match sort_field {
            Some("Random") => "RANDOM()".to_string(),
            Some("PremiereDate") => format!(
                "i.premiere_date IS NULL, i.premiere_date {}, i.sort_name ASC",
                if descending { "DESC" } else { "ASC" }
            ),
            _ => "i.sort_name ASC, i.name ASC".to_string(),
        };

        // 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. Same shape as `search`
        // and `get_favorites`.
        //
        // TRACES: UR-065 | DR-212 | UT-206
        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)
        };

        // 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 {
            ""
        };

        // Decided up front so the listing can use the container index; see
        // `items_listing_sql`.
        let parent_is_library = self.is_library(parent_id).await?;

        let sql = items_listing_sql(
            parent_is_library,
            include_catalog_browse(),
            &type_filter,
            favorites_filter,
            &order_by,
            limit,
            start_index,
        );

        // Children are matched on `container_id` (see `items_listing_sql`); a
        // library parent also matches through the `libraries` EXISTS clause.
        let mut params = vec![
            QueryParam::String(self.server_id.clone()),
            QueryParam::String(parent_id.to_string()), // i.container_id = ?
        ];
        if parent_is_library {
            params.push(QueryParam::String(parent_id.to_string())); // libraries.id = ?
        }
        // Positional order matters: the type placeholders sit in `{type_filter}`,
        // which the statement interpolates immediately after the parent-matching
        // group and before `{favorites_filter}`, so they bind here — after the
        // ids above, before the favourites user id.
        params.extend(type_values.iter().cloned().map(QueryParam::String));
        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())]
        );

        let items = self.with_user_data(cached_items).await;

        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(
            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
                WHERE i.id = ? AND {}",
                downloaded_sql("i", PLAYABLE_TYPES, CONTAINER_TYPES)
            ),
            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!(
                "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
                 WHERE i.server_id = ? AND i.library_id = ?
                 AND {downloaded}
                 -- Collapse leaves into the container that was added: a new
                 -- 14-track album should read as one album, not 14 songs. Only
                 -- drops a leaf when its own container is present in the same
                 -- result, so a standalone track or movie still appears.
                 AND NOT EXISTS (
                     SELECT 1 FROM items parent
                     WHERE parent.id IN (i.album_id, i.season_id, i.series_id, i.parent_id)
                       AND {downloaded_parent}
                 )
                 ORDER BY i.synced_at DESC
                 LIMIT {limit_val}",
                downloaded = downloaded_sql("i", PLAYABLE_TYPES, CONTAINER_TYPES),
                downloaded_parent = downloaded_sql("parent", PLAYABLE_TYPES, CONTAINER_TYPES),
            ),
            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 items = self.with_user_data(cached_items).await;

        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 items = self.with_user_data(cached_items).await;

        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 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
                WHERE {downloaded}
                ORDER BY rp.most_recent_play DESC",
                limit_val,
                downloaded = downloaded_sql("i", "'Audio'", "'MusicAlbum'")
            ),
            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 items = self.with_user_data(cached_items).await;

        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 items = self.with_user_data(cached_items).await;

        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 available = available_sql("i", include_catalog_browse());

        let sql = 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 items_fts fts ON fts.rowid = i.rowid
             WHERE i.server_id = ? AND items_fts MATCH ? AND {}{}
             ORDER BY rank
             LIMIT {}",
            available, 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 = self.with_user_data(cached_items).await;

        // 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>,
        _source_audio_codec: 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 available = available_sql("i", include_catalog_browse());

        let sql = 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
             INNER JOIN user_data ud ON ud.item_id = i.id
             WHERE i.server_id = ?
               AND ud.user_id = ?
               AND ud.is_favorite = 1
               AND {available}{}
             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 items = self.with_user_data(cached_items).await;

        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!(
                "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
                 WHERE i.server_id = ? AND ip.person_id = ? AND {downloaded}
                 ORDER BY i.production_year DESC, i.sort_name ASC
                 LIMIT {}",
                limit,
                downloaded = downloaded_sql("i", PLAYABLE_TYPES, CONTAINER_TYPES)
            ),
            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 items = self.with_user_data(cached_items).await;

        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 {
    // `CATALOG_BROWSE_LOCK` below serialises the tests that flip the
    // process-global `INCLUDE_CATALOG_BROWSE` flag, so its guard is deliberately
    // held across the `.await` of the query under test — that await *is* the
    // critical section. This is not the production deadlock hazard the lint
    // targets: the lock is test-only, uncontended outside these tests, and each
    // `#[tokio::test]` runs on its own single-threaded runtime, so a held guard
    // cannot block another task on the same worker. Restructuring around it
    // would reintroduce the flag race the lock exists to prevent.
    #![allow(clippy::await_holding_lock)]

    use super::*;
    use 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, ()> {
        use crate::utils::lock::MutexSafe;
        CATALOG_BROWSE_LOCK.lock_safe()
    }

    /// 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> {
        Arc::new(RusqliteService::new(create_test_conn()))
    }

    /// The raw connection behind [`create_test_db`], for tests that need to
    /// hold its lock — standing in for a concurrent write.
    fn create_test_conn() -> Arc<Mutex<Connection>> {
        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,
                -- Same rule as schema.rs migration 027.
                container_id TEXT GENERATED ALWAYS AS (
                    CASE item_type
                        WHEN 'Episode' THEN COALESCE(season_id, series_id, parent_id)
                        WHEN 'Season'  THEN COALESCE(series_id, parent_id)
                        WHEN 'Audio'   THEN COALESCE(album_id, parent_id)
                        ELSE parent_id
                    END
                ) VIRTUAL
            );
            CREATE INDEX idx_items_container ON items(container_id, sort_name, name);

            -- 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,
                user_id TEXT,
                file_path TEXT,
                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(Mutex::new(conn))
    }

    /// Helper to create a test MediaItem
    /// The reported bug: offline, a downloaded episode would not play. The
    /// player found the file on disk, then asked the *server* for the item's
    /// playback info — only to read its media-source id — and with no network
    /// that call retried for seven seconds and failed, so playback never began.
    /// A download that needs the internet to play is not a download.
    ///
    /// Driven through the real `HybridRepository` against a server that refuses
    /// connections, because the defect was the hybrid layer sending this call
    /// only to the server.
    ///
    /// TRACES: UR-002, UR-071 | DR-294 | UT-260
    #[tokio::test]
    async fn test_a_downloaded_item_gets_playback_info_without_the_server() {
        let db_service = create_test_db();
        for sql in [
            "INSERT INTO downloads (item_id, user_id, file_path, status) \
             VALUES ('ep-6', 'test-user', '/data/videos/S01E06.mp4', 'completed')",
            // Still downloading: there is no local file to play yet.
            "INSERT INTO downloads (item_id, user_id, file_path, status) \
             VALUES ('ep-7', 'test-user', '/data/videos/S01E07.mp4', 'downloading')",
            // Someone else's download is not this user's local copy.
            "INSERT INTO downloads (item_id, user_id, file_path, status) \
             VALUES ('ep-8', 'other-user', '/data/videos/S01E08.mp4', 'completed')",
        ] {
            db_service.execute(Query::new(sql)).await.unwrap();
        }
        let offline = OfflineRepository::new(
            db_service.clone(),
            "test-server".to_string(),
            "test-user".to_string(),
        );
        let local = OfflineRepository::new(
            db_service,
            "test-server".to_string(),
            "test-user".to_string(),
        );
        // Port 9 on loopback: nothing listens, so every request is refused.
        let online = crate::repository::OnlineRepository::new(
            Arc::new(
                crate::jellyfin::HttpClient::new(crate::jellyfin::HttpConfig::default()).unwrap(),
            ),
            "http://127.0.0.1:9".to_string(),
            "test-user".to_string(),
            "test-token".to_string(),
        );
        let hybrid = crate::repository::HybridRepository::new(online, offline);

        let started = std::time::Instant::now();
        let info = hybrid
            .get_playback_info("ep-6")
            .await
            .expect("a downloaded item must get playback info with no server");
        assert!(
            started.elapsed() < std::time::Duration::from_secs(1),
            "answered locally, not after the network gave up: {:?}",
            started.elapsed()
        );
        assert_eq!(info.stream_url, "/data/videos/S01E06.mp4");
        assert!(info.direct_play && !info.needs_transcoding);
        // Jellyfin's default media source shares the item's id, and a download
        // is always of the default source (no mediaSourceId is requested).
        assert_eq!(info.media_source_id, "ep-6");

        // Not playable locally → still the server's question to answer.
        for not_local in ["ep-7", "ep-8", "never-downloaded"] {
            assert!(
                local
                    .local_playback_info(not_local)
                    .await
                    .unwrap()
                    .is_none(),
                "{not_local} has no completed local file for this user"
            );
        }
    }

    /// Next Up went only to the server, so offline it failed — and the TV
    /// landing page loads it together with Continue Watching and Latest in one
    /// `Promise.all`, so that one failure blanked the whole page ("Failed to
    /// load TV sections: Offline") with its other rows sitting in the cache.
    /// With the server unreachable, the answer is the cache's.
    ///
    /// TRACES: UR-002 | DR-294 | UT-261
    #[tokio::test]
    async fn test_next_up_answers_from_the_cache_when_the_server_is_unreachable() {
        let offline = OfflineRepository::new(
            create_test_db(),
            "test-server".to_string(),
            "test-user".to_string(),
        );
        let online = crate::repository::OnlineRepository::new(
            Arc::new(
                crate::jellyfin::HttpClient::new(crate::jellyfin::HttpConfig::default()).unwrap(),
            ),
            "http://127.0.0.1:9".to_string(),
            "test-user".to_string(),
            "test-token".to_string(),
        );
        let hybrid = crate::repository::HybridRepository::new(online, offline);

        let next_up = hybrid.get_next_up_episodes(None, Some(12)).await;
        assert!(
            next_up.is_ok(),
            "an unreachable server must not fail Next Up offline: {next_up:?}"
        );
    }

    /// The reported bug: offline, "More info" on a downloaded show failed with
    /// "Failed to load item". Its seasons were in the cache the whole time.
    ///
    /// `get_items` gave the cache 100 ms and *discarded* a slower answer, then
    /// waited on the server — which offline fails — and returned the server's
    /// error. The cache was then one SQLite connection behind one mutex, so any
    /// write in progress (the catalog sync that starts at every launch, a
    /// download finishing) pushed a read past 100 ms routinely. Every other cached query
    /// already keeps a slow read alive and waits for it when the server fails;
    /// `get_items`, which the series page calls for the show and each season,
    /// did not.
    ///
    /// TRACES: UR-002 | DR-294 | UT-263
    #[tokio::test]
    async fn test_get_items_waits_for_a_slow_cache_when_the_server_is_unreachable() {
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);
        let conn = create_test_conn();
        let db_service = Arc::new(RusqliteService::new(Arc::clone(&conn)));
        for sql in [
            "INSERT INTO items (id, server_id, name, item_type, synced_at) \
             VALUES ('series-1', 'test-server', 'Show', 'Series', '2026-01-01')",
            "INSERT INTO items (id, server_id, parent_id, name, item_type, synced_at) \
             VALUES ('season-1', 'test-server', 'series-1', 'Season 1', 'Season', '2026-01-01')",
        ] {
            db_service.execute(Query::new(sql)).await.unwrap();
        }
        let offline = OfflineRepository::new(
            db_service,
            "test-server".to_string(),
            "test-user".to_string(),
        );
        let online = crate::repository::OnlineRepository::new(
            Arc::new(
                crate::jellyfin::HttpClient::new(crate::jellyfin::HttpConfig::default()).unwrap(),
            ),
            "http://127.0.0.1:9".to_string(),
            "test-user".to_string(),
            "test-token".to_string(),
        );
        let hybrid = crate::repository::HybridRepository::new(online, offline);

        // Sanity: with nothing holding the database, the season is found. A
        // failure here is the fixture, not the bug.
        let quick = hybrid
            .get_items("series-1", None)
            .await
            .expect("unlocked read");
        assert_eq!(quick.items.len(), 1, "fixture: the season is cached");

        // Now a write holds the connection for 300 ms — longer than the fast path.
        let held = Arc::clone(&conn);
        let writer = std::thread::spawn(move || {
            let _lock = held.lock().unwrap();
            std::thread::sleep(std::time::Duration::from_millis(300));
        });
        std::thread::sleep(std::time::Duration::from_millis(20));

        let slow = hybrid.get_items("series-1", None).await;
        writer.join().unwrap();
        let slow = slow.expect("a slow cache must still answer when the server cannot");
        assert_eq!(slow.items.len(), 1);
        assert_eq!(slow.items[0].id, "season-1");
    }

    /// A hybrid repository whose server refuses every connection, over `conn`.
    fn hybrid_without_server(conn: &Arc<Mutex<Connection>>) -> crate::repository::HybridRepository {
        let offline = OfflineRepository::new(
            Arc::new(RusqliteService::new(Arc::clone(conn))),
            "test-server".to_string(),
            "test-user".to_string(),
        );
        let online = crate::repository::OnlineRepository::new(
            Arc::new(
                crate::jellyfin::HttpClient::new(crate::jellyfin::HttpConfig::default()).unwrap(),
            ),
            "http://127.0.0.1:9".to_string(),
            "test-user".to_string(),
            "test-token".to_string(),
        );
        crate::repository::HybridRepository::new(online, offline)
    }

    /// Hold the database for `ms`, as a concurrent write does.
    fn hold_database(conn: &Arc<Mutex<Connection>>, ms: u64) -> std::thread::JoinHandle<()> {
        let held = Arc::clone(conn);
        let writer = std::thread::spawn(move || {
            let _lock = held.lock().unwrap();
            std::thread::sleep(std::time::Duration::from_millis(ms));
        });
        std::thread::sleep(std::time::Duration::from_millis(20));
        writer
    }

    /// The library list is the first thing an offline launch shows, and it had
    /// the same defect as `get_items`: a cache read slowed past 100 ms by a write
    /// was discarded, the server then failed, and the list came back as a
    /// network error with the libraries on disk.
    ///
    /// TRACES: UR-002 | DR-294 | UT-263
    #[tokio::test]
    async fn test_libraries_wait_for_a_slow_cache_when_the_server_is_unreachable() {
        let conn = create_test_conn();
        conn.lock()
            .unwrap()
            .execute(
                "INSERT INTO libraries (id, server_id, name, collection_type) \
                 VALUES ('lib-tv', 'test-server', 'Shows', 'tvshows')",
                [],
            )
            .unwrap();
        let hybrid = hybrid_without_server(&conn);

        let writer = hold_database(&conn, 300);
        let libs = hybrid.get_libraries().await;
        writer.join().unwrap();

        let libs = libs.expect("a slow cache must still answer when the server cannot");
        assert_eq!(libs.len(), 1);
        assert_eq!(libs[0].id, "lib-tv");
    }

    /// Search and favourites read only the cache — there is no server to fall
    /// back from — so a 100 ms deadline on them just fails them whenever the
    /// database is busy, online or not.
    ///
    /// TRACES: UR-002 | DR-294 | UT-263
    #[tokio::test]
    async fn test_cache_only_reads_wait_out_a_busy_database() {
        let conn = create_test_conn();
        let hybrid = hybrid_without_server(&conn);
        // Fixture sanity: with the database free both answer.
        hybrid
            .search_cache_only("anything", None)
            .await
            .expect("unlocked search");

        let writer = hold_database(&conn, 300);
        let search = hybrid.search_cache_only("anything", None).await;
        writer.join().unwrap();
        search.expect("a busy database delays a cache-only search, it must not fail it");

        let writer = hold_database(&conn, 300);
        let favourites = hybrid
            .get_favorites_cache_only(crate::repository::SearchScope::All, None)
            .await;
        writer.join().unwrap();
        favourites.expect("a busy database delays cache-only favourites, it must not fail them");
    }

    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 season that holds the download —
        // series → season → episode, the same hierarchy the server serves.
        // (It used to return the episode itself, via its `series_id`, beside
        // the seasons; see `a_series_lists_its_seasons_not_their_episodes`.)
        let series_items = repo.get_items("series-1", None).await.unwrap();
        let ids: Vec<&str> = series_items.items.iter().map(|i| i.id.as_str()).collect();
        assert_eq!(
            ids,
            vec!["season-1"],
            "get_items(series_id) should list the season holding the download"
        );
    }

    /// 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::new("music".into(), "Music".into(), "music".into(), None),
            Library::new(
                "movies".into(),
                "Movies".into(),
                "movies".into(),
                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();
    }

    /// Like `insert_item`, but sets `library_id` — which `get_latest_items`
    /// filters on, so rows without it are invisible to that query.
    async fn insert_library_item(
        db: &Arc<RusqliteService>,
        id: &str,
        item_type: &str,
        library_id: &str,
        album_id: Option<&str>,
    ) {
        db.execute(Query::with_params(
            "INSERT INTO items (id, server_id, library_id, name, item_type, album_id, synced_at)
             VALUES (?1, 'test-server', ?2, ?3, ?4, ?5, '2024-01-01')",
            vec![
                QueryParam::String(id.to_string()),
                QueryParam::String(library_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),
            ],
        ))
        .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(),
        )
    }

    /// A newly-synced album appears once in "recently added", not once per track.
    ///
    /// The downloaded-items CTE deliberately matches both the leaves and their
    /// container, which is right for browsing but wrong here: it made a 3-track
    /// album occupy 4 slots in the row. Tracks whose album is itself in the
    /// result are now collapsed into it.
    #[tokio::test]
    async fn test_get_latest_items_collapses_tracks_into_their_album() {
        let db = create_test_db();
        insert_library_item(&db, "album-1", "MusicAlbum", "lib-1", None).await;
        for track in ["track-1", "track-2", "track-3"] {
            insert_library_item(&db, track, "Audio", "lib-1", Some("album-1")).await;
            seed_completed_download(&db, track, 1000).await;
        }
        // A movie has no container, so it must still show up on its own.
        insert_library_item(&db, "movie-1", "Movie", "lib-1", None).await;
        seed_completed_download(&db, "movie-1", 2000).await;

        let repo = make_repo(&db);
        let latest = repo.get_latest_items("lib-1", Some(16)).await.unwrap();
        let ids: Vec<&str> = latest.iter().map(|i| i.id.as_str()).collect();

        assert!(
            !ids.iter().any(|id| id.starts_with("track-")),
            "individual tracks must collapse into their album, got: {ids:?}"
        );
        assert!(ids.contains(&"album-1"), "the album itself is listed");
        assert!(ids.contains(&"movie-1"), "containerless items still listed");
    }

    /// 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: each downloaded library shows **only its own media**.
    ///
    /// Cached items carry no link back to their library (`library_id`/`parent_id`
    /// are NULL — [[offline-libraries-never-cached]]), and the library branch of
    /// the query only asserted that the requested library *exists*, never that
    /// the item belongs to it. So opening any downloaded library listed every
    /// downloaded top-level item on the server: films in the music library,
    /// albums under TV. The library's `collection_type` decides which item types
    /// belong to it, the same mapping `get_downloaded_libraries` already uses.
    ///
    /// TRACES: UR-055 | DR-167 | UT-162
    #[tokio::test]
    async fn test_get_downloaded_items_library_does_not_mix_media_types() {
        let db = create_test_db();
        seed_library(&db, "music-lib", "music").await;
        seed_library(&db, "movie-lib", "movies").await;
        seed_library(&db, "tv-lib", "tvshows").await;

        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, "movie-1", "Movie", None, None, None).await;
        insert_item(&db, "series-1", "Series", None, None, None).await;
        insert_item(&db, "episode-1", "Episode", None, Some("series-1"), None).await;

        seed_completed_download(&db, "track-1", 1000).await;
        seed_completed_download(&db, "movie-1", 2000).await;
        seed_completed_download(&db, "episode-1", 3000).await;

        let repo = make_repo(&db);

        let music: Vec<String> = repo
            .get_downloaded_items("music-lib", None)
            .await
            .unwrap()
            .items
            .iter()
            .map(|i| i.id.clone())
            .collect();
        assert_eq!(
            music,
            vec!["album-1"],
            "the music library must not list films or series; got {:?}",
            music
        );

        let movies: Vec<String> = repo
            .get_downloaded_items("movie-lib", None)
            .await
            .unwrap()
            .items
            .iter()
            .map(|i| i.id.clone())
            .collect();
        assert_eq!(
            movies,
            vec!["movie-1"],
            "the movie library must not list albums or series; got {:?}",
            movies
        );

        let tv: Vec<String> = repo
            .get_downloaded_items("tv-lib", None)
            .await
            .unwrap()
            .items
            .iter()
            .map(|i| i.id.clone())
            .collect();
        assert_eq!(
            tv,
            vec!["series-1"],
            "the TV library must not list albums or films; got {:?}",
            tv
        );
    }

    /// 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();
        let in_series_ids: Vec<&str> = in_series.items.iter().map(|i| i.id.as_str()).collect();
        assert_eq!(
            in_series_ids,
            vec!["season-1"],
            "series drill returns the season — not the season's episodes"
        );
        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-206 — `include_item_types` reaches the listing query as bound
    /// parameters, so a type name can only ever be compared as data.
    ///
    /// Interpolated, the type below closed the `IN (` list and commented out the
    /// rest of the line, leaving `... AND i.item_type IN ('Movie') OR 1=1`, which
    /// is true for every row — the listing then returned the whole cache
    /// regardless of parent or type. Bound, it is just a type name that matches
    /// nothing.
    ///
    /// TRACES: UR-065 | DR-212 | UT-206
    #[tokio::test]
    async fn test_get_items_type_filter_is_bound_not_interpolated() {
        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 injected = repo
            .get_items(
                "lib-1",
                Some(GetItemsOptions {
                    include_item_types: Some(vec!["Movie') OR 1=1 --".to_string()]),
                    ..Default::default()
                }),
            )
            .await
            .expect("a hostile type name must be data, not a broken query");
        assert!(
            injected.items.is_empty(),
            "no cached item has that type, so nothing may come back; got {:?}",
            injected
                .items
                .iter()
                .map(|i| i.id.as_str())
                .collect::<Vec<_>>()
        );

        // A quote on its own is likewise just a character in a type name.
        let quoted = repo
            .get_items(
                "lib-1",
                Some(GetItemsOptions {
                    include_item_types: Some(vec!["Mo'vie".to_string()]),
                    ..Default::default()
                }),
            )
            .await
            .expect("an embedded quote must not break the query");
        assert!(quoted.items.is_empty());
    }

    /// UT-206 — binding the type filter must not disturb the positions of the
    /// parameters around it: the parent ids bind before it and the favourites
    /// user id after it. A misordered vec would silently compare `user_id`
    /// against `item_type`, so this asserts the filters still compose.
    ///
    /// TRACES: UR-065, UR-067 | DR-212 | UT-206
    #[tokio::test]
    async fn test_get_items_binds_multiple_types_in_parameter_order() {
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);

        let db_service = create_test_db();
        seed_favorites(&db_service).await;

        // A favourite album *in lib-1*. The fixture's `album-fav` lives in
        // lib-2, and this test used to expect it back from a lib-1 listing —
        // which only held because the library clause matched every cached row
        // regardless of which library it was in (DR-277). The assertions below
        // still span both requested types, which is what UT-206 is really about;
        // they now do it with an album that is actually in the library.
        db_service
            .execute(Query::new(
                "INSERT INTO items (id, server_id, name, item_type, library_id, synced_at, sort_name) \
                 VALUES ('album-lib1', 'test-server', 'Album In Lib One', 'MusicAlbum', 'lib-1', '2026-01-01', 'Album In Lib One')",
            ))
            .await
            .unwrap();
        db_service
            .execute(Query::new(
                "INSERT INTO user_data (user_id, item_id, is_favorite) VALUES ('test-user', 'album-lib1', 1)",
            ))
            .await
            .unwrap();

        let repo = OfflineRepository::new(
            db_service,
            "test-server".to_string(),
            "test-user".to_string(),
        );

        let both = repo
            .get_items(
                "lib-1",
                Some(GetItemsOptions {
                    include_item_types: Some(vec!["Movie".to_string(), "MusicAlbum".to_string()]),
                    ..Default::default()
                }),
            )
            .await
            .unwrap();
        let mut ids: Vec<&str> = both.items.iter().map(|i| i.id.as_str()).collect();
        ids.sort();
        assert_eq!(ids, vec!["album-lib1", "movie-fav", "movie-plain"]);
        assert!(
            !ids.contains(&"album-fav"),
            "album-fav belongs to lib-2 and must not appear in a lib-1 listing"
        );

        // Two type placeholders *and* the favourites parameter after them.
        let favourites = repo
            .get_items(
                "lib-1",
                Some(GetItemsOptions {
                    include_item_types: Some(vec!["Movie".to_string(), "MusicAlbum".to_string()]),
                    favorites_only: Some(true),
                    ..Default::default()
                }),
            )
            .await
            .unwrap();
        let mut ids: Vec<&str> = favourites.items.iter().map(|i| i.id.as_str()).collect();
        ids.sort();
        assert_eq!(ids, vec!["album-lib1", "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"
        );
    }

    /// UT-152 — the server's watch position is mirrored locally, so an item
    /// watched on another device resumes here.
    ///
    /// The resume check reads only the local `user_data` row, and the mirror
    /// previously carried `is_favorite` alone — so a position set on any other
    /// client never reached this device and cross-device resume silently did
    /// nothing. The `pending_sync` guard is the same conflict rule favourites
    /// use: a local position still waiting to be pushed must not be pulled
    /// backwards by the stale value the server is still reporting.
    ///
    /// TRACES: UR-025, UR-069 | DR-155 | UT-152
    #[tokio::test]
    async fn test_save_to_cache_mirrors_playback_position_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 position = |id: &'static str| {
            let db = db_service.clone();
            async move {
                db.query_optional(
                    Query::with_params(
                        "SELECT playback_position_ticks, 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<i64>>(0)?, row.get::<_, Option<i32>>(1)?)),
                )
                .await
                .unwrap()
            }
        };

        // Watched 20 minutes into this episode on another device.
        let mut watched = create_test_item("ep-1", "Watched Elsewhere", None);
        watched.user_data = Some(UserData {
            playback_position_ticks: Some(12_000_000_000),
            ..Default::default()
        });
        // No user data at all — must not fabricate a position of 0.
        let untouched = create_test_item("ep-2", "No User Data", None);

        repo.save_to_cache("parent-1", &[watched.clone(), untouched])
            .await
            .unwrap();

        assert_eq!(
            position("ep-1").await,
            Some((Some(12_000_000_000), Some(0))),
            "the server's position should be mirrored as synced"
        );
        assert_eq!(
            position("ep-2").await,
            None,
            "an item without UserData should not get an invented position"
        );

        // Watched further here while the server was unreachable: pending_sync = 1.
        db_service
            .execute(Query::with_params(
                "UPDATE user_data SET playback_position_ticks = ?, pending_sync = 1 \
                 WHERE user_id = ? AND item_id = ?",
                vec![
                    QueryParam::Int64(30_000_000_000),
                    QueryParam::String("test-user".to_string()),
                    QueryParam::String("ep-1".to_string()),
                ],
            ))
            .await
            .unwrap();

        // The server still reports the older position; caching must not win.
        repo.save_to_cache("parent-1", &[watched]).await.unwrap();

        assert_eq!(
            position("ep-1").await,
            Some((Some(30_000_000_000), Some(1))),
            "an unsynced local position must not be pulled backwards"
        );
    }

    /// UT-240 — the server's *played* flag is mirrored locally, so an episode
    /// watched anywhere is watched here.
    ///
    /// The mirror carried only the favourite flag and the position, so
    /// `user_data.is_played` was written by nothing but an explicit local
    /// toggle: a cached episode list reported every episode as unwatched, which
    /// is the list `pick_current_episode` reads to decide what is up next
    /// (DR-264), and the list the season view ticks.
    ///
    /// TRACES: UR-025, UR-062 | DR-264 | UT-240
    #[tokio::test]
    async fn test_save_to_cache_mirrors_played_flag_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 played_flag = |id: &'static str| {
            let db = db_service.clone();
            async move {
                db.query_optional(
                    Query::with_params(
                        "SELECT is_played, 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()
            }
        };

        // Watched to the end on another client.
        let mut watched = create_test_item("ep-4", "Watched Elsewhere", None);
        watched.user_data = Some(UserData {
            is_played: Some(true),
            ..Default::default()
        });
        // No user data at all — must not fabricate an "unwatched" record.
        let untouched = create_test_item("ep-5", "No User Data", None);

        repo.save_to_cache("parent-1", &[watched, untouched])
            .await
            .unwrap();

        assert_eq!(
            played_flag("ep-4").await,
            Some((Some(1), Some(0))),
            "the server's played flag should be mirrored as synced"
        );
        assert_eq!(
            played_flag("ep-5").await,
            None,
            "an item without UserData should not get an invented played flag"
        );

        // Marked unwatched here while the server was unreachable.
        db_service
            .execute(Query::with_params(
                "UPDATE user_data SET is_played = 0, pending_sync = 1 \
                 WHERE user_id = ? AND item_id = ?",
                vec![
                    QueryParam::String("test-user".to_string()),
                    QueryParam::String("ep-4".to_string()),
                ],
            ))
            .await
            .unwrap();

        let mut still_played = create_test_item("ep-4", "Watched Elsewhere", None);
        still_played.user_data = Some(UserData {
            is_played: Some(true),
            ..Default::default()
        });
        repo.save_to_cache("parent-1", &[still_played])
            .await
            .unwrap();

        assert_eq!(
            played_flag("ep-4").await,
            Some((Some(0), Some(1))),
            "an unsynced local toggle must survive a cache write"
        );
    }

    /// UT-152 — a server item carrying *only* a position (no favourite flag)
    /// still gets mirrored.
    ///
    /// The mirror used to return early whenever `is_favorite` was absent, which
    /// is exactly the shape of an ordinary watched episode: Jellyfin reports
    /// `PlaybackPositionTicks` with no favourite state. That early return is why
    /// the position never landed.
    ///
    /// TRACES: UR-025 | DR-155 | UT-152
    #[tokio::test]
    async fn test_position_is_mirrored_even_when_no_favourite_flag_is_present() {
        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 mut watched = create_test_item("ep-3", "Position Only", None);
        watched.user_data = Some(UserData {
            is_favorite: None,
            playback_position_ticks: Some(9_000_000_000),
            ..Default::default()
        });

        repo.save_to_cache("parent-1", &[watched]).await.unwrap();

        let stored = db_service
            .query_optional(
                Query::with_params(
                    "SELECT playback_position_ticks FROM user_data \
                     WHERE user_id = ? AND item_id = ?",
                    vec![
                        QueryParam::String("test-user".to_string()),
                        QueryParam::String("ep-3".to_string()),
                    ],
                ),
                |row| row.get::<_, Option<i64>>(0),
            )
            .await
            .unwrap();

        assert_eq!(
            stored,
            Some(Some(9_000_000_000)),
            "a position with no favourite flag must still be mirrored"
        );
    }

    /// A library whose `collection_type` is not one of the three the app has
    /// landing pages for — Books, Photos, Home Videos, Collections, or a mixed
    /// library — must not show the entire server.
    ///
    /// The cache has no item→library link at all (`library_id`/`parent_id` are
    /// NULL, see [[offline-libraries-never-cached]]), so `get_items` matched a
    /// library parent with an EXISTS that never referenced the item:
    ///
    ///     OR EXISTS (SELECT 1 FROM libraries l WHERE l.id = ? AND ...)
    ///
    /// True for every cached row the moment the requested id is any library.
    /// The music/movies/TV landing pages got away with it because each passes
    /// `include_item_types`, which narrowed the result; the generic library page
    /// passes none, so opening a Books library served whatever happened to be
    /// cached — films, albums, episodes. Same defect the downloaded listing had
    /// in DR-167, in the path nobody re-checked.
    ///
    /// TRACES: UR-007, UR-055 | DR-277 | UT-247
    #[tokio::test]
    async fn test_get_items_unknown_library_type_does_not_return_whole_server() {
        // Shared global: other tests flip it, so hold the lock and
        // state it explicitly rather than inheriting whatever ran last.
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);

        let db = create_test_db();

        insert_item(&db, "movie-1", "Movie", None, None, None).await;
        insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
        insert_item(&db, "series-1", "Series", None, None, None).await;

        // Every library kind the app has no landing page for, including the
        // empty `collection_type` Jellyfin sends for a mixed library.
        for collection_type in ["books", "boxsets", "photos", "homevideos", ""] {
            let lib = format!("lib-{collection_type}");
            seed_library(&db, &lib, collection_type).await;

            let repo = make_repo(&db);
            let ids: Vec<String> = repo
                .get_items(&lib, None)
                .await
                .unwrap()
                .items
                .iter()
                .map(|i| i.id.clone())
                .collect();

            assert!(
                ids.is_empty(),
                "a '{collection_type}' library must not serve the server's films, \
                 albums and shows; got {:?}",
                ids
            );
        }
    }

    /// Two libraries of the *same* type are still two libraries. A server with
    /// "TV" and "Shows" — or "Films" and "Kids Films" — must not serve both the
    /// same contents.
    ///
    /// The taxonomy fallback cannot tell them apart: it matches on
    /// `collection_type`, which is identical for both, so every Series on the
    /// server satisfies either one. Only the stored `library_id` can separate
    /// them, which is why populating it is the real fix rather than a nicety.
    ///
    /// TRACES: UR-007 | DR-277 | UT-250
    #[tokio::test]
    async fn test_get_items_two_libraries_of_one_type_are_not_interchangeable() {
        // Shared global: other tests flip it, so hold the lock and
        // state it explicitly rather than inheriting whatever ran last.
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);

        let db = create_test_db();
        seed_library(&db, "tv-lib", "tvshows").await;
        seed_library(&db, "shows-lib", "tvshows").await;

        let repo = make_repo(&db);

        // Seeded through the real write path, because that is what the fix
        // changes: browsing a library is what files its contents under it.
        for (id, lib) in [("series-a", "tv-lib"), ("series-b", "shows-lib")] {
            let mut item = create_test_item(id, id, None);
            item.item_type = "Series".to_string();
            item.kind = crate::domain::MediaKind::Series;
            repo.save_to_cache(lib, &[item]).await.unwrap();
        }
        for (lib, own, other) in [
            ("tv-lib", "series-a", "series-b"),
            ("shows-lib", "series-b", "series-a"),
        ] {
            let ids: Vec<String> = repo
                .get_items(lib, None)
                .await
                .unwrap()
                .items
                .iter()
                .map(|i| i.id.clone())
                .collect();
            assert!(
                ids.contains(&own.to_string()),
                "{lib} should list {own}; got {:?}",
                ids
            );
            assert!(
                !ids.contains(&other.to_string()),
                "{lib} must not list {other}, which lives in the other library; got {:?}",
                ids
            );
        }
    }

    /// Opening an individual collection is a different path and must keep
    /// working: a BoxSet's children carry `parent_id`, which the cache does
    /// store, so they are matched by the ordinary parent link rather than by
    /// the library clause this fix narrowed.
    ///
    /// Worth pinning separately — narrowing the library clause could plausibly
    /// have taken collections with it, and "Collections is empty" would look
    /// identical to the bug it was meant to fix.
    ///
    /// TRACES: UR-007 | DR-277 | UT-249
    #[tokio::test]
    async fn test_get_items_collection_lists_its_own_children() {
        // Shared global: other tests flip it, so hold the lock and
        // state it explicitly rather than inheriting whatever ran last.
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);

        let db = create_test_db();
        seed_library(&db, "boxset-lib", "boxsets").await;

        insert_item(&db, "boxset-1", "BoxSet", None, None, None).await;
        insert_item(&db, "outsider", "Movie", None, None, None).await;

        // A film inside the collection: linked by parent_id, which is what a
        // BoxSet's children actually carry.
        db.execute(Query::with_params(
            "INSERT INTO items (id, server_id, name, item_type, parent_id, synced_at) \
             VALUES ('in-set', 'test-server', 'In The Set', 'Movie', ?1, '2024-01-01')",
            vec![QueryParam::String("boxset-1".to_string())],
        ))
        .await
        .unwrap();

        let repo = make_repo(&db);
        let ids: Vec<String> = repo
            .get_items("boxset-1", None)
            .await
            .unwrap()
            .items
            .iter()
            .map(|i| i.id.clone())
            .collect();

        assert_eq!(
            ids,
            vec!["in-set".to_string()],
            "a collection lists its own children and nothing else; got {:?}",
            ids
        );
    }

    /// The narrowing must not break the libraries that *do* have landing pages:
    /// they reach the same query and must keep returning their own media.
    ///
    /// TRACES: UR-007 | DR-277 | UT-248
    #[tokio::test]
    async fn test_get_items_typed_libraries_still_return_their_own_media() {
        // Shared global: other tests flip it, so hold the lock and
        // state it explicitly rather than inheriting whatever ran last.
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);

        let db = create_test_db();
        seed_library(&db, "music-lib", "music").await;
        seed_library(&db, "movie-lib", "movies").await;
        seed_library(&db, "tv-lib", "tvshows").await;

        insert_item(&db, "album-1", "MusicAlbum", None, None, None).await;
        insert_item(&db, "movie-1", "Movie", None, None, None).await;
        insert_item(&db, "series-1", "Series", None, None, None).await;

        let repo = make_repo(&db);

        for (lib, expected, forbidden) in [
            ("music-lib", "album-1", "movie-1"),
            ("movie-lib", "movie-1", "album-1"),
            ("tv-lib", "series-1", "album-1"),
        ] {
            let ids: Vec<String> = repo
                .get_items(lib, None)
                .await
                .unwrap()
                .items
                .iter()
                .map(|i| i.id.clone())
                .collect();
            assert!(
                ids.contains(&expected.to_string()),
                "{lib} should list {expected}; got {:?}",
                ids
            );
            assert!(
                !ids.contains(&forbidden.to_string()),
                "{lib} must not list {forbidden}; got {:?}",
                ids
            );
        }
    }

    /// Caching a page must not switch off foreign-key enforcement for anyone
    /// else.
    ///
    /// `save_to_cache` turns FK checks off while it writes stub parents. That
    /// pragma is per *connection*, and the connection is shared, so it used to
    /// stay off across every await of the save — any other write that ran in
    /// that window (an orphan row, a cascade that should have fired) went
    /// through unchecked. The toggle must be confined to the cache write alone.
    ///
    /// TRACES: UR-002 | DR-012 | UT-014
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn save_to_cache_does_not_disable_foreign_keys_for_concurrent_writes() {
        // The fixture already has the `test-server` row, so the only thing an
        // orphan below can violate is its parent reference.
        let db_service = create_test_db();

        let repo = OfflineRepository::new(
            db_service.clone(),
            "test-server".to_string(),
            "test-user".to_string(),
        );
        let items: Vec<MediaItem> = (0..1500)
            .map(|i| create_test_item(&format!("item-{i}"), "Item", Some("library-1")))
            .collect();

        let save = tokio::spawn(async move { repo.save_to_cache("library-1", &items).await });

        // Hammer the connection with FK-violating writes while the save runs.
        let mut orphans_accepted = 0;
        let mut attempt = 0;
        while !save.is_finished() {
            let inserted = db_service
                .execute(Query::with_params(
                    "INSERT INTO items (id, server_id, name, item_type, parent_id)
                     VALUES (?, 'test-server', 'Orphan', 'Audio', 'no-such-parent')",
                    vec![QueryParam::String(format!("orphan-{attempt}"))],
                ))
                .await;
            if inserted.is_ok() {
                orphans_accepted += 1;
            }
            attempt += 1;
        }
        save.await.unwrap().unwrap();

        assert!(
            attempt > 0,
            "the save finished before any write could race it"
        );
        assert_eq!(
            orphans_accepted, 0,
            "{orphans_accepted} of {attempt} FK-violating writes were accepted mid-save"
        );
    }

    /// Listings attach user data in batches, and each row must still get its
    /// *own* — across the chunk boundary, and `None` for rows without any.
    ///
    /// TRACES: UR-002 | DR-013
    #[tokio::test]
    async fn listing_user_data_is_batched_per_row() {
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);
        let db_service = create_test_db();
        let repo = OfflineRepository::new(
            db_service.clone(),
            "test-server".to_string(),
            "test-user".to_string(),
        );

        // More rows than one batch; every third has a position of its index.
        let items: Vec<MediaItem> = (0..1203)
            .map(|i| {
                create_test_item(
                    &format!("ep-{i:04}"),
                    &format!("Ep {i:04}"),
                    Some("season-1"),
                )
            })
            .collect();
        repo.save_to_cache("season-1", &items).await.unwrap();
        for i in (0..1203).step_by(3) {
            db_service
                .execute(Query::with_params(
                    "INSERT INTO user_data (user_id, item_id, playback_position_ticks) VALUES ('test-user', ?, ?)",
                    vec![QueryParam::String(format!("ep-{i:04}")), QueryParam::Int64(i as i64)],
                ))
                .await
                .unwrap();
        }

        let result = repo.get_items("season-1", None).await.unwrap();
        assert_eq!(result.items.len(), 1203);
        for item in &result.items {
            let i: i64 = item.id.trim_start_matches("ep-").parse().unwrap();
            let position = item
                .user_data
                .as_ref()
                .and_then(|u| u.playback_position_ticks);
            if i % 3 == 0 {
                assert_eq!(
                    position,
                    Some(i),
                    "{} got someone else's user data",
                    item.id
                );
            } else {
                assert_eq!(position, None, "{} got user data it does not have", item.id);
            }
        }
    }

    /// Listing a series, season or album must look up its children by index,
    /// not walk the whole catalogue.
    ///
    /// The query used to build the set of *every* available item id (a
    /// `UNION` over the whole `items` table) and then filter it to the parent,
    /// and its library clause sat in the same `OR` as the parent columns, which
    /// forced a scan of every row on the server. About 80 ms on a desktop for a
    /// 100k-item cache — several times that on a phone, so the series page's
    /// read missed the cache fast path on every visit. The app never runs
    /// `ANALYZE`, so the plan must be good without statistics.
    ///
    /// TRACES: UR-002 | DR-013
    #[test]
    fn listing_a_non_library_parent_uses_the_container_index() {
        use crate::utils::lock::MutexSafe;
        let db = crate::storage::Database::open_in_memory().unwrap();
        let conn = db.connection();
        let conn = conn.lock_safe();
        for include_catalog in [true, false] {
            let sql = items_listing_sql(
                false,
                include_catalog,
                "",
                "",
                "i.sort_name ASC, i.name ASC",
                100,
                0,
            );
            let mut stmt = conn.prepare(&format!("EXPLAIN QUERY PLAN {sql}")).unwrap();
            let plan: Vec<String> = stmt
                .query_map(rusqlite::params!["srv", "p"], |row| row.get::<_, String>(3))
                .unwrap()
                .map(Result::unwrap)
                .collect();
            let plan = plan.join("\n");
            assert!(
                plan.contains("idx_items_container"),
                "expected a container index lookup, got:\n{plan}"
            );
            assert!(
                !plan.contains("SCAN i") && !plan.contains("idx_items_server"),
                "the listing walks every item on the server:\n{plan}"
            );
        }
    }

    /// A series lists its seasons — not every episode of every season.
    ///
    /// Children were matched on four columns at once (`parent_id`, `album_id`,
    /// `season_id`, `series_id`), and every episode carries its series id, so
    /// a cached series answered with its eleven seasons *and* all ~275
    /// episodes. With the series page's `limit: 100`, episodes whose names sort
    /// before "Season …" could push seasons out of the answer altogether.
    ///
    /// TRACES: UR-002, UR-007 | DR-013
    #[tokio::test]
    async fn a_series_lists_its_seasons_not_their_episodes() {
        let _guard = lock_catalog_browse();
        set_include_catalog_browse(true);
        let db_service = create_test_db();
        for sql in [
            "INSERT INTO items (id, server_id, name, item_type, synced_at) \
             VALUES ('show', 'test-server', 'Show', 'Series', '2026-01-01')",
            "INSERT INTO items (id, server_id, parent_id, series_id, name, item_type, synced_at) \
             VALUES ('s1', 'test-server', 'show', 'show', 'Season 1', 'Season', '2026-01-01')",
            "INSERT INTO items (id, server_id, parent_id, season_id, series_id, name, item_type, synced_at) \
             VALUES ('e1', 'test-server', 's1', 's1', 'show', 'A Pilot', 'Episode', '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(),
        );

        let ids: Vec<String> = repo
            .get_items("show", None)
            .await
            .unwrap()
            .items
            .into_iter()
            .map(|i| i.id)
            .collect();
        assert_eq!(ids, vec!["s1".to_string()]);
    }

    /// Caching an episode (or a track) whose season, series (or album) was
    /// never cached must leave that container reachable: a series lists its
    /// seasons, so without a season row a downloaded episode would be
    /// unreachable from its series page offline.
    ///
    /// TRACES: UR-002, UR-007 | DR-013
    #[tokio::test]
    async fn caching_a_child_creates_its_missing_containers() {
        let db_service = create_test_db();
        let repo = OfflineRepository::new(
            db_service.clone(),
            "test-server".to_string(),
            "test-user".to_string(),
        );
        let mut episode = create_test_item("ep", "Pilot", Some("next-up"));
        episode.item_type = "Episode".to_string();
        episode.season_id = Some("season-9".to_string());
        episode.season_name = Some("Season 9".to_string());
        episode.series_id = Some("show-9".to_string());
        episode.series_name = Some("Show 9".to_string());
        let mut track = create_test_item("trk", "Song", Some("next-up"));
        track.item_type = "Audio".to_string();
        track.album_id = Some("alb-9".to_string());
        track.album_name = Some("Record".to_string());
        repo.save_to_cache("next-up", &[episode, track])
            .await
            .unwrap();

        let row = |id: &'static str| {
            let db_service = db_service.clone();
            async move {
                db_service
                    .query_optional(
                        Query::with_params(
                            "SELECT name, item_type, container_id FROM items WHERE id = ?",
                            vec![QueryParam::String(id.to_string())],
                        ),
                        |r| {
                            Ok((
                                r.get::<_, String>(0)?,
                                r.get::<_, String>(1)?,
                                r.get::<_, Option<String>>(2)?,
                            ))
                        },
                    )
                    .await
                    .unwrap()
            }
        };
        assert_eq!(
            row("season-9").await,
            Some(("Season 9".into(), "Season".into(), Some("show-9".into())))
        );
        assert_eq!(
            row("show-9").await,
            Some(("Show 9".into(), "Series".into(), None))
        );
        assert_eq!(
            row("alb-9").await,
            Some(("Record".into(), "MusicAlbum".into(), None))
        );
    }
}