Apple Podcasts' top charts into the directory weekly, and its categories checked (#252)

With apple_country set, the directory fetches Apple's top 100 podcasts and the top 50 of each of
its 19 categories for that country (itunes.apple.com/{cc}/rss/toppodcasts), looks up each
show's RSS address (/lookup, 100 ids a call, three seconds apart, as Apple asks for no more than
about twenty a minute), and stores them as the snapshot apple-podcasts with no subscribers. The
refresher fetches them like any feed; the most-subscribed chart is unchanged. Apple's primary
genre goes in as the snapshot's category, used only while a feed names none.

The same weekly run reads Apple's genre tree (ws/genres?id=26) and warns when it differs from
ipx_feed's CATEGORIES; on 2026-10-15 they were the same, all 19 and their subcategories.

A syncs table keeps when it last ran: at start if a week has passed, again weekly, an hour after
a failure. keep_instances spares apple-podcasts, which no token holds, and no token may take
the name. A trace a sync, a span each answer.

CLAUDE.md: builds one cargo command at a time, the load looked at before and after, after a
build chained with a test compile took Tower past 200.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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).

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<String> = 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::<Chart>(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<T: serde::de::DeserializeOwned>(client: &reqwest::Client, url: &str) -> Result<T> {
+    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::<T>().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<String, Genre>);
+#[derive(Deserialize)]
+struct Genre {
+    name: String,
+    #[serde(default)]
+    subgenres: BTreeMap<String, Genre>,
+}
+
+fn categories(g: &Genres) -> BTreeMap<String, Vec<String>> {
+    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<String> {
+    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<String, Vec<String>>, ours: &[(&str, &[&str])]) -> (Vec<String>, Vec<String>) {
+    let flat = |c: &str, subs: &mut dyn Iterator<Item = String>| {
+        std::iter::once(c.to_string()).chain(subs.map(|s| format!("{c} / {s}"))).collect::<Vec<_>>()
+    };
+    let theirs: std::collections::BTreeSet<String> =
+        theirs.iter().flat_map(|(c, subs)| flat(c, &mut subs.iter().cloned())).collect();
+    let ours: std::collections::BTreeSet<String> =
+        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<ChartEntry>,
+}
+#[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<Vec<ChartEntry>, D::Error> {
+    #[derive(Deserialize)]
+    #[serde(untagged)]
+    enum OneOrMany {
+        One(ChartEntry),
+        Many(Vec<ChartEntry>),
+    }
+    Ok(match OneOrMany::deserialize(d)? {
+        OneOrMany::One(e) => vec![e],
+        OneOrMany::Many(v) => v,
+    })
+}
+
+fn chart_ids(c: &Chart) -> Vec<String> {
+    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<LookupResult>,
+}
+#[derive(Deserialize)]
+struct LookupResult {
+    #[serde(rename = "feedUrl")]
+    feed_url: Option<String>,
+    #[serde(rename = "primaryGenreName")]
+    genre: Option<String>,
+}
+
+fn lookup_entries(l: &Lookup) -> Vec<SnapshotEntry> {
+    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");
+    }
+}

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<TokenCfg>,
 }
@@ -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)
     }

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;

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 {

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<Option<i64>> {
+        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::<Result<_, _>>()?;
-        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)

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<String>) {
     // 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();