Added offline mode support

Cargo.lock

@@ -1418,7 +1418,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
 
 [[package]]
 name = "furumusic"
-version = "0.4.4"
+version = "0.5.0"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumusic"
-version = "0.4.5"
+version = "0.5.0"
 edition = "2024"
 description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
 

src/jobs/inbox_discover.rs

@@ -12,6 +12,21 @@ const AUDIO_EXTENSIONS: &[&str] = &[
     "mp3", "flac", "ogg", "opus", "aac", "m4a", "wav", "ape", "wv", "wma", "tta", "aiff", "aif",
 ];
 
+/// How long a `failed` review must stay untouched before discover
+/// automatically requeues it (instead of creating a new row per attempt).
+const FAILED_RETRY_COOLDOWN_SECS: i64 = 3600;
+
+/// Leftover files that are safe to purge from inbox folders that no longer
+/// contain any audio (covers, playlists, rip logs and similar sidecar files).
+const JUNK_EXTENSIONS: &[&str] = &[
+    "jpg", "jpeg", "png", "gif", "webp", "bmp", "m3u", "m3u8", "cue", "log", "txt", "nfo", "sfv",
+    "md5", "accurip", "url", "ini", "pdf",
+];
+const JUNK_FILENAMES: &[&str] = &[".ds_store", "thumbs.db", "desktop.ini"];
+
+/// Junk younger than this is kept — an upload might still be in progress.
+const JUNK_MIN_AGE: std::time::Duration = std::time::Duration::from_secs(24 * 3600);
+
 pub struct InboxDiscoverJob;
 
 #[async_trait::async_trait]
