jellytau_lib/commands/

catalog.rs

//! Tauri commands for the offline "browse & queue" feature.
//!
//! TRACES: UR-002, UR-007, UR-024 | JA-004, JA-016 | DR-012, DR-027
//!
//! Two backend pieces support browsing the full server catalog while offline
//! and queueing downloads that fire on reconnect:
//!
//! - [`sync_full_catalog`] walks every library while online and persists all
//!   items to the offline cache so the whole catalog is browsable (greyed out)
//!   offline. It reuses [`HybridRepository::cache_items_from_server`], which in
//!   turn reuses `OfflineRepository::save_to_cache` (sets `synced_at`, which is
//!   what `get_items` branch 3 serves offline).
//! - [`resume_queued_downloads`] resolves and pumps the `pending` download rows
//!   that were queued offline (they have `stream_url IS NULL`), mirroring the
//!   heal-and-pump pattern in `player_preload_upcoming`.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use log::{info, warn};
use tauri::{Emitter, Manager, State};

use crate::commands::download::{pump_download_queue, DownloadManagerWrapper};
use crate::commands::repository::RepositoryManagerWrapper;
use crate::commands::storage::DatabaseWrapper;
use crate::repository::types::GetItemsOptions;
use crate::storage::db_service::{DatabaseService, Query, QueryParam};

/// app_settings key holding the RFC-3339 timestamp of the last successful
/// full-catalog sync.
const LAST_CATALOG_SYNC_KEY: &str = "last_catalog_sync";

/// How long an index stays fresh before a re-index is due.
///
/// This lives in Rust rather than being a frontend constant because it decides
/// *whether the local cache is authoritative* — the same class of decision as
/// `include_catalog_browse`, and squarely the "sync policy" the spec review
/// checklist keeps out of the presentation layer. If it later becomes
/// user-configurable it stays a Rust-owned setting edited through a command.
const CATALOG_INDEX_TTL: Duration = Duration::from_secs(6 * 60 * 60);

/// How often the scheduler wakes to *check* staleness. Far shorter than the TTL
/// because a tick is nearly free — one indexed `app_settings` lookup — and it is
/// what makes the indexer responsive to events it cannot subscribe to: signing
/// in, and coming back online. The TTL, not the tick, decides whether a crawl
/// actually happens.
const CATALOG_INDEX_TICK: Duration = Duration::from_secs(5 * 60);

/// Delay before the first staleness check, to let sign-in complete and the
/// repository be registered. Without it the first check runs against an empty
/// repository manager and a fresh install would sit unindexed until the next
/// tick.
const CATALOG_INDEX_FIRST_CHECK: Duration = Duration::from_secs(15);

/// Kebab-case, per the project's event convention.
pub const CATALOG_INDEX_EVENT: &str = "catalog-index-event";

/// Guards against two passes running at once. Replaces the frontend's
/// `syncInProgress` boolean in `offlineCatalog.ts`, which could not see a pass
/// started by the scheduler.
static INDEX_IN_PROGRESS: AtomicBool = AtomicBool::new(false);

/// Clears [`INDEX_IN_PROGRESS`] however the pass leaves — including on the `?`
/// early return when `get_libraries` fails, which a plain store at the end of
/// the function would leak.
struct IndexPassGuard;

impl Drop for IndexPassGuard {
    fn drop(&mut self) {
        INDEX_IN_PROGRESS.store(false, Ordering::SeqCst);
    }
}

/// Progress of a background index pass, for the staleness hint in the UI.
#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CatalogIndexEvent {
    /// `started` | `finished` | `failed`
    pub state: String,
    pub items_cached: usize,
    pub items_pruned: usize,
    pub libraries_failed: usize,
    /// Present on `failed`.
    pub error: Option<String>,
}

/// Item types worth caching for offline browsing: containers the library
/// landing pages render plus the playable leaves users queue for download.
/// `MusicArtist` and `Playlist` are here because search groups results by them
/// (UR-060's Artists group). Without them in the crawl, the local index can
/// never answer an artist query and those groups can only ever be filled by the
/// server leg. Keep this in step with what `prune_stale_catalog` is allowed to
/// sweep — the crawl is only authoritative for the types it asks for.
///
/// TRACES: UR-065, UR-060 | DR-111
const CATALOG_ITEM_TYPES: &[&str] = &[
    "MusicAlbum",
    "MusicArtist",
    "Movie",
    "Series",
    "Season",
    "Episode",
    "Audio",
    "BoxSet",
    "Playlist",
];

/// Jellyfin item types whose download is a *video* stream rather than an audio
/// one. The download queue stores an opaque `media_type` ('audio'/'video'); this
/// is where the taxonomy that produces it lives, so the frontend never has to
/// know which item types are video.
///
/// TRACES: UR-071 | DR-135
const VIDEO_ITEM_TYPES: &[&str] = &["Movie", "Episode", "Video", "MusicVideo"];

#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CatalogSyncResult {
    /// Total items persisted to the offline cache across all libraries.
    pub items_cached: usize,
    /// Libraries that failed to sync (e.g. server hiccup); best-effort.
    pub libraries_failed: usize,
    /// Entries removed because the server no longer has them. Always 0 when any
    /// library failed, since a partial crawl cannot prove an item is gone.
    pub items_pruned: usize,
}

#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct CatalogSyncStatus {
    /// RFC-3339 timestamp of the last successful sync, if any.
    pub last_synced_at: Option<String>,
}

