diff --git a/.claude/skills/ipx-prod-check/SKILL.md b/.claude/skills/ipx-prod-check/SKILL.md index decd425..d852fc6 100644 --- a/.claude/skills/ipx-prod-check/SKILL.md +++ b/.claude/skills/ipx-prod-check/SKILL.md @@ -307,7 +307,11 @@ checked every twelve hours (#243), 25 at a time, so `feeds checked` over a day s twice the feeds listed; none for over twelve hours means the refresher is stuck. `could not store a feed check` or `could not read which feeds are due` is its database (`ipx_directory` on the `databases` project's Postgres). A config that does not load (`config changed but does not -load`) keeps the old tokens: worth fixing before someone's new token is needed. +load`) keeps the old tokens: worth fixing before someone's new token is needed. Weekly, `Apple +Podcasts synced` (`shows`, `accepted`) brings Apple's charts in as the `apple-podcasts` snapshot +(#252); `Apple Podcasts' sync failed` retries in an hour, worth an issue when it keeps failing, and +`Apple's podcast categories are not ipx's` (`added`, `gone`) is an issue to update `CATEGORIES` in +`crates/feed`. ## Tower diff --git a/CLAUDE.md b/CLAUDE.md index 378b19c..0798b0a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -181,6 +181,19 @@ npx playwright test # 66 browser tests against a real daemon on fixture fe node tests/load/run.js # k6: many people at once against a scratch daemon with 1,500 feeds ``` +**Builds share Tower with everything else, production included.** On 2026-10-16 a build of the +directory followed in the same command by its tests' compile took Tower's load past 200 and the +session was stopped. So: + +* One cargo command at a time: a build, then the tests, as separate commands, never chained or + run in the background beside another, and never two crates' builds at once. Do not tune `-j`. +* Look at the load before and after each (`cat /proc/loadavg`). Above about 24, Tower's twelve + threads twice over, wait for it to come down before starting another. +* Build only the crate the change is in (`cargo build -p ipx-directory`), and run only its tests + until the change is done; the whole workspace's once, at the end. +* Builds and test binaries live under `target/`, a link to `/mnt/fast`; linking through + `/mnt/user` (FUSE) once took the load to 248 on its own. + Things about the browser suite that have cost time: * It starts **its own daemon and database** under `/tmp/ipx-ui-test`, wiped once per run. Playwright diff --git a/crates/directory/CHANGELOG.md b/crates/directory/CHANGELOG.md index 8c199c5..d5ef23a 100644 --- a/crates/directory/CHANGELOG.md +++ b/crates/directory/CHANGELOG.md @@ -10,6 +10,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- With `apple_country` set, Apple Podcasts' top charts for that country (the top 100 and the top 50 of each category) come into the directory every week, beside what servers share, without counting as subscribers; and a warning says when Apple's categories no longer match ipx's (#252). + ### Changed - Each feed is checked every twelve hours instead of six (#243). diff --git a/crates/directory/src/apple.rs b/crates/directory/src/apple.rs new file mode 100644 index 0000000..7318e4a --- /dev/null +++ b/crates/directory/src/apple.rs @@ -0,0 +1,273 @@ +//! Apple Podcasts, weekly (#252): its top charts for a country into the directory, beside what +//! ipx servers share, and its categories checked against the copy ipx keeps (`ipx_feed`). +//! +//! The charts are a snapshot of their own, under `INSTANCE`, with no subscribers: Apple's shows +//! are fetched, categorised and listed like any other feed, and the most subscribed stay what this +//! directory's people subscribe to. Each show's RSS address comes from Apple's lookup API. + +use std::collections::BTreeMap; +use std::time::Duration; + +use anyhow::{Context, Result}; +use serde::Deserialize; +use tracing::Instrument; + +use crate::{Shared, SnapshotEntry}; + +/// What the charts are filed under, as a server's snapshot is under its token's name. +pub const INSTANCE: &str = "apple-podcasts"; +/// How often, and how long to wait after a sync fails before trying again. +const EVERY: i64 = 7 * 86_400; +const RETRY: Duration = Duration::from_secs(3600); +/// The top this many overall, and of each category: about a thousand shows once they overlap. +const TOP: u32 = 100; +const TOP_OF_EACH: u32 = 50; +/// Apple asks for no more than about twenty calls a minute. +const PAUSE: Duration = Duration::from_secs(3); + +/// Syncs when a week has passed since the last, forever. +pub async fn run(dir: Shared, country: String) { + let client = match reqwest::Client::builder() + .user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION"))) + .connect_timeout(Duration::from_secs(10)) + .timeout(Duration::from_secs(60)) + .build() + { + Ok(c) => c, + Err(e) => return tracing::error!(error = %e, "Apple Podcasts' sync could not start"), + }; + loop { + let last = dir.store.synced_at(INSTANCE).await.unwrap_or(None).unwrap_or(0); + let wait = last + EVERY - crate::store::now(); + if wait > 0 { + // Woken hourly at most, so a database restored from before the last sync is noticed. + tokio::time::sleep(Duration::from_secs((wait as u64).min(3600))).await; + continue; + } + let span = tracing::info_span!("apple_sync", country = %country); + match sync(&dir, &client, &country).instrument(span).await { + Ok(()) => { + let _ = dir.store.set_synced(INSTANCE, crate::store::now()).await; + } + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "Apple Podcasts' sync failed; trying again in an hour"); + tokio::time::sleep(RETRY).await; + } + } + } +} + +async fn sync(dir: &Shared, client: &reqwest::Client, country: &str) -> Result<()> { + let genres: Genres = get(client, "https://itunes.apple.com/WebObjects/MZStoreServices.woa/ws/genres?id=26").await?; + let theirs = categories(&genres); + let (added, gone) = differ(&theirs, ipx_feed::categories()); + if !added.is_empty() || !gone.is_empty() { + tracing::warn!( + added = %added.join(", "), gone = %gone.join(", "), + "Apple's podcast categories are not ipx's: the copy in crates/feed (CATEGORIES) wants updating" + ); + } + + // The top overall, then the top of each category, each show once. + let mut ids: Vec = Vec::new(); + let charts = std::iter::once((None, TOP)).chain(genre_ids(&genres).into_iter().map(|g| (Some(g), TOP_OF_EACH))); + for (genre, n) in charts { + let url = match &genre { + Some(g) => format!("https://itunes.apple.com/{country}/rss/toppodcasts/limit={n}/genre={g}/json"), + None => format!("https://itunes.apple.com/{country}/rss/toppodcasts/limit={n}/json"), + }; + match get::(client, &url).await { + Ok(chart) => ids.extend(chart_ids(&chart)), + // One category's chart missing is no reason to keep last week's whole list. + Err(e) => tracing::info!(genre = ?genre, error = %format!("{e:#}"), "an Apple chart did not load"), + } + tokio::time::sleep(PAUSE).await; + } + let mut seen = std::collections::HashSet::new(); + ids.retain(|id| seen.insert(id.clone())); + anyhow::ensure!(!ids.is_empty(), "no chart had a show in it"); + + let mut entries = Vec::new(); + for batch in ids.chunks(100) { + let url = format!("https://itunes.apple.com/lookup?entity=podcast&id={}", batch.join(",")); + let found: Lookup = get(client, &url).await?; + entries.extend(lookup_entries(&found)); + tokio::time::sleep(PAUSE).await; + } + let result = dir.store.snapshot(INSTANCE, &entries).await?; + tracing::info!( + shows = ids.len(), accepted = result.accepted, refused_private = result.refused_private, + refused_blocked = result.refused_blocked, refused_invalid = result.refused_invalid, + "Apple Podcasts synced" + ); + // Fetched now, not when the refresher next looks. + dir.wake.notify_one(); + Ok(()) +} + +async fn get(client: &reqwest::Client, url: &str) -> Result { + let span = tracing::info_span!("apple_get", url = %url); + async { + let resp = client.get(url).send().await.context("asking Apple")?.error_for_status()?; + resp.json::().await.context("reading Apple's answer") + } + .instrument(span) + .await +} + +/// Apple's genres under Podcasts (26): each category's id and name, and its subcategories. +#[derive(Deserialize)] +struct Genres(BTreeMap); +#[derive(Deserialize)] +struct Genre { + name: String, + #[serde(default)] + subgenres: BTreeMap, +} + +fn categories(g: &Genres) -> BTreeMap> { + g.0.get("26") + .map(|podcasts| { + podcasts.subgenres.values().map(|c| (c.name.clone(), c.subgenres.values().map(|s| s.name.clone()).collect())).collect() + }) + .unwrap_or_default() +} + +fn genre_ids(g: &Genres) -> Vec { + g.0.get("26").map(|p| p.subgenres.keys().cloned().collect()).unwrap_or_default() +} + +/// What Apple has that ipx's list lacks, and what ipx's has that Apple no longer does, as +/// "Category" or "Category / Subcategory". +fn differ(theirs: &BTreeMap>, ours: &[(&str, &[&str])]) -> (Vec, Vec) { + let flat = |c: &str, subs: &mut dyn Iterator| { + std::iter::once(c.to_string()).chain(subs.map(|s| format!("{c} / {s}"))).collect::>() + }; + let theirs: std::collections::BTreeSet = + theirs.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().cloned())).collect(); + let ours: std::collections::BTreeSet = + ours.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().map(|s| s.to_string()))).collect(); + (theirs.difference(&ours).cloned().collect(), ours.difference(&theirs).cloned().collect()) +} + +/// A chart, as Apple's iTunes RSS gives it: the shows' Apple ids, in order. +#[derive(Deserialize)] +struct Chart { + feed: ChartFeed, +} +#[derive(Deserialize)] +struct ChartFeed { + // A chart of one show is that show, not a list of one. + #[serde(default, deserialize_with = "one_or_many")] + entry: Vec, +} +#[derive(Deserialize)] +struct ChartEntry { + id: ChartId, +} +#[derive(Deserialize)] +struct ChartId { + attributes: ChartIdAttributes, +} +#[derive(Deserialize)] +struct ChartIdAttributes { + #[serde(rename = "im:id")] + id: String, +} + +fn one_or_many<'de, D: serde::Deserializer<'de>>(d: D) -> Result, D::Error> { + #[derive(Deserialize)] + #[serde(untagged)] + enum OneOrMany { + One(ChartEntry), + Many(Vec), + } + Ok(match OneOrMany::deserialize(d)? { + OneOrMany::One(e) => vec![e], + OneOrMany::Many(v) => v, + }) +} + +fn chart_ids(c: &Chart) -> Vec { + c.feed.entry.iter().map(|e| e.id.attributes.id.clone()).collect() +} + +/// Apple's lookup answer: each show's RSS address and the category Apple files it under. +#[derive(Deserialize)] +struct Lookup { + #[serde(default)] + results: Vec, +} +#[derive(Deserialize)] +struct LookupResult { + #[serde(rename = "feedUrl")] + feed_url: Option, + #[serde(rename = "primaryGenreName")] + genre: Option, +} + +fn lookup_entries(l: &Lookup) -> Vec { + l.results + .iter() + .filter_map(|r| { + let url = r.feed_url.as_deref()?.trim(); + (!url.is_empty()).then(|| SnapshotEntry { + url: url.to_owned(), + subscribers: 0, + // "Podcasts" is the root Apple files a show under when it gives no category. + category: r.genre.clone().filter(|g| g != "Podcasts"), + }) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn apples_categories_are_read_and_compared_with_ipxs() { + let genres: Genres = serde_json::from_str( + r#"{"26": {"name": "Podcasts", "subgenres": { + "1301": {"name": "Arts", "subgenres": {"1482": {"name": "Books"}, "1999": {"name": "Murals"}}}, + "1310": {"name": "Music"}}}}"#, + ) + .unwrap(); + assert_eq!(genre_ids(&genres), ["1301", "1310"]); + let ours: &[(&str, &[&str])] = &[("Arts", &["Books", "Design"]), ("Music", &[]), ("Hobbies", &[])]; + let (added, gone) = differ(&categories(&genres), ours); + assert_eq!(added, ["Arts / Murals"]); + assert_eq!(gone, ["Arts / Design", "Hobbies"]); + // ipx's own copy is Apple's whole list, each category once. + let mine = ipx_feed::categories(); + assert_eq!(mine.len(), 19); + assert!(mine.iter().any(|(c, subs)| *c == "News" && subs.contains(&"Daily News"))); + } + + #[test] + fn a_chart_gives_its_shows_in_order_and_a_lookup_their_addresses() { + let chart: Chart = serde_json::from_str( + r#"{"feed": {"entry": [ + {"id": {"label": "https://x", "attributes": {"im:id": "1200361736"}}}, + {"id": {"label": "https://y", "attributes": {"im:id": "1028908750"}}}]}}"#, + ) + .unwrap(); + assert_eq!(chart_ids(&chart), ["1200361736", "1028908750"]); + let one: Chart = serde_json::from_str(r#"{"feed": {"entry": {"id": {"attributes": {"im:id": "7"}}}}}"#).unwrap(); + assert_eq!(chart_ids(&one), ["7"], "a chart of one"); + let empty: Chart = serde_json::from_str(r#"{"feed": {}}"#).unwrap(); + assert!(chart_ids(&empty).is_empty()); + + let found: Lookup = serde_json::from_str( + r#"{"resultCount": 3, "results": [ + {"trackName": "The Daily", "feedUrl": "https://feeds.simplecast.com/Sl5CSM3S", "primaryGenreName": "Daily News"}, + {"trackName": "No feed"}, + {"trackName": "Rooted", "feedUrl": "https://example.com/rss", "primaryGenreName": "Podcasts"}]}"#, + ) + .unwrap(); + let entries = lookup_entries(&found); + assert_eq!(entries.len(), 2, "a show Apple has no feed for is left out"); + assert_eq!((entries[0].subscribers, entries[0].category.as_deref()), (0, Some("Daily News"))); + assert_eq!(entries[1].category, None, "Podcasts is no category"); + } +} diff --git a/crates/directory/src/config.rs b/crates/directory/src/config.rs index c40d8d9..d2aaed1 100644 --- a/crates/directory/src/config.rs +++ b/crates/directory/src/config.rs @@ -18,6 +18,11 @@ pub struct Config { /// publishes: `0.0.0.0:9188`. Empty serves none. #[serde(default)] pub metrics_bind: String, + /// Apple Podcasts' top charts for this country (its two letters: `ca`, `us`) come into the + /// directory each week, and Apple's categories are checked against ipx's (#252). Empty, or + /// left out, does neither. + #[serde(default)] + pub apple_country: String, #[serde(default)] pub tokens: Vec, } @@ -81,6 +86,14 @@ impl Config { if !seen.insert(t.name.as_str()) { bail!("two tokens are named {:?}", t.name); } + // Apple's charts are filed under it; a server of that name would replace them. + if t.name == crate::apple::INSTANCE { + bail!("{:?} is the name Apple Podcasts' charts are kept under; give the token another", t.name); + } + } + let c = cfg.apple_country.trim(); + if !c.is_empty() && !(c.len() == 2 && c.chars().all(|c| c.is_ascii_alphabetic())) { + bail!("apple_country is a country's two letters, such as \"ca\" or \"us\", not {c:?}"); } Ok(cfg) } diff --git a/crates/directory/src/lib.rs b/crates/directory/src/lib.rs index 699a8f4..982208a 100644 --- a/crates/directory/src/lib.rs +++ b/crates/directory/src/lib.rs @@ -7,6 +7,7 @@ //! The types here are the wire format, shared with ipx, which asks the directory on its //! people's behalf. +pub mod apple; pub mod config; pub mod metrics; pub mod otel; diff --git a/crates/directory/src/main.rs b/crates/directory/src/main.rs index 0d51f29..910e26c 100644 --- a/crates/directory/src/main.rs +++ b/crates/directory/src/main.rs @@ -104,6 +104,11 @@ async fn serve(path: &std::path::Path) -> Result<()> { tokio::spawn(ipx_directory::metrics::serve(metrics)); } tokio::spawn(ipx_directory::refresher(dir.clone())); + let country = cfg.apple_country.trim().to_ascii_lowercase(); + if !country.is_empty() { + tracing::info!(country = %country, "Apple Podcasts' charts are synced weekly"); + tokio::spawn(ipx_directory::apple::run(dir.clone(), country)); + } tokio::spawn(ipx_directory::reloader(dir.clone(), path.to_owned())); axum::serve(listener, ipx_directory::router(dir)) .with_graceful_shutdown(async { diff --git a/crates/directory/src/store.rs b/crates/directory/src/store.rs index bccad57..5f61df3 100644 --- a/crates/directory/src/store.rs +++ b/crates/directory/src/store.rs @@ -84,6 +84,11 @@ const SCHEMA: &[&str] = &[ PRIMARY KEY (instance, feed_id) )", "CREATE INDEX IF NOT EXISTS snapshots_feed ON snapshots (feed_id)", + // When each periodic job last finished: Apple's weekly sync (#252). + "CREATE TABLE IF NOT EXISTS syncs ( + name TEXT PRIMARY KEY, + at BIGINT NOT NULL + )", "CREATE TABLE IF NOT EXISTS blocked ( id TEXT PRIMARY KEY, url TEXT, @@ -164,6 +169,23 @@ impl Store { Ok(self.db.query_all_raw(self.stmt(sql, values)).await?) } + /// When the job `name` last finished, Unix seconds. + pub async fn synced_at(&self, name: &str) -> Result> { + Ok(match self.rows("SELECT at FROM syncs WHERE name = $1", vec![name.into()]).await?.first() { + Some(r) => Some(r.try_get("", "at")?), + None => None, + }) + } + + pub async fn set_synced(&self, name: &str, at: i64) -> Result<()> { + self.exec( + "INSERT INTO syncs (name, at) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET at = excluded.at", + vec![name.into(), at.into()], + ) + .await?; + Ok(()) + } + pub async fn ping(&self) -> Result<()> { self.rows("SELECT 1", vec![]).await?; Ok(()) @@ -241,7 +263,8 @@ impl Store { .iter() .map(|r| r.try_get("", "instance")) .collect::>()?; - for name in present.iter().filter(|n| !names.contains(n)) { + // Apple's charts are no token's, and stay until the next sync replaces them. + for name in present.iter().filter(|n| !names.contains(n) && n.as_str() != crate::apple::INSTANCE) { dropped += self.drop_instance(name).await?; } Ok(dropped) diff --git a/crates/feed/src/lib.rs b/crates/feed/src/lib.rs index 18727fb..1909bf3 100644 --- a/crates/feed/src/lib.rs +++ b/crates/feed/src/lib.rs @@ -1120,6 +1120,11 @@ const ALIASES: &[(&str, &str)] = /// Where a category sits in Apple's list, as (category, subcategory), so the Directory browses a /// category and then its subcategories as Apple's does (#118). A podcast's own is stored as its /// subcategory, and one an admin typed that is not Apple's stands as a category of its own. +/// Apple's categories as ipx has them, for the directory to check against Apple's own (#252). +pub fn categories() -> &'static [(&'static str, &'static [&'static str])] { + CATEGORIES +} + pub fn category_path(name: &str) -> (String, Option) { // A feed's own text, which can end in a stray comma: "Soccer," was a category of its own. let name = name.trim().trim_end_matches([',', ';', '.']).trim_end(); diff --git a/docs/directory.md b/docs/directory.md index 6dc33d9..5dd8a09 100644 --- a/docs/directory.md +++ b/docs/directory.md @@ -81,6 +81,8 @@ bind = "0.0.0.0:8100" # Prometheus metrics, on a port of their own and not the one the tunnel publishes. Empty, or # left out, serves none. metrics_bind = "0.0.0.0:9188" +# Apple Podcasts' top charts for this country, weekly (see below). Empty, or left out: none. +apple_country = "ca" # The environment variable holding the database URL: postgres://... or sqlite://... database_url_env = "DIRECTORY_DATABASE_URL" @@ -100,6 +102,23 @@ production does with `deployment.environment.name=production`. Its version is in `crates/directory/Cargo.toml`, logged at start and answered by `ipx-directory --version`; its changelog is `crates/directory/CHANGELOG.md`. +## Apple Podcasts + +With `apple_country` set, the directory takes Apple Podcasts' top charts for that country once a +week (#252): the top 100 overall and the top 50 of each of Apple's categories, about a thousand +shows. Apple's lookup API gives each one's RSS address, and they go in as a snapshot of their own, +`apple-podcasts`, with no subscribers: fetched, sorted into categories and listed like any other +feed, while the most subscribed stay what the servers sharing here subscribe to. A show Apple drops +from its charts leaves the directory at the next sync, unless a server shares it. An admin can drop +the snapshot as any server's (`DELETE /v1/instances/apple-podcasts`); the next sync brings it back. +No token may be named `apple-podcasts`. + +The same sync compares Apple's list of categories with the one ipx keeps (`CATEGORIES` in +`crates/feed`) and logs a warning naming what differs, for whoever checks the log to update it. +It runs at start when the last sync is a week old, and a failed one tries again an hour later. Each +sync is a trace (`apple_sync`, a span for each of Apple's answers), and a line `Apple Podcasts +synced` says how many shows came and went in. + ## Running one ```yaml