@@ -76,6 +91,10 @@ impl Job for InboxDiscoverJob {
         let mut audio_files = Vec::new();
         collect_audio_files(inbox, &mut audio_files).await?;
 
+        // Purge leftover junk (covers, playlists, logs) from subtrees that no
+        // longer contain audio, so processed uploads don't linger forever.
+        cleanup_inbox_junk(inbox, JUNK_MIN_AGE).await;
+
         log.info(&format!("Found {} audio files in inbox", audio_files.len()));
         if audio_files.is_empty() {
             return Ok(());
@@ -87,6 +106,7 @@ impl Job for InboxDiscoverJob {
         let mut discovered = 0u64;
         let mut skipped_hash = 0u64;
         let mut skipped_existing = 0u64;
+        let mut requeued = 0u64;
 
         for (_folder, files) in &groups {
             for file_path in files {
@@ -94,13 +114,34 @@ impl Job for InboxDiscoverJob {
                     crate::media_paths::path_for_root(&config.agent_inbox_dir, file_path)
                         .unwrap_or_else(|| file_path.to_string_lossy().to_string());
 
-                // Skip if a PendingReview already exists for this path
-                match PendingReview::exists_for_path(&ctx.db, &input_path_str).await {
-                    Ok(true) => {
-                        skipped_existing += 1;
+                // One review row per path: any existing row blocks creating a
+                // new one. A stale "failed" row is requeued in place instead,
+                // so retries don't multiply rows. "rejected" stays rejected.
+                match PendingReview::latest_for_path(&ctx.pool, &input_path_str).await {
+                    Ok(None) => {}
+                    Ok(Some((id, status, updated_at))) => {
+                        if status == "failed" {
+                            let stale = chrono::DateTime::parse_from_rfc3339(&updated_at)
+                                .map(|t| {
+                                    chrono::Utc::now().signed_duration_since(t).num_seconds()
+                                        >= FAILED_RETRY_COOLDOWN_SECS
+                                })
+                                .unwrap_or(true);
+                            if stale {
+                                match PendingReview::requeue_by_ids(&ctx.db, &[id]).await {
+                                    Ok(()) => requeued += 1,
+                                    Err(e) => log.warn(&format!(
+                                        "Failed to requeue review {id} for {input_path_str}: {e}"
+                                    )),
+                                }
+                            } else {
+                                skipped_existing += 1;
+                            }
+                        } else {
+                            skipped_existing += 1;
+                        }
                         continue;
                     }
-                    Ok(false) => {}
                     Err(e) => {
                         log.warn(&format!(
                             "Error checking existing review for {}: {e}",
@@ -215,8 +256,8 @@ impl Job for InboxDiscoverJob {
         }
 
         log.info(&format!(
-            "Discovered {} new files, skipped {} (hash known), skipped {} (already queued)",
-            discovered, skipped_hash, skipped_existing
+            "Discovered {} new files, requeued {} failed, skipped {} (hash known), skipped {} (already tracked)",
+            discovered, requeued, skipped_hash, skipped_existing
         ));
         crate::metrics::record_agent_discover_files(
             audio_files.len() as u64,
@@ -227,7 +268,7 @@ impl Job for InboxDiscoverJob {
 
         // Trigger inbox_process in background if new files were discovered
         // and no orchestrator is already running
-        if discovered > 0 {
+        if discovered + requeued > 0 {
             if crate::jobs::inbox_process::is_orchestrator_running() {
                 log.info(
                     "New files discovered but inbox_process already running, it will pick them up",
@@ -299,3 +340,75 @@ pub fn is_audio_file(name: &str) -> bool {
     let ext = name.rsplit('.').next().unwrap_or("").to_lowercase();
     AUDIO_EXTENSIONS.contains(&ext.as_str())
 }
+
+fn is_junk_file(name: &str) -> bool {
+    let lower = name.to_lowercase();
+    // macOS AppleDouble sidecars ("._track.mp3") and well-known junk names
+    if lower.starts_with("._") || JUNK_FILENAMES.contains(&lower.as_str()) {
+        return true;
+    }
+    let ext = lower.rsplit('.').next().unwrap_or("");
+    JUNK_EXTENSIONS.contains(&ext)
+}
+
+/// Remove leftover sidecar files (covers, playlists, rip logs) from inbox
+/// subtrees that no longer contain any audio, then prune emptied directories.
+///
+/// Junk younger than `min_age` is kept in case an upload is still in
+/// progress, and unknown file types are never touched. Returns `true` when
+/// `dir` still contains something worth keeping (so the caller must not
+/// remove it).
+async fn cleanup_inbox_junk(dir: &Path, min_age: std::time::Duration) -> bool {
+    let mut entries = match tokio::fs::read_dir(dir).await {
+        Ok(e) => e,
+        Err(_) => return true,
+    };
+
+    let mut has_audio = false;
+    let mut keep_other = false;
+    let mut junk: Vec<PathBuf> = Vec::new();
+
+    while let Ok(Some(entry)) = entries.next_entry().await {
+        let name = entry.file_name().to_string_lossy().into_owned();
+        let ft = match entry.file_type().await {
+            Ok(ft) => ft,
+            Err(_) => {
+                keep_other = true;
+                continue;
+            }
+        };
+        if ft.is_dir() {
+            if Box::pin(cleanup_inbox_junk(&entry.path(), min_age)).await {
+                keep_other = true;
+            } else {
+                let _ = tokio::fs::remove_dir(&entry.path()).await;
+            }
+        } else if !name.starts_with('.') && is_audio_file(&name) {
+            // dotfiles are invisible to discovery, so they don't count as audio
+            has_audio = true;
+        } else if is_junk_file(&name) {
+            junk.push(entry.path());
+        } else {
+            keep_other = true;
+        }
+    }
+
+    if has_audio {
+        return true;
+    }
+
+    let mut junk_left = false;
+    for path in junk {
+        let old_enough = tokio::fs::metadata(&path)
+            .await
+            .ok()
+            .and_then(|m| m.modified().ok())
+            .and_then(|t| t.elapsed().ok())
+            .is_some_and(|age| age >= min_age);
+        if !old_enough || tokio::fs::remove_file(&path).await.is_err() {
+            junk_left = true;
+        }
+    }
+
+    keep_other || junk_left
+}

src/jobs/inbox_process.rs

@@ -12,6 +12,11 @@ static ORCHESTRATOR_RUNNING: AtomicBool = AtomicBool::new(false);
 /// PostgreSQL advisory locks use a 64-bit key; this is an arbitrary unique value.
 const ORCHESTRATOR_ADVISORY_LOCK_ID: i64 = 0x4655_5255_4D55_5349; // "FURUMUSI" in hex
 
+/// Maximum number of files sent to the LLM in a single batch call.
+/// Folders with more files are processed in chunks of this size, otherwise
+/// the model's completion window overflows and the JSON response is cut off.
+const MAX_LLM_BATCH_FILES: usize = 20;
+
 /// Check if an orchestrator is currently running (used by inbox_discover to avoid redundant triggers).
 pub fn is_orchestrator_running() -> bool {
     ORCHESTRATOR_RUNNING.load(Ordering::SeqCst)
@@ -214,14 +219,25 @@ impl Job for InboxProcessJob {
                     folder_rel, file_count,
                 ));
 
-                let (ok, fail) =
-                    process_folder_batch(&ctx.db, &config, &ctx.pool, &folder_rel, reviews, log)
-                        .await;
+                // Large folders are split into chunks: a single LLM call for
+                // 100+ files overflows the completion window and the whole
+                // batch fails with a truncated-JSON parse error.
+                for chunk in reviews.chunks(MAX_LLM_BATCH_FILES) {
+                    let (ok, fail) = process_folder_batch(
+                        &ctx.db,
+                        &config,
+                        &ctx.pool,
+                        &folder_rel,
+                        chunk.to_vec(),
+                        log,
+                    )
+                    .await;
 
-                total_ok += ok;
-                total_fail += fail;
+                    total_ok += ok;
+                    total_fail += fail;
+                }
                 log.info(&format!(
-                    "Folder done: {ok} ok, {fail} err. Total so far: {total_ok} ok, {total_fail} err"
+                    "Folder done. Total so far: {total_ok} ok, {total_fail} err"
                 ));
             }
         }
@@ -344,6 +360,7 @@ async fn process_folder_batch(
     log.info("Phase 1: extracting metadata...");
     let mut prepared: Vec<PreparedFile> = Vec::with_capacity(file_count);
     let mut failed_reviews: Vec<PendingReview> = Vec::new();
+    let mut merged_count = 0u64;
 
     for mut review in reviews {
         let stored_input_path = review.input_path_str().to_owned();
@@ -355,9 +372,6 @@ async fn process_folder_batch(
             .unwrap_or("unknown")
             .to_owned();
 
-        // Set status → processing
-        let _ = review.set_processing(db).await;
-
         // Parse context_json
         let mut context: serde_json::Value = review
             .context_json
@@ -365,6 +379,42 @@ async fn process_folder_batch(
             .and_then(|s| serde_json::from_str(s).ok())
             .unwrap_or_default();
 
+        // Resolve duplicates and missing sources before any expensive work.
+        let sha = context
+            .get("sha256")
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_owned();
+        let file_exists = file_path.exists();
+        if !sha.is_empty()
+            && crate::agent::rag::file_hash_exists(pool, &sha)
+                .await
+                .unwrap_or(false)
+        {
+            // Identical content is already in the library — drop the inbox
+            // copy (same as mover::Merged) and close the review.
+            if file_exists {
+                let _ = tokio::fs::remove_file(&file_path).await;
+            }
+            let _ = PendingReview::delete_by_ids(db, &[review.id_val()]).await;
+            log.info(&format!(
+                "{filename}: content already in library (sha256 match) — merged duplicate"
+            ));
+            crate::metrics::record_agent_file_processed("ok", "merged_duplicate");
+            merged_count += 1;
+            continue;
+        }
+        if !file_exists {
+            let msg = format!("{filename}: source file missing: {stored_input_path}");
+            log.error(&msg);
+            let _ = review.set_failed(db, &msg).await;
+            failed_reviews.push(review);
+            continue;
+        }
+
+        // Set status → processing
+        let _ = review.set_processing(db).await;
+
         // Extract metadata (with 60s timeout)
         let path_for_meta = file_path.to_path_buf();
         let metadata_start = std::time::Instant::now();
@@ -444,15 +494,16 @@ async fn process_folder_batch(
     }
 
     log.info(&format!(
-        "Phase 1 done: {} prepared, {} failed metadata",
+        "Phase 1 done: {} prepared, {} merged duplicates, {} failed",
         prepared.len(),
+        merged_count,
         failed_reviews.len(),
     ));
 
     if prepared.is_empty() {
         let duration_ms = batch_start.elapsed().as_millis() as i64;
         let _ = run.set_completed(db, duration_ms, &log.output()).await;
-        return (0, failed_reviews.len() as u64);
+        return (merged_count, failed_reviews.len() as u64);
     }
 
     // Phase 2: RAG lookup (collect unique artist/album queries from all files)
@@ -648,16 +699,17 @@ async fn process_folder_batch(
             let err_msg = format!("Batch LLM call failed: {e}");
             log.error(&err_msg);
             // Mark all files as failed
+            let prepared_count = prepared.len() as u64;
             for mut p in prepared {
                 let _ = p.review.set_failed(db, &err_msg).await;
                 crate::metrics::record_agent_file_processed("failed", "failed");
             }
-            let total_fail_count = failed_reviews.len() as u64 + file_count as u64;
+            let total_fail_count = failed_reviews.len() as u64 + prepared_count;
             let duration_ms = batch_start.elapsed().as_millis() as i64;
             let _ = run
                 .set_failed(db, duration_ms, &log.output(), &err_msg)
                 .await;
-            return (0, total_fail_count);
+            return (merged_count, total_fail_count);
         }
     };
 
@@ -681,7 +733,7 @@ async fn process_folder_batch(
     let completion_per_file = batch_result.completion_tokens / prepared.len().max(1) as u64;
     let duration_per_file = batch_result.duration_ms as i64 / prepared.len().max(1) as i64;
 
-    let mut ok_count = 0u64;
+    let mut ok_count = merged_count;
     let mut fail_count = failed_reviews.len() as u64;
 
     for mut p in prepared {

src/metrics.rs

@@ -883,6 +883,7 @@ const KNOWN_HTTP_ROUTES: &[&str] = &[
     "/api/player/lastfm/now-playing",
     "/api/player/lastfm/scrobble",
     "/api/player/agent-queue",
+    "/api/player/offline/manifest",
     "/api/player/torrents",
     "/api/player/torrents/session/{id}",
     "/api/player/torrents/preview",

src/oidc.rs

@@ -977,27 +977,54 @@ fn safe_mobile_redirect_uri(raw: Option<&str>) -> Option<String> {
     if lower.starts_with("furumi://") || lower.starts_with("furumusic://") {
         return Some(value.to_owned());
     }
+    if is_loopback_http_redirect(&lower) {
+        return Some(value.to_owned());
+    }
     None
 }
 
+/// RFC 8252 §7.3: native apps without a custom URL scheme (the CLI client)
+/// receive the callback on a loopback listener with an ephemeral port.
+fn is_loopback_http_redirect(lower: &str) -> bool {
+    let Some(rest) = lower.strip_prefix("http://") else {
+        return false;
+    };
+    let host_port = rest.split(['/', '?', '#']).next().unwrap_or("");
+    let Some((host, port)) = host_port.rsplit_once(':') else {
+        return false;
+    };
+    matches!(host, "127.0.0.1" | "localhost" | "[::1]")
+        && !port.is_empty()
+        && port.len() <= 5
+        && port.bytes().all(|b| b.is_ascii_digit())
+}
+
 fn mobile_redirect_success(app_redirect_uri: &str, code: &str) -> cot::response::Response {
     let deep_link = append_query_param(app_redirect_uri, "code", code);
+    if is_loopback_http_redirect(&app_redirect_uri.to_ascii_lowercase()) {
+        return auth::redirect(&deep_link);
+    }
     mobile_deep_link_page(
         "success",
         "Sign-in complete",
-        "Furumi should open automatically. You can close this window after the app opens.",
+        "Furumi should open automatically. If it doesn't, use the button or copy the code below.",
         None,
+        Some(code),
         &deep_link,
     )
 }
 
 fn mobile_redirect_error(app_redirect_uri: &str, error: &str) -> cot::response::Response {
     let deep_link = append_query_param(app_redirect_uri, "error", error);
+    if is_loopback_http_redirect(&app_redirect_uri.to_ascii_lowercase()) {
+        return auth::redirect(&deep_link);
+    }
     mobile_deep_link_page(
         "error",
         "Sign-in failed",
-        "Furumi should open automatically and show the sign-in error. You can close this window after the app opens.",
+        "Furumi should open automatically and show the sign-in error.",
         Some(error),
+        None,
         &deep_link,
     )
 }
@@ -1007,6 +1034,7 @@ fn mobile_deep_link_page(
     title: &str,
     message: &str,
     detail: Option<&str>,
+    code: Option<&str>,
     deep_link: &str,
 ) -> cot::response::Response {
     let state_class = html_escape(state);
@@ -1015,6 +1043,15 @@ fn mobile_deep_link_page(
     let detail_html = detail
         .map(|value| format!(r#"<p class="detail">Reason: {}</p>"#, html_escape(value)))
         .unwrap_or_default();
+    let code_html = code
+        .map(|value| {
+            format!(
+                r#"<p class="hint">Signing in from a terminal? Paste this code there:</p>
+    <input class="code" readonly value="{}" onclick="this.select()">"#,
+                html_escape(value)
+            )
+        })
+        .unwrap_or_default();
     let deep_link_html = html_escape(deep_link);
     let deep_link_js =
         serde_json::to_string(deep_link).expect("serializing URL string cannot fail");
@@ -1095,6 +1132,19 @@ fn mobile_deep_link_page(
       font-size: 13px;
       color: #89847c;
     }}
+    .code {{
+      width: 100%;
+      margin-top: 8px;
+      padding: 10px 12px;
+      box-sizing: border-box;
+      border: 1px solid #3a3c42;
+      border-radius: 8px;
+      background: #1a1c20;
+      color: #e8d8a8;
+      font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace;
+      font-size: 13px;
+      text-align: center;
+    }}
   </style>
 </head>
 <body>
@@ -1105,15 +1155,13 @@ fn mobile_deep_link_page(
     {detail_html}
     <a href="{deep_link_html}">Open Furumi</a>
     <p class="hint">If nothing happens, use the button above.</p>
+    {code_html}
   </main>
   <script>
     const deepLink = {deep_link_js};
     window.setTimeout(() => {{
       window.location.href = deepLink;
     }}, 100);
-    window.setTimeout(() => {{
-      window.close();
-    }}, 1800);
   </script>
 </body>
 </html>"#,
@@ -1230,4 +1278,25 @@ mod tests {
         );
         assert!(safe_mobile_redirect_uri(Some("https://example.com/callback")).is_none());
     }
+
+    #[test]
+    fn mobile_oidc_redirect_uri_allows_loopback_http() {
+        assert_eq!(
+            safe_mobile_redirect_uri(Some("http://127.0.0.1:8753/callback")).as_deref(),
+            Some("http://127.0.0.1:8753/callback")
+        );
+        assert_eq!(
+            safe_mobile_redirect_uri(Some("http://localhost:1234/callback")).as_deref(),
+            Some("http://localhost:1234/callback")
+        );
+        assert_eq!(
+            safe_mobile_redirect_uri(Some("http://[::1]:1234/callback")).as_deref(),
+            Some("http://[::1]:1234/callback")
+        );
+        // Non-loopback hosts, missing ports and https stay rejected.
+        assert!(safe_mobile_redirect_uri(Some("http://127.0.0.1/callback")).is_none());
+        assert!(safe_mobile_redirect_uri(Some("http://evil.com:80/callback")).is_none());
+        assert!(safe_mobile_redirect_uri(Some("https://127.0.0.1:80/callback")).is_none());
+        assert!(safe_mobile_redirect_uri(Some("http://127.0.0.1:notaport/x")).is_none());
+    }
 }

src/player/dto.rs

@@ -325,6 +325,45 @@ pub(super) struct UserProfile {
     pub(super) stats: UserStats,
 }
 
+#[derive(Debug, Serialize, JsonSchema)]
+pub(super) struct OfflineManifestResponse {
+    pub(super) generated_at: String,
+    pub(super) tracks: Vec<OfflineTrackManifestItem>,
+    pub(super) playlists: Vec<OfflinePlaylistManifestItem>,
+    pub(super) liked_track_ids: Vec<i64>,
+    pub(super) followed_artist_ids: Vec<i64>,
+}
+
+#[derive(Debug, Serialize, JsonSchema)]
+pub(super) struct OfflineTrackManifestItem {
+    pub(super) id: i64,
+    pub(super) updated_at: String,
+    pub(super) stream_url: String,
+    pub(super) audio_file_id: i64,
+    pub(super) audio_hash: String,
+    pub(super) audio_size_bytes: i64,
+    pub(super) audio_mime_type: String,
+    pub(super) audio_updated_at: String,
+    pub(super) cover_file_id: Option<i64>,
+    pub(super) cover_url: Option<String>,
+    pub(super) cover_hash: Option<String>,
+    pub(super) cover_updated_at: Option<String>,
+}
+
+#[derive(Debug, Serialize, JsonSchema)]
+pub(super) struct OfflinePlaylistManifestItem {
+    pub(super) id: i64,
+    pub(super) title: String,
+    pub(super) description: Option<String>,
+    pub(super) updated_at: String,
+    pub(super) is_own: bool,
+    pub(super) owner_name: Option<String>,
+    pub(super) is_public: bool,
+    pub(super) is_saved: bool,
+    pub(super) kind: String,
+    pub(super) track_ids: Vec<i64>,
+}
+
 #[derive(Debug, Serialize, JsonSchema)]
 pub(super) struct LastfmStatus {
     pub(super) configured: bool,

src/player/mod.rs

@@ -861,6 +861,12 @@ fn native_device_name_from_user_agent(user_agent: Option<&str>) -> Option<String
                 None => "Furumi MacOS".to_string(),
             });
         }
+        if product.eq_ignore_ascii_case("FurumiTUI") || product.eq_ignore_ascii_case("furumi-tui") {
+            return Some(match version.as_deref() {
+                Some(v) => format!("Furumi TUI {v}"),
+                None => "Furumi TUI".to_string(),
+            });
+        }
     }
     None
 }
@@ -890,6 +896,9 @@ fn device_kind_from_user_agent(user_agent: Option<&str>) -> &'static str {
     if ua.contains("furumimac") {
         return "computer";
     }
+    if ua.contains("furumitui/") || ua.contains("furumi-tui/") {
+        return "computer";
+    }
     if ua.contains("iphone") || (ua.contains("android") && ua.contains("mobile")) {
         "phone"
     } else if ua.contains("ipad") || ua.contains("tablet") || ua.contains("android") {
@@ -914,6 +923,22 @@ mod device_tests {
         assert_eq!(device_kind_from_user_agent(user_agent), "phone");
     }
 
+    #[test]
+    fn detects_furumi_tui_native_client() {
+        let user_agent = Some("FurumiTUI/0.1.0 macos");
+
+        assert_eq!(device_name_from_user_agent(user_agent), "Furumi TUI 0.1.0");
+        assert_eq!(device_kind_from_user_agent(user_agent), "computer");
+    }
+
+    #[test]
+    fn detects_furumi_tui_http_user_agent_token() {
+        let user_agent = Some("furumi-tui/0.1.0 (macos)");
+
+        assert_eq!(device_name_from_user_agent(user_agent), "Furumi TUI 0.1.0");
+        assert_eq!(device_kind_from_user_agent(user_agent), "computer");
+    }
+
     #[test]
     fn keeps_browser_fallback_for_generic_android_user_agents() {
         let user_agent = Some("Mozilla/5.0 Android Mobile");
@@ -1023,6 +1048,192 @@ async fn me_handler(
     .into_response()
 }
 
+// ---------------------------------------------------------------------------
+// GET /api/player/offline/manifest
+// ---------------------------------------------------------------------------
+
+async fn offline_manifest_handler(
+    auth_ctx: auth::AuthContext,
+    session: Session,
+    db: Database,
+    pool: &sqlx::PgPool,
+) -> cot::Result<cot::response::Response> {
+    let Some(user) = auth::get_request_user(&auth_ctx, &session, &db).await else {
+        return Ok(json_error(StatusCode::UNAUTHORIZED, "not authenticated"));
+    };
+
+    let generated_at = now_iso_string();
+
+    let track_rows = sqlx::query_as::<_, OfflineTrackManifestRow>(
+        r#"SELECT t.id,
+                  GREATEST(
+                      t.updated_at::text,
+                      r.updated_at::text,
+                      mf.created_at::text,
+                      COALESCE(cover_mf.created_at::text, ''),
+                      COALESCE((
+                          SELECT MAX(a.updated_at::text)
+                            FROM furumusic__track_artist ta
+                            JOIN furumusic__artist a ON a.id = ta.artist_id
+                           WHERE ta.track_id = t.id
+                      ), ''),
+                      COALESCE((
+                          SELECT MAX(a.updated_at::text)
+                            FROM furumusic__release_artist ra
+                            JOIN furumusic__artist a ON a.id = ra.artist_id
+                           WHERE ra.release_id = r.id
+                      ), '')
+                  ) AS updated_at,
+                  mf.id AS audio_file_id,
+                  mf.sha256_hash::text AS audio_hash,
+                  mf.file_size_bytes AS audio_size_bytes,
+                  mf.mime_type::text AS audio_mime_type,
+                  mf.created_at::text AS audio_updated_at,
+                  cover_mf.id AS cover_file_id,
+                  cover_mf.sha256_hash::text AS cover_hash,
+                  cover_mf.created_at::text AS cover_updated_at
+           FROM furumusic__track t
+           JOIN furumusic__release r ON r.id = t.release_id
+           JOIN furumusic__media_file mf ON mf.id = t.audio_file_id
+           LEFT JOIN furumusic__media_file cover_mf
+                  ON cover_mf.id = COALESCE(t.cover_file_id, r.cover_file_id)
+           WHERE t.is_hidden = false AND r.is_hidden = false
+           ORDER BY t.id"#,
+    )
+    .fetch_all(pool)
+    .await
+    .map_err(|e| cot::Error::internal(e.to_string()))?;
+
+    let liked_track_ids = sqlx::query_scalar::<_, i64>(
+        r#"SELECT ult.track_id
+           FROM furumusic__user_liked_track ult
+           JOIN furumusic__track t ON t.id = ult.track_id AND t.is_hidden = false
+           JOIN furumusic__release r ON r.id = t.release_id AND r.is_hidden = false
+           WHERE ult.user_id = $1
+           ORDER BY ult.created_at DESC"#,
+    )
+    .bind(user.id)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| cot::Error::internal(e.to_string()))?;
+
+    let likes_updated_at = sqlx::query_scalar::<_, Option<String>>(
+        "SELECT MAX(created_at::text) FROM furumusic__user_liked_track WHERE user_id = $1",
+    )
+    .bind(user.id)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| cot::Error::internal(e.to_string()))?
+    .unwrap_or_else(|| generated_at.clone());
+
+    let playlist_rows = sqlx::query_as::<_, OfflinePlaylistManifestRow>(
+        r#"SELECT p.id,
+                  p.title::text AS title,
+                  p.description,
+                  p.updated_at::text AS updated_at,
+                  (p.owner_id = $1) AS is_own,
+                  COALESCE(NULLIF(u.display_name, ''), u.username)::text AS owner_name,
+                  p.is_public,
+                  EXISTS (
+                      SELECT 1 FROM furumusic__saved_playlist sp
+                       WHERE sp.user_id = $1 AND sp.playlist_id = p.id
+                  ) AS is_saved,
+                  COALESCE(
+                      array_agg(pt.track_id ORDER BY pt.position)
+                          FILTER (WHERE t.id IS NOT NULL AND r.id IS NOT NULL),
+                      ARRAY[]::bigint[]
+                  ) AS track_ids
+           FROM furumusic__playlist p
+           JOIN furumusic__user u ON u.id = p.owner_id
+           LEFT JOIN furumusic__playlist_track pt ON pt.playlist_id = p.id
+           LEFT JOIN furumusic__track t ON t.id = pt.track_id AND t.is_hidden = false
+           LEFT JOIN furumusic__release r ON r.id = t.release_id AND r.is_hidden = false
+           WHERE p.owner_id = $1
+              OR EXISTS (
+                  SELECT 1 FROM furumusic__saved_playlist sp
+                   WHERE sp.user_id = $1 AND sp.playlist_id = p.id
+              )
+              OR p.is_public = true
+           GROUP BY p.id, u.display_name, u.username
+           ORDER BY
+              CASE WHEN p.owner_id = $1 THEN 0 WHEN p.is_public THEN 2 ELSE 1 END,
+              p.title"#,
+    )
+    .bind(user.id)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| cot::Error::internal(e.to_string()))?;
+
+    let followed_artist_ids = sqlx::query_scalar::<_, i64>(
+        r#"SELECT ufa.artist_id
+           FROM furumusic__user_followed_artist ufa
+           JOIN furumusic__artist a ON a.id = ufa.artist_id AND a.is_hidden = false
+           WHERE ufa.user_id = $1
+           ORDER BY ufa.created_at DESC"#,
+    )
+    .bind(user.id)
+    .fetch_all(pool)
+    .await
+    .map_err(|e| cot::Error::internal(e.to_string()))?;
+
+    let tracks = track_rows
+        .into_iter()
+        .map(|row| OfflineTrackManifestItem {
+            id: row.id,
+            updated_at: row.updated_at,
+            stream_url: format!("/api/player/stream/{}", row.id),
+            audio_file_id: row.audio_file_id,
+            audio_hash: row.audio_hash,
+            audio_size_bytes: row.audio_size_bytes,
+            audio_mime_type: row.audio_mime_type,
+            audio_updated_at: row.audio_updated_at,
+            cover_file_id: row.cover_file_id,
+            cover_url: cover_variant_url(row.cover_file_id, "medium"),
+            cover_hash: row.cover_hash,
+            cover_updated_at: row.cover_updated_at,
+        })
+        .collect();
+
+    let mut playlists = vec![OfflinePlaylistManifestItem {
+        id: -1,
+        title: "Likes".to_string(),
+        description: None,
+        updated_at: likes_updated_at,
+        is_own: true,
+        owner_name: None,
+        is_public: false,
+        is_saved: false,
+        kind: "likes".to_string(),
+        track_ids: liked_track_ids.clone(),
+    }];
+
+    playlists.extend(
+        playlist_rows
+            .into_iter()
+            .map(|row| OfflinePlaylistManifestItem {
+                id: row.id,
+                title: row.title,
+                description: row.description,
+                updated_at: row.updated_at,
+                is_own: row.is_own,
+                owner_name: Some(row.owner_name),
+                is_public: row.is_public,
+                is_saved: row.is_saved,
+                kind: "user".to_string(),
+                track_ids: row.track_ids,
+            }),
+    );
+
+    Json(OfflineManifestResponse {
+        generated_at,
+        tracks,
+        playlists,
+        liked_track_ids,
+        followed_artist_ids,
+    })
+    .into_response()
+}
+
 // ---------------------------------------------------------------------------
 // Last.fm account + scrobbling
 // ---------------------------------------------------------------------------
@@ -3830,7 +4041,11 @@ async fn stream_handler(
 
     // Look up track → audio_file_id → MediaFile
     let media = sqlx::query_as::<_, MediaFileRow>(
-        r#"SELECT mf.file_path, mf.mime_type::text as mime_type, mf.file_size_bytes
+        r#"SELECT mf.file_path,
+                  mf.mime_type::text as mime_type,
+                  mf.file_size_bytes,
+                  mf.sha256_hash::text AS sha256_hash,
+                  mf.created_at::text AS created_at
            FROM furumusic__track t
            JOIN furumusic__media_file mf ON mf.id = t.audio_file_id
            WHERE t.id = $1"#,
@@ -3855,6 +4070,7 @@ async fn stream_handler(
     }
 
     let file_size = media.file_size_bytes as u64;
+    let etag = media_etag("audio", track_id, &media.sha256_hash, None);
 
     // Parse Range header
     let range_header = request.headers().get(RANGE).and_then(|v| v.to_str().ok());
@@ -3874,6 +4090,9 @@ async fn stream_handler(
                 .header(ACCEPT_RANGES, "bytes")
                 .header(CONTENT_RANGE, format!("bytes {start}-{end}/{file_size}"))
                 .header(CONTENT_LENGTH, chunk_size.to_string())
+                .header("ETag", etag.as_str())
+                .header("X-Furumi-Content-Sha256", media.sha256_hash.as_str())
+                .header("X-Furumi-Content-Updated-At", media.created_at.as_str())
                 .body(Body::fixed(data))
                 .expect("valid response");
 
@@ -3892,6 +4111,9 @@ async fn stream_handler(
         .header(CONTENT_TYPE, media.mime_type.as_str())
         .header(ACCEPT_RANGES, "bytes")
         .header(CONTENT_LENGTH, file_size.to_string())
+        .header("ETag", etag.as_str())
+        .header("X-Furumi-Content-Sha256", media.sha256_hash.as_str())
+        .header("X-Furumi-Content-Updated-At", media.created_at.as_str())
         .body(Body::fixed(data))
         .expect("valid response");
 
@@ -4142,7 +4364,12 @@ async fn cover_response(
     };
 
     let media = sqlx::query_as::<_, MediaFileRow>(
-        "SELECT file_path, mime_type::text as mime_type, file_size_bytes FROM furumusic__media_file WHERE id = $1",
+        r#"SELECT file_path,
+                  mime_type::text as mime_type,
+                  file_size_bytes,
+                  sha256_hash::text AS sha256_hash,
+                  created_at::text AS created_at
+           FROM furumusic__media_file WHERE id = $1"#,
     )
     .bind(media_file_id)
     .fetch_optional(pool)
@@ -4160,26 +4387,30 @@ async fn cover_response(
         return Ok(json_error(StatusCode::NOT_FOUND, "file not found on disk"));
     }
 
-    let (response_path, content_type) = variant_name
+    let (response_path, content_type, etag_variant) = variant_name
         .and_then(crate::agent::cover_variants::variant_by_name)
         .map(|variant| {
             let variant_path = crate::agent::cover_variants::variant_path(&full_path, variant);
             if variant_path.exists() {
-                (variant_path, "image/jpeg")
+                (variant_path, "image/jpeg", Some(variant.name))
             } else {
-                (full_path.clone(), media.mime_type.as_str())
+                (full_path.clone(), media.mime_type.as_str(), None)
             }
         })
-        .unwrap_or_else(|| (full_path.clone(), media.mime_type.as_str()));
+        .unwrap_or_else(|| (full_path.clone(), media.mime_type.as_str(), None));
 
     let data = tokio::fs::read(&response_path)
         .await
         .map_err(|e| cot::Error::internal(e.to_string()))?;
+    let etag = media_etag("cover", media_file_id, &media.sha256_hash, etag_variant);
 
     let response = cot::http::Response::builder()
         .status(StatusCode::OK)
         .header(CONTENT_TYPE, content_type)
         .header(CONTENT_LENGTH, data.len().to_string())
+        .header("ETag", etag.as_str())
+        .header("X-Furumi-Content-Sha256", media.sha256_hash.as_str())
+        .header("X-Furumi-Content-Updated-At", media.created_at.as_str())
         .header("Cache-Control", "public, max-age=86400")
         .body(Body::fixed(data))
         .expect("valid response");
@@ -4187,6 +4418,13 @@ async fn cover_response(
     Ok(response)
 }
 
+fn media_etag(kind: &str, id: i64, sha256_hash: &str, variant: Option<&str>) -> String {
+    match variant {
+        Some(variant) => format!("\"{kind}-{id}-{variant}-{sha256_hash}\""),
+        None => format!("\"{kind}-{id}-{sha256_hash}\""),
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Player devices
 // ---------------------------------------------------------------------------
@@ -6246,6 +6484,32 @@ impl App for PlayerApp {
                 },
                 "player_me",
             ),
+            Route::with_handler_and_name(
+                "/offline/manifest",
+                {
+                    let pool = Arc::clone(&pool);
+                    let pool_config = Arc::clone(&pool_config);
+                    get(
+                        move |auth_ctx: auth::AuthContext, session: Session, db: Database| {
+                            let pool = Arc::clone(&pool);
+                            let pool_config = Arc::clone(&pool_config);
+                            async move {
+                                let pg_pool = pool
+                                    .get_or_init(|| async {
+                                        sqlx::postgres::PgPoolOptions::new()
+                                            .max_connections(5)
+                                            .connect(&pool_config.database_url)
+                                            .await
+                                            .expect("player pool")
+                                    })
+                                    .await;
+                                offline_manifest_handler(auth_ctx, session, db, pg_pool).await
+                            }
+                        },
+                    )
+                },
+                "player_offline_manifest",
+            ),
             Route::with_handler_and_name(
                 "/lastfm/status",
                 get({

src/player/rows.rs

@@ -56,6 +56,8 @@ pub(super) struct MediaFileRow {
     pub(super) file_path: String,
     pub(super) mime_type: String,
     pub(super) file_size_bytes: i64,
+    pub(super) sha256_hash: String,
+    pub(super) created_at: String,
 }
 
 #[derive(sqlx::FromRow)]
@@ -287,3 +289,30 @@ pub(super) struct ReleaseInfoRow {
     pub(super) year: Option<i32>,
     pub(super) cover_file_id: Option<i64>,
 }
+
+#[derive(sqlx::FromRow)]
+pub(super) struct OfflineTrackManifestRow {
+    pub(super) id: i64,
+    pub(super) updated_at: String,
+    pub(super) audio_file_id: i64,
+    pub(super) audio_hash: String,
+    pub(super) audio_size_bytes: i64,
+    pub(super) audio_mime_type: String,
+    pub(super) audio_updated_at: String,
+    pub(super) cover_file_id: Option<i64>,
+    pub(super) cover_hash: Option<String>,
+    pub(super) cover_updated_at: Option<String>,
+}
+
+#[derive(sqlx::FromRow)]
+pub(super) struct OfflinePlaylistManifestRow {
+    pub(super) id: i64,
+    pub(super) title: String,
+    pub(super) description: Option<String>,
+    pub(super) updated_at: String,
+    pub(super) is_own: bool,
+    pub(super) owner_name: String,
+    pub(super) is_public: bool,
+    pub(super) is_saved: bool,
+    pub(super) track_ids: Vec<i64>,
+}

src/scheduler/mod.rs

@@ -496,14 +496,24 @@ impl PendingReview {
             .await
     }
 
-    pub async fn exists_for_path(db: &Database, path: &str) -> cot::db::Result<bool> {
-        let all = Self::objects().all(db).await?;
-        let exists = all.iter().any(|r| {
-            let s = r.status.as_str();
-            // "rejected" and "failed" reviews should not block re-discovery
-            s != "rejected" && s != "failed" && r.input_path.as_deref() == Some(path)
-        });
-        Ok(exists)
+    /// Latest review row for an inbox path: `(id, status, updated_at)`.
+    ///
+    /// Used by inbox_discover to decide whether a file needs a new review,
+    /// a requeue of its existing row, or nothing at all — without creating
+    /// a fresh row per retry.
+    pub async fn latest_for_path(
+        pool: &sqlx::PgPool,
+        path: &str,
+    ) -> anyhow::Result<Option<(i64, String, String)>> {
+        let row: Option<(i64, String, String)> = sqlx::query_as(
+            "SELECT id, status::text, updated_at::text \
+             FROM furumusic__pending_review WHERE input_path = $1 \
+             ORDER BY id DESC LIMIT 1",
+        )
+        .bind(path)
+        .fetch_optional(pool)
+        .await?;
+        Ok(row)
     }
 
     /// Mark all "processing" reviews as "failed" — called at scheduler

templates/admin/v2.html

@@ -1613,7 +1613,7 @@ tbody tr:hover {
                             </button>
                             <button class="btn" @click="selectReviewFilter()" :disabled="reviews.total === 0">
                                 <i data-lucide="list-checks"></i>
-                                Select filter
+                                <span x-text="`Select all (${fmt(reviews.total)})`"></span>
                             </button>
                             <button class="btn" @click="clearReviewSelection()" :disabled="selectedReviewCount() === 0">
                                 <i data-lucide="x"></i>