Compare commits

...
4 Commits
Author SHA1 Message Date
Ultradesu 184371afca Added offline mode support
Build and Publish / Build and Publish Docker Image (push) Successful in 3m46s
2026-07-15 18:13:40 +03:00
Ultradesu 716da908c9 Fixed inbox_process task
Build and Publish / Build and Publish Docker Image (push) Successful in 5m34s
2026-07-08 15:22:21 +03:00
Ultradesu 0c120c0868 Added support of TUI player
Build and Publish / Build and Publish Docker Image (push) Successful in 2m38s
2026-06-10 23:20:18 +01:00
Ultradesu d9d0fbb7d1 Added cli client SSO login support
Build and Publish / Build and Publish Docker Image (push) Successful in 3m11s
2026-06-10 13:34:38 +01:00
11 changed files with 621 additions and 44 deletions
Generated
+1 -1
View File
@@ -1418,7 +1418,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
[[package]] [[package]]
name = "furumusic" name = "furumusic"
version = "0.4.4" version = "0.5.0"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"async-trait", "async-trait",
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "furumusic" name = "furumusic"
version = "0.4.5" version = "0.5.0"
edition = "2024" edition = "2024"
description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL" description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
+121 -8
View File
@@ -12,6 +12,21 @@ const AUDIO_EXTENSIONS: &[&str] = &[
"mp3", "flac", "ogg", "opus", "aac", "m4a", "wav", "ape", "wv", "wma", "tta", "aiff", "aif", "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; pub struct InboxDiscoverJob;
#[async_trait::async_trait] #[async_trait::async_trait]
@@ -76,6 +91,10 @@ impl Job for InboxDiscoverJob {
let mut audio_files = Vec::new(); let mut audio_files = Vec::new();
collect_audio_files(inbox, &mut audio_files).await?; 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())); log.info(&format!("Found {} audio files in inbox", audio_files.len()));
if audio_files.is_empty() { if audio_files.is_empty() {
return Ok(()); return Ok(());
@@ -87,6 +106,7 @@ impl Job for InboxDiscoverJob {
let mut discovered = 0u64; let mut discovered = 0u64;
let mut skipped_hash = 0u64; let mut skipped_hash = 0u64;
let mut skipped_existing = 0u64; let mut skipped_existing = 0u64;
let mut requeued = 0u64;
for (_folder, files) in &groups { for (_folder, files) in &groups {
for file_path in files { 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) crate::media_paths::path_for_root(&config.agent_inbox_dir, file_path)
.unwrap_or_else(|| file_path.to_string_lossy().to_string()); .unwrap_or_else(|| file_path.to_string_lossy().to_string());
// Skip if a PendingReview already exists for this path // One review row per path: any existing row blocks creating a
match PendingReview::exists_for_path(&ctx.db, &input_path_str).await { // new one. A stale "failed" row is requeued in place instead,
Ok(true) => { // so retries don't multiply rows. "rejected" stays rejected.
skipped_existing += 1; 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; continue;
} }
Ok(false) => {}
Err(e) => { Err(e) => {
log.warn(&format!( log.warn(&format!(
"Error checking existing review for {}: {e}", "Error checking existing review for {}: {e}",
@@ -215,8 +256,8 @@ impl Job for InboxDiscoverJob {
} }
log.info(&format!( log.info(&format!(
"Discovered {} new files, skipped {} (hash known), skipped {} (already queued)", "Discovered {} new files, requeued {} failed, skipped {} (hash known), skipped {} (already tracked)",
discovered, skipped_hash, skipped_existing discovered, requeued, skipped_hash, skipped_existing
)); ));
crate::metrics::record_agent_discover_files( crate::metrics::record_agent_discover_files(
audio_files.len() as u64, audio_files.len() as u64,
@@ -227,7 +268,7 @@ impl Job for InboxDiscoverJob {
// Trigger inbox_process in background if new files were discovered // Trigger inbox_process in background if new files were discovered
// and no orchestrator is already running // and no orchestrator is already running
if discovered > 0 { if discovered + requeued > 0 {
if crate::jobs::inbox_process::is_orchestrator_running() { if crate::jobs::inbox_process::is_orchestrator_running() {
log.info( log.info(
"New files discovered but inbox_process already running, it will pick them up", "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(); let ext = name.rsplit('.').next().unwrap_or("").to_lowercase();
AUDIO_EXTENSIONS.contains(&ext.as_str()) 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
}
+66 -14
View File
@@ -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. /// 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 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). /// Check if an orchestrator is currently running (used by inbox_discover to avoid redundant triggers).
pub fn is_orchestrator_running() -> bool { pub fn is_orchestrator_running() -> bool {
ORCHESTRATOR_RUNNING.load(Ordering::SeqCst) ORCHESTRATOR_RUNNING.load(Ordering::SeqCst)
@@ -214,14 +219,25 @@ impl Job for InboxProcessJob {
folder_rel, file_count, folder_rel, file_count,
)); ));
let (ok, fail) = // Large folders are split into chunks: a single LLM call for
process_folder_batch(&ctx.db, &config, &ctx.pool, &folder_rel, reviews, log) // 100+ files overflows the completion window and the whole
.await; // 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_ok += ok;
total_fail += fail; total_fail += fail;
}
log.info(&format!( 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..."); log.info("Phase 1: extracting metadata...");
let mut prepared: Vec<PreparedFile> = Vec::with_capacity(file_count); let mut prepared: Vec<PreparedFile> = Vec::with_capacity(file_count);
let mut failed_reviews: Vec<PendingReview> = Vec::new(); let mut failed_reviews: Vec<PendingReview> = Vec::new();
let mut merged_count = 0u64;
for mut review in reviews { for mut review in reviews {
let stored_input_path = review.input_path_str().to_owned(); let stored_input_path = review.input_path_str().to_owned();
@@ -355,9 +372,6 @@ async fn process_folder_batch(
.unwrap_or("unknown") .unwrap_or("unknown")
.to_owned(); .to_owned();
// Set status → processing
let _ = review.set_processing(db).await;
// Parse context_json // Parse context_json
let mut context: serde_json::Value = review let mut context: serde_json::Value = review
.context_json .context_json
@@ -365,6 +379,42 @@ async fn process_folder_batch(
.and_then(|s| serde_json::from_str(s).ok()) .and_then(|s| serde_json::from_str(s).ok())
.unwrap_or_default(); .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) // Extract metadata (with 60s timeout)
let path_for_meta = file_path.to_path_buf(); let path_for_meta = file_path.to_path_buf();
let metadata_start = std::time::Instant::now(); let metadata_start = std::time::Instant::now();
@@ -444,15 +494,16 @@ async fn process_folder_batch(
} }
log.info(&format!( log.info(&format!(
"Phase 1 done: {} prepared, {} failed metadata", "Phase 1 done: {} prepared, {} merged duplicates, {} failed",
prepared.len(), prepared.len(),
merged_count,
failed_reviews.len(), failed_reviews.len(),
)); ));
if prepared.is_empty() { if prepared.is_empty() {
let duration_ms = batch_start.elapsed().as_millis() as i64; let duration_ms = batch_start.elapsed().as_millis() as i64;
let _ = run.set_completed(db, duration_ms, &log.output()).await; 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) // 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}"); let err_msg = format!("Batch LLM call failed: {e}");
log.error(&err_msg); log.error(&err_msg);
// Mark all files as failed // Mark all files as failed
let prepared_count = prepared.len() as u64;
for mut p in prepared { for mut p in prepared {
let _ = p.review.set_failed(db, &err_msg).await; let _ = p.review.set_failed(db, &err_msg).await;
crate::metrics::record_agent_file_processed("failed", "failed"); 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 duration_ms = batch_start.elapsed().as_millis() as i64;
let _ = run let _ = run
.set_failed(db, duration_ms, &log.output(), &err_msg) .set_failed(db, duration_ms, &log.output(), &err_msg)
.await; .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 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 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; let mut fail_count = failed_reviews.len() as u64;
for mut p in prepared { for mut p in prepared {
+1
View File
@@ -883,6 +883,7 @@ const KNOWN_HTTP_ROUTES: &[&str] = &[
"/api/player/lastfm/now-playing", "/api/player/lastfm/now-playing",
"/api/player/lastfm/scrobble", "/api/player/lastfm/scrobble",
"/api/player/agent-queue", "/api/player/agent-queue",
"/api/player/offline/manifest",
"/api/player/torrents", "/api/player/torrents",
"/api/player/torrents/session/{id}", "/api/player/torrents/session/{id}",
"/api/player/torrents/preview", "/api/player/torrents/preview",
+74 -5
View File
@@ -977,27 +977,54 @@ fn safe_mobile_redirect_uri(raw: Option<&str>) -> Option<String> {
if lower.starts_with("furumi://") || lower.starts_with("furumusic://") { if lower.starts_with("furumi://") || lower.starts_with("furumusic://") {
return Some(value.to_owned()); return Some(value.to_owned());
} }
if is_loopback_http_redirect(&lower) {
return Some(value.to_owned());
}
None 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 { fn mobile_redirect_success(app_redirect_uri: &str, code: &str) -> cot::response::Response {
let deep_link = append_query_param(app_redirect_uri, "code", code); 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( mobile_deep_link_page(
"success", "success",
"Sign-in complete", "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, None,
Some(code),
&deep_link, &deep_link,
) )
} }
fn mobile_redirect_error(app_redirect_uri: &str, error: &str) -> cot::response::Response { fn mobile_redirect_error(app_redirect_uri: &str, error: &str) -> cot::response::Response {
let deep_link = append_query_param(app_redirect_uri, "error", error); 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( mobile_deep_link_page(
"error", "error",
"Sign-in failed", "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), Some(error),
None,
&deep_link, &deep_link,
) )
} }
@@ -1007,6 +1034,7 @@ fn mobile_deep_link_page(
title: &str, title: &str,
message: &str, message: &str,
detail: Option<&str>, detail: Option<&str>,
code: Option<&str>,
deep_link: &str, deep_link: &str,
) -> cot::response::Response { ) -> cot::response::Response {
let state_class = html_escape(state); let state_class = html_escape(state);
@@ -1015,6 +1043,15 @@ fn mobile_deep_link_page(
let detail_html = detail let detail_html = detail
.map(|value| format!(r#"<p class="detail">Reason: {}</p>"#, html_escape(value))) .map(|value| format!(r#"<p class="detail">Reason: {}</p>"#, html_escape(value)))
.unwrap_or_default(); .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_html = html_escape(deep_link);
let deep_link_js = let deep_link_js =
serde_json::to_string(deep_link).expect("serializing URL string cannot fail"); serde_json::to_string(deep_link).expect("serializing URL string cannot fail");
@@ -1095,6 +1132,19 @@ fn mobile_deep_link_page(
font-size: 13px; font-size: 13px;
color: #89847c; 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> </style>
</head> </head>
<body> <body>
@@ -1105,15 +1155,13 @@ fn mobile_deep_link_page(
{detail_html} {detail_html}
<a href="{deep_link_html}">Open Furumi</a> <a href="{deep_link_html}">Open Furumi</a>
<p class="hint">If nothing happens, use the button above.</p> <p class="hint">If nothing happens, use the button above.</p>
{code_html}
</main> </main>
<script> <script>
const deepLink = {deep_link_js}; const deepLink = {deep_link_js};
window.setTimeout(() => {{ window.setTimeout(() => {{
window.location.href = deepLink; window.location.href = deepLink;
}}, 100); }}, 100);
window.setTimeout(() => {{
window.close();
}}, 1800);
</script> </script>
</body> </body>
</html>"#, </html>"#,
@@ -1230,4 +1278,25 @@ mod tests {
); );
assert!(safe_mobile_redirect_uri(Some("https://example.com/callback")).is_none()); 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());
}
} }
+39
View File
@@ -325,6 +325,45 @@ pub(super) struct UserProfile {
pub(super) stats: UserStats, 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)] #[derive(Debug, Serialize, JsonSchema)]
pub(super) struct LastfmStatus { pub(super) struct LastfmStatus {
pub(super) configured: bool, pub(super) configured: bool,
+270 -6
View File
@@ -861,6 +861,12 @@ fn native_device_name_from_user_agent(user_agent: Option<&str>) -> Option<String
None => "Furumi MacOS".to_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 None
} }
@@ -890,6 +896,9 @@ fn device_kind_from_user_agent(user_agent: Option<&str>) -> &'static str {
if ua.contains("furumimac") { if ua.contains("furumimac") {
return "computer"; return "computer";
} }
if ua.contains("furumitui/") || ua.contains("furumi-tui/") {
return "computer";
}
if ua.contains("iphone") || (ua.contains("android") && ua.contains("mobile")) { if ua.contains("iphone") || (ua.contains("android") && ua.contains("mobile")) {
"phone" "phone"
} else if ua.contains("ipad") || ua.contains("tablet") || ua.contains("android") { } 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"); 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] #[test]
fn keeps_browser_fallback_for_generic_android_user_agents() { fn keeps_browser_fallback_for_generic_android_user_agents() {
let user_agent = Some("Mozilla/5.0 Android Mobile"); let user_agent = Some("Mozilla/5.0 Android Mobile");
@@ -1023,6 +1048,192 @@ async fn me_handler(
.into_response() .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 // Last.fm account + scrobbling
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -3830,7 +4041,11 @@ async fn stream_handler(
// Look up track → audio_file_id → MediaFile // Look up track → audio_file_id → MediaFile
let media = sqlx::query_as::<_, MediaFileRow>( 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 FROM furumusic__track t
JOIN furumusic__media_file mf ON mf.id = t.audio_file_id JOIN furumusic__media_file mf ON mf.id = t.audio_file_id
WHERE t.id = $1"#, WHERE t.id = $1"#,
@@ -3855,6 +4070,7 @@ async fn stream_handler(
} }
let file_size = media.file_size_bytes as u64; let file_size = media.file_size_bytes as u64;
let etag = media_etag("audio", track_id, &media.sha256_hash, None);
// Parse Range header // Parse Range header
let range_header = request.headers().get(RANGE).and_then(|v| v.to_str().ok()); 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(ACCEPT_RANGES, "bytes")
.header(CONTENT_RANGE, format!("bytes {start}-{end}/{file_size}")) .header(CONTENT_RANGE, format!("bytes {start}-{end}/{file_size}"))
.header(CONTENT_LENGTH, chunk_size.to_string()) .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)) .body(Body::fixed(data))
.expect("valid response"); .expect("valid response");
@@ -3892,6 +4111,9 @@ async fn stream_handler(
.header(CONTENT_TYPE, media.mime_type.as_str()) .header(CONTENT_TYPE, media.mime_type.as_str())
.header(ACCEPT_RANGES, "bytes") .header(ACCEPT_RANGES, "bytes")
.header(CONTENT_LENGTH, file_size.to_string()) .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)) .body(Body::fixed(data))
.expect("valid response"); .expect("valid response");
@@ -4142,7 +4364,12 @@ async fn cover_response(
}; };
let media = sqlx::query_as::<_, MediaFileRow>( 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) .bind(media_file_id)
.fetch_optional(pool) .fetch_optional(pool)
@@ -4160,26 +4387,30 @@ async fn cover_response(
return Ok(json_error(StatusCode::NOT_FOUND, "file not found on disk")); 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) .and_then(crate::agent::cover_variants::variant_by_name)
.map(|variant| { .map(|variant| {
let variant_path = crate::agent::cover_variants::variant_path(&full_path, variant); let variant_path = crate::agent::cover_variants::variant_path(&full_path, variant);
if variant_path.exists() { if variant_path.exists() {
(variant_path, "image/jpeg") (variant_path, "image/jpeg", Some(variant.name))
} else { } 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) let data = tokio::fs::read(&response_path)
.await .await
.map_err(|e| cot::Error::internal(e.to_string()))?; .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() let response = cot::http::Response::builder()
.status(StatusCode::OK) .status(StatusCode::OK)
.header(CONTENT_TYPE, content_type) .header(CONTENT_TYPE, content_type)
.header(CONTENT_LENGTH, data.len().to_string()) .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") .header("Cache-Control", "public, max-age=86400")
.body(Body::fixed(data)) .body(Body::fixed(data))
.expect("valid response"); .expect("valid response");
@@ -4187,6 +4418,13 @@ async fn cover_response(
Ok(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 // Player devices
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
@@ -6246,6 +6484,32 @@ impl App for PlayerApp {
}, },
"player_me", "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( Route::with_handler_and_name(
"/lastfm/status", "/lastfm/status",
get({ get({
+29
View File
@@ -56,6 +56,8 @@ pub(super) struct MediaFileRow {
pub(super) file_path: String, pub(super) file_path: String,
pub(super) mime_type: String, pub(super) mime_type: String,
pub(super) file_size_bytes: i64, pub(super) file_size_bytes: i64,
pub(super) sha256_hash: String,
pub(super) created_at: String,
} }
#[derive(sqlx::FromRow)] #[derive(sqlx::FromRow)]
@@ -287,3 +289,30 @@ pub(super) struct ReleaseInfoRow {
pub(super) year: Option<i32>, pub(super) year: Option<i32>,
pub(super) cover_file_id: Option<i64>, 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>,
}
+18 -8
View File
@@ -496,14 +496,24 @@ impl PendingReview {
.await .await
} }
pub async fn exists_for_path(db: &Database, path: &str) -> cot::db::Result<bool> { /// Latest review row for an inbox path: `(id, status, updated_at)`.
let all = Self::objects().all(db).await?; ///
let exists = all.iter().any(|r| { /// Used by inbox_discover to decide whether a file needs a new review,
let s = r.status.as_str(); /// a requeue of its existing row, or nothing at all — without creating
// "rejected" and "failed" reviews should not block re-discovery /// a fresh row per retry.
s != "rejected" && s != "failed" && r.input_path.as_deref() == Some(path) pub async fn latest_for_path(
}); pool: &sqlx::PgPool,
Ok(exists) 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 /// Mark all "processing" reviews as "failed" — called at scheduler
+1 -1
View File
@@ -1613,7 +1613,7 @@ tbody tr:hover {
</button> </button>
<button class="btn" @click="selectReviewFilter()" :disabled="reviews.total === 0"> <button class="btn" @click="selectReviewFilter()" :disabled="reviews.total === 0">
<i data-lucide="list-checks"></i> <i data-lucide="list-checks"></i>
Select filter <span x-text="`Select all (${fmt(reviews.total)})`"></span>
</button> </button>
<button class="btn" @click="clearReviewSelection()" :disabled="selectedReviewCount() === 0"> <button class="btn" @click="clearReviewSelection()" :disabled="selectedReviewCount() === 0">
<i data-lucide="x"></i> <i data-lucide="x"></i>