/// Walk every library on the server and persist all items to the offline cache
/// so the full catalog is browsable offline (greyed out when not downloaded).
///
/// Best-effort: a library that fails to fetch is counted and skipped rather than
/// aborting the whole sync. Runs libraries sequentially to avoid hammering the
/// server. Uses `Recursive=true` so a single request per library returns the
/// containers and their playable children.
#[tauri::command]
#[specta::specta]
pub async fn sync_full_catalog(
    repository: State<'_, RepositoryManagerWrapper>,
    db: State<'_, DatabaseWrapper>,
    handle: String,
) -> Result<CatalogSyncResult, String> {
    let repo = repository.0.get(&handle).ok_or("Repository not found")?;

    let db_service = {
        let database = db.0.lock().map_err(|e| e.to_string())?;
        Arc::new(database.service())
    };

    run_index_pass(repo, db_service).await
}

/// One full-catalog indexing pass, shared by the [`sync_full_catalog`] command
/// and the background scheduler (DR-109) so there is exactly one implementation
/// and one concurrency guard.
///
/// TRACES: UR-065 | DR-109, DR-110
pub(crate) async fn run_index_pass(
    repo: Arc<crate::repository::HybridRepository>,
    db_service: Arc<crate::storage::db_service::RusqliteService>,
) -> Result<CatalogSyncResult, String> {
    use crate::repository::MediaRepository;

    // One pass at a time. The command and the scheduler can both land here, and
    // two concurrent crawls would double the server load and race on the sweep.
    if INDEX_IN_PROGRESS.swap(true, Ordering::SeqCst) {
        return Err("A catalog index pass is already running".to_string());
    }
    let _guard = IndexPassGuard;

    let libraries = repo.get_libraries().await.map_err(|e| e.to_string())?;
    info!(
        "[Catalog] Full sync starting across {} libraries",
        libraries.len()
    );

    let include_types: Vec<String> = CATALOG_ITEM_TYPES.iter().map(|s| s.to_string()).collect();

    // Taken before the crawl: every row the crawl writes gets a `synced_at`
    // newer than this, so anything still older afterwards is gone server-side.
    let pass_started_at = chrono::Utc::now().to_rfc3339();

    let mut items_cached = 0usize;
    let mut libraries_failed = 0usize;

    for library in &libraries {
        let opts = GetItemsOptions {
            recursive: Some(true),
            include_item_types: Some(include_types.clone()),
            limit: Some(100_000),
            ..Default::default()
        };

        match repo.cache_items_from_server(&library.id, Some(opts)).await {
            Ok(items) => {
                info!(
                    "[Catalog] Cached {} items from library '{}'",
                    items.len(),
                    library.name
                );
                items_cached += items.len();
            }
            Err(e) => {
                warn!(
                    "[Catalog] Failed to sync library '{}': {:?}",
                    library.name, e
                );
                libraries_failed += 1;
            }
        }
    }

    // Propagate server-side deletions — but only after a *complete* crawl.
    // `sync_full_catalog` is best-effort per library, and `items.parent_id` is
    // ON DELETE CASCADE, so sweeping when a library failed to fetch could
    // cascade a whole series away because one request timed out.
    let mut items_pruned = 0usize;
    if libraries_failed == 0 && !libraries.is_empty() {
        match repo
            .prune_stale_catalog(&pass_started_at, &include_types)
            .await
        {
            Ok(removed) => {
                items_pruned = removed;
                if removed > 0 {
                    info!(
                        "[Catalog] Pruned {} entries no longer on the server",
                        removed
                    );
                }
            }
            Err(e) => warn!("[Catalog] Prune of stale catalog entries failed: {:?}", e),
        }
    } else if libraries_failed > 0 {
        info!(
            "[Catalog] Skipping stale-entry prune: {} librar{} failed to sync, so the crawl is not authoritative",
            libraries_failed,
            if libraries_failed == 1 { "y" } else { "ies" }
        );
    }

    // Record the sync time so callers can skip re-syncing too eagerly.
    let now = chrono::Utc::now().to_rfc3339();
    let upsert = Query::with_params(
        "INSERT INTO app_settings (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)
         ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP",
        vec![
            QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string()),
            QueryParam::String(now),
        ],
    );
    if let Err(e) = db_service.execute(upsert).await {
        warn!("[Catalog] Failed to persist last-sync timestamp: {}", e);
    }

    info!(
        "[Catalog] Full sync complete: {} items cached, {} pruned, {} libraries failed",
        items_cached, items_pruned, libraries_failed
    );

    Ok(CatalogSyncResult {
        items_cached,
        libraries_failed,
        items_pruned,
    })
}

/// Whether an index pass is due, given when one last completed.
///
/// Pure so the policy is unit-testable without a clock, a server, or a database.
/// `None` (never indexed) and an unparseable stored value both mean "due" — a
/// corrupt timestamp should trigger a re-index, not silently freeze the catalog.
///
/// TRACES: UR-065 | DR-109 | UT-115
pub(crate) fn index_is_due(
    last_synced_at: Option<&str>,
    now: chrono::DateTime<chrono::Utc>,
    ttl: Duration,
) -> bool {
    let Some(raw) = last_synced_at else {
        return true;
    };
    let Ok(last) = chrono::DateTime::parse_from_rfc3339(raw) else {
        return true;
    };
    now.signed_duration_since(last.with_timezone(&chrono::Utc))
        .to_std()
        .map(|elapsed| elapsed >= ttl)
        // Negative elapsed => the stored stamp is in the future (clock skew).
        // Not due; a future stamp will age into due-ness on its own.
        .unwrap_or(false)
}

