diff --git a/Cargo.lock b/Cargo.lock index 6544778..9b78023 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1418,7 +1418,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" [[package]] name = "furumusic" -version = "0.4.6" +version = "0.4.8" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index d1472b7..d3859d7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "furumusic" -version = "0.4.7" +version = "0.4.8" edition = "2024" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" diff --git a/src/jobs/inbox_discover.rs b/src/jobs/inbox_discover.rs index 21e65ab..18f3b9d 100644 --- a/src/jobs/inbox_discover.rs +++ b/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 = 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 +} diff --git a/src/jobs/inbox_process.rs b/src/jobs/inbox_process.rs index 69557b7..709b121 100644 --- a/src/jobs/inbox_process.rs +++ b/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 = Vec::with_capacity(file_count); let mut failed_reviews: Vec = 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 { diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index c775324..c3e7228 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -496,14 +496,24 @@ impl PendingReview { .await } - pub async fn exists_for_path(db: &Database, path: &str) -> cot::db::Result { - 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> { + 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 diff --git a/templates/admin/v2.html b/templates/admin/v2.html index 8e62618..438bd08 100644 --- a/templates/admin/v2.html +++ b/templates/admin/v2.html @@ -1613,7 +1613,7 @@ tbody tr:hover {