diff --git a/CLAUDE.md b/CLAUDE.md index d9dede4d..e42c6c40 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -44,7 +44,7 @@ bun run build # webapp first, then cargo — the order Tests need no service container: `test_app()` builds a `Config` over a `TempDir` and the suite drives the same router `main` serves. FFmpeg and ffprobe must be on `PATH`. -The integration suite is seventeen targets under `tests/`, one per subject — `auth`, `scanner`, `identity`, `catalog`, `subsonic_contract`, `subsonic_browse`, `subsonic_fields`, `subsonic_methods`, `native_api`, `sync`, `media`, `service`, `cli`, `scrobbling`, `uploads`, `canvas`, `migrations` — nearly all of them sharing `tests/support/mod.rs`. `migrations` is the exception and deliberately so: every other target starts from `test_app()`, which builds its database from nothing, and that is exactly the path an upgrade does not take. It seeds a database at each **released** schema, boots the server on it, and reads the rows back through the services — and it is where anything else that needs a database *behind* this binary belongs, such as proving that a key change reads without migrating and copies the pair aside before it does. `autotests = false`, so each is a target only because `Cargo.toml` declares it: a new file under `tests/` is compiled by nothing until a `[[test]]` entry names it, and the count above is worth checking against that list rather than trusted. `support/mod.rs` is the exception that proves it — shared module code, declared nowhere, a target never. File a new test by the surface it exercises, and check that by what it calls rather than by what its name says. +The integration suite is eighteen targets under `tests/`, one per subject — `auth`, `scanner`, `identity`, `catalog`, `subsonic_contract`, `subsonic_browse`, `subsonic_fields`, `subsonic_methods`, `native_api`, `sync`, `media`, `service`, `cli`, `scrobbling`, `enrichment`, `uploads`, `canvas`, `migrations` — nearly all of them sharing `tests/support/mod.rs`. `migrations` is the exception and deliberately so: every other target starts from `test_app()`, which builds its database from nothing, and that is exactly the path an upgrade does not take. It seeds a database at each **released** schema, boots the server on it, and reads the rows back through the services — and it is where anything else that needs a database *behind* this binary belongs, such as proving that a key change reads without migrating and copies the pair aside before it does. `autotests = false`, so each is a target only because `Cargo.toml` declares it: a new file under `tests/` is compiled by nothing until a `[[test]]` entry names it, and the count above is worth checking against that list rather than trusted. `support/mod.rs` is the exception that proves it — shared module code, declared nowhere, a target never. File a new test by the surface it exercises, and check that by what it calls rather than by what its name says. `cargo build` works without a client build — `build.rs` embeds a placeholder page when `webapp/dist` is absent. A real `vite build` overwrites it. diff --git a/Cargo.toml b/Cargo.toml index e53e6e93..9c99c9a5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -154,3 +154,7 @@ path = "tests/canvas.rs" [[test]] name = "scrobbling" path = "tests/scrobbling.rs" + +[[test]] +name = "enrichment" +path = "tests/enrichment.rs" diff --git a/docs/operations.md b/docs/operations.md index 278010ee..e3e15231 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -169,6 +169,47 @@ Every tunable is a field on `Config` in [`src/config.rs`](../src/config.rs), with its environment variable documented on it. That file is the authority; this page is not a copy of it. +## Switching metadata enrichment on + +**Off by default, and an upgrade leaves it off.** A server that has not been +told otherwise makes no request to any metadata provider. +[RFC-011](rfcs/RFC-011-metadata-enrichment.md) carries the reasoning. + +```bash +WAVEFLOW_METADATA_PROVIDERS=lastfm +WAVEFLOW_METADATA_LASTFM_KEY=… # your own Last.fm API application's key +``` + +| Variable | Default | What it does | +| --- | --- | --- | +| `WAVEFLOW_METADATA_PROVIDERS` | empty | The switch: an ordered, comma-separated list. Empty is off. Today the only provider is `lastfm` | +| `WAVEFLOW_METADATA_LASTFM_KEY` | — | Required when `lastfm` is listed; the server refuses to start without it. No key ships in the binary | +| `WAVEFLOW_METADATA_IMAGES` | on | `off` keeps the lookups and refuses artist image downloads | +| `WAVEFLOW_METADATA_TTL_DAYS` | 30 | How long an answer with content is kept before it is asked again | +| `WAVEFLOW_METADATA_MISS_TTL_DAYS` | 7 | How long "the provider knows nothing about this artist" is kept | +| `WAVEFLOW_METADATA_RATE_PER_SEC` | 4 | Requests per second, per provider | +| `WAVEFLOW_METADATA_REQUEST_TIMEOUT_SECS` | 10 | How long one request may take before it counts as unanswered | + +**Scrobbling's Last.fm key does not switch this on.** An application you +registered for submitting listens is a different consent from lookups about your +whole catalogue. You may place the same key in both variables; each one switches +on only its own feature. + +**What the server starts contacting.** `ws.audioscrobbler.com`, over https, and +only when somebody opens an artist. Nothing is fetched at scan time. Each +request carries the artist's name and your key, and no redirect is followed. +Requests are spaced to the rate above, and simultaneous visits to one artist +share one request. + +**What it stores.** In `waveflow.db`, for each provider and each artist name +asked about: the biography, the ranked list of similar artists by name, whether +the provider knew the artist, and when to ask again. Nothing in it belongs to an +account, and nothing records who asked. A failed request stores nothing, so the +next visit asks again. + +**What it logs.** The provider and a verdict, such as `answered HTTP 503` — +never the request URL, which carries your key. + ## Probes | Path | Answers | diff --git a/docs/rfcs/RFC-011-metadata-enrichment.md b/docs/rfcs/RFC-011-metadata-enrichment.md index 16d05a56..8d472a0c 100644 --- a/docs/rfcs/RFC-011-metadata-enrichment.md +++ b/docs/rfcs/RFC-011-metadata-enrichment.md @@ -1,10 +1,18 @@ # RFC-011 — Server-side metadata enrichment - **Status**: Proposed -- **Implemented by**: nothing yet. The *Status* field above never flips in this - project; this line is the one that says what is running, and today it says - the answer is still the empty container. +- **Implemented by**: in progress, in the order of + [the implementation plan](../metadata-enrichment-plan.md): + InstaZDLL/WaveFlow#855 (the core's clients accept a client, an address and a + key), #270 (the pin moved onto them), and the pull request closing #265 (the + cache and the engine that fills it). No surface answers from the cache yet, + so every container is still the empty one. The *Status* field above never + flips in this project; this line is the one that says what is running. - **Date**: 2026-10-14 +- **Revised**: 2026-10-15, while implementing the cache (#265). Decisions 4, 6 + and 10 rested on attachment points the code does not hold; each correction + below is marked and says what the first version asserted, rather than + erasing it. - **Authors**: the WaveFlow project - **Depends on**: [RFC-002](RFC-002-waveflow-server-v2.md), [RFC-010](RFC-010-external-scrobbling.md) @@ -170,6 +178,14 @@ The cache is therefore keyed on the **canonical name**, as the desktop's own - `metadata_album` — the same shape for album notes, keyed on the provider and the canonical `(artist, title)` pair that `album_match.rs` already computes. +**Revised on 2026-10-15.** The migration that shipped +(`20261015000000_metadata_artist.sql`) differs from this list in three places. +The biography is two columns, `bio_summary` and `bio_full`, because that is how +the providers return it. `image_url` is not there: portraits add their columns +with the fetch that fills them. And `metadata_album` was not created — the +open question on album notes below decides whether it exists, and +`album_match.rs` was not even in the pinned core when this was written. + ### One row per provider, merged on read **The provider belongs in the key, not in a column**, and the first draft of @@ -204,6 +220,16 @@ matching a provider's ranked candidate list back to the name asked for is `select_by_name`. Neither is re-implemented here; both have a desktop issue behind them (#342) recording what naive equality dropped. +**Revised on 2026-10-15: the key is not `normalize_name`.** The paragraph above +asserted it was. Every `artist` row already stores a `canonical_name`, computed +by `waveflow_core::scanner::canonical_name`, and the two functions disagree: +`normalize_name` also folds diacritics and turns `&` into `and`, so "Björk" is +`björk` in the column and `bjork` through it. The cache is keyed on the +column's form, which lets the read path join +`metadata_artist.name_canonical = artist.canonical_name` with nothing +recomputed. `normalize_name` and `select_by_name` keep the job they were written +for — matching a provider's candidates — where a provider returns candidates. + **Nothing in these tables is account data**, which is what makes an instance-wide cache correct rather than merely convenient. Anything that ever becomes account-specific — a per-person override, a hidden artist — belongs in @@ -267,6 +293,13 @@ for someone to reload the page. The mechanism already exists and decision 8 explains why it may now carry an artist: RFC-007's library event stream, whose decision 6 declined an artist event for a reason this RFC removes. +**Revised on 2026-10-15: only one client is told.** The paragraph above implied +every client is. The desktop is, because it mirrors the library through that +feed. The web client is not: the only event stream it reads is a scan's +progress. It re-asks instead, a bounded number of times while the answer says +the lookup is pending, and then stops — which is the web client's step of the +plan, not a property this mechanism gives it. + ## Decision 7 — similar artists resolve to artists the caller can see, and to nothing else `artist.getSimilar` returns names. Subsonic's `similarArtist` entries carry an @@ -415,6 +448,16 @@ them empty. - `getArtistInfo` / `getArtistInfo2` — `biography`, `similarArtist` entries (decision 7), and `smallImageUrl` / `mediumImageUrl` / `largeImageUrl` pointing at this server's own artwork route (decision 8). + + **Revised on 2026-10-15: the three image URLs stay absent.** The bullet above + asserted a route they could point at. Neither artwork route serves a URL a + client loads as-is: `media::artwork` wants the native bearer, and + `getCoverArt` wants Subsonic credentials in the query string, which would put + a member's token into a URL that clients cache and log. A client finds the + portrait through `coverArt`, which every artist element already carries. The + compatibility campaign measures whether a client in the matrix reads + `largeImageUrl` at all; a sealed artwork ticket can be added then, while a URL + scheme cannot be withdrawn once clients have cached it. - `getSimilarSongs` / `getSimilarSongs2` — tracks by the neighbours the caller can see. - `getAlbumInfo` / `getAlbumInfo2` — `notes` beside the `musicBrainzId` they diff --git a/migrations-v2/20261015000000_metadata_artist.sql b/migrations-v2/20261015000000_metadata_artist.sql new file mode 100644 index 00000000..3da05afb --- /dev/null +++ b/migrations-v2/20261015000000_metadata_artist.sql @@ -0,0 +1,53 @@ +-- What the metadata providers said about an artist — RFC-011 decisions 4 and 5. +-- +-- **Instance-wide, and nothing in it is account data.** A biography is the same +-- text whoever asks, and `artist` rows are per library, so keying on +-- `artist.id` would fetch the same answer once per library. The key is the +-- name instead, in the form `artist.canonical_name` already stores +-- (`waveflow_core::scanner::canonical_name`), so the read path joins on that +-- column with nothing recomputed. RFC-011 first named +-- `name_match::normalize_name` here; the two disagree — "Björk" is `björk` in +-- the column and `bjork` through it — and its erratum says so. +-- +-- **The provider is part of the key.** Each provider keeps its own row, its +-- own outcome and its own clock, and the fields are merged at read time in the +-- operator's order. One row per name would have to pick a winner when written, +-- and would let one provider's "never heard of them" silence another that has +-- a biography. +-- +-- `provider` carries no CHECK: the vocabulary is the server's `Provider` enum, +-- and widening a CHECK in SQLite means rebuilding the table. `outcome` does, +-- because its two words are decision 5's and do not grow — the third outcome, +-- unreached, is the absence of a row. +-- +-- Album notes (`metadata_album`) wait for the RFC's open question on them. +-- Portraits add their columns with the fetch that fills them. +CREATE TABLE metadata_artist ( + provider TEXT NOT NULL, + name_canonical TEXT NOT NULL, + outcome TEXT NOT NULL CHECK (outcome IN ('hit', 'miss')), + bio_summary TEXT, + bio_full TEXT, + fetched_at INTEGER NOT NULL, + expires_at INTEGER NOT NULL, + PRIMARY KEY (provider, name_canonical) +) STRICT, WITHOUT ROWID; + +-- The ranked neighbours, as the provider gave them. Kept by name and resolved +-- to local artists only when read, through the caller's own libraries +-- (decision 7): an id here would name a row in somebody's library. +-- +-- Rewritten whole with its parent row, in the same transaction, so a list is +-- never half one answer and half the next. The cascade only matters to a row +-- deleted outright; a refresh updates the parent in place. +CREATE TABLE metadata_artist_similar ( + provider TEXT NOT NULL, + name_canonical TEXT NOT NULL, + rank INTEGER NOT NULL, + similar_name TEXT NOT NULL, + similar_canonical TEXT NOT NULL, + score REAL NOT NULL, + PRIMARY KEY (provider, name_canonical, rank), + FOREIGN KEY (provider, name_canonical) + REFERENCES metadata_artist(provider, name_canonical) ON DELETE CASCADE +) STRICT, WITHOUT ROWID; diff --git a/src/config.rs b/src/config.rs index 08b90814..0f3b3d0a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -189,6 +189,75 @@ pub struct ScrobbleDestination { pub fingerprint: String, } +/// Metadata enrichment, as RFC-011 decision 3 lists its settings. +/// +/// **Off unless `providers` names one**, and that is the only switch: an empty +/// list makes no outbound request of any kind, and nothing else here can +/// contradict it. Separate from scrobbling on purpose — an operator who placed +/// a Last.fm application for submissions has not thereby agreed to lookups +/// about the whole catalogue, so `lastfm_key` is its own setting even when it +/// holds the same value. +#[derive(Clone)] +pub struct MetadataSettings { + /// `WAVEFLOW_METADATA_PROVIDERS`: an ordered, comma-separated list, empty + /// by default. The order is the order fields are merged in when read. + pub providers: Vec, + /// `WAVEFLOW_METADATA_IMAGES`, on by default, `off` to keep the lookups and + /// refuse decision 8's image fetch. A narrowing only: it switches nothing + /// on while `providers` is empty. + pub images: bool, + /// `WAVEFLOW_METADATA_LASTFM_KEY`, the operator's own application key. + /// Required when `providers` names `lastfm`: no key ships in this binary. + pub lastfm_key: Option, + /// `WAVEFLOW_METADATA_TTL_DAYS`: how long an answer with content stands. + pub ttl: Duration, + /// `WAVEFLOW_METADATA_MISS_TTL_DAYS`: how long "the provider knows nothing" + /// stands. Shorter than `ttl`, and a setting of its own, because it is the + /// answer most likely to change. + pub miss_ttl: Duration, + /// `WAVEFLOW_METADATA_RATE_PER_SEC`: requests per second, per provider. + pub rate_per_sec: u32, + /// `WAVEFLOW_METADATA_REQUEST_TIMEOUT_SECS`: how long one request to a + /// provider may take before it counts as unreached. + pub request_timeout: Duration, +} + +impl MetadataSettings { + /// Enrichment switched off — what a server that is merely upgraded runs. + pub fn off() -> Self { + Self { + providers: Vec::new(), + images: true, + lastfm_key: None, + ttl: Duration::from_secs(30 * 24 * 60 * 60), + miss_ttl: Duration::from_secs(7 * 24 * 60 * 60), + rate_per_sec: 4, + request_timeout: Duration::from_secs(10), + } + } +} + +/// Written by hand so the key never reaches a log: a `Config` is printed at +/// startup under `RUST_LOG=debug`, and a Last.fm key travels in a query string +/// where anyone holding it can spend the operator's quota. +impl std::fmt::Debug for MetadataSettings { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("MetadataSettings") + .field("providers", &self.providers) + .field("images", &self.images) + .field( + "lastfm_key", + &self.lastfm_key.as_ref().map(|_| "[redacted]"), + ) + .field("ttl", &self.ttl) + .field("miss_ttl", &self.miss_ttl) + .field("rate_per_sec", &self.rate_per_sec) + .field("request_timeout", &self.request_timeout) + .finish() + } +} + #[derive(Clone)] pub struct Config { pub bind_addr: SocketAddr, @@ -352,6 +421,8 @@ pub struct Config { /// flag says "this address is mine", not "send my members' tokens in clear /// wherever I point you". pub outbound_allow_plaintext: bool, + /// Metadata enrichment — RFC-011. See [`MetadataSettings`]. + pub metadata: MetadataSettings, pub allowed_origins: Vec, /// How the catalogue decides which row a scanned file belongs to. /// @@ -398,6 +469,7 @@ impl std::fmt::Debug for Config { .field("destinations", &self.destinations) .field("lastfm", &self.lastfm) .field("outbound_allow_plaintext", &self.outbound_allow_plaintext) + .field("metadata", &self.metadata) .field("allowed_origins", &self.allowed_origins) .field("pid", &self.pid) .finish() @@ -590,6 +662,8 @@ impl Config { ), }; + let metadata = parse_metadata_settings()?; + let allowed_origins = std::env::var("WAVEFLOW_ALLOWED_ORIGINS") .unwrap_or_default() .split(',') @@ -664,6 +738,7 @@ impl Config { destinations, lastfm, outbound_allow_plaintext, + metadata, allowed_origins, pid, }) @@ -766,6 +841,10 @@ impl Config { // two tests here stand on. lastfm: None, outbound_allow_plaintext: true, + // Off, as production defaults: the suite never reaches a real + // provider, and the targets that exercise enrichment build their + // own engine against a listener of their own. + metadata: MetadataSettings::off(), allowed_origins: Vec::new(), // The real defaults, so the whole test suite exercises the specs // production runs under rather than a simplified stand-in. @@ -972,6 +1051,66 @@ fn normalize_public_url(value: &str) -> anyhow::Result { Ok(parsed.origin().ascii_serialization()) } +/// Reads RFC-011 decision 3's settings. +fn parse_metadata_settings() -> anyhow::Result { + let defaults = MetadataSettings::off(); + let lastfm_key = read_trimmed_env("WAVEFLOW_METADATA_LASTFM_KEY"); + let providers = parse_metadata_providers( + &std::env::var("WAVEFLOW_METADATA_PROVIDERS").unwrap_or_default(), + lastfm_key.is_some(), + )?; + let days = |name: &str, default: Duration| -> anyhow::Result { + let days: u64 = parse_positive_env(name, default.as_secs() / (24 * 60 * 60))?; + days.checked_mul(24 * 60 * 60) + .map(Duration::from_secs) + .ok_or_else(|| anyhow::anyhow!("invalid {name}: too large")) + }; + Ok(MetadataSettings { + providers, + images: parse_bool_env("WAVEFLOW_METADATA_IMAGES", defaults.images)?, + lastfm_key, + ttl: days("WAVEFLOW_METADATA_TTL_DAYS", defaults.ttl)?, + miss_ttl: days("WAVEFLOW_METADATA_MISS_TTL_DAYS", defaults.miss_ttl)?, + rate_per_sec: parse_positive_env("WAVEFLOW_METADATA_RATE_PER_SEC", defaults.rate_per_sec)?, + request_timeout: Duration::from_secs(parse_positive_env( + "WAVEFLOW_METADATA_REQUEST_TIMEOUT_SECS", + defaults.request_timeout.as_secs(), + )?), + }) +} + +/// `WAVEFLOW_METADATA_PROVIDERS`, in the operator's order. +/// +/// A provider named without its key, an unknown name and a name given twice +/// all refuse to boot. Each is a typo, and a server that started with +/// enrichment quietly narrower than its operator wrote is the silent failure +/// these settings must not have — the operator learns it here, not from an +/// artist page that never fills. +fn parse_metadata_providers( + list: &str, + has_lastfm_key: bool, +) -> anyhow::Result> { + let mut providers = Vec::new(); + for name in list + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + { + let provider = crate::enrichment::Provider::parse(name) + .ok_or_else(|| anyhow::anyhow!("invalid WAVEFLOW_METADATA_PROVIDERS entry {name:?}"))?; + if providers.contains(&provider) { + anyhow::bail!("WAVEFLOW_METADATA_PROVIDERS names {name:?} twice"); + } + if provider == crate::enrichment::Provider::LastFm && !has_lastfm_key { + anyhow::bail!( + "WAVEFLOW_METADATA_PROVIDERS names lastfm without WAVEFLOW_METADATA_LASTFM_KEY" + ); + } + providers.push(provider); + } + Ok(providers) +} + fn parse_env(name: &str, default: &str) -> anyhow::Result where T: std::str::FromStr, @@ -1072,6 +1211,7 @@ fn default_pid_specs() -> crate::pid::PidSpecs { #[cfg(test)] mod tests { use super::normalize_public_url; + use super::parse_metadata_providers; use super::DEFAULT_DB_ACQUIRE_TIMEOUT_MS; use super::{parse_destinations, DEFAULT_SCROBBLE_DESTINATION}; use super::{validate_canvas, CanvasLimits}; @@ -1089,6 +1229,25 @@ mod tests { } } + /// RFC-011 decision 3: empty is off, and every typo refuses to boot rather + /// than starting enrichment narrower than the operator wrote. + #[test] + fn metadata_providers_are_read_strictly() { + use crate::enrichment::Provider; + + assert_eq!(parse_metadata_providers("", false).unwrap(), []); + assert_eq!(parse_metadata_providers(" , ", false).unwrap(), []); + assert_eq!( + parse_metadata_providers(" lastfm ", true).unwrap(), + [Provider::LastFm] + ); + // No key ships in the binary: naming the provider without one is a + // misconfiguration, not a quiet off. + assert!(parse_metadata_providers("lastfm", false).is_err()); + assert!(parse_metadata_providers("lastfm,lastfm", true).is_err()); + assert!(parse_metadata_providers("last.fm", true).is_err()); + } + #[test] fn upload_limits_that_contradict_each_other_are_refused() { assert!(validate_uploads(&workable()).is_ok()); diff --git a/src/database.rs b/src/database.rs index d60c5667..d0557d46 100644 --- a/src/database.rs +++ b/src/database.rs @@ -1900,6 +1900,136 @@ async fn insert_audit( Ok(()) } +/// RFC-011's cache, which [`crate::enrichment`] fills and reads. +/// +/// No user id and no tenancy anywhere in it: what a provider says about a name +/// is the same whoever asks, and nothing here is account data (decision 4). +/// Tenancy applies where a cached name meets the catalogue, which is the +/// reader's query, not these. +impl Database { + /// Every provider's row for one canonical name, in no particular order. + /// + /// A row from a provider this build does not know — written by a later + /// version the instance was downgraded from — is passed over rather than + /// refused: it is a fact about a provider nobody configured here. + pub async fn metadata_artist_rows( + &self, + name_canonical: &str, + ) -> Result, sqlx::Error> { + let rows = sqlx::query( + "SELECT provider, outcome, bio_summary, bio_full, fetched_at, expires_at \ + FROM metadata_artist WHERE name_canonical = ?", + ) + .bind(name_canonical) + .fetch_all(&self.pool) + .await?; + let mut cached = Vec::with_capacity(rows.len()); + for row in rows { + let provider: String = row.try_get("provider")?; + let Some(provider) = crate::enrichment::Provider::parse(&provider) else { + continue; + }; + let outcome: String = row.try_get("outcome")?; + let outcome = crate::enrichment::Outcome::parse(&outcome).ok_or_else(|| { + sqlx::Error::Decode(format!("unknown enrichment outcome {outcome:?}").into()) + })?; + cached.push(crate::enrichment::CachedArtist { + provider, + outcome, + bio_summary: row.try_get("bio_summary")?, + bio_full: row.try_get("bio_full")?, + fetched_at: row.try_get("fetched_at")?, + expires_at: row.try_get("expires_at")?, + }); + } + Ok(cached) + } + + /// One provider's ranked neighbours for one canonical name, best first. + pub async fn metadata_artist_similar( + &self, + provider: crate::enrichment::Provider, + name_canonical: &str, + ) -> Result, sqlx::Error> { + let rows = sqlx::query( + "SELECT similar_name, similar_canonical, score FROM metadata_artist_similar \ + WHERE provider = ? AND name_canonical = ? ORDER BY rank", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .fetch_all(&self.pool) + .await?; + rows.into_iter() + .map(|row| { + Ok(crate::enrichment::CachedSimilar { + name: row.try_get("similar_name")?, + name_canonical: row.try_get("similar_canonical")?, + score: row.try_get("score")?, + }) + }) + .collect() + } + + /// Records what a provider answered: its row and its whole neighbour list, + /// in one transaction. + /// + /// The writer gate is taken here and nowhere earlier — the request that + /// produced `answer` is over before this is called, so no network I/O ever + /// runs under it. + pub(crate) async fn write_metadata_artist( + &self, + provider: crate::enrichment::Provider, + name_canonical: &str, + answer: &crate::enrichment::ArtistAnswer, + fetched_at: i64, + expires_at: i64, + ) -> Result<(), sqlx::Error> { + let _writer = self.writer_guard().await; + let mut tx = self.pool.begin().await?; + sqlx::query( + "INSERT INTO metadata_artist \ + (provider, name_canonical, outcome, bio_summary, bio_full, fetched_at, expires_at) \ + VALUES (?, ?, ?, ?, ?, ?, ?) \ + ON CONFLICT (provider, name_canonical) DO UPDATE SET \ + outcome = excluded.outcome, bio_summary = excluded.bio_summary, \ + bio_full = excluded.bio_full, fetched_at = excluded.fetched_at, \ + expires_at = excluded.expires_at", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .bind(answer.outcome().as_str()) + .bind(answer.bio_summary.as_deref()) + .bind(answer.bio_full.as_deref()) + .bind(fetched_at) + .bind(expires_at) + .execute(&mut *tx) + .await?; + sqlx::query( + "DELETE FROM metadata_artist_similar WHERE provider = ? AND name_canonical = ?", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .execute(&mut *tx) + .await?; + for (rank, similar) in answer.similar.iter().enumerate() { + sqlx::query( + "INSERT INTO metadata_artist_similar \ + (provider, name_canonical, rank, similar_name, similar_canonical, score) \ + VALUES (?, ?, ?, ?, ?, ?)", + ) + .bind(provider.as_str()) + .bind(name_canonical) + .bind(rank as i64) + .bind(&similar.name) + .bind(&similar.name_canonical) + .bind(similar.score) + .execute(&mut *tx) + .await?; + } + tx.commit().await + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/enrichment.rs b/src/enrichment.rs new file mode 100644 index 00000000..7f2b99eb --- /dev/null +++ b/src/enrichment.rs @@ -0,0 +1,511 @@ +//! Metadata enrichment — RFC-011: the cache, and the engine that fills it. +//! +//! **Demand-driven and never on the answering path** (decision 6). A reader asks +//! [`Enrichment::artist`] for what the cache holds; for every configured +//! provider with nothing fresh, that call schedules a fill in the background +//! and returns at once. The first visit to an artist shows nothing; the next +//! one shows what came back. +//! +//! **Three outcomes** (decision 5), written on the provider's own row: a *hit* +//! stands for the long TTL, a *miss* — the provider answered and knows nothing +//! — for the short one, and *unreached* writes nothing at all, so the next +//! visit asks again. A network blip must never become "this artist has no +//! biography" for a month. +//! +//! **Bounded per provider**: requests are spaced to the configured rate, and +//! concurrent first visits for one name collapse onto a single fill. The +//! writer gate is taken for the row write only, after the requests are over. +//! +//! **Nothing about a request is logged but the provider and a verdict.** A +//! Last.fm request carries the operator's key in its query string; neither the +//! URL nor a response body is ever written to a log (decision 9). + +use std::{ + sync::Arc, + time::{Duration, Instant}, +}; + +use dashmap::DashMap; +use waveflow_core::metadata::{lastfm::LastfmClient, LookupError}; + +use crate::{config::MetadataSettings, database::Database}; + +/// Last.fm's API root. A constant, not a setting (decision 9): there is no +/// operator field, so there is no URL to validate. +const LASTFM_BASE_URL: &str = "https://ws.audioscrobbler.com/2.0/"; + +/// How many neighbours are asked for and kept. Subsonic's `getArtistInfo` +/// defaults to twenty; asking for more leaves room for the ones a caller cannot +/// see, which decision 7 drops at read time. +const SIMILAR_LIMIT: u32 = 50; + +/// How many fills may be waiting or running at once, across providers. +/// +/// Pacing reserves a slot for every request and refuses none, so without this +/// a page that opens two thousand artists would queue two thousand fills — +/// a quarter of an hour at the default rate — and an outage, which writes no +/// row, would let every further visit add more. Beyond the ceiling nothing is +/// scheduled; the visit after asks again, which is what an unreached lookup +/// does anyway. +pub const MAX_IN_FLIGHT: usize = 64; + +/// A provider an operator can name in `WAVEFLOW_METADATA_PROVIDERS`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum Provider { + LastFm, +} + +impl Provider { + pub fn parse(value: &str) -> Option { + match value { + "lastfm" => Some(Self::LastFm), + _ => None, + } + } + + /// The name in the setting, in the cache's `provider` column, and in logs. + pub fn as_str(self) -> &'static str { + match self { + Self::LastFm => "lastfm", + } + } +} + +/// What a provider said, when it said something. Unreached has no variant: it +/// is the absence of a row. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + Hit, + Miss, +} + +impl Outcome { + pub fn parse(value: &str) -> Option { + match value { + "hit" => Some(Self::Hit), + "miss" => Some(Self::Miss), + _ => None, + } + } + + pub fn as_str(self) -> &'static str { + match self { + Self::Hit => "hit", + Self::Miss => "miss", + } + } +} + +/// One provider's cached row for one name. +#[derive(Debug, Clone)] +pub struct CachedArtist { + pub provider: Provider, + pub outcome: Outcome, + pub bio_summary: Option, + pub bio_full: Option, + pub fetched_at: i64, + pub expires_at: i64, +} + +/// One neighbour in a provider's cached list. +#[derive(Debug, Clone)] +pub struct CachedSimilar { + pub name: String, + pub name_canonical: String, + pub score: f64, +} + +/// What one fill brought back from one provider, before it is written. +#[derive(Debug, Default)] +pub(crate) struct ArtistAnswer { + pub(crate) bio_summary: Option, + pub(crate) bio_full: Option, + pub(crate) similar: Vec, +} + +impl ArtistAnswer { + /// A hit carries something; an answer with nothing in it is a miss. + pub(crate) fn outcome(&self) -> Outcome { + if self.bio_summary.is_some() || self.bio_full.is_some() || !self.similar.is_empty() { + Outcome::Hit + } else { + Outcome::Miss + } + } +} + +/// Where each provider is reached. [`Endpoints::default`] is production; the +/// test suite points these at listeners of its own, which is the only reason +/// the type exists. +#[derive(Debug, Clone)] +pub struct Endpoints { + pub lastfm: String, +} + +impl Default for Endpoints { + fn default() -> Self { + Self { + lastfm: LASTFM_BASE_URL.to_owned(), + } + } +} + +/// The engine, or its absence. +/// +/// Cheap to clone; every clone drives the same engine. Disabled when no +/// provider is configured, and then every call returns at once having read and +/// sent nothing. +#[derive(Clone)] +pub struct Enrichment { + engine: Option>, +} + +struct Engine { + db: Database, + /// In the operator's order, which is the order fields are merged in. + sources: Vec, + ttl_ms: i64, + miss_ttl_ms: i64, + request_timeout: Duration, + /// The fills in flight, by provider and canonical name. A visit that finds + /// its key here schedules nothing: the fill already running answers it. + in_flight: DashMap<(Provider, String), ()>, + /// How many fills are waiting or running: the count [`MAX_IN_FLIGHT`] + /// bounds, and the one [`Enrichment::idle`] waits on. + running: tokio::sync::watch::Sender, +} + +struct Source { + provider: Provider, + client: Client, + pace: Pace, +} + +enum Client { + LastFm { client: LastfmClient, key: String }, +} + +impl Enrichment { + /// Enrichment switched off. + pub fn disabled() -> Self { + Self { engine: None } + } + + /// The engine `settings` describe, against the providers' real addresses. + pub fn new(db: Database, settings: &MetadataSettings) -> anyhow::Result { + Self::with_endpoints(db, settings, Endpoints::default()) + } + + /// [`Self::new`], with the providers reached at `endpoints`. + /// + /// Only the configured providers are built. A key placed with no provider + /// naming it builds nothing — the provider list is the switch, and a key + /// is not consent (decision 3). + pub fn with_endpoints( + db: Database, + settings: &MetadataSettings, + endpoints: Endpoints, + ) -> anyhow::Result { + let mut sources = Vec::new(); + for &provider in &settings.providers { + let http = crate::outbound::client(settings.request_timeout) + .map_err(|_| anyhow::anyhow!("the outbound client could not be built"))?; + let client = match provider { + Provider::LastFm => { + let key = settings.lastfm_key.clone().ok_or_else(|| { + anyhow::anyhow!("metadata provider lastfm is configured without a key") + })?; + Client::LastFm { + client: LastfmClient::with_http(http, endpoints.lastfm.clone())?, + key, + } + } + }; + sources.push(Source { + provider, + client, + pace: Pace::new(settings.rate_per_sec), + }); + } + if sources.is_empty() { + return Ok(Self::disabled()); + } + Ok(Self { + engine: Some(Arc::new(Engine { + db, + sources, + ttl_ms: duration_ms(settings.ttl), + miss_ttl_ms: duration_ms(settings.miss_ttl), + request_timeout: settings.request_timeout, + in_flight: DashMap::new(), + running: tokio::sync::watch::channel(0).0, + })), + }) + } + + pub fn is_enabled(&self) -> bool { + self.engine.is_some() + } + + /// What the cache holds about the artist called `name`, one row per + /// configured provider that has one, in the operator's order — and a fill + /// scheduled for every configured provider whose row is missing or stale. + /// + /// Stale rows are returned, not withheld (decision 5): a slightly old + /// biography beats none, and serving it keeps a provider outage invisible. + /// Nothing here waits on a provider. + pub async fn artist(&self, name: &str) -> Result, sqlx::Error> { + let Some(engine) = &self.engine else { + return Ok(Vec::new()); + }; + let canonical = waveflow_core::scanner::canonical_name(name); + if canonical.is_empty() { + return Ok(Vec::new()); + } + let rows = engine.db.metadata_artist_rows(&canonical).await?; + let now = crate::authentication::now_ms(); + let mut ordered = Vec::new(); + for (index, source) in engine.sources.iter().enumerate() { + let row = rows.iter().find(|row| row.provider == source.provider); + if row.is_none_or(|row| row.expires_at <= now) { + Arc::clone(engine).schedule(index, name, &canonical); + } + ordered.extend(row.cloned()); + } + Ok(ordered) + } + + /// Waits until no fill is running. + /// + /// For the test suite, which has to know a fill is over before it can say + /// what the fill wrote — or that it wrote nothing. + #[doc(hidden)] + pub async fn idle(&self) { + let Some(engine) = &self.engine else { + return; + }; + let mut running = engine.running.subscribe(); + // The sender lives as long as the engine, which `self` holds. + let _ = running.wait_for(|count| *count == 0).await; + } +} + +impl Engine { + /// Starts a fill for one provider and one name, unless one is running or + /// [`MAX_IN_FLIGHT`] are. + fn schedule(self: Arc, index: usize, name: &str, canonical: &str) { + let key = (self.sources[index].provider, canonical.to_owned()); + // A place is reserved before the key is claimed, in one step under the + // counter's lock: a length read then an insert would let two visits on + // two threads both see one place left and both take it. + let reserved = self.running.send_if_modified(|count| { + let free = *count < MAX_IN_FLIGHT; + if free { + *count += 1; + } + free + }); + if !reserved { + return; + } + if self.in_flight.insert(key.clone(), ()).is_some() { + self.running.send_modify(|count| *count -= 1); + return; + } + // Released however the fill ends — a panic, or a task dropped before + // its first poll included, which is why it is built here rather than + // inside the task. A key left behind would stop that name from ever + // being asked again. + let flight = Flight { engine: self, key }; + let name = name.to_owned(); + tokio::spawn(async move { + flight.engine.fill(index, &name, &flight.key.1).await; + drop(flight); + }); + } + + async fn fill(&self, index: usize, name: &str, canonical: &str) { + let source = &self.sources[index]; + let answer = match &source.client { + Client::LastFm { client, key } => self.ask_lastfm(source, client, key, name).await, + }; + let answer = match answer { + Ok(answer) => answer, + Err(verdict) => { + tracing::warn!( + provider = source.provider.as_str(), + verdict = %verdict, + "artist lookup unreached; nothing cached" + ); + return; + } + }; + let fetched_at = crate::authentication::now_ms(); + let outcome = answer.outcome(); + let ttl = match outcome { + Outcome::Hit => self.ttl_ms, + Outcome::Miss => self.miss_ttl_ms, + }; + if let Err(error) = self + .db + .write_metadata_artist( + source.provider, + canonical, + &answer, + fetched_at, + fetched_at.saturating_add(ttl), + ) + .await + { + tracing::warn!( + provider = source.provider.as_str(), + error = %error, + "artist lookup answered but could not be cached" + ); + return; + } + tracing::debug!( + provider = source.provider.as_str(), + outcome = outcome.as_str(), + "artist lookup cached" + ); + } + + /// `artist.getInfo` and `artist.getSimilar`. Both must be reached for the + /// answer to count: half an answer written as a whole one would cache the + /// missing half as "nothing" for the TTL. + async fn ask_lastfm( + &self, + source: &Source, + client: &LastfmClient, + key: &str, + name: &str, + ) -> Result { + source.pace.wait().await; + let info = self.bounded(client.artist_get_info(name, key)).await?; + source.pace.wait().await; + let similar = self + .bounded(client.artist_get_similar(name, key, SIMILAR_LIMIT)) + .await?; + let own = waveflow_core::scanner::canonical_name(name); + let similar = similar + .into_iter() + .filter_map(|neighbour| { + let name_canonical = waveflow_core::scanner::canonical_name(&neighbour.name); + (!name_canonical.is_empty() && name_canonical != own).then(|| CachedSimilar { + name: neighbour.name, + name_canonical, + score: f64::from(neighbour.match_score), + }) + }) + .collect(); + let (bio_summary, bio_full) = info + .map(|info| (info.bio_summary, info.bio_full)) + .unwrap_or_default(); + Ok(ArtistAnswer { + bio_summary, + bio_full, + similar, + }) + } + + /// One request, under the request timeout; anything but an answer is a + /// verdict. + async fn bounded( + &self, + request: impl std::future::Future>, + ) -> Result { + match tokio::time::timeout(self.request_timeout, request).await { + Ok(Ok(answer)) => Ok(answer), + Ok(Err(error)) => Err(Verdict::of(&error)), + Err(_) => Err(Verdict::Timeout), + } + } +} + +/// Releases a fill's key and count when the fill ends. +struct Flight { + engine: Arc, + key: (Provider, String), +} + +impl Drop for Flight { + fn drop(&mut self) { + self.engine.in_flight.remove(&self.key); + self.engine.running.send_modify(|count| *count -= 1); + } +} + +/// Why a lookup was unreached, in words that carry nothing from the request or +/// the response — no URL, no body, no key. +#[derive(Debug)] +enum Verdict { + Timeout, + Transport, + Refused { code: i64 }, + Status { status: u16 }, + TooLarge, + Undecodable, +} + +impl Verdict { + fn of(error: &LookupError) -> Self { + match error { + LookupError::Http(error) if error.is_timeout() => Self::Timeout, + LookupError::Http(_) => Self::Transport, + LookupError::Refused { code } => Self::Refused { code: *code }, + LookupError::Status { status } => Self::Status { status: *status }, + LookupError::TooLarge { .. } => Self::TooLarge, + LookupError::Decode(_) => Self::Undecodable, + } + } +} + +impl std::fmt::Display for Verdict { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Timeout => formatter.write_str("timed out"), + Self::Transport => formatter.write_str("transport failure"), + Self::Refused { code } => write!(formatter, "refused with code {code}"), + Self::Status { status } => write!(formatter, "answered HTTP {status}"), + Self::TooLarge => formatter.write_str("answer over the size ceiling"), + Self::Undecodable => formatter.write_str("answer not understood"), + } + } +} + +/// Spaces one provider's requests to its rate: a token bucket that holds one +/// token, so there is never a burst to spend. +/// +/// The reservation is taken under the lock and the wait happens outside it, so +/// callers queue in order without one sleeping caller holding up the rest. +struct Pace { + interval: Duration, + next: std::sync::Mutex, +} + +impl Pace { + fn new(rate_per_sec: u32) -> Self { + Self { + interval: Duration::from_secs(1) / rate_per_sec.max(1), + next: std::sync::Mutex::new(Instant::now()), + } + } + + async fn wait(&self) { + let slot = { + let mut next = self + .next + .lock() + .unwrap_or_else(|poison| poison.into_inner()); + let slot = (*next).max(Instant::now()); + *next = slot + self.interval; + slot + }; + tokio::time::sleep_until(slot.into()).await; + } +} + +fn duration_ms(duration: Duration) -> i64 { + i64::try_from(duration.as_millis()).unwrap_or(i64::MAX) +} diff --git a/src/lib.rs b/src/lib.rs index 896c856f..b853c549 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,9 +6,11 @@ pub mod catalog; pub mod cli; pub mod config; pub mod database; +pub mod enrichment; pub mod lyrics; pub mod media; pub mod oauth; +pub mod outbound; pub mod pid; pub mod playlist_file; pub mod scanner; @@ -76,6 +78,9 @@ pub struct AppState { /// The body ceiling the canvas route carries, derived from /// `WAVEFLOW_CANVAS_MAX_BYTES` so the two cannot disagree. pub canvas_body_limit: usize, + /// Metadata enrichment — RFC-011. Disabled unless the operator named a + /// provider. + pub enrichment: enrichment::Enrichment, } /// Roughly one offer's worth of JSON — a sixty-four character hash, a size, a @@ -596,8 +601,9 @@ pub async fn initialize(config: &Config) -> anyhow::Result { scanner.clone(), config, ); - // The only outbound surface this server has, and it is switched on by the - // operator naming a destination rather than by anybody linking an account. + // One of the two outbound surfaces this server has — metadata enrichment, + // below, is the other — and it is switched on by the operator naming a + // destination rather than by anybody linking an account. // Nothing leaves until both are true: an adapter with no authorisation // behind it is never handed a listen. RFC-010 decision 11. for destination in &config.destinations { @@ -673,6 +679,9 @@ pub async fn initialize(config: &Config) -> anyhow::Result { // the background task already runs leaves a window, short and sufficient, // in which listens leave for yesterday's destination. RFC-010 decision 10. services.reconcile_scrobble_links().await?; + // Off unless the operator named a provider, and then it still makes no + // request until somebody asks about an artist (RFC-011 decisions 3 and 6). + let enrichment = enrichment::Enrichment::new(db.clone(), &config.metadata)?; Ok(AppState { db, auth, @@ -691,6 +700,7 @@ pub async fn initialize(config: &Config) -> anyhow::Result { // Refused at startup if it does not fit, so this conversion cannot be // the place that decides what a ceiling falls back to. canvas_body_limit: usize::try_from(config.canvas.max_bytes).unwrap_or(usize::MAX), + enrichment, }) } diff --git a/src/outbound.rs b/src/outbound.rs new file mode 100644 index 00000000..2f9ac95a --- /dev/null +++ b/src/outbound.rs @@ -0,0 +1,47 @@ +//! The HTTP client every outbound request in this server is made with. +//! +//! Two subsystems call out: scrobbling (RFC-010), to destinations an operator +//! names, and metadata enrichment (RFC-011), to providers whose addresses are +//! constants. They share this client so the rules below are written once and +//! cannot drift apart — RFC-011 decision 9 adopts RFC-010 decision 10 rather +//! than restating it, and a second builder would be the place it quietly +//! stopped being true. + +use std::time::Duration; + +/// How much of the budget a connection may spend before the rest of the request +/// has had any of it. +/// +/// A third, so a destination that accepts slowly still leaves room for the +/// request itself rather than consuming the whole deadline in the handshake. +const CONNECT_SHARE: u32 = 3; + +/// The client. +/// +/// Each property below is RFC-010 decision 10 written down: +/// +/// - **No redirect is followed.** `reqwest` follows up to ten by default, and +/// the tenth can be anywhere. A destination that answers `302` to an internal +/// address would otherwise have this server fetch it and hand back what it +/// found, which is the shape of every request-forgery there is. +/// - **No proxy is read from the environment.** An operator's `HTTP_PROXY`, +/// set for something else entirely, must not silently become the route every +/// request takes. +/// - **Only the connection is bounded here.** The request as a whole is bounded +/// once, by the caller's own `tokio::time::timeout`, and deliberately not a +/// second time by this client. +/// +/// That last one was two equal deadlines racing until a review noticed. Both +/// ended at `Ambiguous`, so the scrobbling queue looked the same either way — +/// but only the drain's own deadline records the link as stalled, and that +/// record is what stops one silent destination from spending an entry per row. +/// Which of the two fired was a coin toss, so which protection applied was a +/// coin toss. +pub fn client(timeout: Duration) -> Result { + reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(timeout / CONNECT_SHARE) + .no_proxy() + .user_agent(concat!("WaveFlowServer/", env!("CARGO_PKG_VERSION"))) + .build() +} diff --git a/src/scrobblers/mod.rs b/src/scrobblers/mod.rs index 41045e35..fe70ab5c 100644 --- a/src/scrobblers/mod.rs +++ b/src/scrobblers/mod.rs @@ -5,10 +5,11 @@ //! That line is what decision 6 exists to draw, and putting an adapter beside //! the drain would rub it out within a release. //! -//! **This is the only place in the server that makes an outbound request.** -//! Decision 10 calls that the real risk of the whole RFC — a server that calls -//! a URL is a server somebody can make call a URL — so the rules live here, -//! once, rather than in each adapter. +//! Decision 10 calls outbound requests the real risk of the whole RFC — a +//! server that calls a URL is a server somebody can make call a URL — so the +//! rules live once, in [`crate::outbound`], rather than in each adapter. That +//! module is shared with metadata enrichment (RFC-011), the server's only other +//! outbound surface. use std::time::Duration; @@ -118,41 +119,10 @@ pub fn destination_fingerprint(url: &url::Url) -> String { )) } -/// How much of the budget a connection may spend before the rest of the request -/// has had any of it. -/// -/// A third, so a destination that accepts slowly still leaves room for the -/// submission itself rather than consuming the whole deadline in the handshake. -const CONNECT_SHARE: u32 = 3; - -/// The client every adapter shares. -/// -/// Each property below is decision 10 written down: -/// -/// - **No redirect is followed.** `reqwest` follows up to ten by default, and -/// the tenth can be anywhere. A destination that answers `302` to an internal -/// address would otherwise have this server fetch it and hand back what it -/// found, which is the shape of every request-forgery there is. -/// - **No proxy is read from the environment.** An operator's `HTTP_PROXY`, -/// set for something else entirely, must not silently become the route every -/// listen takes. -/// - **Only the connection is bounded here.** The request as a whole is bounded -/// once, by the drain's own `tokio::time::timeout`, and deliberately not a -/// second time by this client. -/// -/// That last one was two equal deadlines racing until a review noticed. Both -/// ended at `Ambiguous`, so the queue looked the same either way — but only the -/// drain's own deadline records the link as stalled, and that record is what -/// stops one silent destination from spending an entry per row. Which of the two -/// fired was a coin toss, so which protection applied was a coin toss. +/// The client every adapter shares: [`crate::outbound::client`], which holds +/// decision 10's rules and is shared with metadata enrichment. pub fn outbound_client(timeout: Duration) -> Result { - reqwest::Client::builder() - .redirect(reqwest::redirect::Policy::none()) - .connect_timeout(timeout / CONNECT_SHARE) - .no_proxy() - .user_agent(concat!("WaveFlowServer/", env!("CARGO_PKG_VERSION"))) - .build() - .map_err(|_| OutboundError::Unbuildable) + crate::outbound::client(timeout).map_err(|_| OutboundError::Unbuildable) } /// Checks a destination an operator configured, before it is ever called. diff --git a/tests/enrichment.rs b/tests/enrichment.rs new file mode 100644 index 00000000..0988764c --- /dev/null +++ b/tests/enrichment.rs @@ -0,0 +1,467 @@ +//! Metadata enrichment's cache and engine — RFC-011 decisions 3, 5, 6 and 9. +//! +//! Every test points the engine at a fake Last.fm on a loopback listener of its +//! own; the suite never reaches a real provider. The fake counts what it is +//! asked, which is what most of these assertions are about: enrichment's +//! failures are requests made that should not have been, or answers cached that +//! were never given. + +// Not every target uses every fixture, and a shared module is not dead +// code for being partly unused here. +#[allow(dead_code)] +mod support; + +use std::{ + collections::HashMap, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::Duration, +}; + +use axum::{ + extract::{Query, State}, + http::{header, StatusCode}, + response::{IntoResponse, Response}, + routing::get, + Router, +}; +use waveflow_server::{ + config::MetadataSettings, + enrichment::{Endpoints, Enrichment, Outcome, Provider}, +}; + +#[derive(Clone, Copy)] +enum Mode { + /// A biography and two neighbours. + Hit, + /// Last.fm's code 6 to both methods: it knows nothing of this artist. + Miss, + /// A failure status with a body that is not Last.fm's. + Unavailable, + /// A `302` to `/followed`, which answers like `Hit`. + Redirect, + /// A well-formed answer three mebibytes long. + Oversized, +} + +#[derive(Clone)] +struct Fake { + mode: Mode, + delay: Duration, + info: Arc, + similar: Arc, + followed: Arc, +} + +struct Listener { + fake: Fake, + base_url: String, +} + +async fn fake_lastfm(mode: Mode, delay: Duration) -> Listener { + let fake = Fake { + mode, + delay, + info: Arc::default(), + similar: Arc::default(), + followed: Arc::default(), + }; + let router = Router::new() + .route("/2.0/", get(answer)) + .route("/followed", get(followed)) + .with_state(fake.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); + Listener { + fake, + base_url: format!("http://{address}/2.0/"), + } +} + +async fn answer( + State(fake): State, + Query(query): Query>, +) -> Response { + let method = query.get("method").map(String::as_str).unwrap_or_default(); + match method { + "artist.getinfo" => fake.info.fetch_add(1, Ordering::SeqCst), + "artist.getsimilar" => fake.similar.fetch_add(1, Ordering::SeqCst), + _ => return StatusCode::BAD_REQUEST.into_response(), + }; + tokio::time::sleep(fake.delay).await; + match fake.mode { + Mode::Hit => found(method), + Mode::Miss => { + json(r#"{"error":6,"message":"The artist you supplied could not be found"}"#.to_owned()) + } + Mode::Unavailable => (StatusCode::SERVICE_UNAVAILABLE, "down").into_response(), + Mode::Redirect => (StatusCode::FOUND, [(header::LOCATION, "/followed")]).into_response(), + Mode::Oversized => json(format!( + r#"{{"artist":{{"name":"Björk","bio":{{"summary":"{}"}}}}}}"#, + "a".repeat(3 * 1024 * 1024) + )), + } +} + +async fn followed(State(fake): State) -> Response { + fake.followed.fetch_add(1, Ordering::SeqCst); + found("artist.getinfo") +} + +fn found(method: &str) -> Response { + json(if method == "artist.getinfo" { + r#"{"artist":{"name":"Björk","bio":{"summary":"Icelandic singer.","content":"Icelandic singer, at length."}}}"#.to_owned() + } else { + r#"{"similarartists":{"artist":[ + {"name":"Sigur Rós","match":"0.9","image":[]}, + {"name":"Múm","match":"0.75","image":[]} + ]}}"# + .to_owned() + }) +} + +fn json(body: String) -> Response { + ([(header::CONTENT_TYPE, "application/json")], body).into_response() +} + +/// A key made up for the run, so no credential sits in the source and a log +/// search for it cannot match anything but this run's own output. +fn run_key() -> String { + uuid::Uuid::new_v4().simple().to_string() +} + +fn settings(providers: Vec, key: &str) -> MetadataSettings { + MetadataSettings { + providers, + lastfm_key: Some(key.to_owned()), + // Fast enough that pacing never decides a test. + rate_per_sec: 1_000, + request_timeout: Duration::from_secs(5), + ..MetadataSettings::off() + } +} + +fn engine( + state: &waveflow_server::AppState, + settings: &MetadataSettings, + listener: &Listener, +) -> Enrichment { + Enrichment::with_endpoints( + state.db.clone(), + settings, + Endpoints { + lastfm: listener.base_url.clone(), + }, + ) + .unwrap() +} + +fn canonical(name: &str) -> String { + waveflow_core::scanner::canonical_name(name) +} + +/// Decision 3: the provider list is the switch. A key placed without a +/// provider naming it — the shape an operator who set up scrobbling first +/// leaves behind — is not consent, and asks nothing. +#[tokio::test] +async fn an_empty_provider_list_asks_nothing_even_with_a_key() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Hit, Duration::ZERO).await; + let enrichment = engine(&state, &settings(Vec::new(), &run_key()), &listener); + + assert!(!enrichment.is_enabled()); + assert!(enrichment.artist("Björk").await.unwrap().is_empty()); + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 0); + assert_eq!(listener.fake.similar.load(Ordering::SeqCst), 0); + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); +} + +/// The ordinary path, and decision 6's cost stated as a test: the first visit +/// answers from an empty cache, the second has what came back. +#[tokio::test] +async fn a_hit_is_cached_with_its_biography_and_ranked_neighbours() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Hit, Duration::ZERO).await; + let settings = settings(vec![Provider::LastFm], &run_key()); + let enrichment = engine(&state, &settings, &listener); + + assert!(enrichment.artist("Björk").await.unwrap().is_empty()); + enrichment.idle().await; + let rows = enrichment.artist("Björk").await.unwrap(); + + assert_eq!(rows.len(), 1); + let row = &rows[0]; + assert_eq!(row.provider, Provider::LastFm); + assert_eq!(row.outcome, Outcome::Hit); + assert_eq!(row.bio_summary.as_deref(), Some("Icelandic singer.")); + assert_eq!( + row.bio_full.as_deref(), + Some("Icelandic singer, at length.") + ); + assert_eq!( + row.expires_at - row.fetched_at, + settings.ttl.as_millis() as i64 + ); + let similar = state + .db + .metadata_artist_similar(Provider::LastFm, &canonical("Björk")) + .await + .unwrap(); + let names: Vec<_> = similar.iter().map(|s| s.name.as_str()).collect(); + assert_eq!(names, ["Sigur Rós", "Múm"]); + assert_eq!(similar[0].name_canonical, canonical("Sigur Rós")); + // A fresh row is served, not asked again. + enrichment.idle().await; + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); +} + +/// Decision 5: "the provider knows nothing" is an answer, cached — for the +/// short window, because it is the answer most likely to change. +#[tokio::test] +async fn a_provider_that_knows_nothing_is_cached_as_a_miss_for_the_miss_ttl() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Miss, Duration::ZERO).await; + let settings = settings(vec![Provider::LastFm], &run_key()); + let enrichment = engine(&state, &settings, &listener); + + enrichment.artist("Nobody Knows").await.unwrap(); + enrichment.idle().await; + + let rows = state + .db + .metadata_artist_rows(&canonical("Nobody Knows")) + .await + .unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].outcome, Outcome::Miss); + assert_eq!(rows[0].bio_summary, None); + assert_eq!( + rows[0].expires_at - rows[0].fetched_at, + settings.miss_ttl.as_millis() as i64, + "a miss stands for the miss TTL, not the hit TTL" + ); +} + +/// Decision 5's third outcome: a provider that could not be asked writes no +/// row, so the next visit asks again instead of reading the outage as an +/// artist without a biography. +#[tokio::test] +async fn a_provider_that_cannot_be_reached_writes_no_row() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Unavailable, Duration::ZERO).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 2); +} + +/// Decision 6: ten clients opening the same page are one request. +#[tokio::test] +async fn ten_concurrent_first_visits_make_one_request() { + let (_temp, _config, state) = support::test_app().await; + // Slow enough that every visit lands while the first fill is still out. + let listener = fake_lastfm(Mode::Hit, Duration::from_millis(300)).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + let visits = (0..10).map(|_| enrichment.artist("Björk")); + for answer in futures_util::future::join_all(visits).await { + assert!(answer.unwrap().is_empty()); + } + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); + assert_eq!(listener.fake.similar.load(Ordering::SeqCst), 1); +} + +/// Pacing refuses no request, so the ceiling on fills in flight is what bounds +/// the queue: one artist past it is not asked, and its next visit will be. +#[tokio::test] +async fn no_more_than_the_ceiling_of_fills_is_ever_queued() { + let (_temp, _config, state) = support::test_app().await; + // Slow enough that every fill is still out when the last visit lands. + let listener = fake_lastfm(Mode::Hit, Duration::from_secs(1)).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + let names: Vec = (0..=waveflow_server::enrichment::MAX_IN_FLIGHT) + .map(|index| format!("Artist {index}")) + .collect(); + let visits = names.iter().map(|name| enrichment.artist(name)); + for answer in futures_util::future::join_all(visits).await { + answer.unwrap(); + } + enrichment.idle().await; + + assert_eq!( + listener.fake.info.load(Ordering::SeqCst), + waveflow_server::enrichment::MAX_IN_FLIGHT + ); +} + +/// Decision 9: no redirect is followed. Followed, this one would have been +/// cached as a hit from wherever the provider pointed. +#[tokio::test] +async fn a_redirect_from_the_provider_is_not_followed() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Redirect, Duration::ZERO).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); + assert_eq!(listener.fake.followed.load(Ordering::SeqCst), 0); + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); +} + +/// Decision 9: a response is bounded. Read whole, this one would have been +/// cached as a three-mebibyte biography. +#[tokio::test] +async fn an_answer_over_the_ceiling_is_refused() { + let (_temp, _config, state) = support::test_app().await; + let listener = fake_lastfm(Mode::Oversized, Duration::ZERO).await; + let enrichment = engine( + &state, + &settings(vec![Provider::LastFm], &run_key()), + &listener, + ); + + enrichment.artist("Björk").await.unwrap(); + enrichment.idle().await; + + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1); + assert!(state + .db + .metadata_artist_rows(&canonical("Björk")) + .await + .unwrap() + .is_empty()); +} + +/// A writer for the log, kept so the test can read it back. +#[derive(Clone, Default)] +struct Captured(Arc>>); + +impl std::io::Write for Captured { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } +} + +impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Captured { + type Writer = Self; + + fn make_writer(&'a self) -> Self::Writer { + self.clone() + } +} + +/// The log of this whole test binary, at every level. +/// +/// Global rather than scoped to one test: a subscriber installed with +/// `set_default` covers one thread, and tracing caches each callsite's interest +/// for the whole process — with the other tests running beside it, a scoped +/// capture kept two lines out of hundreds. Every test here writes into this +/// one; the assertion below looks for its own run's key, which no other test +/// knows. +fn captured_log() -> &'static Captured { + static LOG: std::sync::OnceLock = std::sync::OnceLock::new(); + LOG.get_or_init(|| { + let captured = Captured::default(); + tracing_subscriber::fmt() + .with_max_level(tracing::Level::TRACE) + .with_writer(captured.clone()) + .with_ansi(false) + .init(); + captured + }) +} + +/// Decision 9 and `CLAUDE.md`: the key travels in Last.fm's query string, and +/// must reach no log line — not on success, not on a refusal, not on a +/// connection that never opened — at any level, the HTTP stack's own included. +#[tokio::test] +async fn the_key_never_reaches_the_log() { + let captured = captured_log(); + + let (_temp, _config, state) = support::test_app().await; + let key = run_key(); + let settings = settings(vec![Provider::LastFm], &key); + // One artist per mode: asked about the same one, the second engine would + // find the first one's fresh row and send nothing — and the 503 checked + // below would then only be found if another test happened to log one. + for (mode, artist) in [(Mode::Hit, "Björk"), (Mode::Unavailable, "Múm")] { + let listener = fake_lastfm(mode, Duration::ZERO).await; + let enrichment = engine(&state, &settings, &listener); + enrichment.artist(artist).await.unwrap(); + enrichment.idle().await; + assert_eq!(listener.fake.info.load(Ordering::SeqCst), 1, "{artist}"); + } + // Nothing listens on port 1 of loopback: the connection itself fails. + let closed = Enrichment::with_endpoints( + state.db.clone(), + &settings, + Endpoints { + lastfm: "http://127.0.0.1:1/2.0/".to_owned(), + }, + ) + .unwrap(); + closed.artist("Sigur Rós").await.unwrap(); + closed.idle().await; + + let log = String::from_utf8(captured.0.lock().unwrap().clone()).unwrap(); + // The capture works: both failures were logged, by provider and verdict. + // A closed port is refused on Linux and left to the connect timeout on + // Windows, so the second verdict is one of two. + assert!(log.contains("answered HTTP 503"), "{log}"); + assert!( + log.contains("verdict=transport failure") || log.contains("verdict=timed out"), + "{log}" + ); + assert!(!log.contains(&key), "the key reached the log:\n{log}"); +}