/// Read the last-sync timestamp straight from `app_settings`.
async fn read_last_sync(
    db_service: &Arc<crate::storage::db_service::RusqliteService>,
) -> Option<String> {
    db_service
        .query_optional(
            Query::with_params(
                "SELECT value FROM app_settings WHERE key = ?",
                vec![QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string())],
            ),
            |row| row.get(0),
        )
        .await
        .ok()
        .flatten()
}

/// Start the background catalog indexer.
///
/// Replaces the frontend's startup-only `syncCatalog()` call: index freshness is
/// sync policy and belongs in Rust (see the layer assignment in
/// docs/architecture/03-data-flow.md, "Search Flow"). Ticks every
/// [`CATALOG_INDEX_TICK`] and
/// runs a pass when a repository exists, the server is reachable, and the index
/// is older than [`CATALOG_INDEX_TTL`].
///
/// TRACES: UR-065 | DR-109, IR-030
pub fn spawn_catalog_indexer(app: tauri::AppHandle) {
    tauri::async_runtime::spawn(async move {
        // Check shortly after launch, then on every tick — not tick-then-check,
        // which would leave a fresh install unindexed for a full tick.
        tokio::time::sleep(CATALOG_INDEX_FIRST_CHECK).await;

        loop {
            if let Err(e) = maybe_run_scheduled_pass(&app).await {
                // Never fatal — a failed pass leaves the existing index in place
                // and we retry on the next tick.
                warn!("[Catalog] Scheduled index pass skipped: {}", e);
            }

            tokio::time::sleep(CATALOG_INDEX_TICK).await;
        }
    });
}

/// One scheduler tick: check the preconditions, then index if due.
async fn maybe_run_scheduled_pass(app: &tauri::AppHandle) -> Result<(), String> {
    if INDEX_IN_PROGRESS.load(Ordering::SeqCst) {
        return Ok(());
    }

    let db_service = {
        let db = app.state::<DatabaseWrapper>();
        let database = db.0.lock().map_err(|e| e.to_string())?;
        Arc::new(database.service())
    };

    if !index_is_due(
        read_last_sync(&db_service).await.as_deref(),
        chrono::Utc::now(),
        CATALOG_INDEX_TTL,
    ) {
        return Ok(());
    }

    // Offline: leave the index alone. The crawl would fail every library and,
    // more importantly, a partial crawl must never reach the deletion sweep.
    {
        let monitor = app.state::<crate::commands::connectivity::ConnectivityMonitorWrapper>();
        let monitor = monitor.0.lock().await;
        if !monitor.get_status().await.is_server_reachable {
            return Ok(());
        }
    }

    let repo = {
        let manager = app.state::<RepositoryManagerWrapper>();
        let handles = manager.0.handles();
        let Some(handle) = handles.first() else {
            // Not signed in yet.
            return Ok(());
        };
        manager.0.get(handle).ok_or("Repository not found")?
    };

    info!("[Catalog] Index is stale; starting a scheduled pass");
    let _ = app.emit(
        CATALOG_INDEX_EVENT,
        CatalogIndexEvent {
            state: "started".to_string(),
            items_cached: 0,
            items_pruned: 0,
            libraries_failed: 0,
            error: None,
        },
    );

    match run_index_pass(repo, db_service).await {
        Ok(result) => {
            let _ = app.emit(
                CATALOG_INDEX_EVENT,
                CatalogIndexEvent {
                    state: "finished".to_string(),
                    items_cached: result.items_cached,
                    items_pruned: result.items_pruned,
                    libraries_failed: result.libraries_failed,
                    error: None,
                },
            );
            Ok(())
        }
        Err(e) => {
            let _ = app.emit(
                CATALOG_INDEX_EVENT,
                CatalogIndexEvent {
                    state: "failed".to_string(),
                    items_cached: 0,
                    items_pruned: 0,
                    libraries_failed: 0,
                    error: Some(e.clone()),
                },
            );
            Err(e)
        }
    }
}

/// Report the last-synced timestamp so the UI can show a hint / decide whether
/// to trigger a fresh sync.
#[tauri::command]
#[specta::specta]
pub async fn catalog_sync_status(
    db: State<'_, DatabaseWrapper>,
) -> Result<CatalogSyncStatus, String> {
    let db_service = {
        let database = db.0.lock().map_err(|e| e.to_string())?;
        Arc::new(database.service())
    };

    let query = Query::with_params(
        "SELECT value FROM app_settings WHERE key = ?",
        vec![QueryParam::String(LAST_CATALOG_SYNC_KEY.to_string())],
    );
    let last_synced_at: Option<String> = db_service
        .query_optional(query, |row| row.get(0))
        .await
        .map_err(|e| e.to_string())?;

    Ok(CatalogSyncStatus { last_synced_at })
}

/// Control whether offline library queries reveal the full synced catalog
/// (greyed-out, non-downloaded media) or only downloaded/local media.
///
/// The frontend calls this from the "Show all server media" toggle: pass `true`
/// when online, or when offline with the toggle on; pass `false` when offline
/// with the toggle off so library pages show downloaded media only. Fixes the
/// bug where offline library pages showed every server item regardless of the
/// toggle.
#[tauri::command]
#[specta::specta]
pub fn set_show_server_catalog(show: bool) {
    crate::repository::offline::set_include_catalog_browse(show);
}

