Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9ab27c921c | ||
|
|
bd1d3a4ebc |
No files matched your search
Generated
+1
-1
@@ -1988,7 +1988,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
|
||||
|
||||
[[package]]
|
||||
name = "furumusic"
|
||||
version = "0.10.8"
|
||||
version = "0.11.1"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"async-stream",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "furumusic"
|
||||
version = "0.11.0"
|
||||
version = "0.11.1"
|
||||
edition = "2024"
|
||||
description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
|
||||
|
||||
|
||||
@@ -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
|
||||
```
|
||||
@@ -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<v2::LastfmPopularityRunRequest>| {
|
||||
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<views::LastfmPopularityForm>| {
|
||||
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<i64> {
|
||||
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)
|
||||
}
|
||||
@@ -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<LastfmPopularityRunRequest>,
|
||||
) -> cot::Result<cot::response::Response> {
|
||||
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,
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
pub async fn lastfm_popularity_run(
|
||||
_admin: AuthenticatedUser,
|
||||
db: &Database,
|
||||
pool: &sqlx::PgPool,
|
||||
form: RequestForm<LastfmPopularityForm>,
|
||||
) -> cot::Result<cot::http::Response<Body>> {
|
||||
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,
|
||||
|
||||
@@ -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" , "Заполнить только пустые";
|
||||
|
||||
+371
-182
@@ -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<String>,
|
||||
lastfm_updated_at: Option<String>,
|
||||
}
|
||||
|
||||
// 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<LastfmTrack>,
|
||||
@@ -23,8 +58,69 @@ struct LastfmTrackInfoResponse {
|
||||
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct LastfmTrack {
|
||||
listeners: Option<String>,
|
||||
playcount: Option<String>,
|
||||
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<bool> {
|
||||
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<Vec<TrackLookupRow>> {
|
||||
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<i64> = 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<bool> {
|
||||
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<Option<(i64, i64)>> {
|
||||
) -> Result<LookupOutcome, LookupError> {
|
||||
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<LookupOutcome, LookupError> {
|
||||
// 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::<i64>()
|
||||
.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::<i64>()
|
||||
.unwrap_or(0);
|
||||
let playcount = info
|
||||
.playcount
|
||||
.as_deref()
|
||||
.unwrap_or("0")
|
||||
.parse::<i64>()
|
||||
.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;
|
||||
@@ -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::<LastfmPopularityOptions>(r#"{"refresh_existing":"true"}"#).is_err()
|
||||
);
|
||||
assert!(
|
||||
serde_json::from_str::<LastfmPopularityOptions>(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<TrackLookupRow> {
|
||||
fetch_candidates(
|
||||
conn,
|
||||
0,
|
||||
i64::MAX,
|
||||
LastfmPopularityOptions {
|
||||
refresh_existing: full,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
fn ids(rows: &[TrackLookupRow]) -> Vec<i64> {
|
||||
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<f64> =
|
||||
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::<Vec<_>>());
|
||||
}
|
||||
|
||||
#[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<f64>) = 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<f64>) = 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());
|
||||
}
|
||||
@@ -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);
|
||||
@@ -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,
|
||||
];
|
||||
}
|
||||
@@ -13,7 +13,7 @@
|
||||
</table>
|
||||
|
||||
<div style="margin: 1rem 0; display: flex; gap: .5rem; align-items: flex-end;">
|
||||
{% if job.name_str() != "metadata_backfill" %}
|
||||
{% if job.name_str() != "metadata_backfill" && job.name_str() != "lastfm_popularity" %}
|
||||
<form method="post" action="/admin/jobs/{{ job.name_str() }}/run" style="margin:0;">
|
||||
<button type="submit" style="padding:.4rem 1rem; background:#007bff; color:#fff; border:none; border-radius:4px; cursor:pointer;">{{ t.jobs_run_now }}</button>
|
||||
</form>
|
||||
@@ -29,6 +29,16 @@
|
||||
{% endif %}
|
||||
</div>
|
||||
|
||||
{% if job.name_str() == "lastfm_popularity" %}
|
||||
<form method="post" action="/admin/jobs/lastfm_popularity/run-options" style="margin:1rem 0;">
|
||||
<p>{{ t.jobs_lastfm_light_help }}</p>
|
||||
<label style="display:block; margin-bottom:1rem;">
|
||||
<input type="checkbox" name="refresh_existing"> {{ t.jobs_lastfm_full }}
|
||||
</label>
|
||||
<button type="submit">{{ t.jobs_run_now }}</button>
|
||||
</form>
|
||||
{% endif %}
|
||||
|
||||
{% if job.name_str() == "metadata_backfill" %}
|
||||
<h2>{{ t.jobs_metadata_backfill_options }}</h2>
|
||||
<form method="post" action="/admin/jobs/metadata_backfill/run-options" style="margin:0 0 1.5rem; padding:1rem; background:#fff; border:1px solid #e0e0e0; border-radius:6px;">
|
||||
|
||||
+19
-2
@@ -2292,13 +2292,20 @@ tbody tr:hover {
|
||||
<label><input type="radio" value="overwrite" x-model="artworkBackfillOptions.mode" /> Search all and replace existing</label>
|
||||
</div>
|
||||
</div>
|
||||
<div class="metadata-backfill-options" x-show="isLastfmPopularityJob(activeJob)">
|
||||
<div class="mode-row">
|
||||
<label><input type="radio" value="light" x-model="lastfmPopularityOptions.mode" /> Light: unchecked tracks only</label>
|
||||
<label><input type="radio" value="full" x-model="lastfmPopularityOptions.mode" /> Full: refresh all visible tracks</label>
|
||||
</div>
|
||||
<p class="job-param-note">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.</p>
|
||||
</div>
|
||||
<div class="job-param-note" x-show="activeJob && !jobHasParameterForm(activeJob)">
|
||||
This task has no manual parameters.
|
||||
</div>
|
||||
<div class="toolbar" style="margin-bottom:12px">
|
||||
<button class="btn primary" type="button" @click="runJob(activeJob)">
|
||||
<i data-lucide="play"></i>
|
||||
<span x-text="isMetadataBackfillJob(activeJob) ? 'Run with options' : 'Run now'"></span>
|
||||
<span x-text="jobHasParameterForm(activeJob) ? 'Run with options' : 'Run now'"></span>
|
||||
</button>
|
||||
<button class="btn" type="button" @click="toggleJob(activeJob)">
|
||||
<i data-lucide="power"></i>
|
||||
@@ -3811,6 +3818,7 @@ function adminV2() {
|
||||
musicbrainz_tags: true,
|
||||
mode: 'fill_missing'
|
||||
},
|
||||
lastfmPopularityOptions: { mode: 'light' },
|
||||
artworkBackfillOptions: {
|
||||
mode: 'missing'
|
||||
},
|
||||
@@ -4624,6 +4632,11 @@ function adminV2() {
|
||||
method: 'POST',
|
||||
body: JSON.stringify(this.artworkBackfillPayload())
|
||||
})
|
||||
: this.isLastfmPopularityJob(job)
|
||||
? await this.request(`${this.apiBase}/jobs/lastfm_popularity/run-options`, {
|
||||
method: 'POST',
|
||||
body: JSON.stringify({ refresh_existing: this.lastfmPopularityOptions.mode === 'full' })
|
||||
})
|
||||
: await this.request(`${this.apiBase}/jobs/${encodeURIComponent(job.name)}/run`, { method: 'POST' });
|
||||
this.showToast(`Run #${result.run_id} started`);
|
||||
this.activeJobName = job.name;
|
||||
@@ -4647,8 +4660,12 @@ function adminV2() {
|
||||
return job && job.name === 'artwork_backfill';
|
||||
},
|
||||
|
||||
isLastfmPopularityJob(job) {
|
||||
return job && job.name === 'lastfm_popularity';
|
||||
},
|
||||
|
||||
jobHasParameterForm(job) {
|
||||
return this.isMetadataBackfillJob(job) || this.isArtworkBackfillJob(job);
|
||||
return this.isMetadataBackfillJob(job) || this.isArtworkBackfillJob(job) || this.isLastfmPopularityJob(job);
|
||||
},
|
||||
|
||||
metadataBackfillPayload() {
|
||||
|
||||
Reference in new issue
Block a user