diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 0954feb..fd8cfec 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -34,9 +34,14 @@ jobs: restore-keys: cargo- # The cover-art cache is append-only and grows every run, so it # must save each time — but it's a few KB, unlike the target dir. + # The mirrored image cache (priv/cache/img) gets the same + # treatment: once a cover/artist photo is downloaded it never + # needs the network again. - uses: actions/cache@v4 with: - path: priv/cache/cover-cache.json + path: | + priv/cache/cover-cache.json + priv/cache/img key: cover-cache-${{ github.run_id }} restore-keys: cover-cache- diff --git a/README.md b/README.md index 1fd3607..7929edf 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,12 @@ # karitham.dev -Personal site. Gleam SSG that fetches from the AT Protocol (Bluesky PDS + Tangled) at build time, hydrates client-side with a Lustre component tree. +Personal site. Gleam SSG that fetches from the AT Protocol (Bluesky PDS + Tangled) at build time, hydrates client-side with a Lustre component tree. Album covers, artist photos, and the profile avatar/banner are mirrored into the site at build time, so the visitor's browser never has to fetch from Cover Art Archive, Wikimedia, or the PDS directly — every image is served from `/img/...` on the site itself. ## Build ```sh nix develop # or: direnv reload +just refresh # data pipeline: plays + covers + mirrored images just build # codegen → client JS → static site → ./dist/ just test # run all tests just clean # wipe build artifacts @@ -24,6 +25,20 @@ Needs Gleam 1.17+ and Erlang/OTP 28+ — `nix develop` provides everything. BLOG_URL="http://localhost:8000" just build ``` +## Refreshing listening data + +The Music section's stats, covers, and page links come from a data pipeline that runs before `just build`: + +```sh +just refresh +``` + +`refresh` downloads your play history as a CAR file from the PDS, aggregates top-N artists/albums/tracks per time range, resolves album covers (Cover Art Archive), artist photos (MusicBrainz → Wikidata → Wikimedia Commons), and MusicBrainz page links, then writes `priv/cache/plays-stats.json` and mirrors every image into `priv/cache/img/` for the SSG to copy into `dist/img/`. + +Every lookup is cache-first with per-endpoint rate limiting, jittered retries, and `Retry-After` handling; with a warm cache `refresh` is fully offline and only new plays touch the network. A failed or interrupted run never corrupts the caches (atomic writes), and the next run simply re-does what didn't finish. + +CI runs `just refresh && just build` on every push and caches `priv/cache/cover-cache.json` + `priv/cache/img` between runs, so deploys only resolve and download what's new. The whole `priv/cache` dir is gitignored and wiped by `just clean`. + ## Adding a post ```sh @@ -60,13 +75,14 @@ error if something's wrong. ## Layout -Three Gleam packages sharing generated types, decoders, and view code: +Four components: -| Package | Role | -| ------------- | --------------------------------------------------------------------------------- | -| **`shared/`** | Model types, generated decoders, view components (compiled to both Erlang and JS) | -| **root** | SSG — fetches data, renders, writes `dist/` | -| **`client/`** | Browser bundle — fetches fresh data on page load, polls plays every 30s | +| Component | Role | +| ----------------------- | ----------------------------------------------------------------------------------------------------------------------- | +| **`shared/`** | Model types, generated decoders, view components (compiled to both Erlang and JS) | +| **root** | SSG — fetches data, mirrors images, renders, writes `dist/` | +| **`client/`** | Browser bundle — fetches fresh data on page load (profile images rewritten to the local mirrors), polls plays every 30s | +| **`tools/parse-plays`** | Rust CLI — CAR → play stats, cover/artist resolution, image mirroring | ## Tests diff --git a/client/src/browser.gleam b/client/src/browser.gleam index 2ae2dfb..d486fb3 100644 --- a/client/src/browser.gleam +++ b/client/src/browser.gleam @@ -4,6 +4,9 @@ pub fn fetch_text(url: String, callback: fn(String) -> Nil) -> Nil @external(javascript, "./browser_ffi.mjs", "set_inner_html") pub fn set_inner_html(id: String, html: String) -> Nil +@external(javascript, "./browser_ffi.mjs", "rewrite_remote_images") +pub fn rewrite_remote_images() -> Nil + @external(javascript, "./browser_ffi.mjs", "set_attribute") pub fn set_attribute(id: String, name: String, value: String) -> Nil diff --git a/client/src/browser_ffi.mjs b/client/src/browser_ffi.mjs index ce52610..c08d123 100644 --- a/client/src/browser_ffi.mjs +++ b/client/src/browser_ffi.mjs @@ -24,6 +24,30 @@ export function set_inner_html(id, html) { } } +// Point every in the document at its local mirror. The map +// is the `#image-rewrites` JSON script tag the SSG embeds (remote URL +// -> local /img path); URLs not in the map (e.g. a freshly-changed +// avatar the build hasn't mirrored yet) keep their remote src and +// fall back to loading from the original host. +export function rewrite_remote_images() { + var script = document.getElementById("image-rewrites"); + if (!script) return; + var map; + try { + map = JSON.parse(script.textContent); + } catch (_) { + console.warn("rewrite_remote_images: bad #image-rewrites JSON"); + return; + } + var imgs = document.querySelectorAll("img[src]"); + for (var i = 0; i < imgs.length; i++) { + var src = imgs[i].getAttribute("src"); + if (src && Object.prototype.hasOwnProperty.call(map, src)) { + imgs[i].setAttribute("src", map[src]); + } + } +} + export function set_attribute(id, name, value) { var el = document.getElementById(id); if (el) { diff --git a/client/src/refresh.gleam b/client/src/refresh.gleam index 13ed9d4..a21382f 100644 --- a/client/src/refresh.gleam +++ b/client/src/refresh.gleam @@ -44,11 +44,15 @@ fn fetch_profile() -> Nil { fn on_profile(text: String) -> Nil { case fetch.decode_profile(text) { - Ok(profile) -> + Ok(profile) -> { browser.set_inner_html( "profile-section", dynamic.render(profile_view.profile(profile)), ) + // The fresh profile carries the PDS's remote avatar/banner URLs; + // point them at the local mirrors from the build. + browser.rewrite_remote_images() + } Error(reason) -> browser.log_error("decode_profile failed: " <> string.inspect(reason)) } diff --git a/justfile b/justfile index 4a3ed95..decdb9f 100644 --- a/justfile +++ b/justfile @@ -31,15 +31,18 @@ new *ARGS: # Fetch fresh listening data from PDS, then derive the stats the site embeds. # One pass: CAR in, stats out — the ~100MB raw dump is never materialized. # `--covers` enables MusicBrainz/Cover Art Archive lookups for pairs not yet -# cached; with a warm cache the stats step is offline. +# cached; with a warm cache the stats step is offline. `--images` mirrors the +# resolved cover/artist images into priv/cache/img so the browser never hits +# Cover Art Archive / Wikimedia at page load; a warm image cache is also offline. refresh: #!/usr/bin/env bash set -euo pipefail - mkdir -p priv/cache + mkdir -p priv/cache priv/cache/img curl -sL "https://eurosky.social/xrpc/com.atproto.sync.getRepo?did=did:plc:kcgwlowulc3rac43lregdawo" \ -o priv/cache/repo.car cd tools/parse-plays && cargo run --release -- refresh ../../priv/cache/repo.car \ - ../../priv/cache/plays-stats.json --covers ../../priv/cache/cover-cache.json + ../../priv/cache/plays-stats.json --covers ../../priv/cache/cover-cache.json \ + --images ../../priv/cache/img # Wipe build artifacts. clean: diff --git a/src/build.gleam b/src/build.gleam index 0939d95..a838726 100644 --- a/src/build.gleam +++ b/src/build.gleam @@ -1,19 +1,21 @@ import api import data/fetch +import data/images import data/model.{type Post, type SiteData, SiteData} import dynamic import encode import filepath import gen/actor/defs.{type ProfileViewDetailed} import gleam/io +import gleam/json import gleam/list -import gleam/option.{type Option, None, Some} +import gleam/option.{type Option, None, Some, map as option_map} import gleam/result import gleam/string import hydration.{HydrationModel} -import lustre/attribute.{class, id} +import lustre/attribute.{class, id, type_} import lustre/element.{type Element, fragment, text, to_document_string} -import lustre/element/html.{div, h2} +import lustre/element/html.{div, h2, script} import simplifile import view/components/post_view import view/layout @@ -30,6 +32,12 @@ pub fn build() { io.println("Fetching data...") let site_data = fetch.fetch_all() + // Mirror the profile's avatar/banner blobs into the site so the + // browser never hits the PDS for them; the returned profile points + // at the local copies and `rewrites` lets the client do the same. + let profile_images = images.mirror_profile_images(site_data.profile) + let site_data = SiteData(..site_data, profile: profile_images.profile) + // Drafts are excluded from the SSG output but the user should // know they exist (so they don't lose work or wonder where // their post went). @@ -41,7 +49,7 @@ pub fn build() { let published_data = SiteData(..site_data, posts: published) - write_index(published_data) + write_index(published_data, profile_images.rewrites) write_posts(published, published_data.profile) write_posts(drafts, published_data.profile) write_style() @@ -50,6 +58,7 @@ pub fn build() { copy_post_assets(published) copy_post_assets(drafts) copy_favicons() + copy_image_cache() copy_client_js() io.println("Done! Site generated in " <> dist_dir) @@ -59,10 +68,10 @@ fn log_draft(post: Post) -> Nil { io.println(" [draft] " <> post.slug <> " — " <> post.title) } -fn write_index(data: SiteData) { +fn write_index(data: SiteData, rewrites: List(#(String, String))) { let og_image: Option(String) = case data.profile.banner { - Some(img) -> Some(img) - None -> data.profile.avatar + Some(img) -> Some(absolutize_img(img)) + None -> option_map(data.profile.avatar, absolutize_img) } let description = case data.profile.description { @@ -87,8 +96,18 @@ fn write_index(data: SiteData) { ), ]) + // The client re-fetches the profile on page load and re-renders it + // with the PDS's remote avatar/banner URLs; this map lets it point + // those at the local mirrors instead. + let rewrites_script = + script( + [type_("application/json"), id("image-rewrites")], + encode_rewrites(rewrites), + ) + let content = fragment([ + rewrites_script, dynamic, div([class("section")], [ div([class("section-header")], [ @@ -103,7 +122,7 @@ fn write_index(data: SiteData) { description: description, image: og_image, url: api.site_url() <> "/", - logo: data.profile.avatar, + logo: option_map(data.profile.avatar, absolutize_img), page_type: layout.Website, ) @@ -123,7 +142,7 @@ fn write_single_post(post: Post, profile: ProfileViewDetailed) { let title = post.title <> " - Kar" let og_image: Option(String) = case post.image { - "" -> profile.avatar + "" -> option_map(profile.avatar, absolutize_img) img -> Some(resolve_og_image_url(post.slug, img)) } @@ -247,6 +266,37 @@ fn copy_client_js() { } } +/// Copy the mirrored cover/artist images from the refresh cache into +/// the site so the browser serves them locally instead of hitting +/// Cover Art Archive / Wikimedia at page load. No-op without a cache. +fn copy_image_cache() { + case simplifile.is_directory("priv/cache/img") { + Ok(True) -> copy_dir("priv/cache/img", dist_dir <> "/img") + _ -> Nil + } +} + +/// The remote→local rewrite map as a JSON object, embedded in +/// `#image-rewrites` for the client (client/browser_ffi.mjs). +fn encode_rewrites(rewrites: List(#(String, String))) -> String { + rewrites + |> list.map(fn(pair) { + let #(remote, local) = pair + #(remote, json.string(local)) + }) + |> json.object + |> json.to_string +} + +/// OG/Twitter image tags must be absolute URLs for crawlers; the +/// mirrored images are root-relative paths. +fn absolutize_img(img: String) -> String { + case string.starts_with(img, "/") { + True -> api.site_url() <> img + False -> img + } +} + fn resolve_og_image_url(slug: String, img: String) -> String { case string.starts_with(img, "http://") || string.starts_with(img, "https://") diff --git a/src/data/images.gleam b/src/data/images.gleam new file mode 100644 index 0000000..598a393 --- /dev/null +++ b/src/data/images.gleam @@ -0,0 +1,82 @@ +//// Build-time mirroring of the profile's avatar/banner blobs. +//// +//// The browser re-fetches the profile on page load (client/refresh.gleam) +//// and would otherwise pull the avatar/banner straight from the PDS on +//// every visit. Instead the SSG downloads them once per build into +//// `dist/img/profile/` and rewrites the profile data to point at the +//// local copies; the remote→local rewrite map is embedded in the page +//// (`#image-rewrites`) so the client can do the same for its fresh +//// render. A failed download keeps the remote URL — the page still +//// works, just with the extra PDS request. + +import data/transport +import gen/actor/defs.{type ProfileViewDetailed, ProfileViewDetailed} +import gleam/list +import gleam/option.{type Option, None, Some} +import gleam/string +import simplifile + +pub type ProfileImages { + ProfileImages(profile: ProfileViewDetailed, rewrites: List(#(String, String))) +} + +const img_dir = "dist/img/profile" + +/// Download the avatar/banner blobs (if any) and return a profile whose +/// image fields point at the local copies, plus the rewrite map. +pub fn mirror_profile_images(profile: ProfileViewDetailed) -> ProfileImages { + let _ = simplifile.create_directory_all(img_dir) + let avatar = mirror_one("avatar", profile.avatar) + let banner = mirror_one("banner", profile.banner) + ProfileImages( + profile: ProfileViewDetailed( + ..profile, + avatar: avatar.local_url, + banner: banner.local_url, + ), + rewrites: list.append(avatar.rewrites, banner.rewrites), + ) +} + +type MirrorResult { + MirrorResult(local_url: Option(String), rewrites: List(#(String, String))) +} + +fn mirror_one(name: String, remote: Option(String)) -> MirrorResult { + case remote { + None -> MirrorResult(local_url: None, rewrites: []) + Some(url) -> { + let fallback = MirrorResult(local_url: remote, rewrites: []) + case transport.fetch_image(url) { + Error(_) -> fallback + Ok(#(bits, content_type)) -> { + let filename = name <> "." <> ext_for_content_type(content_type) + let path = "/img/profile/" <> filename + case + simplifile.write_bits(to: img_dir <> "/" <> filename, bits: bits) + { + Ok(Nil) -> + MirrorResult(local_url: Some(path), rewrites: [#(url, path)]) + Error(_) -> fallback + } + } + } + } + } +} + +fn ext_for_content_type(content_type: String) -> String { + let content_type = string.lowercase(content_type) + ext_for(content_type, [#("png", "png"), #("webp", "webp"), #("gif", "gif")]) +} + +fn ext_for(content_type: String, pairs: List(#(String, String))) -> String { + case pairs { + [] -> "jpg" + [#(needle, ext), ..rest] -> + case string.contains(content_type, needle) { + True -> ext + False -> ext_for(content_type, rest) + } + } +} diff --git a/src/data/transport.gleam b/src/data/transport.gleam index b973118..7bac3df 100644 --- a/src/data/transport.gleam +++ b/src/data/transport.gleam @@ -1,26 +1,113 @@ //// HTTP body fetcher for the SSG (Erlang target). //// -//// `fetch_body` does a GET and returns the response body as a string. -//// Status and error handling stay here so the shared decoders in -//// `shared/src/fetch.gleam` can stay pure. +//// `fetch_body` does a GET and returns the response body as a string; +//// `fetch_image` does a GET and returns raw bytes plus the response's +//// Content-Type (for picking a file extension). Status and error +//// handling stay here so the shared decoders in `shared/src/fetch.gleam` +//// can stay pure. +//// +//// Both retry transient failures (network errors, 429/5xx) a few times +//// with a short fixed delay, so a blip from the PDS or a CDN doesn't +//// blank out a section for this build. import gleam/http/request +import gleam/http/response import gleam/httpc import gleam/int import gleam/result import gleam/string +@external(erlang, "transport_ffi", "sleep") +fn sleep(ms: Int) -> Nil + +/// Retryable vs permanent failure. Transport errors and 429/5xx are +/// transient; 4xx and invalid URLs are permanent. +type HttpError { + Transient(String) + Permanent(String) +} + +const max_attempts = 3 + +const retry_delay_ms = 500 + /// GET `url` and return the response body. Returns an error string on /// network failure, non-2xx status, or invalid URL — the SSG logs /// these and continues with an empty value where appropriate. pub fn fetch_body(url: String) -> Result(String, String) { + retry(fn() { fetch_body_once(url) }) +} + +/// GET `url` and return the raw bytes and Content-Type header. Used to +/// mirror remote images (avatar/banner blobs) into the built site so +/// the visitor's browser never hits the PDS for them. +pub fn fetch_image(url: String) -> Result(#(BitArray, String), String) { + retry(fn() { fetch_image_once(url) }) +} + +fn retry(f: fn() -> Result(a, HttpError)) -> Result(a, String) { + retry_from(f, max_attempts) +} + +fn retry_from( + f: fn() -> Result(a, HttpError), + attempts: Int, +) -> Result(a, String) { + case f() { + Ok(value) -> Ok(value) + Error(Permanent(reason)) -> Error(reason) + Error(Transient(reason)) -> + case attempts <= 1 { + True -> Error(reason) + False -> { + sleep(retry_delay_ms) + retry_from(f, attempts - 1) + } + } + } +} + +fn fetch_body_once(url: String) -> Result(String, HttpError) { use req <- result.try( request.to(url) - |> result.replace_error("invalid url: " <> url), + |> result.replace_error("invalid url: " <> url) + |> result.map_error(fn(e) { Permanent(e) }), + ) + use resp <- result.try( + httpc.send(req) |> result.map_error(fn(e) { Transient(string.inspect(e)) }), ) - use resp <- result.try(httpc.send(req) |> result.map_error(string.inspect)) case resp.status >= 200 && resp.status < 300 { True -> Ok(resp.body) - False -> Error("HTTP " <> int.to_string(resp.status) <> ": " <> resp.body) + False -> { + let reason = "HTTP " <> int.to_string(resp.status) <> ": " <> resp.body + classify_status(resp.status, reason) + } + } +} + +fn fetch_image_once(url: String) -> Result(#(BitArray, String), HttpError) { + use req <- result.try( + request.to(url) + |> result.map(fn(req) { request.set_body(req, <<>>) }) + |> result.replace_error("invalid url: " <> url) + |> result.map_error(fn(e) { Permanent(e) }), + ) + use resp <- result.try( + httpc.send_bits(req) + |> result.map_error(fn(e) { Transient(string.inspect(e)) }), + ) + case resp.status >= 200 && resp.status < 300 { + False -> classify_status(resp.status, "HTTP " <> int.to_string(resp.status)) + True -> { + let content_type = response.get_header(resp, "content-type") + Ok(#(resp.body, result.unwrap(content_type, ""))) + } + } +} + +fn classify_status(status: Int, reason: String) -> Result(a, HttpError) { + case status == 429 || status >= 500 { + True -> Error(Transient(reason)) + False -> Error(Permanent(reason)) } } diff --git a/src/data/transport_ffi.erl b/src/data/transport_ffi.erl new file mode 100644 index 0000000..768fbe0 --- /dev/null +++ b/src/data/transport_ffi.erl @@ -0,0 +1,7 @@ +-module(transport_ffi). +-export([sleep/1]). + +%% Blocking sleep for the SSG's retry backoff. Runs on the build +%% machine, so blocking the process is fine. +sleep(Milliseconds) -> + timer:sleep(Milliseconds). diff --git a/tools/parse-plays/src/covers.rs b/tools/parse-plays/src/covers.rs index d9022a2..7e6e267 100644 --- a/tools/parse-plays/src/covers.rs +++ b/tools/parse-plays/src/covers.rs @@ -1,37 +1,43 @@ //! Cover art resolution: the impure boundary between the pure stats -//! core and MusicBrainz / Cover Art Archive. +//! core and MusicBrainz / Cover Art Archive / Wikidata. //! -//! `resolve` is cache-first: with a cache path, known pairs are served -//! instantly and only missing pairs hit the network — two parallel -//! phases (MusicBrainz searches, then Cover Art Archive fetches) with -//! per-endpoint rate limiting and 503/429 retry. The cache is updated -//! atomically after every successful lookup so interrupted runs keep -//! their work. Without a cache path nothing is fetched at all. - -use crate::stats; +//! All five `resolve_*` phases share one `Cache` (a mutex-guarded map, +//! loaded once and persisted once by the caller) so they can safely +//! run concurrently. Each phase is cache-first: known keys are served +//! instantly and only missing entries hit the network — parallel +//! workers with per-endpoint rate limiting, and every request goes +//! through `net::retry` (jittered backoff, `Retry-After` honor, and +//! retries on transport errors, not just HTTP status codes). + +use crate::{net, stats}; use rayon::prelude::*; use std::collections::HashMap; use std::sync::Mutex; use std::time::{Duration, Instant}; -const USER_AGENT: &str = "karitham-blog/0.1.0 (https://karitham.dev)"; +pub(crate) const USER_AGENT: &str = "karitham-blog/0.1.0 (https://karitham.dev)"; + +/// The shared lookup cache. One entry per resolved key (see the +/// `*_cache_key` helpers); the caller loads it once and persists it +/// once after all phases complete. +pub(crate) type Cache = Mutex>; /// Global rate limiter shared by workers: at most `rate_per_sec` /// acquisitions per second across all threads. -struct Limiter { +pub(crate) struct Limiter { min_interval: Duration, last: Mutex, } impl Limiter { - fn new(rate_per_sec: f64) -> Self { + pub(crate) fn new(rate_per_sec: f64) -> Self { Self { min_interval: Duration::from_secs_f64(1.0 / rate_per_sec), last: Mutex::new(Instant::now()), } } - fn acquire(&self) { + pub(crate) fn acquire(&self) { let mut last = self.last.lock().unwrap(); let since = Instant::now().duration_since(*last); if since < self.min_interval { @@ -41,16 +47,72 @@ impl Limiter { } } -fn is_retryable(code: u16) -> bool { +pub(crate) fn is_retryable(code: u16) -> bool { code == 429 || code == 500 || code == 503 } +static RETRY_POLICY: net::RetryPolicy = net::RetryPolicy { + max_attempts: 5, + base_delay: Duration::from_millis(500), + max_delay: Duration::from_secs(8), +}; + +/// Load the cache from disk; a missing or corrupt file is an empty +/// cache, never a failure. +pub(crate) fn load_cache(path: &str) -> HashMap { + std::fs::read_to_string(path) + .ok() + .and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or_default() +} + +/// Persist the cache atomically (tmp + rename) so a crash mid-write +/// never corrupts the previous state. +pub(crate) fn save_cache_atomically(path: &str, cache: &HashMap) { + let Ok(serialized) = serde_json::to_string_pretty(cache) else { + return; + }; + let tmp = format!("{path}.tmp"); + if std::fs::write(&tmp, serialized).is_ok() { + let _ = std::fs::rename(&tmp, path); + } +} + fn clean_query(s: &str) -> String { - s.replace('"', "").replace('\\', "") + s.replace(['"', '\\'], "") +} + +/// GET a JSON API with the given query params, retried through +/// `net::retry`. A 200 that fails to parse is a permanent miss — the +/// endpoint answered, the shape just isn't what we expected. +fn get_json(agent: &ureq::Agent, url: &str, params: &[(&str, &str)]) -> Option { + net::retry(&RETRY_POLICY, || { + let mut req = agent.get(url); + for (key, value) in params { + req = req.query(key, value); + } + match req.call() { + Ok(resp) if resp.status() == 200 => match resp.into_string() { + Ok(body) => match serde_json::from_str(&body) { + Ok(value) => net::Attempt::Done(value), + Err(_) => net::Attempt::Stop, + }, + Err(_) => net::Attempt::Stop, + }, + Ok(_) => net::Attempt::Stop, + Err(ureq::Error::Status(code, resp)) if is_retryable(code) => { + net::Attempt::Again(net::retry_after(&resp)) + } + // Transport-level failures (timeouts, resets, DNS) are + // transient — retry rather than dropping the link for this + // whole build. + Err(_) => net::Attempt::Again(None), + } + }) } /// MusicBrainz release search → best-guess release MBID. Retries on -/// rate-limit/5xx with exponential backoff, which lets us run a bit +/// rate-limit/5xx with jittered backoff, which lets us run a bit /// hotter than the nominal 1 req/s and have MB push back politely. fn musicbrainz_release(agent: &ureq::Agent, artist: &str, album: &str) -> Option { let query = format!( @@ -58,49 +120,33 @@ fn musicbrainz_release(agent: &ureq::Agent, artist: &str, album: &str) -> Option clean_query(album), clean_query(artist) ); - for attempt in 0..3u32 { - match agent - .get("https://musicbrainz.org/ws/2/release") - .query("query", &query) - .query("fmt", "json") - .query("limit", "1") - .call() - { - Ok(resp) => { - let body: serde_json::Value = - serde_json::from_str(&resp.into_string().ok()?).ok()?; - return body["releases"] - .as_array()? - .first()? - .get("id")? - .as_str() - .map(str::to_string); - } - Err(ureq::Error::Status(code, _)) if is_retryable(code) => { - std::thread::sleep(Duration::from_secs(1 << attempt)); - } - Err(_) => return None, - } - } - None + get_json( + agent, + "https://musicbrainz.org/ws/2/release", + &[("query", &query), ("fmt", "json"), ("limit", "1")], + ) + .and_then(|body| { + body["releases"] + .as_array()? + .first()? + .get("id")? + .as_str() + .map(str::to_string) + }) } -/// Cover Art Archive front image for a release, 500px. Same retry -/// policy as the MusicBrainz search; a non-2xx response (e.g. 404 — -/// no front art) is a permanent miss. +/// Cover Art Archive front image for a release, 500px. A non-2xx +/// response (e.g. 404 — no front art) is a permanent miss. fn coverart_url(agent: &ureq::Agent, mbid: &str) -> Option { let url = format!("https://coverartarchive.org/release/{mbid}/front-500"); - for attempt in 0..3u32 { - match agent.get(&url).call() { - Ok(resp) if resp.status() == 200 => return Some(url), - Ok(_) => return None, - Err(ureq::Error::Status(code, _)) if is_retryable(code) => { - std::thread::sleep(Duration::from_secs(1 << attempt)); - } - Err(_) => return None, + net::retry(&RETRY_POLICY, || match agent.get(&url).call() { + Ok(resp) if resp.status() == 200 => net::Attempt::Done(url.clone()), + Ok(_) => net::Attempt::Stop, + Err(ureq::Error::Status(code, resp)) if is_retryable(code) => { + net::Attempt::Again(net::retry_after(&resp)) } - } - None + Err(_) => net::Attempt::Again(None), + }) } fn cache_key(artist: &str, album: &str) -> String { @@ -142,55 +188,34 @@ fn track_url_cache_key(artist: &str, track: &str) -> String { ) } -fn load_cache(path: &str) -> HashMap { - std::fs::read_to_string(path) - .ok() - .and_then(|s| serde_json::from_str(&s).ok()) - .unwrap_or_default() -} - -fn save_cache_atomically(path: &str, cache: &HashMap) { - let Ok(serialized) = serde_json::to_string_pretty(cache) else { - return; - }; - let tmp = format!("{path}.tmp"); - if std::fs::write(&tmp, serialized).is_ok() { - let _ = std::fs::rename(&tmp, path); - } -} - /// Resolve cover URLs for the given (artist, album, optional MBID) -/// triples. See the module docs for the cache-first, parallel, -/// retried behavior. Known MBIDs skip the MusicBrainz search phase, -/// but when a provided MBID has no Cover Art Archive image we fall -/// back to a name search — piper/lazuli MBIDs sometimes point at -/// release variants without art while the canonical release has it. +/// triples. Known MBIDs skip the MusicBrainz search phase, but when a +/// provided MBID has no Cover Art Archive image we fall back to a name +/// search — piper/lazuli MBIDs sometimes point at release variants +/// without art while the canonical release has it. pub fn resolve( pairs: &[(String, String, Option)], - cache_path: Option<&str>, + cache: &Cache, ) -> HashMap<(String, String), String> { - let Some(path) = cache_path else { - return HashMap::new(); - }; - - let mut cache = load_cache(path); - let mut out: HashMap<(String, String), String> = HashMap::new(); - let missing: Vec<(String, String, Option)> = pairs - .iter() - .filter_map( - |(artist, album, mbid)| match cache.get(&cache_key(artist, album)) { - Some(url) => { - out.insert( - (stats::normalize(artist), stats::normalize(album)), - url.clone(), - ); - None - } - None => Some((artist.clone(), album.clone(), mbid.clone())), - }, - ) - .collect(); + let missing: Vec<(String, String, Option)> = { + let cache = cache.lock().unwrap(); + pairs + .iter() + .filter_map( + |(artist, album, mbid)| match cache.get(&cache_key(artist, album)) { + Some(url) => { + out.insert( + (stats::normalize(artist), stats::normalize(album)), + url.clone(), + ); + None + } + None => Some((artist.clone(), album.clone(), mbid.clone())), + }, + ) + .collect() + }; if missing.is_empty() { return out; @@ -244,16 +269,18 @@ pub fn resolve( }) .collect(); - for (artist, album, _, url) in &caa_outcomes { - let Some(url) = url else { continue }; - cache.insert(cache_key(artist, album), url.clone()); - // Keys are normalized so the pure `cover` lookup in main - // (which normalizes both sides) finds them regardless of case. - out.insert( - (stats::normalize(artist), stats::normalize(album)), - url.clone(), - ); - save_cache_atomically(path, &cache); + { + let mut cache = cache.lock().unwrap(); + for (artist, album, _, url) in &caa_outcomes { + let Some(url) = url else { continue }; + cache.insert(cache_key(artist, album), url.clone()); + // Keys are normalized so the pure `cover` lookup in main + // (which normalizes both sides) finds them regardless of case. + out.insert( + (stats::normalize(artist), stats::normalize(album)), + url.clone(), + ); + } } // Fallback: provided MBIDs with no art get a name search — the @@ -282,10 +309,10 @@ pub fn resolve( }) .collect(); + let mut cache = cache.lock().unwrap(); for (artist, album, url) in caa2 { cache.insert(cache_key(&artist, &album), url.clone()); out.insert((stats::normalize(&artist), stats::normalize(&album)), url); - save_cache_atomically(path, &cache); } } @@ -295,24 +322,21 @@ pub fn resolve( /// Resolve real artist images via the Wikidata chain (MusicBrainz /// artist search → url-rels → P18 → Commons). Cache-first, same /// append-only pattern. -pub fn resolve_artists(artists: &[String], cache_path: Option<&str>) -> HashMap { - let Some(path) = cache_path else { - return HashMap::new(); - }; - - let mut cache = load_cache(path); - +pub fn resolve_artists(artists: &[String], cache: &Cache) -> HashMap { let mut out: HashMap = HashMap::new(); - let missing: Vec = artists - .iter() - .filter_map(|name| match cache.get(&artist_cache_key(name)) { - Some(url) => { - out.insert(stats::normalize(name), url.clone()); - None - } - None => Some(name.clone()), - }) - .collect(); + let missing: Vec = { + let cache = cache.lock().unwrap(); + artists + .iter() + .filter_map(|name| match cache.get(&artist_cache_key(name)) { + Some(url) => { + out.insert(stats::normalize(name), url.clone()); + None + } + None => Some(name.clone()), + }) + .collect() + }; if missing.is_empty() { return out; @@ -344,10 +368,10 @@ pub fn resolve_artists(artists: &[String], cache_path: Option<&str>) -> HashMap< }) .collect(); + let mut cache = cache.lock().unwrap(); for (name, url) in results { cache.insert(artist_cache_key(&name), url.clone()); out.insert(stats::normalize(&name), url); - save_cache_atomically(path, &cache); } out @@ -356,92 +380,58 @@ pub fn resolve_artists(artists: &[String], cache_path: Option<&str>) -> HashMap< /// MusicBrainz artist search → best-guess artist MBID. fn musicbrainz_artist_mbid(agent: &ureq::Agent, name: &str) -> Option { let query = format!("artist:\"{}\"", clean_query(name)); - for attempt in 0..3u32 { - match agent - .get("https://musicbrainz.org/ws/2/artist") - .query("query", &query) - .query("fmt", "json") - .query("limit", "1") - .call() - { - Ok(resp) => { - let body: serde_json::Value = - serde_json::from_str(&resp.into_string().ok()?).ok()?; - return body["artists"] - .as_array()? - .first()? - .get("id")? - .as_str() - .map(str::to_string); - } - Err(ureq::Error::Status(code, _)) if is_retryable(code) => { - std::thread::sleep(Duration::from_secs(1 << attempt)); - } - Err(_) => return None, - } - } - None + get_json( + agent, + "https://musicbrainz.org/ws/2/artist", + &[("query", &query), ("fmt", "json"), ("limit", "1")], + ) + .and_then(|body| { + body["artists"] + .as_array()? + .first()? + .get("id")? + .as_str() + .map(str::to_string) + }) } /// MB artist lookup with URL relations → the artist's Wikidata QID. fn wikidata_qid(agent: &ureq::Agent, mbid: &str) -> Option { - for attempt in 0..3u32 { - let resp = match agent - .get(&format!("https://musicbrainz.org/ws/2/artist/{mbid}")) - .query("inc", "url-rels") - .query("fmt", "json") - .call() - { - Ok(resp) => resp, - Err(ureq::Error::Status(code, _)) if is_retryable(code) => { - std::thread::sleep(Duration::from_secs(1 << attempt)); - continue; - } - Err(_) => return None, - }; - let body: serde_json::Value = serde_json::from_str(&resp.into_string().ok()?).ok()?; - for relation in body["relations"].as_array()? { - if relation["type"].as_str() == Some("wikidata") { - let url = relation["url"]["resource"].as_str()?; - // https://www.wikidata.org/wiki/Q130798 - return url.rsplit('/').next().map(str::to_string); - } + let body = get_json( + agent, + &format!("https://musicbrainz.org/ws/2/artist/{mbid}"), + &[("inc", "url-rels"), ("fmt", "json")], + )?; + for relation in body["relations"].as_array()? { + if relation["type"].as_str() == Some("wikidata") { + let url = relation["url"]["resource"].as_str()?; + // https://www.wikidata.org/wiki/Q130798 + return url.rsplit('/').next().map(str::to_string); } - return None; } None } /// Wikidata `P18` (image) claim → Commons filename. fn wikidata_p18(agent: &ureq::Agent, qid: &str) -> Option { - for attempt in 0..3u32 { - match agent - .get("https://www.wikidata.org/w/api.php") - .query("action", "wbgetclaims") - .query("property", "P18") - .query("entity", qid) - .query("format", "json") - .call() - { - Ok(resp) => { - let body: serde_json::Value = - serde_json::from_str(&resp.into_string().ok()?).ok()?; - return body["claims"]["P18"] - .as_array()? - .first()? - .get("mainsnak")? - .get("datavalue")? - .get("value")? - .as_str() - .map(str::to_string); - } - Err(ureq::Error::Status(code, _)) if is_retryable(code) => { - std::thread::sleep(Duration::from_secs(1 << attempt)); - } - Err(_) => return None, - } - } - None + let body = get_json( + agent, + "https://www.wikidata.org/w/api.php", + &[ + ("action", "wbgetclaims"), + ("property", "P18"), + ("entity", qid), + ("format", "json"), + ], + )?; + body["claims"]["P18"] + .as_array()? + .first()? + .get("mainsnak")? + .get("datavalue")? + .get("value")? + .as_str() + .map(str::to_string) } /// Commons image URL for a filename, resized to 600px. @@ -473,31 +463,19 @@ fn musicbrainz_recording(agent: &ureq::Agent, artist: &str, track: &str) -> Opti clean_query(track), clean_query(artist) ); - for attempt in 0..3u32 { - match agent - .get("https://musicbrainz.org/ws/2/recording") - .query("query", &query) - .query("fmt", "json") - .query("limit", "1") - .call() - { - Ok(resp) => { - let body: serde_json::Value = - serde_json::from_str(&resp.into_string().ok()?).ok()?; - return body["recordings"] - .as_array()? - .first()? - .get("id")? - .as_str() - .map(str::to_string); - } - Err(ureq::Error::Status(code, _)) if is_retryable(code) => { - std::thread::sleep(Duration::from_secs(1 << attempt)); - } - Err(_) => return None, - } - } - None + get_json( + agent, + "https://musicbrainz.org/ws/2/recording", + &[("query", &query), ("fmt", "json"), ("limit", "1")], + ) + .and_then(|body| { + body["recordings"] + .as_array()? + .first()? + .get("id")? + .as_str() + .map(str::to_string) + }) } /// Resolve MusicBrainz release pages for the top albums. Cache-first, @@ -506,30 +484,27 @@ fn musicbrainz_recording(agent: &ureq::Agent, artist: &str, track: &str) -> Opti /// (artist, album) pairs. pub fn resolve_album_urls( pairs: &[(String, String, Option)], - cache_path: Option<&str>, + cache: &Cache, ) -> HashMap<(String, String), String> { - let Some(path) = cache_path else { - return HashMap::new(); - }; - - let mut cache = load_cache(path); - let mut out: HashMap<(String, String), String> = HashMap::new(); - let missing: Vec<(String, String, Option)> = pairs - .iter() - .filter_map( - |(artist, album, mbid)| match cache.get(&album_url_cache_key(artist, album)) { - Some(url) => { - out.insert( - (stats::normalize(artist), stats::normalize(album)), - url.clone(), - ); - None + let missing: Vec<(String, String, Option)> = { + let cache = cache.lock().unwrap(); + pairs + .iter() + .filter_map(|(artist, album, mbid)| { + match cache.get(&album_url_cache_key(artist, album)) { + Some(url) => { + out.insert( + (stats::normalize(artist), stats::normalize(album)), + url.clone(), + ); + None + } + None => Some((artist.clone(), album.clone(), mbid.clone())), } - None => Some((artist.clone(), album.clone(), mbid.clone())), - }, - ) - .collect(); + }) + .collect() + }; if missing.is_empty() { return out; @@ -557,10 +532,10 @@ pub fn resolve_album_urls( }) .collect(); + let mut cache = cache.lock().unwrap(); for (artist, album, url) in results { cache.insert(album_url_cache_key(&artist, &album), url.clone()); out.insert((stats::normalize(&artist), stats::normalize(&album)), url); - save_cache_atomically(path, &cache); } out @@ -569,27 +544,21 @@ pub fn resolve_album_urls( /// Resolve MusicBrainz artist pages for the top artists. Cache-first; /// a miss does the same artist search as the image pipeline. Keys are /// normalized artist names. -pub fn resolve_artist_urls( - artists: &[String], - cache_path: Option<&str>, -) -> HashMap { - let Some(path) = cache_path else { - return HashMap::new(); - }; - - let mut cache = load_cache(path); - +pub fn resolve_artist_urls(artists: &[String], cache: &Cache) -> HashMap { let mut out: HashMap = HashMap::new(); - let missing: Vec = artists - .iter() - .filter_map(|name| match cache.get(&artist_url_cache_key(name)) { - Some(url) => { - out.insert(stats::normalize(name), url.clone()); - None - } - None => Some(name.clone()), - }) - .collect(); + let missing: Vec = { + let cache = cache.lock().unwrap(); + artists + .iter() + .filter_map(|name| match cache.get(&artist_url_cache_key(name)) { + Some(url) => { + out.insert(stats::normalize(name), url.clone()); + None + } + None => Some(name.clone()), + }) + .collect() + }; if missing.is_empty() { return out; @@ -614,10 +583,10 @@ pub fn resolve_artist_urls( }) .collect(); + let mut cache = cache.lock().unwrap(); for (name, url) in results { cache.insert(artist_url_cache_key(&name), url.clone()); out.insert(stats::normalize(&name), url); - save_cache_atomically(path, &cache); } out @@ -628,30 +597,27 @@ pub fn resolve_artist_urls( /// normalized (artist, track) pairs. pub fn resolve_track_urls( tracks: &[(String, String)], - cache_path: Option<&str>, + cache: &Cache, ) -> HashMap<(String, String), String> { - let Some(path) = cache_path else { - return HashMap::new(); - }; - - let mut cache = load_cache(path); - let mut out: HashMap<(String, String), String> = HashMap::new(); - let missing: Vec<(String, String)> = tracks - .iter() - .filter_map( - |(artist, track)| match cache.get(&track_url_cache_key(artist, track)) { - Some(url) => { - out.insert( - (stats::normalize(artist), stats::normalize(track)), - url.clone(), - ); - None - } - None => Some((artist.clone(), track.clone())), - }, - ) - .collect(); + let missing: Vec<(String, String)> = { + let cache = cache.lock().unwrap(); + tracks + .iter() + .filter_map( + |(artist, track)| match cache.get(&track_url_cache_key(artist, track)) { + Some(url) => { + out.insert( + (stats::normalize(artist), stats::normalize(track)), + url.clone(), + ); + None + } + None => Some((artist.clone(), track.clone())), + }, + ) + .collect() + }; if missing.is_empty() { return out; @@ -677,10 +643,10 @@ pub fn resolve_track_urls( }) .collect(); + let mut cache = cache.lock().unwrap(); for (artist, track, url) in results { cache.insert(track_url_cache_key(&artist, &track), url.clone()); out.insert((stats::normalize(&artist), stats::normalize(&track)), url); - save_cache_atomically(path, &cache); } out diff --git a/tools/parse-plays/src/images.rs b/tools/parse-plays/src/images.rs new file mode 100644 index 0000000..800d4ec --- /dev/null +++ b/tools/parse-plays/src/images.rs @@ -0,0 +1,276 @@ +//! Build-time image mirroring: download every remote image referenced +//! by the stats JSON into a local cache directory and rewrite the +//! `image` fields to point at the local copies. +//! +//! The visitor's browser should never have to hit Cover Art Archive, +//! Wikimedia Commons (the Wikidata chain), or similar third-party +//! hosts at page load — those are slow and every page view burns +//! their bandwidth. Everything the Music section shows is mirrored +//! here once per `just refresh`, then served from `/img/...` beside +//! the rest of the site. Downloads that fail keep their remote URL in +//! the JSON so the tile still renders. + +use crate::{ + covers::{Limiter, USER_AGENT, is_retryable}, + net, +}; +use rayon::prelude::*; +use std::collections::{HashMap, HashSet}; +use std::io::Read; +use std::path::Path; +use std::time::Duration; + +/// Generous cap — covers and artist photos are a few hundred KB. +const MAX_IMAGE_BYTES: u64 = 8 * 1024 * 1024; +const RATE_PER_SEC: f64 = 8.0; + +static RETRY_POLICY: net::RetryPolicy = net::RetryPolicy { + max_attempts: 5, + base_delay: Duration::from_millis(500), + max_delay: Duration::from_secs(8), +}; + +/// 16 hex chars of fnv-1a 64. Collision-proof enough for a few +/// hundred URLs and stable across runs, so the cache is append-only. +pub fn short_hash(s: &str) -> String { + let mut hash: u64 = 0xcbf29ce484222325; + for b in s.as_bytes() { + hash ^= *b as u64; + hash = hash.wrapping_mul(0x100000001b3); + } + format!("{hash:016x}") +} + +fn ext_for_content_type(ct: &str) -> &'static str { + let ct = ct.to_ascii_lowercase(); + if ct.contains("png") { + "png" + } else if ct.contains("webp") { + "webp" + } else if ct.contains("gif") { + "gif" + } else { + "jpg" + } +} + +/// Snapshot the cache dir: hash → local URL path for every complete +/// `.` file. Leftover `.tmp` files from a killed run are +/// swept and never treated as cache entries. +fn scan_cache(dir: &Path) -> (HashMap, Vec) { + let mut by_hash = HashMap::new(); + let mut stale = Vec::new(); + if let Ok(entries) = std::fs::read_dir(dir) { + for entry in entries.flatten() { + let name = entry.file_name().to_string_lossy().into_owned(); + if name.ends_with(".tmp") { + stale.push(entry.path()); + continue; + } + if let Some((hash, _)) = name.split_once('.') + && hash.len() == 16 + { + by_hash.insert(hash.to_string(), format!("/img/{name}")); + } + } + } + (by_hash, stale) +} + +/// Download `url` into `dir`, returning the local URL path +/// (`/img/.`), or `None` on failure. A non-2xx response +/// (404 — no image) is a permanent miss, not worth retrying. +fn fetch_image(agent: &ureq::Agent, dir: &Path, url: &str) -> Option { + let hash = short_hash(url); + net::retry(&RETRY_POLICY, || { + match agent.get(url).call() { + Ok(resp) if resp.status() == 200 => { + let ct = resp + .header("content-type") + .map(str::to_string) + .unwrap_or_default(); + let mut buf = Vec::new(); + let read = resp + .into_reader() + .take(MAX_IMAGE_BYTES + 1) + .read_to_end(&mut buf); + // Empty or oversized: don't write, and don't let a + // truncated copy become the cached artifact. + match read { + Ok(n) if !buf.is_empty() && n as u64 <= MAX_IMAGE_BYTES => { + let name = format!("{hash}.{}", ext_for_content_type(&ct)); + // Write to a temp file and rename so a crash + // mid-write never leaves a truncated file that + // later builds mistake for a complete cache entry. + let tmp = dir.join(format!("{name}.tmp")); + match std::fs::write(&tmp, &buf) + .and_then(|_| std::fs::rename(&tmp, dir.join(&name))) + { + Ok(_) => net::Attempt::Done(format!("/img/{name}")), + Err(_) => net::Attempt::Stop, + } + } + _ => net::Attempt::Stop, + } + } + Ok(_) => net::Attempt::Stop, + Err(ureq::Error::Status(code, resp)) if is_retryable(code) => { + net::Attempt::Again(net::retry_after(&resp)) + } + // Transport-level failures (timeouts, resets) are transient. + Err(_) => net::Attempt::Again(None), + } + }) +} + +/// Mirror every unique remote `url` into `dir` (cache-first) and +/// return the remote→local path map. Downloads run in parallel under +/// a shared rate limiter; a failed download is simply absent from the +/// map so the caller can keep the remote URL. +pub fn download_many(urls: &[String], dir: &Path) -> HashMap { + let _ = std::fs::create_dir_all(dir); + let (by_hash, stale) = scan_cache(dir); + for stale_path in stale { + let _ = std::fs::remove_file(stale_path); + } + + let unique: HashSet<&String> = urls.iter().collect(); + let agent = ureq::AgentBuilder::new() + .user_agent(USER_AGENT) + .timeout(Duration::from_secs(30)) + .build(); + let limiter = &Limiter::new(RATE_PER_SEC); + + let urls: Vec<&String> = unique.into_iter().collect(); + // Parallel downloads under a shared limiter: with server latency + // far above the limiter interval, workers overlap the waits. + let results: Vec<(String, Option)> = urls + .par_iter() + .map(|url| { + let url = *url; + let hash = short_hash(url); + match by_hash.get(&hash) { + Some(local) => (url.clone(), Some(local.clone())), + None => { + limiter.acquire(); + (url.clone(), fetch_image(&agent, dir, url)) + } + } + }) + .collect(); + + let mut rewrites = HashMap::new(); + for (url, local) in results { + if let Some(local) = local { + rewrites.insert(url, local); + } + } + rewrites +} + +/// Point every `image` field at its local mirror, when one exists. +pub fn apply_rewrites(value: &mut serde_json::Value, rewrites: &HashMap) { + match value { + serde_json::Value::Array(items) => { + for item in items { + apply_rewrites(item, rewrites); + } + } + serde_json::Value::Object(map) => { + if let Some(serde_json::Value::String(url)) = map.get("image") + && let Some(local) = rewrites.get(url) + { + map.insert("image".into(), serde_json::Value::String(local.clone())); + } + for (_, v) in map { + apply_rewrites(v, rewrites); + } + } + _ => {} + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn short_hash_is_stable() { + let a = short_hash("https://example.com/a.png"); + let b = short_hash("https://example.com/a.png"); + let c = short_hash("https://example.com/b.png"); + assert_eq!(a.len(), 16); + assert_eq!(a, b); + assert_ne!(a, c); + } + + #[test] + fn ext_maps_content_types() { + assert_eq!(ext_for_content_type("image/jpeg"), "jpg"); + assert_eq!(ext_for_content_type("image/png"), "png"); + assert_eq!(ext_for_content_type("image/webp"), "webp"); + assert_eq!(ext_for_content_type("IMAGE/GIF"), "gif"); + assert_eq!(ext_for_content_type("application/octet-stream"), "jpg"); + } + + #[test] + fn scan_cache_indexes_complete_files_only() { + let dir = std::env::temp_dir().join(format!("parse-plays-scan-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let hash = short_hash("https://example.com/cover"); + std::fs::write(dir.join(format!("{hash}.png")), b"x").unwrap(); + std::fs::write(dir.join(format!("{hash}.jpg.tmp")), b"partial").unwrap(); + std::fs::write(dir.join("other.txt"), b"junk").unwrap(); + + let (by_hash, stale) = scan_cache(&dir); + assert_eq!(by_hash.len(), 1); + assert_eq!(by_hash.get(&hash).unwrap(), &format!("/img/{hash}.png")); + assert_eq!(stale.len(), 1); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn download_many_hits_cache_without_network() { + let dir = std::env::temp_dir().join(format!("parse-plays-dl-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let url = "https://example.com/cover".to_string(); + let hash = short_hash(&url); + std::fs::write(dir.join(format!("{hash}.jpg")), b"x").unwrap(); + + let map = download_many(&[url.clone()], &dir); + assert_eq!(map.get(&url).unwrap(), &format!("/img/{hash}.jpg")); + std::fs::remove_dir_all(&dir).ok(); + } + + #[test] + fn applies_rewrites_to_image_fields() { + let mut doc = json!({ + "ranges": { + "1m": { + "artists": [{ "name": "A", "image": "https://a.example/1.jpg" }], + "albums": [ + { "name": "B", "image": "https://b.example/2.png" }, + { "name": "C", "image": "/img/abc.jpg" }, + ], + } + } + }); + let rewrites = HashMap::from([( + "https://a.example/1.jpg".to_string(), + "/img/1.jpg".to_string(), + )]); + apply_rewrites(&mut doc, &rewrites); + let image = |name: &str, i: usize| { + doc["ranges"]["1m"][name][i]["image"] + .as_str() + .unwrap() + .to_string() + }; + assert_eq!(image("artists", 0), "/img/1.jpg"); + // Unmapped remote URL stays as-is so the tile still renders. + assert_eq!(image("albums", 0), "https://b.example/2.png"); + // Already-local reference untouched. + assert_eq!(image("albums", 1), "/img/abc.jpg"); + } +} diff --git a/tools/parse-plays/src/main.rs b/tools/parse-plays/src/main.rs index f1f213f..0a1e9f9 100644 --- a/tools/parse-plays/src/main.rs +++ b/tools/parse-plays/src/main.rs @@ -19,9 +19,12 @@ mod car; mod covers; +mod images; +mod net; mod stats; use serde_json::json; +use std::collections::HashMap; #[tokio::main] async fn main() { @@ -33,8 +36,12 @@ async fn main() { _ => { eprintln!("usage:"); eprintln!(" parse-plays car — dump plays JSON to stdout"); - eprintln!(" parse-plays stats [--covers ]"); - eprintln!(" parse-plays refresh [--covers ]"); + eprintln!( + " parse-plays stats [--covers ] [--images ]" + ); + eprintln!( + " parse-plays refresh [--covers ] [--images ]" + ); std::process::exit(2); } }; @@ -44,6 +51,28 @@ async fn main() { } } +/// Parse the `--covers ` / `--images ` options that trail +/// the positional arguments. Unknown tokens are ignored. +fn parse_options(args: &[String]) -> (Option, Option) { + let mut covers = None; + let mut images = None; + let mut i = 0; + while i < args.len() { + match args[i].as_str() { + "--covers" => { + covers = args.get(i + 1).cloned(); + i += 2; + } + "--images" => { + images = args.get(i + 1).cloned(); + i += 2; + } + _ => i += 1, + } + } + (covers, images) +} + async fn cmd_car(args: &[String]) -> Result<(), String> { let car_path = args.first().ok_or("missing ")?; let plays = car::read_plays(car_path).await?; @@ -54,10 +83,7 @@ async fn cmd_car(args: &[String]) -> Result<(), String> { fn cmd_stats(args: &[String]) -> Result<(), String> { let plays_path = args.first().ok_or("missing ")?; let stats_out = args.get(1).ok_or("missing ")?; - let cache_path = match args.get(2).map(String::as_str) { - Some("--covers") => args.get(3).map(String::as_str), - _ => None, - }; + let (cache_path, images_dir) = parse_options(&args[2..]); // Gather: raw history from the previous pipeline step. `-` reads // stdin so `parse-plays car ... | parse-plays stats - ...` avoids @@ -79,16 +105,18 @@ fn cmd_stats(args: &[String]) -> Result<(), String> { let agg = stats::aggregate(&plays, chrono::Utc::now().date_naive()); // Commit (impure): resolve covers through the cache, then serialize. - write_stats(&agg, cache_path, stats_out) + write_stats( + &agg, + cache_path.as_deref(), + stats_out, + images_dir.as_deref(), + ) } async fn cmd_refresh(args: &[String]) -> Result<(), String> { let car_path = args.first().ok_or("missing ")?; let stats_out = args.get(1).ok_or("missing ")?; - let cache_path = match args.get(2).map(String::as_str) { - Some("--covers") => args.get(3).map(String::as_str), - _ => None, - }; + let (cache_path, images_dir) = parse_options(&args[2..]); // Gather: raw history straight from the CAR, no JSON round-trip. let plays = car::read_plays(car_path).await?; @@ -97,16 +125,23 @@ async fn cmd_refresh(args: &[String]) -> Result<(), String> { let agg = stats::aggregate(&plays, chrono::Utc::now().date_naive()); // Commit (impure). - write_stats(&agg, cache_path, stats_out) + write_stats( + &agg, + cache_path.as_deref(), + stats_out, + images_dir.as_deref(), + ) } /// Shared tail of `stats` and `refresh`: resolve covers, artist -/// images, and MusicBrainz page links for the top-N entries, serialize -/// the grids, write the stats file. +/// images, and MusicBrainz page links for the top-N entries, mirror +/// the remote images into a local dir, serialize the grids, write the +/// stats file. fn write_stats( agg: &stats::Aggregated, cache_path: Option<&str>, stats_out: &str, + images_dir: Option<&str>, ) -> Result<(), String> { let pairs = stats::needed_pairs(agg); let artists = stats::needed_artists(agg); @@ -118,11 +153,52 @@ fn write_stats( artists.len(), pairs.len() + artists.len() + tracks.len() ); - let cover_map = covers::resolve(&pairs, cache_path); - let artist_map = covers::resolve_artists(&artists, cache_path); - let album_url_map = covers::resolve_album_urls(&pairs, cache_path); - let artist_url_map = covers::resolve_artist_urls(&artists, cache_path); - let track_url_map = covers::resolve_track_urls(&tracks, cache_path); + + // Shared lookup cache: loaded once, written by every phase under + // the mutex, persisted once after all phases. Without a cache path + // we still share an (empty) cache — behavior is unchanged. + let cache = std::sync::Mutex::new(cache_path.map(covers::load_cache).unwrap_or_default()); + + // Stage 1: the image sources — cover art and artist photos. These + // two are the slow ones (MusicBrainz + Wikidata chains), so they + // run first and in parallel. + let (cover_map, artist_map) = rayon::join( + || covers::resolve(&pairs, &cache), + || covers::resolve_artists(&artists, &cache), + ); + + // Stage 2: mirror the images in parallel with resolving the + // MusicBrainz page links — the downloads only need the two image + // maps from stage 1, so the URL lookups overlap them instead of + // running serially before them. + let image_urls: Vec = cover_map + .values() + .chain(artist_map.values()) + .cloned() + .collect(); + let mut download_map: HashMap = HashMap::new(); + let mut album_url_map = HashMap::new(); + let mut artist_url_map = HashMap::new(); + let mut track_url_map = HashMap::new(); + rayon::scope(|s| { + if let Some(dir) = images_dir { + s.spawn(|_| { + download_map = images::download_many(&image_urls, std::path::Path::new(dir)); + }); + } + s.spawn(|_| album_url_map = covers::resolve_album_urls(&pairs, &cache)); + s.spawn(|_| artist_url_map = covers::resolve_artist_urls(&artists, &cache)); + s.spawn(|_| track_url_map = covers::resolve_track_urls(&tracks, &cache)); + }); + if let Some(dir) = images_dir { + eprintln!("mirrored images into {dir}"); + } + + // Persist the merged cache once, after every phase has finished. + if let Some(path) = cache_path { + covers::save_cache_atomically(path, &cache.into_inner().unwrap()); + } + let cover = |artist: &str, album: &str| -> String { cover_map .get(&(stats::normalize(artist), stats::normalize(album))) @@ -162,7 +238,10 @@ fn write_stats( }; let stats_json = stats::build_ranges(agg, &lookups); - let doc = json!({ "ranges": stats_json }); + let mut doc = json!({ "ranges": stats_json }); + // Point the image fields at the local mirrors; any URL missing + // from the map (failed download) keeps its remote value. + images::apply_rewrites(&mut doc, &download_map); std::fs::write( stats_out, serde_json::to_string_pretty(&doc).expect("serialize"), diff --git a/tools/parse-plays/src/net.rs b/tools/parse-plays/src/net.rs new file mode 100644 index 0000000..019c90d --- /dev/null +++ b/tools/parse-plays/src/net.rs @@ -0,0 +1,150 @@ +//! Shared retry/backoff for external HTTP lookups. +//! +//! Every endpoint we touch (MusicBrainz, Cover Art Archive, Wikidata, +//! Wikimedia Commons) is rate-limited and occasionally flakes. Retrying +//! with jittered exponential backoff keeps us polite while tolerating +//! transient failures: the same error that used to permanently drop a +//! link from this build is now retried, and the jitter stops a herd of +//! parallel workers from hammering the endpoint at identical instants. + +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +/// How hard to try, and how long to wait between attempts. +#[derive(Clone, Copy)] +pub struct RetryPolicy { + pub max_attempts: u32, + pub base_delay: Duration, + pub max_delay: Duration, +} + +impl Default for RetryPolicy { + fn default() -> Self { + Self { + max_attempts: 5, + base_delay: Duration::from_millis(500), + max_delay: Duration::from_secs(8), + } + } +} + +/// One outcome of a single attempt. +pub enum Attempt { + /// Got a usable answer. + Done(T), + /// Permanent failure — retrying cannot help (4xx, missing data). + Stop, + /// Transient failure; `Some(n)` = server's `Retry-After` seconds. + Again(Option), +} + +/// Run `attempt` up to `policy.max_attempts` times with jittered +/// exponential backoff. `Stop` aborts immediately; `Again(None)` waits +/// the backoff, `Again(Some(n))` honors the server's `Retry-After` +/// (capped at `max_delay`). Returns `None` when attempts are exhausted. +pub fn retry(policy: &RetryPolicy, mut attempt: impl FnMut() -> Attempt) -> Option { + let mut delay = policy.base_delay; + for i in 0..policy.max_attempts { + match attempt() { + Attempt::Done(value) => return Some(value), + Attempt::Stop => return None, + Attempt::Again(retry_after) => { + if i + 1 == policy.max_attempts { + return None; + } + let wait = retry_after + .map(Duration::from_secs) + .unwrap_or(delay) + .min(policy.max_delay); + std::thread::sleep(wait + jitter(wait)); + } + } + delay = (delay * 2).min(policy.max_delay); + } + None +} + +/// Add 0–25% of `base` as jitter so parallel workers don't retry in +/// lockstep. Seeded from the clock; good enough to decorrelate ~dozens +/// of retries without pulling in a RNG dependency. +fn jitter(base: Duration) -> Duration { + let ms = base.as_millis() as u64; + let salt = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.subsec_nanos() as u64) + .unwrap_or(0); + Duration::from_millis(ms + salt % (ms / 4 + 1)) +} + +/// `Retry-After` header value as seconds, when it is a plain integer. +pub fn retry_after(resp: &ureq::Response) -> Option { + resp.header("retry-after") + .and_then(|v| v.trim().parse().ok()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn retries_transient_then_succeeds() { + let mut calls = 0; + let out = retry(&RetryPolicy::default(), || { + calls += 1; + match calls { + 1..=3 => Attempt::Again(None), + _ => Attempt::Done("ok"), + } + }); + assert_eq!(out, Some("ok")); + assert_eq!(calls, 4); + } + + #[test] + fn stops_on_permanent_failure() { + let mut calls = 0; + let out: Option = retry(&RetryPolicy::default(), || { + calls += 1; + Attempt::Stop + }); + assert_eq!(out, None); + assert_eq!(calls, 1); + } + + #[test] + fn gives_up_after_max_attempts() { + let policy = RetryPolicy { + max_attempts: 3, + ..RetryPolicy::default() + }; + let mut calls = 0; + let out: Option = retry(&policy, || { + calls += 1; + Attempt::Again(None) + }); + assert_eq!(out, None); + assert_eq!(calls, 3); + } + + #[test] + fn honors_retry_after() { + let mut calls = 0; + let out: Option = retry(&RetryPolicy::default(), || { + calls += 1; + match calls { + 1 => Attempt::Again(Some(1)), + _ => Attempt::Done(7), + } + }); + assert_eq!(out, Some(7)); + assert_eq!(calls, 2); + } + + #[test] + fn jitter_is_bounded() { + for _ in 0..100 { + let j = jitter(Duration::from_millis(1000)); + assert!(j >= Duration::from_millis(1000)); + assert!(j < Duration::from_millis(1500)); + } + } +}