jellytau_lib/commands/

repository.rs

//! Tauri commands for repository access
//! Uses handle-based system: UUID -> Arc<HybridRepository>
//!
//! TRACES: UR-007, UR-008, UR-023, UR-034, UR-035, UR-036 | IR-022, IR-024, JA-004, JA-005, JA-006, JA-029, JA-030, JA-031

use crate::utils::lock::MutexSafe;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use log::{debug, error, info, warn};
use serde::{Deserialize, Serialize};
use tauri::{AppHandle, Emitter, State};
use uuid::Uuid;

use crate::domain::rank_search_results;
use crate::jellyfin::HttpClient;
use crate::repository::capabilities::ServerCapabilities;
use crate::repository::{
    series_progress, types::*, HybridRepository, MediaRepository, OfflineRepository,
    OnlineRepository, StreamSelection,
};

/// Repository handle manager
pub struct RepositoryManager {
    repositories: Arc<Mutex<HashMap<String, Arc<HybridRepository>>>>,
}

impl RepositoryManager {
    pub fn new() -> Self {
        Self {
            repositories: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    pub fn create(&self, handle: String, repository: HybridRepository) {
        let mut repos = self.repositories.lock_safe();
        repos.insert(handle, Arc::new(repository));
    }

    pub fn get(&self, handle: &str) -> Option<Arc<HybridRepository>> {
        let repos = self.repositories.lock_safe();
        repos.get(handle).cloned()
    }

    /// Handles of every live repository.
    ///
    /// The background catalog indexer (DR-109) runs outside any command, so it
    /// has no handle passed in and needs to discover one. In practice there is a
    /// single signed-in repository; returning all of them avoids inventing an
    /// "active" concept the rest of the code does not have.
    ///
    /// TRACES: UR-065 | DR-109
    pub fn handles(&self) -> Vec<String> {
        let repos = self.repositories.lock_safe();
        repos.keys().cloned().collect()
    }

    pub fn destroy(&self, handle: &str) {
        let mut repos = self.repositories.lock_safe();
        repos.remove(handle);
    }
}

/// Wrapper for Tauri state
pub struct RepositoryManagerWrapper(pub RepositoryManager);

/// Read the server's reported version and resolve it into capabilities.
///
/// Never fails: a server row that is missing, or carries a version this build
/// cannot parse, yields the conservative generation rather than an error. A
/// client that refused to start because it did not recognise a version string
/// would be the exact failure UR-085 exists to remove.
///
/// TRACES: UR-085 | IR-035, DR-280
async fn server_capabilities(
    db: &Arc<crate::storage::db_service::RusqliteService>,
    server_id: &str,
) -> ServerCapabilities {
    use crate::storage::db_service::{DatabaseService, Query, QueryParam};

    let reported: Option<String> = db
        .query_one(
            Query::with_params(
                "SELECT version FROM servers WHERE id = ?1",
                vec![QueryParam::String(server_id.to_string())],
            ),
            |row| row.get::<_, Option<String>>(0),
        )
        .await
        .ok()
        .flatten();

    match reported {
        Some(version) => ServerCapabilities::from_reported(version.as_str()),
        None => {
            debug!("[REPO] No server version recorded for {server_id}; assuming current target");
            ServerCapabilities::assumed()
        }
    }
}

/// Drop the cached catalog if the server changed generation since we last looked.
///
/// Returns whether anything was invalidated, which is what the tests assert on.
///
/// The first run after this feature ships records the generation and invalidates
/// nothing: a NULL column means "never recorded", not "changed". Making the
/// absence of information trigger a full re-fetch would charge every existing
/// user bandwidth for a server upgrade that has not happened.
///
/// TRACES: UR-085 | DR-284
async fn invalidate_cache_on_generation_change(
    db: &Arc<crate::storage::db_service::RusqliteService>,
    server_id: &str,
    generation: crate::repository::capabilities::ServerGeneration,
) -> bool {
    use crate::storage::db_service::{DatabaseService, Query, QueryParam};

    let current = format!("{generation:?}");

    let previous: Option<String> = db
        .query_one(
            Query::with_params(
                "SELECT catalog_generation FROM servers WHERE id = ?1",
                vec![QueryParam::String(server_id.to_string())],
            ),
            |row| row.get::<_, Option<String>>(0),
        )
        .await
        .ok()
        .flatten();

    let changed = matches!(previous.as_deref(), Some(prev) if prev != current);

    if changed {
        warn!(
            "[REPO] Server generation changed ({:?} -> {}); dropping the cached catalog so it \
             is re-fetched under the new generation's shapes",
            previous, current
        );
        if let Err(e) = db
            .execute(Query::with_params(
                "UPDATE items SET synced_at = NULL WHERE server_id = ?1",
                vec![QueryParam::String(server_id.to_string())],
            ))
            .await
        {
            // Not fatal: stale-but-parseable rows are better than refusing to
            // start, and the next successful sync overwrites them anyway.
            error!("[REPO] Failed to invalidate cached catalog: {e}");
        }
    }

    if previous.as_deref() != Some(current.as_str()) {
        if let Err(e) = db
            .execute(Query::with_params(
                "UPDATE servers SET catalog_generation = ?1 WHERE id = ?2",
                vec![
                    QueryParam::String(current),
                    QueryParam::String(server_id.to_string()),
                ],
            ))
            .await
        {
            error!("[REPO] Failed to record server generation: {e}");
        }
    }

    changed
}

/// Create a new repository instance
/// Returns a handle (UUID) for accessing the repository
#[tauri::command]
#[specta::specta]
// Four of the eight arguments are Tauri `State<'_, _>` injections, not caller
// input. Folding the remaining four into a struct would change the IPC contract
// and the generated TypeScript for no readability gain.
#[allow(clippy::too_many_arguments)]
pub async fn repository_create(
    manager: State<'_, RepositoryManagerWrapper>,
    player: State<'_, crate::commands::player::PlayerStateWrapper>,
    db: State<'_, crate::commands::storage::DatabaseWrapper>,
    connectivity: State<'_, crate::commands::connectivity::ConnectivityMonitorWrapper>,
    server_url: String,
    user_id: String,
    access_token: String,
    server_id: String,
) -> Result<String, String> {
    info!("[REPO] repository_create called for user: {}", user_id);

    // Create HTTP client for online repository
    debug!("[REPO] Creating HTTP client...");
    let http_config = crate::jellyfin::HttpConfig::default();
    let http_client = HttpClient::new(http_config).map_err(|e| {
        error!("[REPO] HTTP client creation failed: {}", e);
        e.to_string()
    })?;
    debug!("[REPO] HTTP client created successfully");

    // Grab a connectivity reporter so the online repository's server outcomes
    // drive the reachability state the UI observes (source of truth for the
    // offline/online banner). See docs/architecture/07-connectivity.md.
    let connectivity_reporter = {
        let monitor = connectivity.0.lock().await;
        monitor.reporter()
    };

    // Create offline repository with async-safe database service
    debug!("[REPO] Creating database service...");
    let db_service = {
        let database = db.0.lock().map_err(|e| {
            error!("[REPO] Database lock failed: {}", e);
            e.to_string()
        })?;
        debug!("[REPO] Database lock acquired, getting service...");
        Arc::new(database.service())
    }; // Lock is released here
    debug!("[REPO] Database service created");

    // Resolve what this server can do, from the version it reported at connect.
    // `AuthManager::connect_to_server` already parsed it and `storage` already
    // persisted it, so this costs one indexed read and no extra round trip.
    //
    // A missing or unreadable version is not an error: `from_reported` treats it
    // as the older generation, whose request shapes also work on the newer one.
    //
    // TRACES: UR-085 | IR-035, DR-280
    let capabilities = server_capabilities(&db_service, &server_id).await;
    info!(
        "[REPO] Server generation: {:?} (reported {:?})",
        capabilities.generation,
        capabilities.version.as_ref().map(|v| v.raw.as_str())
    );

    // A server upgraded underneath us means the cached catalog was parsed under
    // a different generation's assumptions. TRACES: UR-085 | DR-284
    invalidate_cache_on_generation_change(&db_service, &server_id, capabilities.generation).await;

    // Create online repository wired to connectivity reporting
    debug!("[REPO] Creating online repository...");
    let online = OnlineRepository::new(
        Arc::new(http_client),
        server_url,
        user_id.clone(),
        access_token,
    )
    .with_connectivity(connectivity_reporter)
    .with_capabilities(capabilities);
    debug!("[REPO] Online repository created");

    debug!("[REPO] Creating offline repository...");
    let offline = OfflineRepository::new(db_service, server_id, user_id);
    debug!("[REPO] Offline repository created");

    // Create hybrid repository
    debug!("[REPO] Creating hybrid repository...");
    let hybrid = HybridRepository::new(online, offline);
    debug!("[REPO] Hybrid repository created");

    // Generate handle and store repository
    let uuid = Uuid::new_v4();
    let handle = format!("{}", uuid);
    info!("[REPO] Generated handle: {}", handle);

    // Store repository synchronously
    debug!("[REPO] Storing repository...");
    manager.0.create(handle.clone(), hybrid);
    info!("[REPO] Repository stored successfully");

    // Give the player controller a repository for next-episode lookups. The
    // Android playback-ended callback has no repository handle, so without
    // this the episode autoplay countdown never triggers there.
    if let Some(repo) = manager.0.get(&handle) {
        let controller = player.0.lock().await;
        controller.set_repository(repo);
    }

    Ok(handle)
}

/// Destroy a repository instance
#[tauri::command]
#[specta::specta]
pub async fn repository_destroy(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
) -> Result<(), String> {
    manager.0.destroy(&handle);
    Ok(())
}

/// Get libraries
#[tauri::command]
#[specta::specta]
pub async fn repository_get_libraries(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
) -> Result<Vec<Library>, String> {
    debug!("[REPO] get_libraries called with handle: {}", handle);
    let repo = manager.0.get(&handle).ok_or_else(|| {
        error!("[REPO] Repository not found for handle: {}", handle);
        "Repository not found".to_string()
    })?;
    debug!("[REPO] Repository found, fetching libraries...");
    repo.as_ref().get_libraries().await.map_err(|e| {
        error!("[REPO] Error fetching libraries: {:?}", e);
        format!("{:?}", e)
    })
}

/// Get items in a container (library, folder, album, etc.)
#[tauri::command]
#[specta::specta]
pub async fn repository_get_items(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    parent_id: String,
    options: Option<GetItemsOptions>,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_items(&parent_id, options)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get a single item by ID
#[tauri::command]
#[specta::specta]
pub async fn repository_get_item(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<MediaItem, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_item(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Downloaded-only browse: libraries that contain downloaded content.
///
/// Backs the Downloads "Downloaded" surface. Never merges server results and is
/// authoritative — an empty list means nothing is downloaded.
///
/// TRACES: UR-055 | DR-082
#[tauri::command]
#[specta::specta]
pub async fn repository_get_downloaded_libraries(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
) -> Result<Vec<Library>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.get_downloaded_libraries()
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Downloaded-only browse: items under a container that are on the device.
///
/// TRACES: UR-055 | DR-082, DR-083
#[tauri::command]
#[specta::specta]
pub async fn repository_get_downloaded_items(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    parent_id: String,
    options: Option<GetItemsOptions>,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.get_downloaded_items(&parent_id, options)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// On-disk usage of downloaded content (device total, per-item/container bytes).
///
/// TRACES: UR-056 | DR-085
#[tauri::command]
#[specta::specta]
pub async fn repository_get_download_disk_usage(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
) -> Result<DownloadDiskUsage, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.get_download_disk_usage()
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Query the optional JRay plugin for the actors on screen at time `t`
/// (seconds) in an item. Returns an empty list when JRay isn't installed or
/// has no data for the item, so the caller can render nothing without error.
#[tauri::command]
#[specta::specta]
pub async fn repository_jray_actors_at(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    t: f64,
) -> Result<Vec<crate::repository::JRayActor>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_jray_actors(&item_id, t)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get latest items in a library
#[tauri::command]
#[specta::specta]
pub async fn repository_get_latest_items(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    parent_id: String,
    limit: Option<usize>,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_latest_items(&parent_id, limit)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get resume items (continue watching/listening).
///
/// The home screen's Continue Watching row and every library's "pick up where
/// you left off" hero come through here; each item carries its own resume
/// position in `UserData`.
///
/// TRACES: UR-019, UR-023, UR-034 | IR-024, JA-013, JA-015 | DR-026, DR-038
#[tauri::command]
#[specta::specta]
pub async fn repository_get_resume_items(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    parent_id: Option<String>,
    limit: Option<usize>,
) -> Result<Vec<MediaItem>, String> {
    debug!("[REPO] get_resume_items called with handle: {}", handle);
    let repo = manager.0.get(&handle).ok_or_else(|| {
        error!("[REPO] Repository not found for handle: {}", handle);
        "Repository not found".to_string()
    })?;
    debug!("[REPO] Repository found, fetching resume items...");
    repo.as_ref()
        .get_resume_items(parent_id.as_deref(), limit)
        .await
        .map_err(|e| {
            error!("[REPO] Error fetching resume items: {:?}", e);
            format!("{:?}", e)
        })
}

/// Get next up episodes.
///
/// TRACES: UR-023, UR-034 | IR-024, JA-014 | DR-026
#[tauri::command]
#[specta::specta]
pub async fn repository_get_next_up_episodes(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    series_id: Option<String>,
    limit: Option<usize>,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_next_up_episodes(series_id.as_deref(), limit)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Every episode of a series, across all seasons, in series order.
///
/// Jellyfin hangs episodes off season folders — except for "flat" series whose
/// children are episodes directly. Both shapes are provider vocabulary, so the
/// fan-out and its fallback live in Rust rather than being reimplemented in the
/// frontend (which is what it used to do).
///
/// TRACES: UR-062 | DR-101
#[tauri::command]
#[specta::specta]
pub async fn repository_get_series_episodes(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    series_id: String,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    series_progress::fetch_series_episodes(repo.as_ref(), &series_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// The episode a viewer should land on when they open a series.
///
/// "Current" is domain policy, not layout: an episode in progress, else the
/// server's Next Up for the series, else the first unwatched episode, else the
/// first. The third rung is what makes this work offline, where Next Up is
/// always empty. Returns `None` only when the series has no episodes at all.
///
/// TRACES: UR-062 | DR-101
#[tauri::command]
#[specta::specta]
pub async fn repository_get_series_current_episode(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    series_id: String,
) -> Result<Option<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    series_progress::resolve_current_episode(repo.as_ref(), &series_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// A series' episodes and the viewer's current episode, from one season
/// fan-out. The series page used to ask for these as two commands, each of
/// which walked every season.
///
/// TRACES: UR-062 | DR-101, DR-295
#[tauri::command]
#[specta::specta]
pub async fn repository_get_series_view(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    series_id: String,
) -> Result<series_progress::SeriesView, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    series_progress::resolve_series_view(repo.as_ref(), &series_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Erase the viewer's watch history for an item.
///
/// Clears the played flag and the resume position; on a series or season the
/// server applies it to everything inside. A series cleared this way is "never
/// watched" again, so `repository_get_series_current_episode` returns its
/// premiere. Requires the server — offline this fails rather than diverging
/// local state the next sync would overwrite.
///
/// TRACES: UR-064 | DR-106
#[tauri::command]
#[specta::specta]
pub async fn repository_clear_watch_history(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<(), String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .clear_watch_history(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get recently played audio
#[tauri::command]
#[specta::specta]
pub async fn repository_get_recently_played_audio(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    limit: Option<usize>,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_recently_played_audio(limit)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get resume movies
#[tauri::command]
#[specta::specta]
pub async fn repository_get_resume_movies(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    limit: Option<usize>,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_resume_movies(limit)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get albums the user hasn't listened to recently ("rediscover")
#[tauri::command]
#[specta::specta]
pub async fn repository_get_rediscover_albums(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    parent_id: Option<String>,
    limit: Option<usize>,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_rediscover_albums(parent_id.as_deref(), limit)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get genres for a library
#[tauri::command]
#[specta::specta]
pub async fn repository_get_genres(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    parent_id: Option<String>,
) -> Result<Vec<Genre>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_genres(parent_id.as_deref())
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Tauri event name carrying the merged (cache + server) search results.
pub const SEARCH_EVENT_NAME: &str = "search-event";

/// Payload for the deferred, merged search results pushed to the frontend.
///
/// `request_id` matches the value the frontend passed to `repository_search`,
/// letting it discard updates from queries that have since been superseded.
#[derive(specta::Type, Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SearchUpdateEvent {
    pub request_id: u32,
    pub result: SearchResult,
}

/// Search for items.
///
/// Resolves `SearchOptions::scope` into concrete Jellyfin item types before
/// dispatching, so scope taxonomy stays in Rust.
///
/// TRACES: UR-049, UR-050 | DR-063
#[tauri::command]
#[specta::specta]
pub async fn repository_search(
    app: AppHandle,
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    query: String,
    options: Option<SearchOptions>,
    request_id: u32,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;

    // Expand the opaque scope into item types HERE — once, before the cache and
    // server paths diverge — so both phases filter identically. Doing it later
    // (or in only one path) makes offline results disagree with online ones.
    // The frontend sends `scope` and never names a Jellyfin item type for
    // search; see docs/specs/scoped-search-boundary.md.
    let options = options.map(|mut o| {
        o.resolve_scope();
        o
    });

    // Phase 1: instant local results from the cache (downloaded content) so the
    // UI can render immediately while the server is still being queried.
    let mut cache_result = repo
        .search_cache_only(&query, options.clone())
        .await
        .unwrap_or_else(|e| {
            debug!("[Search] Cache search miss/timeout: {:?}", e);
            SearchResult {
                items: Vec::new(),
                total_record_count: 0,
            }
        });

    // Neither backend orders by *where* the query matched, so a mid-word hit
    // ("Sparks" for "parks") can outrank a prefix hit ("Parks and Recreation").
    // Both phases are ranked with the same rules so the list does not reshuffle
    // when the server results land.
    rank_search_results(&mut cache_result.items, &query);

    // Phase 2: query the live server in the background, merge with the cache,
    // and push the union to the frontend via a `search-event`. Tagged with
    // `request_id` so the frontend can discard results from superseded queries.
    let repo_bg = repo.clone();
    let cache_for_merge = cache_result.clone();
    tauri::async_runtime::spawn(async move {
        match repo_bg.search_server_only(&query, options).await {
            Ok(server_result) => {
                let mut merged =
                    HybridRepository::merge_search_results(cache_for_merge, server_result);
                // Rank the union, not each half: a server-only prefix match must
                // be able to outrank a cached mid-word one.
                rank_search_results(&mut merged.items, &query);
                let event = SearchUpdateEvent {
                    request_id,
                    result: merged,
                };
                if let Err(e) = app.emit(SEARCH_EVENT_NAME, &event) {
                    error!("[Search] Failed to emit search update: {}", e);
                }
            }
            Err(e) => {
                // Server failed — the cache results are already on screen, so
                // just log. (Offline / unreachable server falls here.)
                warn!(
                    "[Search] Server search failed, keeping cache results: {:?}",
                    e
                );
            }
        }
    });

    Ok(cache_result)
}

/// Get playback info for an item
#[tauri::command]
#[specta::specta]
pub async fn repository_get_playback_info(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<PlaybackInfo, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_playback_info(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get a video stream URL.
///
/// There is no start-position parameter on purpose: the URL is an HLS playlist
/// covering the whole item, and a position on it makes the server reject every
/// segment with `400` (DR-181). Callers resume by seeking after load.
///
/// TRACES: UR-004 | DR-181 | UT-182
#[tauri::command]
#[specta::specta]
pub async fn repository_get_video_stream_url(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    media_source_id: Option<String>,
    audio_stream_index: Option<i32>,
) -> Result<String, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_video_stream_url(&item_id, media_source_id.as_deref(), audio_stream_index)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Decide what stream to play for a video, and describe it.
///
/// Replaces `repository_get_video_stream_url` for playback. The returned
/// [`StreamSelection`] carries the transport explicitly, so the frontend picks
/// its loader from a tagged enum instead of testing the URL for `.m3u8`; and it
/// carries the quality ladder as it applies to *this* source, so the picker can
/// stop offering rungs that produce the same bytes as Original.
///
/// No start-position parameter, for the same reason as the URL builder: a
/// position on an HLS playlist is copied onto every segment URI and the server
/// rejects each with `400` (DR-181). Callers resume by seeking after load.
///
/// TRACES: UR-070, UR-079 | DR-225, DR-227, DR-228 | UT-213
#[tauri::command]
#[specta::specta]
pub async fn repository_get_stream_selection(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    media_source_id: Option<String>,
    audio_stream_index: Option<i32>,
) -> Result<StreamSelection, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_stream_selection(&item_id, media_source_id.as_deref(), audio_stream_index)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get an audio-only stream URL for a *video* item (background-audio handoff).
///
/// TRACES: UR-040 | JA-032 | UT-061
#[tauri::command]
#[specta::specta]
pub async fn repository_get_audio_only_stream_url_for_video(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    media_source_id: Option<String>,
    start_time_seconds: Option<f64>,
    audio_stream_index: Option<i32>,
) -> Result<String, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_audio_only_stream_url_for_video(
            &item_id,
            media_source_id.as_deref(),
            start_time_seconds,
            audio_stream_index,
        )
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get audio stream URL for a track
#[tauri::command]
#[specta::specta]
pub async fn repository_get_audio_stream_url(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<String, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_audio_stream_url(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get Live TV channels (broadcast / IPTV) for browsing
#[tauri::command]
#[specta::specta]
pub async fn repository_get_live_tv_channels(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
) -> Result<Vec<MediaItem>, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_live_tv_channels()
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get the root list of plugin "Channels"
#[tauri::command]
#[specta::specta]
pub async fn repository_get_channels(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_channels()
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Open a live stream for a Live TV channel / live item
#[tauri::command]
#[specta::specta]
pub async fn repository_open_live_stream(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<LiveStreamInfo, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .open_live_stream(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Report playback start
#[tauri::command]
#[specta::specta]
pub async fn repository_report_playback_start(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    position_ms: i64,
) -> Result<(), String> {
    let position_ticks = position_ms * 10_000;
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .report_playback_start(&item_id, position_ticks)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Report playback progress
#[tauri::command]
#[specta::specta]
pub async fn repository_report_playback_progress(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    position_ms: i64,
) -> Result<(), String> {
    let position_ticks = position_ms * 10_000;
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .report_playback_progress(&item_id, position_ticks)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Report playback stopped
///
/// A stop-report that cannot reach the server is queued rather than dropped:
/// this is the position the resume point is built from, and losing it is
/// exactly the "it forgot where I was" the sync queue exists to prevent. The
/// drain (DR-131) pushes it on the next reconnect. Queueing is best-effort —
/// failing the command because the *queue* write failed would tell the caller
/// the report was lost when the local position was already saved.
///
/// TRACES: UR-025 | DR-154 | UT-151
#[tauri::command]
#[specta::specta]
pub async fn repository_report_playback_stopped(
    db: State<'_, crate::commands::storage::DatabaseWrapper>,
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    position_ms: i64,
) -> Result<(), String> {
    // Milliseconds across the boundary; the Jellyfin API wants ticks.
    let position_ticks = position_ms * 10_000;
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;

    let result = repo
        .as_ref()
        .report_playback_stopped(&item_id, position_ticks)
        .await;

    if let Err(e) = &result {
        let db_service = {
            let database = db.0.lock().map_err(|err| err.to_string())?;
            Arc::new(database.service())
        };
        let user_id = repo.user_id().to_string();
        if let Err(queue_err) = crate::commands::sync_drain::enqueue_playback_stopped(
            &db_service,
            &user_id,
            &item_id,
            position_ticks,
        )
        .await
        {
            warn!(
                "[Repository] Stop-report for {} failed ({:?}) and could not be queued: {}",
                item_id, e, queue_err
            );
        } else {
            debug!(
                "[Repository] Stop-report for {} failed ({:?}); queued for the next reconnect",
                item_id, e
            );
        }
    }

    result.map_err(|e| format!("{:?}", e))
}

/// Get image URL for an item
#[tauri::command]
#[specta::specta]
pub fn repository_get_image_url(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    image_type: ImageType,
    options: Option<ImageOptions>,
) -> Result<String, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    Ok(repo.as_ref().get_image_url(&item_id, image_type, options))
}

/// Get subtitle URL for a media item
#[tauri::command]
#[specta::specta]
#[allow(dead_code)]
pub fn repository_get_subtitle_url(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    media_source_id: String,
    stream_index: i32,
    format: String,
) -> Result<String, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    Ok(repo
        .as_ref()
        .get_subtitle_url(&item_id, &media_source_id, stream_index, &format))
}

/// Get video download URL with quality preset
#[tauri::command]
#[specta::specta]
#[allow(dead_code)]
pub async fn repository_get_video_download_url(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    quality: String,
    media_source_id: Option<String>,
) -> Result<String, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    // Async because the audio-codec policy has to know what the source's audio
    // is before it can decide whether the file may be copied verbatim (DR-171).
    // The frontend calls this exactly as before — the decision stays in Rust.
    Ok(crate::repository::resolve_video_download_url(
        repo.as_ref(),
        &item_id,
        &quality,
        media_source_id.as_deref(),
    )
    .await)
}

/// Mark an item as favorite
#[tauri::command]
#[specta::specta]
pub async fn repository_mark_favorite(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<(), String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .mark_favorite(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Tauri event announcing that favourite state changed behind the UI's back —
/// either because the server disagreed with the cache on a background refresh,
/// or because pending offline toggles were pushed on reconnect.
///
/// TRACES: UR-069 | DR-120
pub const FAVORITES_CHANGED_EVENT: &str = "favorites-changed";

/// Payload for [`FAVORITES_CHANGED_EVENT`] — the ids whose favourite state
/// actually flipped, so the frontend refreshes those rather than everything.
///
/// TRACES: UR-069 | DR-120 | UT-107
#[derive(specta::Type, Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FavoritesChangedEvent {
    pub item_ids: Vec<String>,
}

/// Ids whose favourite state differs between what we showed and what the server
/// has — favourited elsewhere since the cache was written, or un-favourited
/// elsewhere.
///
/// Pulled out of the command so the "emit nothing when nothing changed" rule is
/// testable: an unchanged set must leave a quiet page quiet rather than
/// triggering a refetch on every visit.
///
/// TRACES: UR-069 | DR-120 | UT-107
fn changed_favorite_ids(
    cached: &std::collections::HashSet<String>,
    server: &std::collections::HashSet<String>,
) -> Vec<String> {
    let mut changed: Vec<String> = server.symmetric_difference(cached).cloned().collect();
    // Deterministic order so the event payload does not depend on hash seeding.
    changed.sort();
    changed
}

/// Everything the viewer has favourited, across libraries, narrowed by scope.
///
/// Two-phase like `repository_search`: the local answer returns immediately and
/// a background server pass emits `favorites-changed` when the server's set
/// differs. Without the second phase a favourite marked in another client shows
/// up only on the *second* visit to the page, since the cache-first read hands
/// back local rows and the refresh is invisible to the frontend.
///
/// TRACES: UR-067 | DR-115, DR-120, JA-033 | UT-107
#[tauri::command]
#[specta::specta]
pub async fn repository_get_favorites(
    app: AppHandle,
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    scope: SearchScope,
    options: Option<GetItemsOptions>,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;

    let cache_result = repo
        .get_favorites_cache_only(scope, options.clone())
        .await
        .unwrap_or_else(|e| {
            debug!("[Favorites] Cache miss/timeout: {:?}", e);
            SearchResult {
                items: Vec::new(),
                total_record_count: 0,
            }
        });

    // With "Show all server media" off the local answer is authoritative
    // (DR-080) — don't go behind the user's back to the server.
    if !crate::repository::offline::include_catalog_browse() {
        return Ok(cache_result);
    }

    // Nothing cached yet — a fresh install, or a viewer whose favourites were
    // all marked on another client. Returning the empty result here paints
    // "Nothing favourited yet — tap the heart on anything you like", which is a
    // *wrong* answer, corrected a server round trip later when the background
    // refresh fires `favorites-changed`. Ask the repository for a real answer
    // instead: its `get_favorites` is exactly this read — cache first, server on
    // a miss, saving through — and it applies the same DR-080 gate.
    //
    // TRACES: UR-067 | DR-115
    if !cache_result.has_content() {
        debug!("[Favorites] Nothing cached; answering from the server");
        return repo
            .get_favorites(scope, options)
            .await
            .map_err(|e| format!("{:?}", e));
    }

    let repo_bg = repo.clone();
    let cached_ids: std::collections::HashSet<String> =
        cache_result.items.iter().map(|i| i.id.clone()).collect();
    tauri::async_runtime::spawn(async move {
        match repo_bg.get_favorites_server_only(scope, options).await {
            Ok(server_result) => {
                let server_ids: std::collections::HashSet<String> =
                    server_result.items.iter().map(|i| i.id.clone()).collect();
                let changed = changed_favorite_ids(&cached_ids, &server_ids);

                if !changed.is_empty() {
                    let event = FavoritesChangedEvent { item_ids: changed };
                    if let Err(e) = app.emit(FAVORITES_CHANGED_EVENT, &event) {
                        error!("[Favorites] Failed to emit change event: {}", e);
                    }
                }
            }
            Err(e) => {
                warn!(
                    "[Favorites] Server refresh failed, keeping cached favourites: {:?}",
                    e
                );
            }
        }
    });

    Ok(cache_result)
}

/// Unmark an item as favorite
#[tauri::command]
#[specta::specta]
pub async fn repository_unmark_favorite(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
) -> Result<(), String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .unmark_favorite(&item_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get person details
#[tauri::command]
#[specta::specta]
pub async fn repository_get_person(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    person_id: String,
) -> Result<MediaItem, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_person(&person_id)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get items by person (actor, director, etc.)
#[tauri::command]
#[specta::specta]
pub async fn repository_get_items_by_person(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    person_id: String,
    options: Option<GetItemsOptions>,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_items_by_person(&person_id, options)
        .await
        .map_err(|e| format!("{:?}", e))
}

/// Get similar/related items for a media item
#[tauri::command]
#[specta::specta]
pub async fn repository_get_similar_items(
    manager: State<'_, RepositoryManagerWrapper>,
    handle: String,
    item_id: String,
    limit: Option<usize>,
) -> Result<SearchResult, String> {
    let repo = manager.0.get(&handle).ok_or("Repository not found")?;
    repo.as_ref()
        .get_similar_items(&item_id, limit)
        .await
        .map_err(|e| format!("{:?}", e))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_repository_manager_creation() {
        let manager = RepositoryManager::new();
        // A freshly created manager holds no repositories
        assert!(manager.get("any-handle").is_none());
    }

    fn ids(values: &[&str]) -> std::collections::HashSet<String> {
        values.iter().map(|v| v.to_string()).collect()
    }

    /// UT-107 — the background refresh reports only what actually changed.
    ///
    /// TRACES: UR-069 | DR-120 | UT-107
    #[test]
    fn test_changed_favorite_ids_reports_both_directions() {
        // Favourited in another client since we cached.
        assert_eq!(
            changed_favorite_ids(&ids(&["a"]), &ids(&["a", "b"])),
            vec!["b".to_string()]
        );

        // Un-favourited in another client.
        assert_eq!(
            changed_favorite_ids(&ids(&["a", "b"]), &ids(&["a"])),
            vec!["b".to_string()]
        );

        // Both at once, in a stable order.
        assert_eq!(
            changed_favorite_ids(&ids(&["a", "b"]), &ids(&["b", "c"])),
            vec!["a".to_string(), "c".to_string()]
        );
    }

    /// An unchanged set emits nothing — otherwise every visit to the page would
    /// fire an event and trigger a pointless refetch.
    ///
    /// TRACES: UR-069 | DR-120 | UT-107
    #[test]
    fn test_changed_favorite_ids_is_empty_when_nothing_moved() {
        assert!(changed_favorite_ids(&ids(&["a", "b"]), &ids(&["b", "a"])).is_empty());
        assert!(changed_favorite_ids(&ids(&[]), &ids(&[])).is_empty());
    }

    #[test]
    fn test_repository_manager_wrapper_structure() {
        let manager = RepositoryManager::new();
        let wrapper = RepositoryManagerWrapper(manager);
        // The wrapper exposes the underlying manager, which starts empty
        assert!(wrapper.0.get("any-handle").is_none());
    }

    #[test]
    fn test_repository_manager_get_nonexistent() {
        let manager = RepositoryManager::new();
        // Getting a non-existent repository should return None
        let result = manager.get("nonexistent-handle");
        assert!(result.is_none());
    }

    #[test]
    fn test_uuid_handle_generation() {
        let uuid = Uuid::new_v4();
        let handle = format!("{}", uuid);
        // UUID should convert to a non-empty string
        assert!(!handle.is_empty());
    }

    #[test]
    fn test_uuid_handles_are_unique() {
        let handle1 = format!("{}", Uuid::new_v4());
        let handle2 = format!("{}", Uuid::new_v4());
        // Two generated UUIDs should be different
        assert_ne!(handle1, handle2);
    }

    #[test]
    fn test_uuid_handle_format() {
        let uuid = Uuid::new_v4();
        let handle = format!("{}", uuid);
        // UUID should have standard format with hyphens
        let parts: Vec<&str> = handle.split('-').collect();
        assert_eq!(parts.len(), 5);
    }

    #[test]
    fn test_repository_manager_destroy_nonexistent() {
        let manager = RepositoryManager::new();
        // Destroying a non-existent repository should not panic
        manager.destroy("nonexistent-handle");
    }

    #[test]
    fn test_repository_manager_is_send_sync() {
        // Verify RepositoryManager can be used in async contexts
        fn is_send_sync<T: Send + Sync>() {}
        is_send_sync::<RepositoryManager>();
    }

    #[test]
    fn test_repository_manager_wrapper_is_send_sync() {
        // Verify RepositoryManagerWrapper is Send + Sync
        fn is_send_sync<T: Send + Sync>() {}
        is_send_sync::<RepositoryManagerWrapper>();
    }

    #[test]
    fn test_multiple_manager_instances() {
        let manager1 = RepositoryManager::new();
        let manager2 = RepositoryManager::new();

        // Multiple manager instances should be independent
        let handle1_nonexistent = manager1.get("test");
        let handle2_nonexistent = manager2.get("test");

        assert!(handle1_nonexistent.is_none());
        assert!(handle2_nonexistent.is_none());
    }

    #[test]
    fn test_handle_string_properties() {
        let uuid = Uuid::new_v4();
        let handle = format!("{}", uuid);

        // Handle should be alphanumeric with hyphens
        for c in handle.chars() {
            assert!(c.is_alphanumeric() || c == '-');
        }
    }

    #[test]
    fn test_repository_manager_concurrent_access() {
        let manager = Arc::new(RepositoryManager::new());
        let mut handles = vec![];

        // Verify manager can be wrapped in Arc for concurrent access
        for _ in 0..3 {
            let mgr = Arc::clone(&manager);
            let handle = std::thread::spawn(move || {
                let result = mgr.get("test");
                assert!(result.is_none());
            });
            handles.push(handle);
        }

        for h in handles {
            h.join().unwrap();
        }
    }
}

#[cfg(test)]
mod generation_change_tests {
    use super::*;
    use crate::repository::capabilities::ServerGeneration;
    use crate::storage::db_service::{DatabaseService, Query, QueryParam, RusqliteService};

    async fn db_with_server() -> Arc<RusqliteService> {
        let conn = rusqlite::Connection::open_in_memory().expect("in-memory db");
        for (_, sql) in crate::storage::schema::MIGRATIONS {
            conn.execute_batch(sql).expect("migration");
        }
        let db = Arc::new(RusqliteService::new(Arc::new(std::sync::Mutex::new(conn))));
        db.execute(Query::with_params(
            "INSERT INTO servers (id, name, url, version) VALUES (?1, ?2, ?3, ?4)",
            vec![
                QueryParam::String("srv-1".into()),
                QueryParam::String("Home".into()),
                QueryParam::String("https://example.test".into()),
                QueryParam::String("10.11.5".into()),
            ],
        ))
        .await
        .expect("seed server");
        db
    }

    async fn recorded(db: &Arc<RusqliteService>) -> Option<String> {
        db.query_one(
            Query::new("SELECT catalog_generation FROM servers WHERE id = 'srv-1'"),
            |row| row.get::<_, Option<String>>(0),
        )
        .await
        .ok()
        .flatten()
    }

    /// The first look records the generation and invalidates nothing. A NULL
    /// column means "never recorded", not "changed" — treating it as a change
    /// would charge every existing user a full re-fetch on upgrade.
    ///
    /// TRACES: UR-085 | DR-284
    #[tokio::test]
    async fn the_first_look_records_without_invalidating() {
        let db = db_with_server().await;
        let invalidated =
            invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;

        assert!(!invalidated, "a first sighting is not a change");
        assert_eq!(recorded(&db).await.as_deref(), Some("V10_11"));
    }

    /// Seeing the same generation again is not a change either.
    ///
    /// TRACES: UR-085 | DR-284
    #[tokio::test]
    async fn an_unchanged_generation_does_not_invalidate() {
        let db = db_with_server().await;
        invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;
        let invalidated =
            invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;

        assert!(!invalidated);
        assert_eq!(recorded(&db).await.as_deref(), Some("V10_11"));
    }

    /// An actual upgrade drops the cached catalog and records the new
    /// generation, so the next browse re-fetches under the new shapes.
    ///
    /// TRACES: UR-085 | DR-284
    #[tokio::test]
    async fn a_real_upgrade_invalidates_and_records() {
        let db = db_with_server().await;
        invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V10_11).await;

        let invalidated =
            invalidate_cache_on_generation_change(&db, "srv-1", ServerGeneration::V12Plus).await;

        assert!(invalidated, "10.11 -> 12.x is a generation change");
        assert_eq!(recorded(&db).await.as_deref(), Some("V12Plus"));
    }

    /// A server row that is missing entirely must not panic or invalidate.
    ///
    /// TRACES: UR-085 | DR-284
    #[tokio::test]
    async fn an_unknown_server_is_harmless() {
        let db = db_with_server().await;
        let invalidated =
            invalidate_cache_on_generation_change(&db, "no-such-server", ServerGeneration::V12Plus)
                .await;
        assert!(!invalidated);
    }
}