fix offline mode and layout bugs

src-tauri/src/commands/catalog.rs

@@ -0,0 +1,440 @@
+//! 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)
+    };
+
+    // 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"));
+    }
+}

src-tauri/src/commands/download/mod.rs

@@ -1268,13 +1268,50 @@ fn spawn_download_worker(
             debug!("   Unregistered download {}. Active downloads: {}", download_id, active.len());
         }
 
+        // The pump runs downloads in the background, so the terminal status MUST
+        // be persisted to the DB here — the frontend event handler only writes it
+        // when that download happens to be loaded in its store, which is not the
+        // case for auto-pumped rows (or any completion while the downloads page is
+        // closed). `check_for_local_download` filters on status = 'completed', so a
+        // missed write leaves finished files unrecognized: albums never show as
+        // downloaded and playback never switches from the (expiring) stream to the
+        // local file, cutting tracks off mid-play.
+        let db_service = {
+            let db = app.state::<DatabaseWrapper>();
+            let database = match db.0.lock() {
+                Ok(d) => d,
+                Err(e) => {
+                    error!("[pump] Failed to lock database after download {}: {}", download_id, e);
+                    return;
+                }
+            };
+            Arc::new(database.service())
+        };
+
         match result {
             Ok(res) => {
                 info!("Download completed successfully: {} bytes", res.bytes_downloaded);
+                let file_path = target_path.to_string_lossy().to_string();
+
+                let update = Query::with_params(
+                    "UPDATE downloads SET status = 'completed', progress = 1.0, \
+                     bytes_downloaded = ?, file_size = ?, file_path = ?, \
+                     completed_at = CURRENT_TIMESTAMP WHERE id = ?",
+                    vec![
+                        QueryParam::Int64(res.bytes_downloaded as i64),
+                        QueryParam::Int64(res.bytes_downloaded as i64),
+                        QueryParam::String(file_path.clone()),
+                        QueryParam::Int64(download_id),
+                    ],
+                );
+                if let Err(e) = db_service.execute(update).await {
+                    error!("[pump] Failed to persist completed status for download {}: {}", download_id, e);
+                }
+
                 let completed_event = DownloadEvent::Completed {
                     download_id,
                     item_id,
-                    file_path: target_path.to_string_lossy().to_string(),
+                    file_path,
                 };
                 match app.emit("download-event", completed_event) {
                     Ok(_) => debug!("   Completed event emitted successfully"),
@@ -1283,6 +1320,18 @@ fn spawn_download_worker(
             }
             Err(e) => {
                 error!("Download failed: {:?}", e);
+
+                let update = Query::with_params(
+                    "UPDATE downloads SET status = 'failed', error_message = ? WHERE id = ?",
+                    vec![
+                        QueryParam::String(e.to_string()),
+                        QueryParam::Int64(download_id),
+                    ],
+                );
+                if let Err(db_err) = db_service.execute(update).await {
+                    error!("[pump] Failed to persist failed status for download {}: {}", download_id, db_err);
+                }
+
                 let failed_event = DownloadEvent::Failed {
                     download_id,
                     item_id,
@@ -1296,17 +1345,6 @@ fn spawn_download_worker(
         }
 
         // A slot just freed — start the next pending download (if any).
-        let db_service = {
-            let db = app.state::<DatabaseWrapper>();
-            let database = match db.0.lock() {
-                Ok(d) => d,
-                Err(e) => {
-                    error!("[pump] Failed to lock database after download {}: {}", download_id, e);
-                    return;
-                }
-            };
-            Arc::new(database.service())
-        };
         pump_download_queue(app.clone(), db_service, active_downloads).await;
     });
 }

src-tauri/src/commands/mod.rs

@@ -2,6 +2,7 @@
 // TRACES: UR-002, UR-003, UR-004, UR-005, UR-009, UR-011, UR-012, UR-017, UR-019, UR-025 |
 //         DR-015, DR-017, DR-021, DR-028
 pub mod auth;
+pub mod catalog;
 pub mod connectivity;
 pub mod conversions;
 pub mod device;
@@ -17,6 +18,7 @@ pub mod storage;
 pub mod sync;
 
 pub use auth::*;
+pub use catalog::*;
 pub use connectivity::*;
 pub use conversions::*;
 pub use device::*;