#[derive(specta::Type, Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ResumeQueuedResult {
    /// Rows whose stream URL was resolved and are now pump-eligible.
    pub resolved: usize,
    /// Rows that couldn't be resolved (item metadata / URL lookup failed).
    pub failed: usize,
}

/// Requeue video downloads that were fetched as audio.
///
/// Before [`resolve_pending_download_urls`] consulted the item's type, a row
/// with no `media_type` — which is every row queued from a media card, since
/// `download_item` does not record one — resolved against
/// `get_audio_stream_url`. A movie queued that way completed with an audio-only
/// transcode on disk, so playing it offline could only ever fail. Those rows are
/// identifiable after the fact (no `media_type`, but a video item), so reset them
/// to pending with no URL and let the resolver fetch the real video.
///
/// Rows carrying an explicit `media_type` were resolved correctly and are left
/// alone, as are genuine audio downloads.
///
/// Returns the number of rows requeued.
///
/// TRACES: UR-071 | DR-136 | UT-126
pub(crate) async fn requeue_mistyped_video_downloads(
    db_service: &Arc<crate::storage::db_service::RusqliteService>,
) -> Result<usize, String> {
    let video_types = VIDEO_ITEM_TYPES
        .iter()
        .map(|t| format!("'{t}'"))
        .collect::<Vec<_>>()
        .join(", ");

    let query = Query::new(format!(
        "UPDATE downloads
         SET status = 'pending', stream_url = NULL, progress = 0,
             bytes_downloaded = 0, started_at = NULL, completed_at = NULL
         WHERE media_type IS NULL
           AND status = 'completed'
           AND item_id IN (SELECT id FROM items WHERE item_type IN ({video_types}))"
    ));

    let n = db_service.execute(query).await.map_err(|e| e.to_string())? as usize;

    if n > 0 {
        info!(
            "[Catalog] Requeued {} video download(s) that were fetched as audio",
            n
        );
    }
    Ok(n)
}

/// What a resolver hands back for one queued row: the URL to fetch and, for a
/// video, the size predicted for it (see `download::estimate`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct ResolvedDownloadUrl {
    pub url: String,
    pub expected_bytes: Option<u64>,
}

impl From<String> for ResolvedDownloadUrl {
    /// An audio stream URL: served static, so the response states its own
    /// length and nothing needs predicting.
    fn from(url: String) -> Self {
        Self {
            url,
            expected_bytes: None,
        }
    }
}

impl From<crate::repository::ResolvedVideoDownload> for ResolvedDownloadUrl {
    fn from(r: crate::repository::ResolvedVideoDownload) -> Self {
        Self {
            url: r.url,
            expected_bytes: r.expected_bytes,
        }
    }
}

/// Core of [`resume_queued_downloads`], factored out for testing: select every
/// `pending`/`stream_url IS NULL` row, resolve each via `resolve` (returning
/// `None` leaves the row pending), and heal the row so the pump can start it.
/// The `resolve` closure receives `(item_id, media_type, quality_preset)`.
///
/// `only_ids` restricts the sweep to specific download rows. Reconnect passes
/// `None` and heals everything; a bulk enqueue (an album, say) passes the rows
/// it just created, so clicking download on one album cannot also start every
/// unrelated row that has been sitting pending.
pub(crate) async fn resolve_pending_download_urls<F, Fut>(
    db_service: &Arc<crate::storage::db_service::RusqliteService>,
    target_dir: &str,
    only_ids: Option<&[i64]>,
    resolve: F,
) -> Result<ResumeQueuedResult, String>
where
    F: Fn(String, String, String) -> Fut,
    Fut: std::future::Future<Output = Option<ResolvedDownloadUrl>>,
{
    if only_ids.is_some_and(|ids| ids.is_empty()) {
        return Ok(ResumeQueuedResult {
            resolved: 0,
            failed: 0,
        });
    }
    // A row's own media_type wins; otherwise the *item's* type decides. Rows
    // queued from a media card never carry one (`download_item` does not record
    // it), and defaulting that NULL to 'audio' resolved movies against
    // `get_audio_stream_url` — the file on disk was an audio-only transcode, so
    // offline video could never play. Falling back to 'audio' only when the item
    // is unknown keeps the historical behaviour for uncached items.
    // TRACES: UR-071, UR-052 | DR-135
    let video_types = VIDEO_ITEM_TYPES
        .iter()
        .map(|t| format!("'{t}'"))
        .collect::<Vec<_>>()
        .join(", ");
    let id_filter = match only_ids {
        Some(ids) => format!(
            " AND d.id IN ({})",
            ids.iter()
                .map(|id| id.to_string())
                .collect::<Vec<_>>()
                .join(", ")
        ),
        None => String::new(),
    };
    let rows_query = Query::new(format!(
        "SELECT d.id, d.item_id,
                COALESCE(
                    d.media_type,
                    CASE WHEN i.item_type IN ({video_types}) THEN 'video'
                         WHEN i.item_type IS NOT NULL THEN 'audio'
                    END,
                    'audio'),
                COALESCE(d.quality_preset, 'original')
         FROM downloads d
         LEFT JOIN items i ON i.id = d.item_id
         WHERE d.status = 'pending' AND d.stream_url IS NULL{id_filter}"
    ));
    let rows: Vec<(i64, String, String, String)> = db_service
        .query_many(rows_query, |row| {
            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
        })
        .await
        .map_err(|e| e.to_string())?;

    if rows.is_empty() {
        return Ok(ResumeQueuedResult {
            resolved: 0,
            failed: 0,
        });
    }

    info!(
        "[Catalog] Resolving {} offline-queued downloads on reconnect",
        rows.len()
    );

    let mut resolved = 0usize;
    let mut failed = 0usize;

    for (download_id, item_id, media_type, quality) in rows {
        let target = match resolve(item_id.clone(), media_type, quality).await {
            Some(target) => target,
            None => {
                failed += 1;
                continue;
            }
        };

        // Heal the row so the pump can start it. Guard on stream_url IS NULL so a
        // concurrent resolver doesn't clobber an already-started row. The
        // predicted size, when there is one, gives the worker a progress total
        // for a response that carries none (DR-290).
        let expected = target
            .expected_bytes
            .and_then(|n| i64::try_from(n).ok())
            .map_or(QueryParam::Null, QueryParam::Int64);
        let update = Query::with_params(
            "UPDATE downloads SET stream_url = ?, target_dir = ?,
                 file_size = COALESCE(?, file_size)
             WHERE id = ? AND status = 'pending' AND stream_url IS NULL",
            vec![
                QueryParam::String(target.url),
                QueryParam::String(target_dir.to_string()),
                expected,
                QueryParam::Int64(download_id),
            ],
        );
        match db_service.execute(update).await {
            Ok(n) if n > 0 => resolved += 1,
            Ok(_) => {} // already resolved by someone else; not a failure
            Err(e) => {
                warn!(
                    "[Catalog] Failed to persist URL for download {}: {}",
                    download_id, e
                );
                failed += 1;
            }
        }
    }

    Ok(ResumeQueuedResult { resolved, failed })
}

