From bd1d3a4ebcb7ba8fe6a7f16e9bc40d9e6db918f0 Mon Sep 17 00:00:00 2001 From: Aleksandr Bogomiakov Date: Fri, 16 Oct 2026 00:59:09 +0100 Subject: [PATCH] Optimize Last.fm popularity lookups with light and full modes --- README.md | 23 + src/admin/mod.rs | 96 +++++ src/admin/v2.rs | 15 + src/admin/views.rs | 24 ++ src/i18n/phrases.rs | 2 + src/jobs/lastfm_popularity.rs | 553 +++++++++++++++++-------- src/jobs/lastfm_popularity_tests.rs | 337 +++++++++++++++ src/music/lastfm_popularity_lookup.sql | 27 ++ src/music/mod.rs | 30 ++ templates/admin/job_detail.html | 12 +- templates/admin/v2.html | 21 +- 11 files changed, 955 insertions(+), 185 deletions(-) create mode 100644 src/jobs/lastfm_popularity_tests.rs create mode 100644 src/music/lastfm_popularity_lookup.sql diff --git a/README.md b/README.md index af7a047..57d79ee 100644 --- a/README.md +++ b/README.md @@ -131,6 +131,22 @@ The browser interface and JSON API are served by the same application. Import, artwork, metadata enrichment, similarity indexing, and maintenance run as durable background jobs rather than blocking playback requests. +The `lastfm_popularity` job defaults to **light** mode: it requests popularity +only for tracks without a completed lookup or with a changed title or primary +artist. Both matches (including zero counts) and explicit Last.fm “Track not +found” results are skipped on subsequent light runs. HTTP errors, timeouts, +invalid responses, and other API errors remain eligible for retry. A failed +full refresh also makes that track eligible for the next light run. + +Choose **Full: refresh all visible tracks** in the task's admin parameters to +refresh existing statistics and retry previous misses. The options endpoint +`POST /admin/v2/api/jobs/lastfm_popularity/run-options` accepts +`{"refresh_existing": true}`; an empty object, scheduled runs, and the ordinary +run endpoint use light mode. Neither mode limits the number of tracks processed. +An authoritative miss clears current statistics while preserving their history. +The migration preserves existing popularity; legacy misses need one more lookup +because older versions did not persist them. + ## Contributing Bug reports, design discussions, and patches are welcome. Before submitting a @@ -141,3 +157,10 @@ cargo fmt --all -- --check cargo clippy --all-targets -- -D warnings cargo test --all-targets ``` + +Last.fm database tests use temporary tables on a disposable PostgreSQL database: + +```bash +FURUMUSIC_LASTFM_TEST_DATABASE_URL=postgres://user:pass@localhost/test_db \ + cargo test lastfm_popularity -- --include-ignored +``` diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 1284539..fcb5255 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -388,6 +388,32 @@ impl App for AdminApp { }), "admin_v2_artwork_backfill_run_options", ), + Route::with_handler_and_name( + "/v2/api/jobs/lastfm_popularity/run-options", + cot::router::method::post({ + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + move |session: Session, + db: Database, + json: Json| { + 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("admin pool") + }) + .await; + v2::run_lastfm_popularity(session, db, pg_pool, json).await + } + } + }), + "admin_v2_lastfm_popularity_run_options", + ), Route::with_handler_and_name( "/v2/api/jobs/{name}/run", cot::router::method::post({ @@ -1237,6 +1263,33 @@ impl App for AdminApp { }), "admin_metadata_backfill_run", ), + Route::with_handler_and_name( + "/jobs/lastfm_popularity/run-options", + cot::router::method::post({ + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + move |session: Session, db: Database, + form: RequestForm| { + let pool = Arc::clone(&pool); + let pool_config = Arc::clone(&pool_config); + async move { + let admin = match auth::require_admin_or_redirect(&session, &db).await { + Ok(u) => u, + Err(resp) => return Ok(resp), + }; + let pg_pool = pool.get_or_init(|| async { + sqlx::postgres::PgPoolOptions::new() + .max_connections(3) + .connect(&pool_config.database_url) + .await + .expect("admin pool") + }).await; + views::lastfm_popularity_run(admin, &db, pg_pool, form).await + } + } + }), + "admin_lastfm_popularity_run", + ), Route::with_handler_and_name( "/jobs/{name}/run", cot::router::method::post({ @@ -1490,3 +1543,46 @@ impl App for AdminApp { all } } + +/// Start an options-based run for both admin interfaces. Ordinary scheduler +/// entry points continue to call Job::run with light-mode defaults. +async fn start_lastfm_popularity( + db: &Database, + pool: &sqlx::PgPool, + options: crate::jobs::lastfm_popularity::LastfmPopularityOptions, +) -> cot::Result { + use crate::scheduler::{JobContext, JobLog, JobRun}; + + let mut run = JobRun::create_running(db, "lastfm_popularity", "manual") + .await + .map_err(|err| cot::Error::internal(format!("failed to create job run: {err}")))?; + let run_id = run.id_val(); + let (live_config, _) = AppConfig::load_with_db(db).await; + let db = db.clone(); + let pool = pool.clone(); + tokio::spawn(async move { + let start = std::time::Instant::now(); + let ctx = JobContext { + config: Arc::new(live_config), + db: db.clone(), + pool: pool.clone(), + run_id, + registry: Arc::new(JobRegistry::new()), + }; + let mut log = JobLog::with_live_flush(pool, run_id); + let result = + crate::jobs::lastfm_popularity::run_with_options(&ctx, &mut log, options).await; + let duration_ms = start.elapsed().as_millis() as i64; + match result { + Ok(()) => { + let _ = run.set_completed(&db, duration_ms, &log.output()).await; + } + Err(err) => { + let _ = run + .set_failed(&db, duration_ms, &log.output(), &err.to_string()) + .await; + } + } + }); + Ok(run_id) +} diff --git a/src/admin/v2.rs b/src/admin/v2.rs index c9e8b5f..ab526a0 100644 --- a/src/admin/v2.rs +++ b/src/admin/v2.rs @@ -110,6 +110,8 @@ pub struct ArtworkBackfillRunRequest { overwrite_existing: bool, } +pub type LastfmPopularityRunRequest = crate::jobs::lastfm_popularity::LastfmPopularityOptions; + #[derive(Debug, Deserialize)] pub(super) struct UpdateLibraryItemRequest { kind: String, @@ -1734,6 +1736,19 @@ pub async fn run_metadata_backfill( Json(JobRunStartedDto { ok: true, run_id }).into_response() } +pub async fn run_lastfm_popularity( + session: Session, + db: Database, + pool: &PgPool, + Json(options): Json, +) -> cot::Result { + if let Err(response) = require_admin_json(&session, &db).await { + return Ok(response); + } + let run_id = super::start_lastfm_popularity(&db, pool, options).await?; + Json(JobRunStartedDto { ok: true, run_id }).into_response() +} + pub async fn run_artwork_backfill( session: Session, db: Database, diff --git a/src/admin/views.rs b/src/admin/views.rs index 19fdf11..57edc96 100644 --- a/src/admin/views.rs +++ b/src/admin/views.rs @@ -1453,6 +1453,30 @@ pub async fn job_run_now( Ok(auth::redirect(&format!("/admin/jobs/{job_name}"))) } +#[derive(Debug, Form)] +pub struct LastfmPopularityForm { + refresh_existing: Option, +} + +pub async fn lastfm_popularity_run( + _admin: AuthenticatedUser, + db: &Database, + pool: &sqlx::PgPool, + form: RequestForm, +) -> cot::Result> { + let RequestForm(result) = form; + let FormResult::Ok(data) = result else { + return Ok(auth::redirect("/admin/jobs/lastfm_popularity")); + }; + let options = crate::jobs::lastfm_popularity::LastfmPopularityOptions { + refresh_existing: data.refresh_existing.is_some(), + }; + let run_id = super::start_lastfm_popularity(db, pool, options).await?; + Ok(auth::redirect(&format!( + "/admin/jobs/lastfm_popularity/runs/{run_id}" + ))) +} + pub async fn metadata_backfill_run( _admin: AuthenticatedUser, db: &Database, diff --git a/src/i18n/phrases.rs b/src/i18n/phrases.rs index 3ff1a62..577e7ca 100644 --- a/src/i18n/phrases.rs +++ b/src/i18n/phrases.rs @@ -194,6 +194,8 @@ translations! { jobs_back_to_list: "Back to jobs" , "Назад к заданиям"; jobs_run_detail: "Run detail" , "Детали запуска"; jobs_back_to_job: "Back to job" , "Назад к заданию"; + jobs_lastfm_light_help: "Light mode checks only unchecked tracks, changed titles or artists, and failed lookups. Previous matches and confirmed misses are skipped." , "Лайт-режим проверяет только непроверенные треки, изменённые названия или исполнителей и ошибки прошлых запросов. Найденные и подтверждённо отсутствующие треки пропускаются."; + jobs_lastfm_full: "Full refresh: check all visible tracks, including previous matches and misses" , "Полное обновление: проверить все видимые треки, включая ранее найденные и отсутствующие"; jobs_metadata_backfill_options: "Metadata backfill options" , "Параметры обновления метадаты"; jobs_metadata_backfill_fields: "Fields to update" , "Поля для обновления"; jobs_metadata_backfill_fill_missing: "Fill missing only" , "Заполнить только пустые"; diff --git a/src/jobs/lastfm_popularity.rs b/src/jobs/lastfm_popularity.rs index 8056697..109a878 100644 --- a/src/jobs/lastfm_popularity.rs +++ b/src/jobs/lastfm_popularity.rs @@ -1,19 +1,54 @@ +use std::time::Duration; + +use anyhow::Context; +use reqwest::{Client, StatusCode}; use serde::Deserialize; +use sqlx::PgConnection; use crate::scheduler::{Job, JobContext, JobLog}; pub struct LastfmPopularityJob; -const LASTFM_REQUEST_DELAY: std::time::Duration = std::time::Duration::from_millis(1200); +const LASTFM_API_URL: &str = "https://ws.audioscrobbler.com/2.0/"; +const LASTFM_REQUEST_DELAY: Duration = Duration::from_millis(1200); +const TRACK_PAGE_SIZE: i64 = 500; +const ADVISORY_LOCK_ID: i64 = 0x4c41_5354_464d_504f; // "LASTFMPO" + +/// Scheduled and ordinary manual runs use light mode. Full mode is opt-in. +#[derive(Debug, Clone, Copy, Default, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct LastfmPopularityOptions { + pub refresh_existing: bool, +} #[derive(Debug, sqlx::FromRow)] struct TrackLookupRow { id: i64, title: String, artist_name: Option, - lastfm_updated_at: Option, } +// Store the lookup identity alongside the result so every metadata editing path +// (including artist renames and link changes) automatically invalidates it. +const TRACK_CANDIDATES_SQL: &str = r#" + SELECT t.id, t.title::text AS title, primary_artist.name AS artist_name + FROM furumusic__track t + LEFT JOIN LATERAL ( + SELECT a.name::text AS name + FROM furumusic__track_artist ta + JOIN furumusic__artist a ON a.id = ta.artist_id + WHERE ta.track_id = t.id AND ta.role <> 'featuring' + ORDER BY ta.position, ta.artist_id + LIMIT 1 + ) primary_artist ON true + WHERE t.is_hidden = false AND t.id > $1 AND t.id <= $2 + AND ($3 OR t.lastfm_checked_at IS NULL + OR t.lastfm_lookup_title IS DISTINCT FROM t.title::text + OR t.lastfm_lookup_artist IS DISTINCT FROM primary_artist.name) + ORDER BY t.id + LIMIT $4 +"#; + #[derive(Debug, Deserialize)] struct LastfmTrackInfoResponse { track: Option, @@ -23,8 +58,69 @@ struct LastfmTrackInfoResponse { #[derive(Debug, Deserialize)] struct LastfmTrack { - listeners: Option, - playcount: Option, + listeners: String, + playcount: String, +} + +#[derive(Debug, PartialEq)] +enum LookupOutcome { + Found { listeners: i64, playcount: i64 }, + NotFound, +} + +#[derive(Debug)] +enum LookupError { + Http(StatusCode), + Api { code: i32, message: String }, + Transport(reqwest::Error), + InvalidResponse(String), +} + +impl LookupError { + fn should_stop_run(&self) -> bool { + match self { + Self::Http(status) => matches!(status.as_u16(), 401 | 403 | 429), + Self::Api { code, .. } => matches!(code, 2 | 3 | 4 | 5 | 9 | 10 | 11 | 13 | 26 | 29), + _ => false, + } + } +} + +impl std::fmt::Display for LookupError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Http(status) => write!(f, "Last.fm HTTP {status}"), + Self::Api { code, message } => write!(f, "Last.fm API error {code}: {message}"), + Self::Transport(err) => write!(f, "Last.fm request failed: {err}"), + Self::InvalidResponse(message) => write!(f, "Invalid Last.fm response: {message}"), + } + } +} + +impl std::error::Error for LookupError {} + +#[derive(Default)] +struct PopularityStats { + considered: u64, + requested: u64, + updated: u64, + not_found: u64, + skipped: u64, + failed: u64, +} + +impl PopularityStats { + fn summary(&self) -> String { + format!( + "{} considered, {} requested, {} updated, {} not found, {} skipped, {} failed", + self.considered, + self.requested, + self.updated, + self.not_found, + self.skipped, + self.failed + ) + } } #[async_trait::async_trait] @@ -34,7 +130,7 @@ impl Job for LastfmPopularityJob { } fn description(&self) -> &'static str { - "Update Last.fm playcount/listener popularity for library tracks" + "Fetch Last.fm popularity for unchecked tracks; optionally refresh all tracks" } fn default_cron(&self) -> &'static str { @@ -43,159 +139,236 @@ impl Job for LastfmPopularityJob { } async fn run(&self, ctx: &JobContext, log: &mut JobLog) -> anyhow::Result<()> { - let api_key = ctx.config.lastfm_api_key.trim(); - if api_key.is_empty() { - log.warn("lastfm_api_key is not configured, skipping Last.fm popularity update"); - return Ok(()); - } - - let tracks = sqlx::query_as::<_, TrackLookupRow>( - r#"SELECT t.id, - t.title::text AS title, - t.lastfm_updated_at::text AS lastfm_updated_at, - ( - SELECT a.name::text - FROM furumusic__track_artist ta - JOIN furumusic__artist a ON a.id = ta.artist_id - WHERE ta.track_id = t.id AND ta.role <> 'featuring' - ORDER BY ta.position - LIMIT 1 - ) AS artist_name - FROM furumusic__track t - WHERE t.is_hidden = false - ORDER BY t.lastfm_updated_at IS NOT NULL, t.lastfm_updated_at ASC, t.id ASC"#, - ) - .fetch_all(&ctx.pool) - .await?; - - if tracks.is_empty() { - log.info("No visible tracks found for Last.fm popularity update"); - return Ok(()); - } - - log.info(&format!( - "Starting Last.fm popularity update for {} visible tracks; oldest or missing ratings are processed first; request delay is {} ms; rating formula is ln(playcount + 1) * ln(listeners + 1)", - tracks.len(), - LASTFM_REQUEST_DELAY.as_millis() - )); - - let client = reqwest::Client::builder() - .user_agent("furumusic-lastfm-popularity/0.1") - .timeout(std::time::Duration::from_secs(15)) - .build()?; - let mut updated = 0u64; - let mut skipped = 0u64; - let mut failed = 0u64; - - for (index, track) in tracks.iter().enumerate() { - let Some(artist) = track - .artist_name - .as_deref() - .map(str::trim) - .filter(|v| !v.is_empty()) - else { - skipped += 1; - log.warn(&format!( - "Skipping track {} \"{}\": no primary artist", - track.id, track.title - )); - continue; - }; - - log.info(&format!( - "Last.fm lookup {}/{}: track {} \"{}\" by \"{}\" (previous update: {})", - index + 1, - tracks.len(), - track.id, - track.title, - artist, - track.lastfm_updated_at.as_deref().unwrap_or("never") - )); - let result = fetch_track_info(&client, api_key, artist, &track.title).await; - match result { - Ok(Some((listeners, playcount))) => { - let rating = popularity_rating(listeners, playcount); - let fetched_at = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); - sqlx::query( - r#"UPDATE furumusic__track - SET lastfm_listeners = $2, - lastfm_playcount = $3, - lastfm_rating = $4, - lastfm_updated_at = $5 - WHERE id = $1"#, - ) - .bind(track.id) - .bind(listeners) - .bind(playcount) - .bind(rating) - .bind(&fetched_at) - .execute(&ctx.pool) - .await?; - sqlx::query( - r#"INSERT INTO furumusic__track_popularity_history - (track_id, source, listeners, playcount, rating, fetched_at) - VALUES ($1, 'lastfm', $2, $3, $4, $5)"#, - ) - .bind(track.id) - .bind(listeners) - .bind(playcount) - .bind(rating) - .bind(&fetched_at) - .execute(&ctx.pool) - .await?; - updated += 1; - log.info(&format!( - "Updated track {} \"{}\" by \"{}\": listeners={listeners}, playcount={playcount}, rating={rating:.4}", - track.id, track.title, artist - )); - } - Ok(None) => { - skipped += 1; - log.warn(&format!( - "Last.fm has no usable match for track {} \"{}\" by \"{}\"", - track.id, track.title, artist - )); - } - Err(err) if err.to_string().contains("Last.fm rate limit exceeded") => { - failed += 1; - log.error("Last.fm rate limit exceeded; stopping this run early"); - break; - } - Err(err) => { - failed += 1; - log.warn(&format!( - "Last.fm lookup failed for track {} \"{}\" / \"{}\": {err}", - track.id, artist, track.title - )); - } - } - - if (index + 1) % 50 == 0 { - log.info(&format!( - "Last.fm progress: {}/{} tracks, {updated} updated, {skipped} skipped, {failed} failed", - index + 1, - tracks.len() - )); - } - tokio::time::sleep(LASTFM_REQUEST_DELAY).await; - } - - log.info(&format!( - "Last.fm popularity update finished: {updated} updated, {skipped} skipped, {failed} failed, {} considered", - tracks.len() - )); - Ok(()) + run_with_options(ctx, log, LastfmPopularityOptions::default()).await } } +pub async fn run_with_options( + ctx: &JobContext, + log: &mut JobLog, + options: LastfmPopularityOptions, +) -> anyhow::Result<()> { + let api_key = ctx.config.lastfm_api_key.trim(); + if api_key.is_empty() { + log.warn("lastfm_api_key is not configured, skipping Last.fm popularity update"); + return Ok(()); + } + + let mut conn = ctx.pool.acquire().await?; + // A session lock must never return to the pool, including on cancellation. + // All database work below uses this connection; no transaction spans HTTP. + conn.close_on_drop(); + if !try_lock(&mut conn).await? { + log.info("Last.fm popularity update is already running; skipping this run"); + return Ok(()); + } + + let client = Client::builder() + .user_agent(format!( + "furumusic-lastfm-popularity/{}", + env!("CARGO_PKG_VERSION") + )) + .timeout(Duration::from_secs(15)) + .build()?; + log.info(&format!( + "Last.fm popularity options: mode={}; request delay={} ms", + if options.refresh_existing { + "full" + } else { + "light" + }, + LASTFM_REQUEST_DELAY.as_millis() + )); + let result = process_tracks(&mut conn, &client, api_key, options, log).await; + // Release promptly on normal completion; close_on_drop also covers errors + // and task cancellation without leaking the lock into another run. + conn.close().await?; + result +} + +async fn try_lock(conn: &mut PgConnection) -> anyhow::Result { + Ok(sqlx::query_scalar("SELECT pg_try_advisory_lock($1)") + .bind(ADVISORY_LOCK_ID) + .fetch_one(conn) + .await?) +} + +async fn fetch_candidates( + conn: &mut PgConnection, + after_id: i64, + max_id: i64, + options: LastfmPopularityOptions, +) -> anyhow::Result> { + Ok(sqlx::query_as(TRACK_CANDIDATES_SQL) + .bind(after_id) + .bind(max_id) + .bind(options.refresh_existing) + .bind(TRACK_PAGE_SIZE) + .fetch_all(conn) + .await?) +} + +async fn process_tracks( + conn: &mut PgConnection, + client: &Client, + api_key: &str, + options: LastfmPopularityOptions, + log: &mut JobLog, +) -> anyhow::Result<()> { + // Bound the snapshot by id, not by count: every eligible track at the start + // can be processed, while concurrent imports wait for the next run. + let max_id: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(id), 0) FROM furumusic__track") + .fetch_one(&mut *conn) + .await?; + let mut after_id = 0; + let mut stats = PopularityStats::default(); + let mut next_request_at = tokio::time::Instant::now(); + let mut last_progress = std::time::Instant::now(); + loop { + let tracks = fetch_candidates(conn, after_id, max_id, options).await?; + if tracks.is_empty() { + break; + } + // Invalidate the page before HTTP, preserving its statistics. A failed + // write or cancellation during a full refresh then also leaves work for + // the next light run. This costs one update per page, not per request. + let ids: Vec = tracks.iter().map(|track| track.id).collect(); + mark_pending(conn, &ids).await?; + for track in tracks { + // Advance even after an error so a failing track is tried only once + // per run and cannot starve the remaining candidates. + after_id = track.id; + stats.considered += 1; + let artist = track.artist_name.as_deref().unwrap_or_default(); + if artist.trim().is_empty() || track.title.trim().is_empty() { + stats.skipped += 1; + // Missing local metadata says nothing about the Last.fm catalog. + continue; + } + tokio::time::sleep_until(next_request_at).await; + stats.requested += 1; + let outcome = fetch_track_info(client, api_key, artist, &track.title).await; + next_request_at = tokio::time::Instant::now() + LASTFM_REQUEST_DELAY; + match outcome { + Ok(outcome) => { + if save_outcome(conn, &track, &outcome).await? { + match outcome { + LookupOutcome::Found { .. } => stats.updated += 1, + LookupOutcome::NotFound => stats.not_found += 1, + } + } else { + // Metadata changed or the track was removed during HTTP. + stats.skipped += 1; + } + } + Err(err) => { + stats.failed += 1; + tracing::warn!(track_id = track.id, error = %err, "Last.fm popularity lookup failed"); + if stats.failed <= 5 || err.should_stop_run() { + log.warn(&format!( + "Last.fm lookup failed for track {}: {err}", + track.id + )); + } + if err.should_stop_run() { + log.error(&format!("Last.fm popularity stopped: {}", stats.summary())); + return Err(err.into()); + } + } + } + // Keep live logs bounded: the shared logger rewrites its entire + // accumulated output whenever a new line is added. + if last_progress.elapsed() >= Duration::from_secs(30) { + log.info(&format!("Last.fm progress: {}", stats.summary())); + last_progress = std::time::Instant::now(); + } + } + } + log.info(&format!("Last.fm popularity finished: {}", stats.summary())); + anyhow::ensure!( + stats.failed == 0, + "{} Last.fm lookups failed; they will be retried in light mode", + stats.failed + ); + Ok(()) +} + +async fn mark_pending(conn: &mut PgConnection, track_ids: &[i64]) -> anyhow::Result<()> { + sqlx::query( + "UPDATE furumusic__track SET lastfm_checked_at = NULL \ + WHERE id = ANY($1) AND lastfm_checked_at IS NOT NULL", + ) + .bind(track_ids) + .execute(conn) + .await?; + Ok(()) +} + +async fn save_outcome( + conn: &mut PgConnection, + track: &TrackLookupRow, + outcome: &LookupOutcome, +) -> anyhow::Result { + let (listeners, playcount, rating, updated_at) = match outcome { + LookupOutcome::Found { + listeners, + playcount, + } => ( + Some(*listeners), + Some(*playcount), + Some(popularity_rating(*listeners, *playcount)), + Some(chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string()), + ), + // An authoritative miss replaces the current result, but preserves + // historical observations. Unknown popularity is not a zero rating. + LookupOutcome::NotFound => (None, None, None, None), + }; + // The data-modifying CTE atomically saves the result and its history in one + // round trip. Ignore responses for a lookup identity that is already stale. + let saved: bool = sqlx::query_scalar( + r#" + WITH updated AS ( + UPDATE furumusic__track t + SET lastfm_listeners = $2, lastfm_playcount = $3, + lastfm_rating = $4, lastfm_updated_at = $5, + lastfm_checked_at = CURRENT_TIMESTAMP, + lastfm_lookup_title = $6, lastfm_lookup_artist = $7 + WHERE t.id = $1 AND t.is_hidden = false AND t.title::text = $6 + AND (SELECT a.name::text + FROM furumusic__track_artist ta + JOIN furumusic__artist a ON a.id = ta.artist_id + WHERE ta.track_id = t.id AND ta.role <> 'featuring' + ORDER BY ta.position, ta.artist_id LIMIT 1) IS NOT DISTINCT FROM $7 + RETURNING id + ), history AS ( + INSERT INTO furumusic__track_popularity_history + (track_id, source, listeners, playcount, rating, fetched_at) + SELECT id, 'lastfm', $2::bigint, $3::bigint, $4::double precision, $5::varchar + FROM updated WHERE $2::bigint IS NOT NULL + ) + SELECT EXISTS (SELECT 1 FROM updated) + "#, + ) + .bind(track.id) + .bind(listeners) + .bind(playcount) + .bind(rating) + .bind(updated_at) + .bind(&track.title) + .bind(&track.artist_name) + .fetch_one(conn) + .await?; + Ok(saved) +} + async fn fetch_track_info( - client: &reqwest::Client, + client: &Client, api_key: &str, artist: &str, track: &str, -) -> anyhow::Result> { +) -> Result { let response = client - .get("https://ws.audioscrobbler.com/2.0/") + .get(LASTFM_API_URL) .query(&[ ("method", "track.getInfo"), ("api_key", api_key), @@ -205,45 +378,61 @@ async fn fetch_track_info( ("format", "json"), ]) .send() - .await?; + .await + .map_err(|err| LookupError::Transport(err.without_url()))?; + let status = response.status(); + if !status.is_success() { + return Err(LookupError::Http(status)); + } + let body = response + .bytes() + .await + .map_err(|err| LookupError::Transport(err.without_url()))?; + interpret_response(status, &body) +} - if response.status() == reqwest::StatusCode::NOT_FOUND { - return Ok(None); +fn interpret_response(status: StatusCode, body: &[u8]) -> Result { + // A bare HTTP 404 (or any other HTTP failure) never proves catalog absence. + if !status.is_success() { + return Err(LookupError::Http(status)); } - let response = response.error_for_status()?; - let body: LastfmTrackInfoResponse = response.json().await?; + let body: LastfmTrackInfoResponse = serde_json::from_slice(body) + .map_err(|err| LookupError::InvalidResponse(err.to_string()))?; if let Some(code) = body.error { - if code == 29 { - anyhow::bail!("Last.fm rate limit exceeded"); + let message = body.message.unwrap_or_else(|| "unknown error".to_owned()); + // Last.fm uses error 6 for both missing tracks and invalid parameters. + // Only an explicit track-not-found message is a cacheable miss. + if matches!(code, 6 | 7) + && message + .trim() + .trim_end_matches('.') + .eq_ignore_ascii_case("track not found") + { + return Ok(LookupOutcome::NotFound); } - if code == 6 || code == 7 { - return Ok(None); - } - anyhow::bail!( - "Last.fm API error {code}: {}", - body.message.unwrap_or_else(|| "unknown error".to_string()) - ); + return Err(LookupError::Api { code, message }); } - let Some(info) = body.track else { - return Ok(None); + let info = body + .track + .ok_or_else(|| LookupError::InvalidResponse("missing track".to_owned()))?; + let parse_count = |value: &str, field: &str| { + value + .parse::() + .ok() + .filter(|value| *value >= 0) + .with_context(|| format!("invalid {field}")) + .map_err(|err| LookupError::InvalidResponse(err.to_string())) }; - let listeners = info - .listeners - .as_deref() - .unwrap_or("0") - .parse::() - .unwrap_or(0); - let playcount = info - .playcount - .as_deref() - .unwrap_or("0") - .parse::() - .unwrap_or(0); - Ok(Some((listeners.max(0), playcount.max(0)))) + Ok(LookupOutcome::Found { + listeners: parse_count(&info.listeners, "listeners")?, + playcount: parse_count(&info.playcount, "playcount")?, + }) } fn popularity_rating(listeners: i64, playcount: i64) -> f64 { - let listeners = listeners.max(0) as f64; - let playcount = playcount.max(0) as f64; - playcount.ln_1p() * listeners.ln_1p() + (playcount as f64).ln_1p() * (listeners as f64).ln_1p() } + +#[cfg(test)] +#[path = "lastfm_popularity_tests.rs"] +mod tests; diff --git a/src/jobs/lastfm_popularity_tests.rs b/src/jobs/lastfm_popularity_tests.rs new file mode 100644 index 0000000..9114db8 --- /dev/null +++ b/src/jobs/lastfm_popularity_tests.rs @@ -0,0 +1,337 @@ +use super::*; +use sqlx::Connection; + +#[test] +fn run_options_default_to_light_and_require_explicit_full_mode() { + assert!(!LastfmPopularityOptions::default().refresh_existing); + let light: LastfmPopularityOptions = serde_json::from_str("{}").unwrap(); + assert!(!light.refresh_existing); + let full: LastfmPopularityOptions = + serde_json::from_str(r#"{"refresh_existing":true}"#).unwrap(); + assert!(full.refresh_existing); + assert!( + serde_json::from_str::(r#"{"refresh_existing":"true"}"#).is_err() + ); + assert!( + serde_json::from_str::(r#"{"refresh_exising":true}"#).is_err() + ); +} + +#[test] +fn only_explicit_catalog_misses_are_cacheable() { + for code in [6, 7] { + for message in ["Track not found", "Track not found.", " track NOT found "] { + let body = serde_json::json!({"error": code, "message": message}).to_string(); + assert_eq!( + interpret_response(StatusCode::OK, body.as_bytes()).unwrap(), + LookupOutcome::NotFound + ); + } + } + for (code, message) in [ + (6, "Invalid parameters"), + (6, "Track parameter is required"), + (7, "Invalid resource specified"), + (8, "Operation failed"), + (11, "Service Offline"), + (16, "Temporary error"), + (29, "Track not found"), + ] { + let body = serde_json::json!({"error": code, "message": message}).to_string(); + assert!(interpret_response(StatusCode::OK, body.as_bytes()).is_err()); + } + for status in [ + StatusCode::NOT_FOUND, + StatusCode::TOO_MANY_REQUESTS, + StatusCode::INTERNAL_SERVER_ERROR, + StatusCode::SERVICE_UNAVAILABLE, + ] { + // Even a JSON not-found message cannot turn an HTTP error into a miss. + assert!(interpret_response(status, br#"{"error":6,"message":"Track not found"}"#).is_err()); + } +} + +#[test] +fn invalid_or_incomplete_popularity_is_retryable_and_zero_is_valid() { + for body in [ + "not json", + "{}", + r#"{"track":null}"#, + r#"{"track":{}}"#, + r#"{"track":{"listeners":"1"}}"#, + r#"{"track":{"listeners":"bad","playcount":"2"}}"#, + r#"{"track":{"listeners":"1","playcount":"-1"}}"#, + r#"{"track":{"listeners":"1","playcount":"9223372036854775808"}}"#, + r#"{"track":{"listeners":null,"playcount":"2"}}"#, + ] { + assert!( + interpret_response(StatusCode::OK, body.as_bytes()).is_err(), + "{body}" + ); + } + assert_eq!( + interpret_response( + StatusCode::OK, + br#"{"track":{"listeners":"0","playcount":"0"}}"# + ) + .unwrap(), + LookupOutcome::Found { + listeners: 0, + playcount: 0 + } + ); + assert_eq!(popularity_rating(0, 0), 0.0); + assert!(popularity_rating(10, 100) > popularity_rating(1, 10)); +} + +#[test] +fn global_errors_stop_the_run_without_caching_absence() { + assert!(LookupError::Http(StatusCode::TOO_MANY_REQUESTS).should_stop_run()); + for code in [10, 11, 26, 29] { + assert!( + LookupError::Api { + code, + message: String::new() + } + .should_stop_run() + ); + } + assert!(!LookupError::Http(StatusCode::INTERNAL_SERVER_ERROR).should_stop_run()); + assert!(!LookupError::InvalidResponse("missing track".into()).should_stop_run()); +} + +// These tests use temporary tables on isolated connections, never application +// tables. Run explicitly against a disposable PostgreSQL instance: +// FURUMUSIC_LASTFM_TEST_DATABASE_URL=... cargo test lastfm_popularity -- --include-ignored +async fn database_fixture() -> PgConnection { + let url = std::env::var("FURUMUSIC_LASTFM_TEST_DATABASE_URL") + .expect("set FURUMUSIC_LASTFM_TEST_DATABASE_URL to a disposable PostgreSQL database"); + let mut conn = PgConnection::connect(&url).await.unwrap(); + sqlx::raw_sql( + r#" + CREATE TEMP TABLE furumusic__track ( + id BIGINT PRIMARY KEY, title TEXT NOT NULL, + is_hidden BOOLEAN NOT NULL DEFAULT false, + lastfm_listeners BIGINT, lastfm_playcount BIGINT, + lastfm_rating DOUBLE PRECISION, lastfm_updated_at VARCHAR(32) + ); + CREATE TEMP TABLE furumusic__artist (id BIGINT PRIMARY KEY, name TEXT NOT NULL); + CREATE TEMP TABLE furumusic__track_artist ( + track_id BIGINT, artist_id BIGINT, role TEXT, position INTEGER + ); + CREATE TEMP TABLE furumusic__track_popularity_history ( + id BIGSERIAL PRIMARY KEY, track_id BIGINT NOT NULL, source VARCHAR(32) NOT NULL, + listeners BIGINT NOT NULL, playcount BIGINT NOT NULL, + rating DOUBLE PRECISION NOT NULL, fetched_at VARCHAR(32) NOT NULL + ); + INSERT INTO furumusic__artist VALUES (1, 'Artist'), (2, 'Guest'), (3, 'Other'); + INSERT INTO furumusic__track (id, title, lastfm_rating, lastfm_listeners, lastfm_playcount) + VALUES (1, 'Legacy zero', 0, 0, 0), (2, 'New', NULL, NULL, NULL), + (3, 'Hidden', NULL, NULL, NULL), (4, 'No artist', NULL, NULL, NULL); + UPDATE furumusic__track SET is_hidden = true WHERE id = 3; + INSERT INTO furumusic__track_artist VALUES + (1, 1, 'main', 0), (2, 1, 'main', 0), (3, 1, 'main', 0), (2, 2, 'featuring', -1); + "#, + ) + .execute(&mut conn) + .await + .unwrap(); + // Exercise precisely the same prepared statements as the real migration. + for statement in include_str!("../music/lastfm_popularity_lookup.sql").split(';') { + if !statement.trim().is_empty() { + sqlx::query(statement).execute(&mut conn).await.unwrap(); + } + } + conn +} + +async fn candidates(conn: &mut PgConnection, full: bool) -> Vec { + fetch_candidates( + conn, + 0, + i64::MAX, + LastfmPopularityOptions { + refresh_existing: full, + }, + ) + .await + .unwrap() +} + +fn ids(rows: &[TrackLookupRow]) -> Vec { + rows.iter().map(|row| row.id).collect() +} + +#[tokio::test] +#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"] +async fn database_modes_misses_retries_and_metadata_changes() { + let mut conn = database_fixture().await; + assert_eq!(ids(&candidates(&mut conn, false).await), [2, 4]); + assert_eq!(ids(&candidates(&mut conn, true).await), [1, 2, 4]); + let new = candidates(&mut conn, false).await.remove(0); + assert_eq!(new.artist_name.as_deref(), Some("Artist")); + assert!( + save_outcome(&mut conn, &new, &LookupOutcome::NotFound) + .await + .unwrap() + ); + assert_eq!(ids(&candidates(&mut conn, false).await), [4]); + assert_eq!(ids(&candidates(&mut conn, true).await), [1, 2, 4]); + // A failed full refresh must become eligible for light mode again. + mark_pending(&mut conn, &[new.id]).await.unwrap(); + assert_eq!(ids(&candidates(&mut conn, false).await), [2, 4]); + let found = LookupOutcome::Found { + listeners: 5, + playcount: 10, + }; + assert!(save_outcome(&mut conn, &new, &found).await.unwrap()); + assert_eq!(ids(&candidates(&mut conn, false).await), [4]); + // Unlike a network error, a confirmed miss replaces old current statistics. + assert!( + save_outcome(&mut conn, &new, &LookupOutcome::NotFound) + .await + .unwrap() + ); + let rating: Option = + sqlx::query_scalar("SELECT lastfm_rating FROM furumusic__track WHERE id=2") + .fetch_one(&mut conn) + .await + .unwrap(); + assert_eq!(rating, None); + let history: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM furumusic__track_popularity_history") + .fetch_one(&mut conn) + .await + .unwrap(); + assert_eq!(history, 1); + sqlx::query("UPDATE furumusic__artist SET name='Renamed' WHERE id=1") + .execute(&mut conn) + .await + .unwrap(); + assert_eq!(ids(&candidates(&mut conn, false).await), [1, 2, 4]); + // A response fetched before a concurrent rename cannot mark the new identity checked. + assert!(!save_outcome(&mut conn, &new, &found).await.unwrap()); + let renamed = candidates(&mut conn, false).await.remove(1); + assert!(save_outcome(&mut conn, &renamed, &found).await.unwrap()); + sqlx::query("UPDATE furumusic__track SET title='New title' WHERE id=2") + .execute(&mut conn) + .await + .unwrap(); + assert_eq!(ids(&candidates(&mut conn, false).await), [1, 2, 4]); + assert!(!save_outcome(&mut conn, &renamed, &found).await.unwrap()); + let retitled = candidates(&mut conn, false).await.remove(1); + assert!(save_outcome(&mut conn, &retitled, &found).await.unwrap()); + sqlx::query("UPDATE furumusic__track_artist SET artist_id=3 WHERE track_id=2 AND role='main'") + .execute(&mut conn) + .await + .unwrap(); + assert_eq!(ids(&candidates(&mut conn, false).await), [1, 2, 4]); + assert!(!save_outcome(&mut conn, &retitled, &found).await.unwrap()); + let relinked = candidates(&mut conn, false).await.remove(1); + assert_eq!(relinked.artist_name.as_deref(), Some("Other")); + assert!(save_outcome(&mut conn, &relinked, &found).await.unwrap()); + sqlx::query("DELETE FROM furumusic__track_artist WHERE track_id=2 AND role='main'") + .execute(&mut conn) + .await + .unwrap(); + assert_eq!(candidates(&mut conn, false).await[1].artist_name, None); +} + +#[tokio::test] +#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"] +async fn database_pagination_does_not_limit_runs_or_repeat_failures() { + let mut conn = database_fixture().await; + sqlx::query("INSERT INTO furumusic__track (id, title) SELECT n, 'Track ' || n FROM generate_series(5, 1105) n") + .execute(&mut conn).await.unwrap(); + sqlx::query("INSERT INTO furumusic__track_artist SELECT id, 1, 'main', 0 FROM furumusic__track WHERE id>=5") + .execute(&mut conn).await.unwrap(); + let mut after_id = 0; + let mut seen = Vec::new(); + loop { + let page = fetch_candidates( + &mut conn, + after_id, + 1105, + LastfmPopularityOptions::default(), + ) + .await + .unwrap(); + if page.is_empty() { + break; + } + assert!(page.len() <= TRACK_PAGE_SIZE as usize); + for row in page { + after_id = row.id; + // Simulate failed requests: all candidates remain pending in the DB. + mark_pending(&mut conn, &[row.id]).await.unwrap(); + seen.push(row.id); + } + } + assert_eq!(seen.len(), 1103); + assert!(seen.windows(2).all(|pair| pair[0] < pair[1])); + assert_eq!(seen.last(), Some(&1105)); + let bounded = fetch_candidates(&mut conn, 1000, 1010, LastfmPopularityOptions::default()) + .await + .unwrap(); + assert_eq!(ids(&bounded), (1001..=1010).collect::>()); +} + +#[tokio::test] +#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"] +async fn database_result_and_history_are_atomic() { + let mut conn = database_fixture().await; + let track = candidates(&mut conn, false).await.remove(0); + sqlx::query("ALTER TABLE furumusic__track_popularity_history ADD CHECK (playcount > 100)") + .execute(&mut conn) + .await + .unwrap(); + assert!( + save_outcome( + &mut conn, + &track, + &LookupOutcome::Found { + listeners: 5, + playcount: 10 + } + ) + .await + .is_err() + ); + let state: (bool, Option) = sqlx::query_as( + "SELECT lastfm_checked_at IS NULL, lastfm_rating FROM furumusic__track WHERE id=2", + ) + .fetch_one(&mut conn) + .await + .unwrap(); + assert_eq!(state, (true, None)); +} + +#[tokio::test] +#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"] +async fn database_interrupted_full_refresh_remains_pending_without_losing_statistics() { + let mut conn = database_fixture().await; + let page = candidates(&mut conn, true).await; + let track_ids = ids(&page); + mark_pending(&mut conn, &track_ids).await.unwrap(); + assert_eq!(track_ids, vec![1, 2, 4]); + assert_eq!(ids(&candidates(&mut conn, false).await), track_ids); + let state: (bool, Option) = sqlx::query_as( + "SELECT lastfm_checked_at IS NULL, lastfm_rating FROM furumusic__track WHERE id=1", + ) + .fetch_one(&mut conn) + .await + .unwrap(); + assert_eq!(state, (true, Some(0.0))); +} + +#[tokio::test] +#[ignore = "requires FURUMUSIC_LASTFM_TEST_DATABASE_URL"] +async fn database_lock_is_shared_by_light_and_full_runs() { + let url = std::env::var("FURUMUSIC_LASTFM_TEST_DATABASE_URL").unwrap(); + let mut first = PgConnection::connect(&url).await.unwrap(); + let mut second = PgConnection::connect(&url).await.unwrap(); + assert!(try_lock(&mut first).await.unwrap()); + assert!(!try_lock(&mut second).await.unwrap()); + first.close().await.unwrap(); + assert!(try_lock(&mut second).await.unwrap()); +} diff --git a/src/music/lastfm_popularity_lookup.sql b/src/music/lastfm_popularity_lookup.sql new file mode 100644 index 0000000..5c4b3ca --- /dev/null +++ b/src/music/lastfm_popularity_lookup.sql @@ -0,0 +1,27 @@ +ALTER TABLE furumusic__track + ADD COLUMN IF NOT EXISTS lastfm_checked_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS lastfm_lookup_title TEXT, + ADD COLUMN IF NOT EXISTS lastfm_lookup_artist TEXT; + +CREATE INDEX IF NOT EXISTS idx_track_artist_lastfm_lookup + ON furumusic__track_artist (track_id, position, artist_id) + WHERE role <> 'featuring'; + +-- Preserve existing observations, including valid zero-valued popularity. +-- Legacy misses cannot be backfilled because they were only written to logs. +UPDATE furumusic__track t + SET lastfm_checked_at = CURRENT_TIMESTAMP, + lastfm_lookup_title = t.title::text, + lastfm_lookup_artist = ( + SELECT a.name::text + FROM furumusic__track_artist ta + JOIN furumusic__artist a ON a.id = ta.artist_id + WHERE ta.track_id = t.id AND ta.role <> 'featuring' + ORDER BY ta.position, ta.artist_id + LIMIT 1 + ) + WHERE t.lastfm_checked_at IS NULL + AND (t.lastfm_updated_at IS NOT NULL + OR t.lastfm_rating IS NOT NULL + OR t.lastfm_listeners IS NOT NULL + OR t.lastfm_playcount IS NOT NULL); diff --git a/src/music/mod.rs b/src/music/mod.rs index ea010f5..ee3db2f 100644 --- a/src/music/mod.rs +++ b/src/music/mod.rs @@ -2823,6 +2823,35 @@ pub mod db_migrations { &[Operation::custom(add_playback_coordination_columns).build()]; } + // -- M0049: persist Last.fm lookup results and their metadata identity -- + + #[cot::db::migrations::migration_op] + async fn add_lastfm_lookup_state(ctx: migrations::MigrationContext<'_>) -> cot::db::Result<()> { + // Cot prepares raw queries, so execute these static DDL statements + // separately rather than sending a multi-statement prepared query. + for statement in include_str!("lastfm_popularity_lookup.sql").split(';') { + if !statement.trim().is_empty() { + ctx.db.raw(statement).await?; + } + } + Ok(()) + } + + #[derive(Debug, Copy, Clone)] + pub struct M0049AddLastfmLookupState; + + impl migrations::Migration for M0049AddLastfmLookupState { + const APP_NAME: &'static str = "furumusic"; + const MIGRATION_NAME: &'static str = "m_0049_add_lastfm_lookup_state"; + const DEPENDENCIES: &'static [migrations::MigrationDependency] = + &[migrations::MigrationDependency::migration( + "furumusic", + "m_0048_add_playback_coordination_columns", + )]; + const OPERATIONS: &'static [Operation] = + &[Operation::custom(add_lastfm_lookup_state).build()]; + } + pub const MIGRATIONS: &[&SyncDynMigration] = &[ &M0006CreateMediaFile, &M0007CreateArtist, @@ -2862,5 +2891,6 @@ pub mod db_migrations { &M0046CreateLocalUploadHistory, &M0047CreateYouTubeImportMediaLinks, &M0048AddPlaybackCoordinationColumns, + &M0049AddLastfmLookupState, ]; } diff --git a/templates/admin/job_detail.html b/templates/admin/job_detail.html index 69d87ff..acfc776 100644 --- a/templates/admin/job_detail.html +++ b/templates/admin/job_detail.html @@ -13,7 +13,7 @@
- {% if job.name_str() != "metadata_backfill" %} + {% if job.name_str() != "metadata_backfill" && job.name_str() != "lastfm_popularity" %}
@@ -29,6 +29,16 @@ {% endif %}
+{% if job.name_str() == "lastfm_popularity" %} +
+

{{ t.jobs_lastfm_light_help }}

+ + +
+{% endif %} + {% if job.name_str() == "metadata_backfill" %}

{{ t.jobs_metadata_backfill_options }}

diff --git a/templates/admin/v2.html b/templates/admin/v2.html index 3cb466a..fde09d0 100644 --- a/templates/admin/v2.html +++ b/templates/admin/v2.html @@ -2292,13 +2292,20 @@ tbody tr:hover { +
+
+ + +
+

Light mode skips previous matches and tracks confirmed missing. Changed titles or artists and failed lookups are checked again. Full mode also retries previous matches and misses.

+
This task has no manual parameters.