ipx/crates/directory/src/apple.rs

//! 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");
    }
}