/// Resolve the stream URL for every download row that was queued while offline
/// (`status = 'pending' AND stream_url IS NULL`), then pump the queue so they
/// start. Call this on reconnect.
///
/// Audio rows resolve via `get_audio_stream_url`; video rows (media_type =
/// 'video') via the pure `get_video_download_url` builder using the row's stored
/// `quality_preset` — mirroring `enqueue_video_downloads`. Rows whose URL can't
/// be resolved are left pending (they retry on the next reconnect).
#[tauri::command]
#[specta::specta]
pub async fn resume_queued_downloads(
    repository: State<'_, RepositoryManagerWrapper>,
    db: State<'_, DatabaseWrapper>,
    download_manager: State<'_, DownloadManagerWrapper>,
    app: tauri::AppHandle,
    handle: String,
) -> Result<ResumeQueuedResult, String> {
    use crate::repository::MediaRepository;

    let repo = repository.0.get(&handle).ok_or("Repository not found")?;

    // The pump needs a target_dir; use the same storage root the other download
    // paths use (the database's parent directory — see `storage_get_path`).
    let (db_service, target_dir) = {
        let database = db.0.lock().map_err(|e| e.to_string())?;
        let target_dir = database
            .path()
            .parent()
            .ok_or_else(|| "Database path has no parent directory".to_string())?
            .to_string_lossy()
            .to_string();
        (Arc::new(database.service()), target_dir)
    };

    // Recover stale downloads: rows left in 'downloading' when the app was killed
    // mid-transfer are orphaned — nothing ever restarts them, so they show as
    // permanently "downloading". Reset them to 'pending' and clear the stale
    // stream_url so they get re-resolved and restarted from scratch below.
    let recover_query = Query::new(
        "UPDATE downloads SET status = 'pending', stream_url = NULL, progress = 0, \
         bytes_downloaded = 0, started_at = NULL \
         WHERE status = 'downloading'",
    );
    match db_service.execute(recover_query).await {
        Ok(n) if n > 0 => info!("[Catalog] Reset {} stale 'downloading' rows to pending", n),
        Ok(_) => {}
        Err(e) => warn!("[Catalog] Failed to reset stale downloads: {}", e),
    }

    // Repair rows that completed as audio because their media_type was missing;
    // they hold an audio-only transcode where a video should be, so requeue them
    // for the resolver below. TRACES: UR-071 | DR-136
    if let Err(e) = requeue_mistyped_video_downloads(&db_service).await {
        warn!(
            "[Catalog] Failed to requeue mis-typed video downloads: {}",
            e
        );
    }

    // Resolve each row's URL against the (now reachable) repository.
    let repo_for_resolve = Arc::clone(&repo);
    let outcome = resolve_pending_download_urls(
        &db_service,
        &target_dir,
        None,
        move |item_id: String, media_type: String, quality: String| {
            let repo = Arc::clone(&repo_for_resolve);
            async move {
                if media_type == "video" {
                    Some(
                        crate::repository::resolve_video_download(
                            repo.as_ref(),
                            &item_id,
                            &quality,
                            None,
                        )
                        .await
                        .into(),
                    )
                } else {
                    match repo.get_audio_stream_url(&item_id).await {
                        Ok(url) => Some(url.into()),
                        Err(e) => {
                            warn!(
                                "[Catalog] Failed to resolve audio URL for {}: {:?}",
                                item_id, e
                            );
                            None
                        }
                    }
                }
            }
        },
    )
    .await
    .map_err(|e| e.to_string())?;

    let ResumeQueuedResult { resolved, failed } = outcome;

    // Kick the pump so the newly-resolved rows actually start.
    if resolved > 0 {
        let active_downloads = {
            let manager = download_manager.0.lock().map_err(|e| e.to_string())?;
            manager.get_active_downloads()
        };
        pump_download_queue(app, db_service, active_downloads).await;
    }

    info!(
        "[Catalog] Resume complete: {} resolved, {} failed",
        resolved, failed
    );

    Ok(ResumeQueuedResult { resolved, failed })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::storage::db_service::RusqliteService;
    use crate::utils::lock::MutexSafe;
    use rusqlite::Connection;
    use std::sync::Mutex;

    /// The re-index policy. Pure, so it is testable without a clock, a server or
    /// a database — which is the reason it was factored out of the scheduler.
    ///
    /// TRACES: UR-065 | DR-109 | UT-115
    #[test]
    fn test_index_is_due() {
        let ttl = Duration::from_secs(6 * 60 * 60);
        let now = chrono::DateTime::parse_from_rfc3339("2026-08-04T12:00:00+00:00")
            .unwrap()
            .with_timezone(&chrono::Utc);

        // Never indexed => due. This is the first-run case.
        assert!(index_is_due(None, now, ttl));

        // Indexed 7 hours ago => past the 6h TTL => due.
        assert!(index_is_due(Some("2026-08-04T05:00:00+00:00"), now, ttl));

        // Indexed 1 hour ago => fresh => not due. This is what stops the
        // scheduler re-crawling every tick.
        assert!(!index_is_due(Some("2026-08-04T11:00:00+00:00"), now, ttl));

        // Exactly at the TTL boundary counts as due.
        assert!(index_is_due(Some("2026-08-04T06:00:00+00:00"), now, ttl));

        // A corrupt stored value must trigger a re-index, not freeze the
        // catalog forever behind an unparseable timestamp.
        assert!(index_is_due(Some("not-a-timestamp"), now, ttl));
        assert!(index_is_due(Some(""), now, ttl));

        // A timestamp in the future (clock skew, or a restored backup) is not
        // due — it ages into due-ness rather than causing a crawl every tick.
        assert!(!index_is_due(Some("2026-08-05T00:00:00+00:00"), now, ttl));

        // Offsets other than UTC are compared as instants, not as strings.
        assert!(!index_is_due(Some("2026-08-04T13:30:00+02:00"), now, ttl));
    }

    fn test_db() -> Arc<RusqliteService> {
        let conn = Connection::open_in_memory().unwrap();
        conn.execute_batch(
            r#"
            CREATE TABLE downloads (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                item_id TEXT NOT NULL,
                status TEXT NOT NULL,
                stream_url TEXT,
                target_dir TEXT,
                media_type TEXT,
                quality_preset TEXT,
                file_size INTEGER,
                progress REAL DEFAULT 0,
                bytes_downloaded INTEGER DEFAULT 0,
                started_at TEXT,
                completed_at TEXT
            );
            CREATE TABLE items (
                id TEXT PRIMARY KEY,
                item_type TEXT
            );
            "#,
        )
        .unwrap();
        Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
    }

    async fn insert_item(db: &Arc<RusqliteService>, item_id: &str, item_type: &str) {
        db.execute(Query::with_params(
            "INSERT INTO items (id, item_type) VALUES (?, ?)",
            vec![
                QueryParam::String(item_id.to_string()),
                QueryParam::String(item_type.to_string()),
            ],
        ))
        .await
        .unwrap();
    }

    async fn insert_download(
        db: &Arc<RusqliteService>,
        item_id: &str,
        status: &str,
        stream_url: Option<&str>,
        media_type: Option<&str>,
    ) {
        let q = Query::with_params(
            "INSERT INTO downloads (item_id, status, stream_url, media_type) VALUES (?, ?, ?, ?)",
            vec![
                QueryParam::String(item_id.to_string()),
                QueryParam::String(status.to_string()),
                stream_url
                    .map(|s| QueryParam::String(s.to_string()))
                    .unwrap_or(QueryParam::Null),
                media_type
                    .map(|s| QueryParam::String(s.to_string()))
                    .unwrap_or(QueryParam::Null),
            ],
        );
        db.execute(q).await.unwrap();
    }

    async fn get_row(
        db: &Arc<RusqliteService>,
        item_id: &str,
    ) -> (String, Option<String>, Option<String>) {
        let q = Query::with_params(
            "SELECT status, stream_url, target_dir FROM downloads WHERE item_id = ?",
            vec![QueryParam::String(item_id.to_string())],
        );
        db.query_one(q, |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)))
            .await
            .unwrap()
    }

    /// IT-017: a download queued from a greyed-out offline catalog entry
    /// (pending, `stream_url IS NULL`) persists, and on reconnect its URL is
    /// resolved and the row is healed (URL + target dir) so the pump can start
    /// it — while already-resolved rows are left untouched.
    ///
    /// TRACES: UR-052, UR-011 | IT-017
    #[tokio::test]
    async fn resolves_offline_queued_row_and_leaves_resolved_rows_untouched() {
        let db = test_db();
        // A row queued offline: pending with no URL yet.
        insert_download(&db, "queued-1", "pending", None, None).await;
        // An already-resolved pending row: must NOT be touched.
        insert_download(&db, "already", "pending", Some("http://existing/url"), None).await;
        // A completed row: irrelevant.
        insert_download(&db, "done", "completed", Some("http://done/url"), None).await;

        let out = resolve_pending_download_urls(
            &db,
            "/data/downloads",
            None,
            |item_id, _mt, _q| async move { Some(format!("http://resolved/{item_id}").into()) },
        )
        .await
        .unwrap();

        assert_eq!(out.resolved, 1);
        assert_eq!(out.failed, 0);

        // The offline-queued row now has a URL + target dir and stays pending.
        let (status, url, target) = get_row(&db, "queued-1").await;
        assert_eq!(status, "pending");
        assert_eq!(url.as_deref(), Some("http://resolved/queued-1"));
        assert_eq!(target.as_deref(), Some("/data/downloads"));

        // The already-resolved row is unchanged (not re-resolved).
        let (_s, url2, _t) = get_row(&db, "already").await;
        assert_eq!(url2.as_deref(), Some("http://existing/url"));
    }

    /// A bulk enqueue resolves only the rows it just created. Downloading one
    /// album must not also start every unrelated row that has been sitting
    /// pending with no URL (the smart cache leaves plenty of those).
    ///
    /// TRACES: UR-018, UR-055 | DR-173 | UT-171
    #[tokio::test]
    async fn only_ids_restricts_the_sweep_to_the_given_rows() {
        let db = test_db();
        insert_download(&db, "mine", "pending", None, Some("audio")).await;
        insert_download(&db, "someone-elses", "pending", None, Some("audio")).await;

        let mine: i64 = db
            .query_one(
                Query::new("SELECT id FROM downloads WHERE item_id = 'mine'"),
                |row| row.get(0),
            )
            .await
            .unwrap();

        let out = resolve_pending_download_urls(
            &db,
            "/data",
            Some(&[mine]),
            |item_id, _mt, _q| async move { Some(format!("http://resolved/{item_id}").into()) },
        )
        .await
        .unwrap();

        assert_eq!(out.resolved, 1);
        assert_eq!(out.failed, 0);

        let (_s, url, _t) = get_row(&db, "mine").await;
        assert_eq!(url.as_deref(), Some("http://resolved/mine"));

        let (status, other_url, _t) = get_row(&db, "someone-elses").await;
        assert_eq!(status, "pending");
        assert_eq!(
            other_url, None,
            "a scoped resolve must leave unrelated pending rows alone"
        );
    }

    /// An empty id list resolves nothing — it must not fall through to "sweep
    /// everything", which is what an unguarded `IN ()` would amount to.
    ///
    /// TRACES: UR-018, UR-055 | DR-173 | UT-171
    #[tokio::test]
    async fn an_empty_id_list_resolves_nothing() {
        let db = test_db();
        insert_download(&db, "untouched", "pending", None, Some("audio")).await;

        let out =
            resolve_pending_download_urls(&db, "/data", Some(&[]), |item_id, _mt, _q| async move {
                Some(format!("http://resolved/{item_id}").into())
            })
            .await
            .unwrap();

        assert_eq!(out.resolved, 0);
        let (_s, url, _t) = get_row(&db, "untouched").await;
        assert_eq!(url, None);
    }

    #[tokio::test]
    async fn counts_unresolvable_rows_as_failed_and_leaves_them_pending() {
        let db = test_db();
        insert_download(&db, "bad", "pending", None, None).await;

        // Resolver returns None (e.g. server lookup failed).
        let out =
            resolve_pending_download_urls(&db, "/data", None, |_id, _mt, _q| async move { None })
                .await
                .unwrap();

        assert_eq!(out.resolved, 0);
        assert_eq!(out.failed, 1);

        // Still pending with no URL, so a later reconnect can retry it.
        let (status, url, _t) = get_row(&db, "bad").await;
        assert_eq!(status, "pending");
        assert_eq!(url, None);
    }

    /// A movie queued from a media card has no `media_type` — `download_item`
    /// never records one. Defaulting that NULL to 'audio' resolved the row
    /// against `get_audio_stream_url`, so the "downloaded movie" on disk was an
    /// audio-only transcode and offline video playback could never work. The
    /// item's own type is the authority.
    ///
    /// TRACES: UR-071, UR-052 | DR-135 | UT-125
    #[tokio::test]
    async fn null_media_type_resolves_from_the_item_type_not_audio() {
        let db = test_db();
        insert_item(&db, "movie-1", "Movie").await;
        insert_item(&db, "ep-1", "Episode").await;
        insert_item(&db, "track-1", "Audio").await;
        for id in ["movie-1", "ep-1", "track-1"] {
            insert_download(&db, id, "pending", None, None).await;
        }

        let seen = Arc::new(Mutex::new(Vec::new()));
        let seen_c = Arc::clone(&seen);
        resolve_pending_download_urls(&db, "/data", None, move |item_id, media_type, _q| {
            let seen = Arc::clone(&seen_c);
            async move {
                seen.lock_safe().push((item_id.clone(), media_type));
                Some(format!("http://resolved/{item_id}").into())
            }
        })
        .await
        .unwrap();

        let seen = seen.lock_safe().clone();
        let of = |id: &str| {
            seen.iter()
                .find(|(i, _)| i == id)
                .map(|(_, m)| m.clone())
                .unwrap()
        };
        assert_eq!(of("movie-1"), "video", "a Movie must download as video");
        assert_eq!(of("ep-1"), "video", "an Episode must download as video");
        assert_eq!(of("track-1"), "audio", "a track is still audio");
    }

    /// An unknown item (never cached locally) has no type to derive from, so it
    /// keeps the historical audio default rather than failing the row.
    ///
    /// TRACES: UR-071 | DR-135 | UT-125
    #[tokio::test]
    async fn unknown_item_falls_back_to_audio() {
        let db = test_db();
        insert_download(&db, "ghost", "pending", None, None).await;

        let seen = Arc::new(Mutex::new(String::new()));
        let seen_c = Arc::clone(&seen);
        resolve_pending_download_urls(&db, "/data", None, move |_id, media_type, _q| {
            let seen = Arc::clone(&seen_c);
            async move {
                *seen.lock_safe() = media_type;
                Some("http://x".to_string().into())
            }
        })
        .await
        .unwrap();

        assert_eq!(*seen.lock_safe(), "audio");
    }

    /// An explicit `media_type` on the row always wins over the item's type.
    ///
    /// TRACES: UR-071 | DR-135 | UT-125
    #[tokio::test]
    async fn explicit_media_type_beats_the_item_type() {
        let db = test_db();
        insert_item(&db, "odd", "Audio").await;
        insert_download(&db, "odd", "pending", None, Some("video")).await;

        let seen = Arc::new(Mutex::new(String::new()));
        let seen_c = Arc::clone(&seen);
        resolve_pending_download_urls(&db, "/data", None, move |_id, media_type, _q| {
            let seen = Arc::clone(&seen_c);
            async move {
                *seen.lock_safe() = media_type;
                Some("http://x".to_string().into())
            }
        })
        .await
        .unwrap();

        assert_eq!(*seen.lock_safe(), "video");
    }

    /// Rows already downloaded under the audio default hold an audio-only
    /// transcode on disk, so they play as a broken video forever. They are
    /// identifiable — no `media_type` but a video item — and are requeued so the
    /// resolver fetches the real video. Correctly-typed rows and genuine audio
    /// downloads must be left alone.
    ///
    /// TRACES: UR-071 | DR-136 | UT-126
    #[tokio::test]
    async fn requeues_video_downloaded_under_the_audio_default() {
        let db = test_db();
        insert_item(&db, "movie-1", "Movie").await;
        insert_item(&db, "track-1", "Audio").await;
        insert_item(&db, "movie-ok", "Movie").await;
        // Mis-downloaded: completed, no media_type, video item.
        insert_download(&db, "movie-1", "completed", Some("http://audio/url"), None).await;
        // A real audio download: untouched.
        insert_download(&db, "track-1", "completed", Some("http://audio/ok"), None).await;
        // A correctly-typed video download: untouched.
        insert_download(
            &db,
            "movie-ok",
            "completed",
            Some("http://video/ok"),
            Some("video"),
        )
        .await;

        let requeued = requeue_mistyped_video_downloads(&db).await.unwrap();
        assert_eq!(requeued, 1);

        let (status, url, _t) = get_row(&db, "movie-1").await;
        assert_eq!(status, "pending", "the mis-typed row must download again");
        assert_eq!(url, None, "its audio URL must be cleared so it re-resolves");

        let (status, url, _t) = get_row(&db, "track-1").await;
        assert_eq!(status, "completed", "a real audio download is untouched");
        assert_eq!(url.as_deref(), Some("http://audio/ok"));

        let (status, _u, _t) = get_row(&db, "movie-ok").await;
        assert_eq!(status, "completed", "a correct video download is untouched");
    }

    /// A transcode answers with no `Content-Length`, so the worker's only
    /// chance at a progress total is the size predicted at resolve time. That
    /// prediction has to reach the row, and only where there is one — an
    /// audio row's `None` must not null out a size the row already holds.
    ///
    /// TRACES: UR-071 | DR-290 | UT-253
    #[tokio::test]
    async fn resolving_persists_the_predicted_size_without_erasing_a_known_one() {
        let db = test_db();
        insert_download(&db, "film", "pending", None, Some("video")).await;
        insert_download(&db, "track", "pending", None, Some("audio")).await;
        db.execute(Query::with_params(
            "UPDATE downloads SET file_size = 777 WHERE item_id = ?",
            vec![QueryParam::String("track".to_string())],
        ))
        .await
        .unwrap();

        resolve_pending_download_urls(&db, "/data", None, |item_id, media_type, _q| async move {
            Some(ResolvedDownloadUrl {
                url: format!("http://resolved/{item_id}"),
                expected_bytes: (media_type == "video").then_some(1_500_000_000),
            })
        })
        .await
        .unwrap();

        let size = |item: &'static str| {
            let db = Arc::clone(&db);
            async move {
                db.query_one(
                    Query::with_params(
                        "SELECT file_size FROM downloads WHERE item_id = ?",
                        vec![QueryParam::String(item.to_string())],
                    ),
                    |row| row.get::<_, Option<i64>>(0),
                )
                .await
                .unwrap()
            }
        };
        assert_eq!(size("film").await, Some(1_500_000_000));
        assert_eq!(size("track").await, Some(777));
    }

    #[tokio::test]
    async fn video_rows_use_media_type_in_resolver() {
        let db = test_db();
        insert_download(&db, "vid-1", "pending", None, Some("video")).await;

        let out = resolve_pending_download_urls(
            &db,
            "/data",
            None,
            |item_id, media_type, _q| async move {
                assert_eq!(media_type, "video");
                Some(format!("http://transcode/{item_id}").into())
            },
        )
        .await
        .unwrap();

        assert_eq!(out.resolved, 1);
        let (_s, url, _t) = get_row(&db, "vid-1").await;
        assert_eq!(url.as_deref(), Some("http://transcode/vid-1"));
    }
}