jellytau/src-tauri/src/commands/catalog.rs

//! Tauri commands for the offline "browse & queue" feature.
//!
//! 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::Arc;

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

use crate::commands::repository::RepositoryManagerWrapper;
use crate::commands::download::{pump_download_queue, DownloadManagerWrapper};
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";

/// Item types worth caching for offline browsing: containers the library
/// landing pages render plus the playable leaves users queue for download.
const CATALOG_ITEM_TYPES: &[&str] = &[
    "MusicAlbum",
    "Movie",
    "Series",
    "Season",
    "Episode",
    "Audio",
    "BoxSet",
];

#[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,
}

#[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> {
    use crate::repository::MediaRepository;

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

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

    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;
            }
        }
    }

    // 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, {} libraries failed",
        items_cached, libraries_failed
    );

    Ok(CatalogSyncResult {
        items_cached,
        libraries_failed,
    })
}

/// 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 })
}

#[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,
}

/// 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)`.
pub(crate) async fn resolve_pending_download_urls<F, Fut>(
    db_service: &Arc<crate::storage::db_service::RusqliteService>,
    target_dir: &str,
    resolve: F,
) -> Result<ResumeQueuedResult, String>
where
    F: Fn(String, String, String) -> Fut,
    Fut: std::future::Future<Output = Option<String>>,
{
    let rows_query = Query::new(
        "SELECT id, item_id, COALESCE(media_type, 'audio'), COALESCE(quality_preset, 'original')
         FROM downloads
         WHERE status = 'pending' AND stream_url IS NULL",
    );
    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 stream_url = match resolve(item_id.clone(), media_type, quality).await {
            Some(url) => url,
            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.
        let update = Query::with_params(
            "UPDATE downloads SET stream_url = ?, target_dir = ?
             WHERE id = ? AND status = 'pending' AND stream_url IS NULL",
            vec![
                QueryParam::String(stream_url),
                QueryParam::String(target_dir.to_string()),
                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;

    use crate::repository::HybridRepository;

    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),
    }

    // 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,
        move |item_id: String, media_type: String, quality: String| {
            let repo = Arc::clone(&repo_for_resolve);
            async move {
                if media_type == "video" {
                    Some(<HybridRepository as MediaRepository>::get_video_download_url(
                        repo.as_ref(),
                        &item_id,
                        &quality,
                        None,
                    ))
                } else {
                    match repo.get_audio_stream_url(&item_id).await {
                        Ok(url) => Some(url),
                        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 rusqlite::Connection;
    use std::sync::Mutex;

    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
            );
            "#,
        )
        .unwrap();
        Arc::new(RusqliteService::new(Arc::new(Mutex::new(conn))))
    }

    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()
    }

    #[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", |item_id, _mt, _q| async move {
            Some(format!("http://resolved/{item_id}"))
        })
        .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"));
    }

    #[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", |_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);
    }

    #[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", |item_id, media_type, _q| async move {
            assert_eq!(media_type, "video");
            Some(format!("http://transcode/{item_id}"))
        })
        .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"));
    }
}