lift limit on cid

src/budget.rs

@@ -0,0 +1,241 @@
+//! Per-upstream request budget — a shared token bucket.
+//!
+//! Lifted from the card feeder's `tmdb_budget.rs` (its `TmdbBudget` half; the
+//! per-indexer registry has no counterpart here) and renamed, because the
+//! mechanism is upstream-agnostic and the music feeder needs three independent
+//! instances rather than one global.
+//!
+//! **Why three, not one.** MusicBrainz's 1 req/s is a hard published ceiling
+//! enforced per source IP; ListenBrainz and the Internet Archive are far more
+//! generous. A single shared bucket would either throttle all three to 1/s or
+//! let MusicBrainz calls burst past a limit that gets the whole box blocked.
+//!
+//! Design notes, unchanged from the original:
+//! - **Lazy refill.** Tokens accrue on each `acquire` from elapsed wall-clock,
+//!   so there is no background timer task and no idle cost.
+//! - **Cancel-safe.** The token is decremented synchronously under the lock
+//!   right before `acquire` returns [`Lease::Granted`]; there is no `.await`
+//!   between the decision to grant and the return, so a dropped `acquire`
+//!   future (consumer disconnected) never spends a token.
+//! - **Global pause on throttle.** One upstream 429/503 freezes *all* grants
+//!   on that bucket until the Retry-After window elapses, via
+//!   [`RateBudget::note_throttled`].
+//!
+//! The critical sections hold a plain `std::sync::Mutex` (no awaits inside),
+//! while the waiting happens outside the lock on a `tokio::time::sleep` raced
+//! against a `tokio::sync::Notify`.
+
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use tokio::sync::Notify;
+
+/// Outcome of a [`RateBudget::acquire`] call.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Lease {
+    /// A token was granted; the caller may make one upstream call.
+    Granted,
+    /// The wait deadline elapsed before a token became available. The caller
+    /// degrades best-effort — it does **not** call the upstream anyway.
+    DeadlineExceeded,
+}
+
+struct BucketState {
+    tokens: f64,
+    capacity: f64,
+    refill_per_sec: f64,
+    last_refill: Instant,
+    /// When set and in the future, all grants are frozen.
+    paused_until: Option<Instant>,
+}
+
+impl BucketState {
+    /// Accrue tokens for elapsed time since the last refill. Called under the
+    /// lock before inspecting `tokens`.
+    fn refill(&mut self, now: Instant) {
+        let elapsed = now.saturating_duration_since(self.last_refill).as_secs_f64();
+        if elapsed > 0.0 {
+            self.tokens = (self.tokens + elapsed * self.refill_per_sec).min(self.capacity);
+            self.last_refill = now;
+        }
+    }
+}
+
+/// Shared token bucket gating one upstream's API calls across the process.
+pub struct RateBudget {
+    state: Mutex<BucketState>,
+    /// Woken whenever a grant *might* now be possible (a token accrued, or a
+    /// pause was cleared). Waiters re-check under the lock after waking; a
+    /// missed wake only costs an extra bounded sleep.
+    wake: Notify,
+}
+
+impl RateBudget {
+    /// Build a budget sustaining `refill_per_sec` grants/second with a burst
+    /// ceiling of `capacity` tokens. The bucket starts full, so the first
+    /// burst is served immediately.
+    pub fn new(refill_per_sec: f64, capacity: f64) -> Arc<Self> {
+        Arc::new(Self {
+            state: Mutex::new(BucketState {
+                tokens: capacity,
+                capacity,
+                refill_per_sec,
+                last_refill: Instant::now(),
+                paused_until: None,
+            }),
+            wake: Notify::new(),
+        })
+    }
+
+    /// Wait for one token, giving up after `deadline`. Cancel-safe: dropping
+    /// the returned future before it resolves [`Lease::Granted`] spends no
+    /// token.
+    pub async fn acquire(&self, deadline: Duration) -> Lease {
+        let give_up_at = Instant::now() + deadline;
+        loop {
+            // Decide-or-compute-wait, all under the lock (no awaits inside).
+            let sleep_for = {
+                let mut st = self.state.lock().expect("rate budget mutex poisoned");
+                let now = Instant::now();
+                if let Some(until) = st.paused_until {
+                    if now >= until {
+                        st.paused_until = None;
+                    }
+                }
+                let paused = st.paused_until;
+                if paused.is_none() {
+                    st.refill(now);
+                    if st.tokens >= 1.0 {
+                        // Atomic grant: decrement and return with no await between.
+                        st.tokens -= 1.0;
+                        return Lease::Granted;
+                    }
+                }
+                if now >= give_up_at {
+                    return Lease::DeadlineExceeded;
+                }
+                let until_deadline = give_up_at.saturating_duration_since(now);
+                let wait = match paused {
+                    // Frozen: wake when the pause is scheduled to lift.
+                    Some(until) => until.saturating_duration_since(now),
+                    // Throttled: wake when the next whole token should accrue.
+                    None => {
+                        let needed = 1.0 - st.tokens;
+                        let secs = if st.refill_per_sec > 0.0 {
+                            needed / st.refill_per_sec
+                        } else {
+                            until_deadline.as_secs_f64()
+                        };
+                        Duration::from_secs_f64(secs.max(0.0))
+                    }
+                };
+                wait.min(until_deadline)
+            };
+            tokio::select! {
+                _ = tokio::time::sleep(sleep_for) => {}
+                _ = self.wake.notified() => {}
+            }
+            if Instant::now() >= give_up_at {
+                return Lease::DeadlineExceeded;
+            }
+        }
+    }
+
+    /// Feed back an upstream throttle response (429, or MusicBrainz's 503):
+    /// pause all grants on this bucket until `now + retry_after`. Extends an
+    /// existing pause but never shortens it.
+    pub fn note_throttled(&self, retry_after: Duration) {
+        let until = Instant::now() + retry_after;
+        {
+            let mut st = self.state.lock().expect("rate budget mutex poisoned");
+            st.paused_until = Some(match st.paused_until {
+                Some(existing) if existing > until => existing,
+                _ => until,
+            });
+        }
+        // Wake waiters so they recompute their sleep against the new pause.
+        self.wake.notify_waiters();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test(start_paused = true)]
+    async fn grants_up_to_capacity_immediately() {
+        let b = RateBudget::new(3.0, 3.0);
+        for _ in 0..3 {
+            assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn next_token_waits_for_refill() {
+        let b = RateBudget::new(2.0, 2.0); // 2/s sustained
+        for _ in 0..2 {
+            assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
+        }
+        // Bucket empty; next grant must wait ~0.5 s for one token at 2/s.
+        let start = Instant::now();
+        assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
+        assert!(
+            start.elapsed() >= Duration::from_millis(450),
+            "expected to wait ~0.5s for refill, waited {:?}",
+            start.elapsed()
+        );
+    }
+
+    /// The MusicBrainz shape specifically: 1/s with a burst of 2 must not let
+    /// three calls through inside one second. This is the property that keeps
+    /// the box off their block-list.
+    #[tokio::test(start_paused = true)]
+    async fn musicbrainz_shape_holds_one_per_second() {
+        let b = RateBudget::new(1.0, 1.0);
+        let start = Instant::now();
+        for _ in 0..4 {
+            assert_eq!(b.acquire(Duration::from_secs(30)).await, Lease::Granted);
+        }
+        // 4 grants from a 2-token bucket refilling at 1/s ⇒ at least 2 s.
+        assert!(
+            start.elapsed() >= Duration::from_millis(1_900),
+            "4 calls at 1/s (burst 2) must take ~2s, took {:?}",
+            start.elapsed()
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn deadline_exceeded_when_drained() {
+        let b = RateBudget::new(0.001, 1.0); // effectively no refill within the test
+        assert_eq!(b.acquire(Duration::from_secs(1)).await, Lease::Granted);
+        assert_eq!(
+            b.acquire(Duration::from_millis(200)).await,
+            Lease::DeadlineExceeded
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn note_throttled_pauses_grants_even_with_tokens_available() {
+        let b = RateBudget::new(100.0, 100.0);
+        b.note_throttled(Duration::from_secs(2));
+        assert_eq!(
+            b.acquire(Duration::from_millis(500)).await,
+            Lease::DeadlineExceeded
+        );
+        // After the pause window, grants resume.
+        assert_eq!(b.acquire(Duration::from_secs(5)).await, Lease::Granted);
+    }
+
+    /// A pause is extended, never shortened — otherwise a late short retry
+    /// would cancel an earlier long one and walk straight back into the block.
+    #[tokio::test(start_paused = true)]
+    async fn a_shorter_retry_after_does_not_shorten_an_existing_pause() {
+        let b = RateBudget::new(100.0, 100.0);
+        b.note_throttled(Duration::from_secs(10));
+        b.note_throttled(Duration::from_secs(1));
+        assert_eq!(
+            b.acquire(Duration::from_secs(3)).await,
+            Lease::DeadlineExceeded
+        );
+    }
+}

src/cache.rs

@@ -13,7 +13,7 @@
 use std::path::Path;
 use std::sync::Arc;
 
-use redb::{Database, ReadableTableMetadata, TableDefinition};
+use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};
 
 pub const CACHE_FILENAME: &str = "cache.redb";
 
@@ -53,6 +53,44 @@ const SUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("s
 /// OpenSubtitles search results keyed by `"<tmdb_id>\x01<lang3>"`. Value is the
 /// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel.
 const OPENSUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("opensubtitles");
+
+
+// ── music-family tables (MusicBrainz / Internet Archive / YouTube / Jamendo) ──
+
+/// Ranked MusicBrainz search results keyed by `"<entity>\x01<normalised text>"`.
+const MB_SEARCH_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_search");
+/// Release-group detail keyed by its MBID.
+const MB_RELEASE_GROUP_TABLE: TableDefinition<'_, &str, &str> =
+    TableDefinition::new("mb_release_group");
+/// Artist detail keyed by its MBID.
+const MB_ARTIST_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_artist");
+/// A release-group's canonical track list keyed by the release-group MBID.
+const MB_TRACKLIST_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_tracklist");
+/// Internet Archive per-item file listing keyed by the item identifier.
+const IA_ITEM_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("ia_item");
+/// An artist's YouTube channel keys, keyed by artist MBID. See
+/// `youtube::ChannelKeys`.
+///
+/// ⚠ **Only ever holds a POSITIVE result.** MusicBrainz answers `503` when
+/// crowded and the client reports every failure as `None`, so caching a
+/// negative would freeze a transport hiccup into "this artist has no channel"
+/// — the exact fabrication the match probe hit (study §7.5).
+const YT_CHANNEL_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("yt_channel");
+/// Piped `music_songs` results keyed by the normalised search text.
+const YT_SEARCH_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("yt_search");
+/// Jamendo per-track detail keyed by the Jamendo track id. Permanent: a track
+/// id names an immutable recording, and its licence, artist credit and
+/// download URL do not change under it.
+const JAMENDO_TRACK_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("jamendo_track");
+/// Jamendo API calls spent, keyed by UTC `YYYY-MM`.
+///
+/// ⚠ **This table is not a cache — it is an accounting ledger, and losing it
+/// loses money.** Every other table here can be deleted to reclaim disk with
+/// no consequence beyond a slower next request. Deleting this one resets the
+/// month's spend to zero while Jamendo's own counter keeps climbing, so the
+/// feeder cheerfully spends a quota it has already exhausted and every call
+/// fails for the rest of the month. See `consts::JAMENDO_MONTHLY_QUOTA`.
+const JAMENDO_QUOTA_TABLE: TableDefinition<'_, &str, u64> = TableDefinition::new("jamendo_quota");
 /// Small durable plugin scratch values that must survive a restart but aren't
 /// content-derived — today, the torznab plugin's Prowlarr indexer-set
 /// fingerprint, so a set that changed *while the feeder was down* is still
@@ -113,6 +151,15 @@ impl MidhashCache {
             tx.open_table(SUBTITLES_TABLE)?;
             tx.open_table(OPENSUBTITLES_TABLE)?;
             tx.open_table(MISC_TABLE)?;
+            tx.open_table(MB_SEARCH_TABLE)?;
+            tx.open_table(MB_RELEASE_GROUP_TABLE)?;
+            tx.open_table(MB_ARTIST_TABLE)?;
+            tx.open_table(MB_TRACKLIST_TABLE)?;
+            tx.open_table(IA_ITEM_TABLE)?;
+            tx.open_table(YT_CHANNEL_TABLE)?;
+            tx.open_table(YT_SEARCH_TABLE)?;
+            tx.open_table(JAMENDO_TRACK_TABLE)?;
+            tx.open_table(JAMENDO_QUOTA_TABLE)?;
             tx.commit()?;
         }
         Ok(Self { db: Arc::new(db) })
@@ -120,6 +167,47 @@ impl MidhashCache {
 
     str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE);
 
+    // ── music-family accessors ──
+    str_table_accessors!(get_mb_search, put_mb_search, MB_SEARCH_TABLE);
+    str_table_accessors!(get_mb_release_group, put_mb_release_group, MB_RELEASE_GROUP_TABLE);
+    str_table_accessors!(get_mb_artist, put_mb_artist, MB_ARTIST_TABLE);
+    str_table_accessors!(get_mb_tracklist, put_mb_tracklist, MB_TRACKLIST_TABLE);
+    str_table_accessors!(get_ia_item, put_ia_item, IA_ITEM_TABLE);
+    str_table_accessors!(get_yt_channel, put_yt_channel, YT_CHANNEL_TABLE);
+    str_table_accessors!(get_yt_search, put_yt_search, YT_SEARCH_TABLE);
+    str_table_accessors!(get_jamendo_track, put_jamendo_track, JAMENDO_TRACK_TABLE);
+
+    pub fn jamendo_quota_spend(
+        &self,
+        month: &str,
+        budget: u64,
+    ) -> Result<Option<u64>, redb::Error> {
+        let tx = self.db.begin_write()?;
+        let spent_after;
+        {
+            let mut table = tx.open_table(JAMENDO_QUOTA_TABLE)?;
+            let spent = table.get(month)?.map(|v| v.value()).unwrap_or(0);
+            if spent >= budget {
+                drop(table);
+                // Nothing was written; abort rather than commit an empty txn.
+                tx.abort()?;
+                return Ok(None);
+            }
+            spent_after = spent + 1;
+            table.insert(month, spent_after)?;
+        }
+        tx.commit()?;
+        Ok(Some(spent_after))
+    }
+
+    /// Calls already spent in `month`. Read-only — for health and the config
+    /// surface, never on the request path.
+    pub fn jamendo_quota_used(&self, month: &str) -> Result<u64, redb::Error> {
+        let tx = self.db.begin_read()?;
+        let table = tx.open_table(JAMENDO_QUOTA_TABLE)?;
+        Ok(table.get(month)?.map(|v| v.value()).unwrap_or(0))
+    }
+
     str_table_accessors!(
         /// Read a durable plugin scratch value (see [`MISC_TABLE`]).
         get_misc,
@@ -351,3 +439,68 @@ mod tests {
         assert_eq!(cache.entry_count().unwrap(), 2);
     }
 }
+
+/// Cache key for a search: entity type plus the normalised query text.
+///
+/// Normalisation is lowercase + whitespace-collapse only. Deliberately not
+/// more aggressive: MusicBrainz treats punctuation as significant (`Re:ZERO`,
+/// `M!LK`, `!!!` are all real artist names), so stripping it would collapse
+/// distinct queries onto one cache entry and serve the wrong answer.
+pub fn search_key(entity: &str, text: &str) -> String {
+    let normalised = text.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase();
+    format!("{entity}\u{1}{normalised}")
+}
+
+#[cfg(test)]
+mod music_table_tests {
+    use super::*;
+    use tempfile::tempdir;
+
+    #[test]
+    fn round_trips_each_table_independently() {
+        let dir = tempdir().expect("tempdir");
+        let c = MidhashCache::open(dir.path()).expect("open");
+
+        c.put_mb_search("k", "v-search").unwrap();
+        c.put_mb_release_group("k", "v-rg").unwrap();
+        c.put_mb_artist("k", "v-artist").unwrap();
+        c.put_mb_tracklist("k", "v-tracks").unwrap();
+        c.put_ia_item("k", "v-ia").unwrap();
+
+        // Same key in five tables must not collide.
+        assert_eq!(c.get_mb_search("k").unwrap().as_deref(), Some("v-search"));
+        assert_eq!(c.get_mb_release_group("k").unwrap().as_deref(), Some("v-rg"));
+        assert_eq!(c.get_mb_artist("k").unwrap().as_deref(), Some("v-artist"));
+        assert_eq!(c.get_mb_tracklist("k").unwrap().as_deref(), Some("v-tracks"));
+        assert_eq!(c.get_ia_item("k").unwrap().as_deref(), Some("v-ia"));
+        assert!(c.get_mb_search("absent").unwrap().is_none());
+    }
+
+    #[test]
+    fn search_key_normalises_case_and_whitespace_only() {
+        assert_eq!(
+            search_key("release-group", "  Kind   OF Blue "),
+            search_key("release-group", "kind of blue")
+        );
+        // Entity namespacing keeps an artist search off a release-group entry.
+        assert_ne!(
+            search_key("artist", "miles davis"),
+            search_key("release-group", "miles davis")
+        );
+    }
+
+    /// Punctuation is significant to MusicBrainz, so it must survive
+    /// normalisation — otherwise two genuinely different works share a cache
+    /// entry and one of them is served the other's answer.
+    #[test]
+    fn search_key_preserves_punctuation() {
+        assert_ne!(
+            search_key("artist", "m!lk"),
+            search_key("artist", "mlk"),
+        );
+        assert_ne!(
+            search_key("release-group", "re:zero"),
+            search_key("release-group", "rezero"),
+        );
+    }
+}

src/common.rs

@@ -286,3 +286,43 @@ mod tests {
         assert!(decode_infohash("").is_none());
     }
 }
+
+/// Percent-encode for a URL **path or query value**, escaping space as `%20`.
+///
+/// ⚠ **Not interchangeable with [`urlencode`]**, which encodes space as `+`.
+/// The `+` form is correct for `application/x-www-form-urlencoded` bodies and
+/// wrong for a Lucene-backed search API — MusicBrainz's `/ws/2` treats `+` as
+/// the "required term" metacharacter, so a `+`-encoded space silently changes
+/// the query's meaning rather than failing loudly. Use this one for anything
+/// going into a query *value*; use `urlencode` for form bodies.
+pub fn percent_encode(s: &str) -> String {
+    let mut out = String::with_capacity(s.len() * 2);
+    for b in s.bytes() {
+        match b {
+            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
+                out.push(b as char)
+            }
+            b' ' => out.push_str("%20"),
+            _ => out.push_str(&format!("%{b:02X}")),
+        }
+    }
+    out
+}
+
+/// Normalise a variable-precision date (`YYYY`, `YYYY-MM`, `YYYY-MM-DD`) to a
+/// full `YYYY-MM-DD`, or `None` if it is not one.
+///
+/// ⚠ Validate the shape, never infer it from the length: a first cut matched on
+/// `len()` alone, which turned the seven-character string "garbage" into the
+/// `releasedate` "garbage-01" — a malformed value written to the shared hash
+/// where nothing downstream would question it.
+pub fn normalise_date(raw: &str) -> Option<String> {
+    let digits = |s: &str, n: usize| s.len() == n && s.chars().all(|c| c.is_ascii_digit());
+    let parts: Vec<&str> = raw.trim().split('-').collect();
+    match parts.as_slice() {
+        [y] if digits(y, 4) => Some(format!("{y}-01-01")),
+        [y, m] if digits(y, 4) && digits(m, 2) => Some(format!("{y}-{m}-01")),
+        [y, m, d] if digits(y, 4) && digits(m, 2) && digits(d, 2) => Some(format!("{y}-{m}-{d}")),
+        _ => None,
+    }
+}

src/domain.rs

@@ -0,0 +1,549 @@
+//! `contentKind` → `domain` / `workForm` — the routing tables from
+//! METADATA_KEYS.md §1.
+//!
+//! `domain` answers **"which application is this record destined for?"**. It
+//! is the coarsest facet in the registry and the key every client filters its
+//! catalogue on — **one value per app**, no alternation:
+//!
+//! ```text
+//! meta-watch   →  domain:screen
+//! meta-listen  →  domain:music
+//! meta-read    →  domain:literature
+//! ```
+//!
+//! `workForm` answers the *other* question the old `film`/`tv` split was
+//! secretly encoding: **is this work complete in itself, or one instalment of
+//! a run?** (`standalone` | `serial`). It is cross-domain — a comic issue is
+//! `serial` in `literature` exactly as an episode is `serial` in `screen` —
+//! which is what lets a client ask for it independently of the wall.
+//!
+//! # Why they are stored and not derived on read
+//!
+//! `contentKind`'s vocabulary is deliberately open — a new feeder may mint a
+//! kind and ship without a fleet-wide update. That only stays safe because the
+//! domain travels *with* the record: a peer that has never heard of
+//! `podcastEpisode` still routes it correctly, because it never has to
+//! interpret the token. Deriving on read would silently drop every record
+//! whose kind the reader predates. The same argument applies to `workForm`,
+//! plus a second one: the two kinds that most need it (`pack` and `card`) are
+//! exactly the two the kind alone cannot decide.
+//!
+//! # ⚠ The binding is permanent and one-way
+//!
+//! Add kinds freely; **never re-map an existing kind to a different domain**.
+//! Because the value is stored, two peers on different versions that disagree
+//! about where `audiobook` belongs produce a split-brain that redeploying
+//! cannot heal — the wrong domain is already written into records across the
+//! mesh. If a kind genuinely has to move, mint a new kind.
+//!
+//! The `film`+`tv` → `screen` merge (METADATA_KEYS.md §14.17) is the one
+//! deliberate exception, taken once and early and paid for with a one-shot
+//! meta-core corpus sweep. It is not a precedent.
+//!
+//! # Why `domain` and not one kind-key per medium
+//!
+//! `record_matches` is AND-across-keys / OR-within-key, and a non-matching key
+//! fails the record outright. A per-medium split (`videoKind` / `audioKind` /
+//! `documentKind`) would force meta-listen to express "music video **or**
+//! audio track" as an OR across two *different* keys — which the query
+//! language cannot represent at all. One domain key, OR-ed within itself,
+//! always can. `workForm` is one key everywhere for the same reason.
+
+/// The `contentKind` → `domain` table. A kind absent from this list has no
+/// domain and is **not independently routable** — correct for sidecars
+/// (subtitles, artwork) and for un-anchored hits, a producer bug for anything
+/// else.
+///
+/// ⚠ `pack` is deliberately absent. A season pack and an album release are the
+/// same structural thing, so the kind alone cannot tell `screen` from `music`;
+/// the writer stamps `pack`'s domain from its own context (the same reason
+/// `derive_file_type` cannot have a second `pack` arm).
+///
+/// ⚠ `podcast` / `podcastEpisode` are absent too: the `spoken` domain was
+/// retired unused (METADATA_KEYS.md §14.17), and the PR that first *writes*
+/// those kinds picks their domain.
+const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
+    // screen — everything meta-watch consumes, films and serials alike. The
+    // standalone/serial split lives in `workForm`, not here.
+    ("movie", "screen"),
+    ("series", "screen"),
+    ("episode", "screen"),
+    // music
+    ("track", "music"),
+    ("album", "music"),
+    ("artist", "music"),
+    ("musicVideo", "music"),
+    ("djMix", "music"),
+    ("liveSet", "music"),
+    // literature — an audiobook's identity is the book (same author, same ISBN
+    // family, same series as the ebook), so meta-read groups the two editions
+    // on one page. The player does not decide the domain.
+    ("book", "literature"),
+    ("audiobook", "literature"),
+    ("comic", "literature"),
+    ("manga", "literature"),
+    ("magazine", "literature"),
+    // science
+    ("paper", "science"),
+];
+
+/// The `contentKind` → `workForm` table (METADATA_KEYS.md §1).
+///
+/// `standalone` = a work complete in itself. `serial` = an instalment of an
+/// ongoing run, **or the container or headline for one** — which is why a
+/// `series` card and an `episode` are both `serial`.
+///
+/// ⚠ `pack` is absent for the same reason it is absent above: a season pack is
+/// `serial`, an album release is `standalone`, and only the writer knows which.
+const WORK_FORM_BY_CONTENT_KIND: &[(&str, &str)] = &[
+    // screen
+    ("movie", "standalone"),
+    ("series", "serial"),
+    ("episode", "serial"),
+    // music — an album is a closed work, not an ongoing run, so every rung of
+    // the music ladder is standalone.
+    ("track", "standalone"),
+    ("album", "standalone"),
+    ("artist", "standalone"),
+    ("musicVideo", "standalone"),
+    ("djMix", "standalone"),
+    ("liveSet", "standalone"),
+    // literature — the split that proves `workForm` is not a screen-only axis.
+    ("book", "standalone"),
+    ("audiobook", "standalone"),
+    ("comic", "serial"),
+    ("manga", "serial"),
+    ("magazine", "serial"),
+    // science
+    ("paper", "standalone"),
+];
+
+/// Resolve the domain a `contentKind` belongs to.
+///
+/// Returns `None` for `pack` (not derivable from the kind — the writer supplies
+/// it), for format/sidecar kinds that route nowhere, and for any kind this
+/// build has never heard of. A caller that already knows the domain from its
+/// own context must prefer that over this table.
+pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> {
+    lookup(DOMAIN_BY_CONTENT_KIND, content_kind)
+}
+
+/// Resolve the work form a `contentKind` sits on. Same contract as
+/// [`domain_for_content_kind`]: `None` means "not derivable", never a guess.
+pub fn work_form_for_content_kind(content_kind: &str) -> Option<&'static str> {
+    lookup(WORK_FORM_BY_CONTENT_KIND, content_kind)
+}
+
+fn lookup(table: &'static [(&'static str, &'static str)], key: &str) -> Option<&'static str> {
+    let k = key.trim();
+    table
+        .iter()
+        .find(|(kind, _)| kind.eq_ignore_ascii_case(k))
+        .map(|(_, value)| *value)
+}
+
+/// Kinds whose `domain` the kind→domain table deliberately cannot decide, so
+/// the **writer** supplies it (METADATA_KEYS.md §1). Today that is `pack`
+/// alone: a season pack and an album release are the same structural thing, and
+/// only the producer knows which — indexer-feeder's `bt.rs` reads the domain off
+/// the per-file kind *before* overwriting it with `pack` precisely so the value
+/// survives.
+///
+/// They must ride along in every `domain:` → `contentKind:` expansion. The
+/// expansion is only ever a **pre-filter** — the real `domain:` term is enforced
+/// afterwards by [`crate::query_eval::record_matches`] — so widening it is free,
+/// while omitting these kinds would drop exactly the records whose `domain` is
+/// the most authoritative one on the wire.
+pub const WRITER_SUPPLIED_DOMAIN_KINDS: &[&str] = &["pack"];
+
+/// The `workForm` twin of [`WRITER_SUPPLIED_DOMAIN_KINDS`], and for the same
+/// reason: a season pack is `serial`, an album release is `standalone`.
+pub const WRITER_SUPPLIED_WORK_FORM_KINDS: &[&str] = &["pack"];
+
+/// The inverse: every `contentKind` that belongs to `domain`.
+///
+/// Used at the gateway-routing boundary. A gateway declares the *content kinds*
+/// it serves (`served_content_kinds`), not domains, so a `domain:` filter is
+/// matched by expanding it here rather than by adding a `served_domains` field
+/// to the heartbeat — which would be a wire-protocol change requiring every
+/// feeder manifest and every peer to move at once.
+///
+/// ⚠ Since the `film`+`tv` merge this returns **several** kinds for `screen`
+/// (`movie`, `series`, `episode`) where the two old values were disjoint. A
+/// caller that reads only the first element silently means "movie" — scan the
+/// whole list, or narrow with a `workForm:` term first.
+pub fn content_kinds_for_domain(domain: &str) -> Vec<&'static str> {
+    inverse(DOMAIN_BY_CONTENT_KIND, domain)
+}
+
+/// The inverse of the `workForm` table. Same routing-boundary role, same
+/// scan-the-whole-list warning as [`content_kinds_for_domain`].
+pub fn content_kinds_for_work_form(work_form: &str) -> Vec<&'static str> {
+    inverse(WORK_FORM_BY_CONTENT_KIND, work_form)
+}
+
+fn inverse(table: &'static [(&'static str, &'static str)], value: &str) -> Vec<&'static str> {
+    let v = value.trim();
+    table
+        .iter()
+        .filter(|(_, val)| val.eq_ignore_ascii_case(v))
+        .map(|(kind, _)| *kind)
+        .collect()
+}
+
+/// Rewrite a query so a `domain:` / `workForm:` filter can reach a plugin that
+/// has never heard of either key.
+///
+/// Returns `None` when the query carries neither — the common path, and cheaper
+/// than cloning.
+///
+/// # Why translate instead of teaching every plugin
+///
+/// `record_matches` fails a record outright when a filter key is missing from
+/// its fields, and `domain` is missing from every record a not-yet-updated
+/// feeder emits (`workForm` from every record written before it existed).
+/// Most feeders pin the SDK by git tag, so they cannot pick up a new key on our
+/// schedule. Translating `domain:literature` into the `contentKind:`
+/// alternation it covers means the wire query a plugin sees never mentions
+/// either key — old plugins keep working unchanged, while the dispatcher still
+/// enforces the real filters afterwards, against records it has just stamped.
+///
+/// # The two axes intersect
+///
+/// `domain:screen workForm:serial` covers `series` and `episode` — the
+/// **intersection**, not the union. Widening to the union would hand a serial
+/// query the whole film corpus as a pre-filter; the post-stamp check would then
+/// throw it away, having already paid for it upstream.
+///
+/// An explicit `contentKind:` filter already in the query is left alone: it is
+/// the narrower of the three, and the `domain`/`workForm` terms are still
+/// enforced post-stamp, so the AND semantics hold either way.
+pub fn expand_routing_filters(
+    query: &crate::query::GatewayQuery,
+) -> Option<crate::query::GatewayQuery> {
+    let domains = query.filters.get("domain");
+    let work_forms = query.filters.get("workForm");
+    if domains.is_none() && work_forms.is_none() {
+        return None;
+    }
+
+    let by_domain = domains.map(|vs| expand_values(vs, content_kinds_for_domain));
+    let by_work_form = work_forms.map(|vs| expand_values(vs, content_kinds_for_work_form));
+
+    let mut out = query.clone();
+    out.filters.remove("domain");
+    out.filters.remove("workForm");
+    if out.filters.contains_key("contentKind") {
+        return Some(out);
+    }
+
+    let mut kinds: Vec<&'static str> = match (by_domain, by_work_form) {
+        (Some(d), Some(w)) => d.into_iter().filter(|k| w.contains(k)).collect(),
+        (Some(d), None) => d,
+        (None, Some(w)) => w,
+        (None, None) => unreachable!("guarded above"),
+    };
+    // Domain-ambiguous kinds come along whenever the expansion is non-empty
+    // — see [`WRITER_SUPPLIED_DOMAIN_KINDS`]. Skipped when nothing mapped,
+    // so an unknown value still expands to nothing and leaves the filter
+    // out entirely (below) rather than narrowing to `contentKind:pack`.
+    if !kinds.is_empty() {
+        for k in WRITER_SUPPLIED_DOMAIN_KINDS
+            .iter()
+            .chain(WRITER_SUPPLIED_WORK_FORM_KINDS.iter())
+        {
+            if !kinds.contains(k) {
+                kinds.push(k);
+            }
+        }
+    }
+    // An unknown value — or two axes with no kind in common — expands to
+    // nothing. Inserting an empty alternation would drop every record at the
+    // plugin; leaving the filter out lets the plugin answer and the post-stamp
+    // check do the (correct, empty) filtering.
+    if !kinds.is_empty() {
+        out.filters.insert(
+            "contentKind".to_string(),
+            kinds.into_iter().map(|k| k.to_string()).collect(),
+        );
+    }
+    Some(out)
+}
+
+/// Deprecated alias for [`expand_routing_filters`], kept so a caller pinned to
+/// an older SDK keeps compiling. It expands `workForm:` too — the name is the
+/// only thing that is stale.
+#[deprecated(note = "renamed to expand_routing_filters; it now expands workForm: as well")]
+pub fn expand_domain_filter(
+    query: &crate::query::GatewayQuery,
+) -> Option<crate::query::GatewayQuery> {
+    expand_routing_filters(query)
+}
+
+fn expand_values(
+    values: &[String],
+    expand: fn(&str) -> Vec<&'static str>,
+) -> Vec<&'static str> {
+    let mut out: Vec<&'static str> = Vec::new();
+    for v in values {
+        for k in expand(v) {
+            if !out.contains(&k) {
+                out.push(k);
+            }
+        }
+    }
+    out
+}
+
+/// Stamp `domain` on a field map that already carries a `contentKind`.
+///
+/// The convention from METADATA_KEYS.md §1: **whoever writes `contentKind`
+/// writes `domain` in the same breath**. An existing `domain` is never
+/// overwritten — a writer that knows better than the table (a `pack`, say) has
+/// already set it.
+pub fn stamp_domain(fields: &mut std::collections::BTreeMap<String, String>) {
+    stamp(fields, "domain", domain_for_content_kind);
+}
+
+/// Stamp `workForm` the same way, from the same `contentKind`. Call it
+/// wherever [`stamp_domain`] is called — the registry treats the two as one
+/// breath.
+pub fn stamp_work_form(fields: &mut std::collections::BTreeMap<String, String>) {
+    stamp(fields, "workForm", work_form_for_content_kind);
+}
+
+fn stamp(
+    fields: &mut std::collections::BTreeMap<String, String>,
+    key: &str,
+    derive: fn(&str) -> Option<&'static str>,
+) {
+    if fields.contains_key(key) {
+        return;
+    }
+    let Some(kind) = fields.get("contentKind").cloned() else {
+        return;
+    };
+    if let Some(value) = derive(&kind) {
+        fields.insert(key.to_string(), value.to_string());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::BTreeMap;
+
+    #[test]
+    fn kinds_map_to_their_domain() {
+        assert_eq!(domain_for_content_kind("movie"), Some("screen"));
+        assert_eq!(domain_for_content_kind("episode"), Some("screen"));
+        assert_eq!(domain_for_content_kind("track"), Some("music"));
+        // The case that proves medium and domain are different axes.
+        assert_eq!(domain_for_content_kind("musicVideo"), Some("music"));
+        // The case that proves the player does not decide the domain.
+        assert_eq!(domain_for_content_kind("audiobook"), Some("literature"));
+    }
+
+    #[test]
+    fn film_and_tv_are_one_domain_now() {
+        // The merge (METADATA_KEYS.md §14.17): meta-watch's wall is one term.
+        assert_eq!(
+            domain_for_content_kind("movie"),
+            domain_for_content_kind("series")
+        );
+        // ...and the split it used to carry lives here instead.
+        assert_eq!(work_form_for_content_kind("movie"), Some("standalone"));
+        assert_eq!(work_form_for_content_kind("series"), Some("serial"));
+        assert_eq!(work_form_for_content_kind("episode"), Some("serial"));
+    }
+
+    #[test]
+    fn work_form_is_not_a_screen_only_axis() {
+        // The point of the key: it means the same thing in every domain.
+        assert_eq!(work_form_for_content_kind("comic"), Some("serial"));
+        assert_eq!(work_form_for_content_kind("book"), Some("standalone"));
+        assert_eq!(work_form_for_content_kind("album"), Some("standalone"));
+    }
+
+    #[test]
+    fn the_spoken_domain_is_gone() {
+        // Retired unused. The kinds stay reserved with no domain until the PR
+        // that writes them picks one.
+        assert_eq!(domain_for_content_kind("podcast"), None);
+        assert_eq!(domain_for_content_kind("podcastEpisode"), None);
+        assert!(content_kinds_for_domain("spoken").is_empty());
+    }
+
+    #[test]
+    fn pack_has_no_derivable_domain_or_work_form() {
+        assert_eq!(domain_for_content_kind("pack"), None);
+        assert_eq!(work_form_for_content_kind("pack"), None);
+    }
+
+    #[test]
+    fn unknown_kind_is_none_not_a_guess() {
+        assert_eq!(domain_for_content_kind("hologram"), None);
+        assert_eq!(work_form_for_content_kind("hologram"), None);
+    }
+
+    #[test]
+    fn expansion_is_the_inverse() {
+        let music = content_kinds_for_domain("music");
+        assert!(music.contains(&"track"));
+        assert!(music.contains(&"musicVideo"));
+        assert!(!music.contains(&"audiobook"));
+        for kind in content_kinds_for_domain("literature") {
+            assert_eq!(domain_for_content_kind(kind), Some("literature"));
+        }
+        for kind in content_kinds_for_work_form("serial") {
+            assert_eq!(work_form_for_content_kind(kind), Some("serial"));
+        }
+    }
+
+    #[test]
+    fn screen_expands_to_every_rung_not_just_the_first() {
+        // The regression the merge creates for any caller reading `.first()`.
+        let screen = content_kinds_for_domain("screen");
+        assert!(screen.contains(&"movie"));
+        assert!(screen.contains(&"series"));
+        assert!(screen.contains(&"episode"));
+    }
+
+    fn q(filters: &[(&str, &[&str])]) -> crate::query::GatewayQuery {
+        crate::query::GatewayQuery {
+            raw_text: String::new(),
+            free_text: String::new(),
+            filters: filters
+                .iter()
+                .map(|(k, vs)| {
+                    (
+                        k.to_string(),
+                        vs.iter().map(|v| v.to_string()).collect::<Vec<_>>(),
+                    )
+                })
+                .collect(),
+            ranges: Vec::new(),
+            negations: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn no_routing_filter_is_left_untouched() {
+        assert!(expand_routing_filters(&q(&[("fileType", &["video"])])).is_none());
+    }
+
+    #[test]
+    fn domain_becomes_the_content_kinds_it_covers() {
+        let out = expand_routing_filters(&q(&[("domain", &["literature"])])).unwrap();
+        // The plugin never sees `domain:` — that is the whole point.
+        assert!(!out.filters.contains_key("domain"));
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"book".to_string()));
+        assert!(kinds.contains(&"audiobook".to_string()));
+        assert!(!kinds.contains(&"track".to_string()));
+    }
+
+    #[test]
+    fn work_form_alone_also_expands() {
+        let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
+        assert!(!out.filters.contains_key("workForm"));
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"episode".to_string()));
+        assert!(kinds.contains(&"comic".to_string()));
+        assert!(!kinds.contains(&"movie".to_string()));
+    }
+
+    #[test]
+    fn the_two_axes_intersect_they_do_not_union() {
+        let out = expand_routing_filters(&q(&[
+            ("domain", &["screen"][..]),
+            ("workForm", &["serial"][..]),
+        ]))
+        .unwrap();
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"series".to_string()));
+        assert!(kinds.contains(&"episode".to_string()));
+        // The union would have dragged the whole film corpus in.
+        assert!(!kinds.contains(&"movie".to_string()));
+        // A `serial` comic is in `literature`, not on meta-watch's wall.
+        assert!(!kinds.contains(&"comic".to_string()));
+    }
+
+    #[test]
+    fn an_explicit_content_kind_filter_wins() {
+        let out = expand_routing_filters(&q(&[
+            ("domain", &["music"][..]),
+            ("contentKind", &["track"][..]),
+        ]))
+        .unwrap();
+        assert_eq!(
+            out.filters.get("contentKind").map(Vec::as_slice),
+            Some(&["track".to_string()][..])
+        );
+    }
+
+    #[test]
+    fn expansion_carries_the_writer_supplied_kinds() {
+        // `pack` has no domain in the table (a season pack and an album are the
+        // same structural thing), so the writer stamps one. Leaving it out of
+        // the alternation would make a `domain:screen` pre-filter drop the whole
+        // torrent-pack corpus, which `record_matches` would then never see.
+        let out = expand_routing_filters(&q(&[("domain", &["screen"])])).unwrap();
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert!(kinds.contains(&"episode".to_string()));
+        assert!(kinds.contains(&"pack".to_string()));
+        // ...and it rides along on the `workForm` axis too, exactly once.
+        let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
+        let kinds = out.filters.get("contentKind").unwrap();
+        assert_eq!(kinds.iter().filter(|k| *k == "pack").count(), 1);
+    }
+
+    #[test]
+    fn unknown_value_adds_no_alternation() {
+        // An empty alternation would drop every record at the plugin; the
+        // post-stamp check does the (correct, empty) filtering.
+        let out = expand_routing_filters(&q(&[("domain", &["hologram"])])).unwrap();
+        // Not even `pack`: a lone `contentKind:pack` would be a NARROWING of a
+        // filter that should widen to nothing.
+        assert!(!out.filters.contains_key("contentKind"));
+    }
+
+    #[test]
+    fn two_axes_with_nothing_in_common_widen_rather_than_narrow() {
+        // No music kind is `serial`. The pre-filter drops out; the post-stamp
+        // check still enforces both terms.
+        let out = expand_routing_filters(&q(&[
+            ("domain", &["music"][..]),
+            ("workForm", &["serial"][..]),
+        ]))
+        .unwrap();
+        assert!(!out.filters.contains_key("contentKind"));
+    }
+
+    #[test]
+    fn stamp_fills_in_and_never_overwrites() {
+        let mut f: BTreeMap<String, String> = BTreeMap::new();
+        f.insert("contentKind".into(), "episode".into());
+        stamp_domain(&mut f);
+        stamp_work_form(&mut f);
+        assert_eq!(f.get("domain").map(String::as_str), Some("screen"));
+        assert_eq!(f.get("workForm").map(String::as_str), Some("serial"));
+
+        // A pack writer that already knows its context keeps it.
+        let mut p: BTreeMap<String, String> = BTreeMap::new();
+        p.insert("contentKind".into(), "pack".into());
+        p.insert("domain".into(), "music".into());
+        p.insert("workForm".into(), "standalone".into());
+        stamp_domain(&mut p);
+        stamp_work_form(&mut p);
+        assert_eq!(p.get("domain").map(String::as_str), Some("music"));
+        assert_eq!(p.get("workForm").map(String::as_str), Some("standalone"));
+
+        // No contentKind → nothing invented.
+        let mut n: BTreeMap<String, String> = BTreeMap::new();
+        stamp_domain(&mut n);
+        stamp_work_form(&mut n);
+        assert!(n.get("domain").is_none());
+        assert!(n.get("workForm").is_none());
+    }
+}

src/hash.rs

@@ -248,8 +248,12 @@ fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
 ///   merge (a TMDB and a MyAnimeList locator on one meta-core record) safe.
 pub const CARD_LOCATOR_CODEC: u64 = 0x1007;
 
-/// Digest byte ceiling for a card locator, matching meta-share's
-/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash.
+/// Digest byte ceiling for the locators that still carry one — `yt-video`
+/// (via [`kinded_locator_cid`]) and `ext-play`. Matches meta-share's
+/// `MAX_MULTIHASH_SIZE`, whose `CidGeneric<64>` rejects a longer multihash.
+///
+/// ⚠ **[`compute_card_cid`] is deliberately NOT subject to this** — see its
+/// doc. The name is kept because both remaining users were written against it.
 const CARD_LOCATOR_MAX_DIGEST: usize = 64;
 
 /// Encode a self-describing **card locator** CID from a metadata source and
@@ -270,8 +274,28 @@ const CARD_LOCATOR_MAX_DIGEST: usize = 64;
 /// operation driven by an explicit cross-source id mapping — never inferred
 /// here (see `meta-gateway/docs/others/card-tier-search.md` §9).
 ///
-/// Returns `None` when `source` is empty or the digest would overflow the
-/// multihash budget — the caller drops that card.
+/// # No length ceiling, deliberately
+///
+/// A CIDv1 digest length is a varint, so the wire format imposes no limit; the
+/// 64-byte figure elsewhere in this file is Rust's fixed-size `CidGeneric<64>`,
+/// which meta-share needs only because beetswap is
+/// `Behaviour<const MAX_MULTIHASH_SIZE, B>`. A card is never bitswap content
+/// (seeding is gated on sha2-256), so that ceiling does not apply — exactly the
+/// reasoning `METADATA_KEYS.md` §2.1 already records for the `url` locator,
+/// which ships URLs past it today. meta-core's Go `Decode` hand-parses varints
+/// with no cap and reads these correctly.
+///
+/// This is load-bearing for URL-shaped ids: a card keyed on a work's own
+/// address (`("weebcentral.com", "/series/…/some-long-slug")`) runs to ~88
+/// bytes, and capping would silently drop the card rather than mint it.
+///
+/// ⚠ The one consumer that still cares is meta-share's `cid_rank`, which parses
+/// through `MsCid::from_str`. A locator it cannot parse scores `RANK_UNKNOWN`
+/// instead of `RANK_LOCATOR`; harmless for a card (its record carries one CID,
+/// so it wins the canonical election at any rank) but pinned by a golden vector
+/// in `cid-rank-vectors.json` so the two implementations cannot drift unseen.
+///
+/// Returns `None` only when `source` or `id` is empty.
 pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
     if source.is_empty() || id.is_empty() {
         return None;
@@ -284,9 +308,6 @@ pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
     write_pb_varint(source_b.len() as u64, &mut digest);
     digest.extend_from_slice(source_b);
     digest.extend_from_slice(id_b);
-    if digest.len() > CARD_LOCATOR_MAX_DIGEST {
-        return None;
-    }
 
     // CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest]
     let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
@@ -423,6 +444,86 @@ fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
 /// `beetswap::incoming_stream`). The `.nzb` manifest travels separately, as an
 /// ordinary sha2-256 IPFS cid in the record's `manifest` field. See
 /// `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md` §5.
+/// Custom multicodec for the **url-locator** family.
+///
+/// A CID whose *bytes are a URL string* rather than a hash of the file. It lets
+/// a record reference externally-hosted bytes without storing a raw URL field:
+/// the URL is wrapped in a CID, so it flows through every cid-shaped code path
+/// (`cids/*` membership, `/api/file/{cid}/raw`, the reverse index) exactly like
+/// a content hash — honouring `METADATA_KEYS.md` rule #5, "every byte reference
+/// is a `cid`, never a URL".
+///
+/// Contrast `ext-play` (`0x1009`), which is wire-identical apart from the
+/// codec: `0x1006` means **"fetch these bytes once and seed them"**, while
+/// `0x1009` means "open this; there are no bytes for us, ever".
+///
+/// ⚠ **Never route a url-CID through a fixed-64-byte multihash type.** An
+/// archive download URL runs to ~75 bytes, past the 62-ish ceiling of
+/// meta-share's `CidGeneric<64>` bitswap alias. That is fine — it is never
+/// bitswap *content*, and the `/raw` dispatch decodes it off the raw string
+/// before the fixed-size codec dispatch runs — but a consumer that parses it as
+/// an `MsCid` will reject a perfectly good locator.
+pub const URL_LOCATOR_CODEC: u64 = 0x1006;
+
+/// Encode an `http(s)` URL as a `0x1006` identity-multihash CIDv1.
+///
+/// The identity multihash (`0x00`) means "the digest *is* the payload", so
+/// [`decode_url_cid`] recovers the URL verbatim with zero I/O.
+///
+/// `None` for anything that is not an absolute http(s) URL — a relative or
+/// malformed value would encode fine and then fail unresolvably at play time,
+/// which is a far more expensive way to find out.
+pub fn compute_url_cid(url: &str) -> Option<String> {
+    let url = url.trim();
+    if !(url.starts_with("http://") || url.starts_with("https://")) {
+        return None;
+    }
+    Some(identity_locator_cid(URL_LOCATOR_CODEC, url.as_bytes()))
+}
+
+/// The artwork locator for a `cover` / `photo` field.
+///
+/// A thin alias over [`compute_url_cid`] that names the *intent* at the call
+/// sites: artwork is referenced by a locator cid, never by a raw `*_url` field
+/// (METADATA_KEYS rule #5). `None` for a non-http(s) URL, in which case the
+/// caller writes no artwork field at all — better a card the gate hides than
+/// one carrying an unresolvable reference.
+pub fn artwork_locator(url: &str) -> Option<String> {
+    compute_url_cid(url)
+}
+
+/// Recover the URL from a `url` locator cid — the exact inverse of
+/// [`compute_url_cid`], with no I/O.
+///
+/// `None` if `cid` is not a well-formed `0x1006` identity-multihash CIDv1 over
+/// an http(s) URL.
+pub fn decode_url_cid(cid: &str) -> Option<String> {
+    let bytes = base32_lower_decode(cid.strip_prefix('b')?)?;
+    let mut i = 0usize;
+    let mut next = || -> Option<u64> {
+        let mut n = 0u64;
+        let mut shift = 0u32;
+        loop {
+            let b = *bytes.get(i)?;
+            i += 1;
+            n |= u64::from(b & 0x7f) << shift;
+            if b & 0x80 == 0 {
+                return Some(n);
+            }
+            shift += 7;
+            if shift > 63 {
+                return None;
+            }
+        }
+    };
+    if next()? != 1 || next()? != URL_LOCATOR_CODEC || next()? != 0x00 {
+        return None;
+    }
+    let len = next()? as usize;
+    let url = std::str::from_utf8(bytes.get(i..i + len)?).ok()?;
+    (url.starts_with("http://") || url.starts_with("https://")).then(|| url.to_string())
+}
+
 pub const NZB_POSTING_CODEC: u64 = 0x1003;
 
 /// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs.
@@ -987,19 +1088,88 @@ mod tests {
         assert_eq!(&digest[5..], b"tv:95479");
     }
 
-    /// Empty inputs are rejected rather than encoded as an ambiguous locator,
-    /// and an id past the 64-byte multihash budget (meta-share's
-    /// `CidGeneric<64>`) returns `None` so the caller drops the card instead of
-    /// emitting a cid that peer would refuse to parse.
+    /// Empty inputs are rejected rather than encoded as an ambiguous locator.
     #[test]
-    fn card_cid_rejects_empty_and_oversized() {
+    fn card_cid_rejects_empty_halves() {
         assert!(compute_card_cid("", "tv:1").is_none());
         assert!(compute_card_cid("tmdb", "").is_none());
-        let huge = "x".repeat(CARD_LOCATOR_MAX_DIGEST);
-        assert!(compute_card_cid("tmdb", &huge).is_none());
-        // Exactly at the ceiling still encodes.
-        let fits = "x".repeat(CARD_LOCATOR_MAX_DIGEST - 5); // varint(4) + "tmdb"
-        assert!(compute_card_cid("tmdb", &fits).is_some());
+    }
+
+    /// ⚠ A card locator has **no length ceiling**, and this is the test that
+    /// says so on purpose — it used to assert the opposite.
+    ///
+    /// The 64-byte figure is meta-share's `CidGeneric<64>` bitswap alias, and a
+    /// card is never bitswap content. Capping here would silently drop exactly
+    /// the cards a URL-shaped identity exists to mint: a `weebcentral.com` work
+    /// path digests to ~88 bytes. Same reasoning `METADATA_KEYS.md` §2.1
+    /// already records for the `url` locator.
+    #[test]
+    fn card_cid_has_no_length_ceiling() {
+        let long_id = "/series/01J76XYGF1RKSPWA4TQ6GGSVFV/\
+                       Ive-Been-Killing-Slimes-for-300-Years-The-Red-Dragon-Academy-for-Girls";
+        let cid = compute_card_cid("weebcentral.com", long_id).expect("no ceiling");
+        let (source, id) = split_card_digest(&cid);
+        assert!(
+            source.len() + id.len() > CARD_LOCATOR_MAX_DIGEST,
+            "fixture must actually exceed the old cap, got {}",
+            source.len() + id.len()
+        );
+        assert_eq!(source, "weebcentral.com");
+        assert_eq!(id, long_id);
+    }
+
+    /// The host-form identity: the source slot names the site that *owns* the
+    /// work, never the tool that reached it. Pinned as a literal because
+    /// `meta-read`'s `card_cid::encode` carries the mirrored assertion — the two
+    /// addressing one work differently is the failure the whole scheme cannot
+    /// survive.
+    #[test]
+    fn host_form_card_cid_golden_vector() {
+        let cid = compute_card_cid("mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512")
+            .expect("card cid");
+        assert_eq!(
+            cid,
+            "bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
+        );
+    }
+
+    /// `varint(source_len) ‖ source ‖ id` back out of a card locator, so the
+    /// tests above can assert on the halves without a decoder in this crate.
+    fn split_card_digest(cid: &str) -> (String, String) {
+        let raw = base32_lower_decode(&cid[1..]);
+        let mut i = 0usize;
+        let mut rd = || {
+            let (mut n, mut sh) = (0u64, 0u32);
+            loop {
+                let b = raw[i];
+                i += 1;
+                n |= u64::from(b & 0x7f) << sh;
+                if b & 0x80 == 0 {
+                    return n;
+                }
+                sh += 7;
+            }
+        };
+        assert_eq!(rd(), 1, "CIDv1");
+        assert_eq!(rd(), CARD_LOCATOR_CODEC);
+        assert_eq!(rd(), 0, "identity multihash");
+        let len = rd() as usize;
+        let digest = &raw[i..i + len];
+        let mut j = 0usize;
+        let (mut sl, mut sh) = (0usize, 0u32);
+        loop {
+            let b = digest[j];
+            j += 1;
+            sl |= usize::from(b & 0x7f) << sh;
+            if b & 0x80 == 0 {
+                break;
+            }
+            sh += 7;
+        }
+        (
+            String::from_utf8(digest[j..j + sl].to_vec()).unwrap(),
+            String::from_utf8(digest[j + sl..].to_vec()).unwrap(),
+        )
     }
 
     // ---- nzb-posting (0x1003) — the dedup contract -----------------------
@@ -1074,3 +1244,148 @@ mod tests {
         assert_eq!(&decoded[5..], &expect);
     }
 }
+
+/// RFC 4648 base32 (lowercase, no padding) → bytes. Inverse of
+/// [`base32_lower_no_padding`]; used by the identity-multihash decoders.
+fn base32_lower_decode(s: &str) -> Option<Vec<u8>> {
+    let mut acc = 0u16;
+    let mut bits = 0u32;
+    let mut out = Vec::with_capacity(s.len() * 5 / 8);
+    for c in s.bytes() {
+        let v = match c {
+            b'a'..=b'z' => c - b'a',
+            b'2'..=b'7' => c - b'2' + 26,
+            _ => return None,
+        };
+        acc = (acc << 5) | u16::from(v);
+        bits += 5;
+        if bits >= 8 {
+            bits -= 8;
+            out.push((acc >> bits) as u8);
+        }
+    }
+    Some(out)
+}
+
+#[cfg(test)]
+mod url_locator_wire_tests {
+    use super::*;
+
+    /// Decode a base32-lower multibase string back to bytes, so the tests can
+    /// assert the wire layout rather than an opaque string.
+    fn decode(cid: &str) -> Vec<u8> {
+        const A: &str = "abcdefghijklmnopqrstuvwxyz234567";
+        let mut bits = String::new();
+        for c in cid.strip_prefix('b').expect("multibase b").chars() {
+            let i = A.find(c).expect("base32 alphabet");
+            bits.push_str(&format!("{i:05b}"));
+        }
+        bits.as_bytes()
+            .chunks(8)
+            .filter(|c| c.len() == 8)
+            .map(|c| u8::from_str_radix(std::str::from_utf8(c).unwrap(), 2).unwrap())
+            .collect()
+    }
+
+    fn read_varint(data: &[u8], at: &mut usize) -> u64 {
+        let (mut v, mut shift) = (0u64, 0);
+        loop {
+            let b = data[*at];
+            *at += 1;
+            v |= ((b & 0x7f) as u64) << shift;
+            if b & 0x80 == 0 {
+                return v;
+            }
+            shift += 7;
+        }
+    }
+
+    /// Pins the on-the-wire layout `METADATA_KEYS.md` §2.1 specifies. meta-share
+    /// decodes exactly this; a change here is a change to a cross-binary
+    /// contract, not an implementation detail.
+    #[test]
+    fn the_wire_layout_matches_the_specified_encoding() {
+        let url = "https://archive.org/download/item/track.flac";
+        let cid = compute_url_cid(url).expect("cid");
+        let raw = decode(&cid);
+        let mut i = 0;
+        assert_eq!(raw[i], 0x01, "CIDv1");
+        i += 1;
+        assert_eq!(read_varint(&raw, &mut i), URL_LOCATOR_CODEC);
+        assert_eq!(raw[i], 0x00, "identity multihash");
+        i += 1;
+        let len = read_varint(&raw, &mut i) as usize;
+        assert_eq!(len, url.len());
+        assert_eq!(
+            std::str::from_utf8(&raw[i..i + len]).unwrap(),
+            url,
+            "the digest IS the payload — decoding must recover the URL verbatim"
+        );
+    }
+
+    /// ⚠ A real archive URL is ~75 bytes, past the ~62-byte ceiling of a
+    /// fixed-64 multihash type. That is expected and fine (the locator is never
+    /// bitswap content), but it is the thing that trips a consumer parsing it
+    /// as an `MsCid`, so the case is pinned deliberately rather than avoided.
+    #[test]
+    fn a_long_archive_url_still_encodes() {
+        let url = "https://archive.org/download/gd1977-05-08eaton/gd77-05-08eaton-d1t01.ogg";
+        assert!(url.len() > 62, "fixture must exercise the over-64 case");
+        let cid = compute_url_cid(url).expect("cid");
+        let raw = decode(&cid);
+        let mut i = 1;
+        read_varint(&raw, &mut i);
+        i += 1;
+        let len = read_varint(&raw, &mut i) as usize;
+        assert_eq!(std::str::from_utf8(&raw[i..i + len]).unwrap(), url);
+    }
+
+    /// Two URLs are two locators — a url-CID is not byte-derivable, so it is an
+    /// address this one record advertises, never a content identity.
+    #[test]
+    fn distinct_urls_yield_distinct_locators() {
+        assert_ne!(
+            compute_url_cid("https://example.com/a.flac"),
+            compute_url_cid("https://example.com/b.flac")
+        );
+    }
+
+    /// A relative or malformed value would encode fine and then fail
+    /// unresolvably at play time — the expensive place to find out.
+    #[test]
+    fn only_absolute_http_urls_are_accepted() {
+        assert!(compute_url_cid("/download/item/track.flac").is_none());
+        assert!(compute_url_cid("ftp://example.com/x").is_none());
+        assert!(compute_url_cid("").is_none());
+        assert!(compute_url_cid("archive.org/download/x").is_none());
+        assert!(compute_url_cid("http://example.com/x").is_some());
+    }
+}
+
+#[cfg(test)]
+mod roundtrip_tests {
+    use super::*;
+
+    /// Long URLs are the whole reason this locator exists (they overflow the
+    /// fixed-size multihash buffer the `cid` crate uses), so the round trip is
+    /// pinned at a realistic Cover Art Archive length, not a toy one.
+    #[test]
+    fn encode_then_decode_is_identity() {
+        for url in [
+            "https://coverartarchive.org/release/aa997ea0-2936-40bd-884d-3af8a0e064dc/12345-500.jpg",
+            "https://archive.org/services/img/gd1977-05-08",
+            "http://example.test/a.png",
+        ] {
+            let cid = compute_url_cid(url).expect("encodes");
+            assert!(cid.starts_with('b'));
+            assert_eq!(decode_url_cid(&cid).as_deref(), Some(url), "round trip: {url}");
+        }
+    }
+
+    #[test]
+    fn a_non_locator_cid_does_not_decode() {
+        // A card locator (0x1007) must not be mistaken for a url locator.
+        let card = crate::hash::compute_card_cid("musicbrainz", "abc").unwrap();
+        assert_eq!(decode_url_cid(&card), None);
+    }
+}

src/lib.rs

@@ -21,13 +21,16 @@
 /// `metacore-app` service the AppStore feeder apps ship on the shared network.
 pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
 
+pub mod budget;
 pub mod cache;
 pub mod common;
 pub mod config;
+pub mod domain;
 pub mod enrich;
 pub mod filename_meta;
 pub mod hash;
 pub mod lang;
+pub mod licence;
 pub mod meta_core;
 pub mod plugin;
 pub mod query;
@@ -35,7 +38,9 @@ pub mod query_eval;
 pub mod serve;
 pub mod types;
 
+pub use budget::RateBudget;
 pub use config::{ConfigField, ConfigSchema, FieldKind};
+pub use licence::licence_from_url;
 pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher};
 pub use meta_core::FeederStore;
 pub use plugin::{

src/licence.rs

@@ -0,0 +1,31 @@
+//! Licence-URL → SPDX-style identifier.
+//!
+//! Generic across upstreams: Internet Archive, Jamendo and Wikimedia Commons
+//! all publish a Creative Commons *URL*, while `METADATA_KEYS.md` requires an
+//! SPDX-style identifier — a consumer comparing licences should not be doing
+//! string surgery on `http://` vs `https://` and a trailing slash.
+
+/// `METADATA_KEYS.md` requires an SPDX-style identifier rather than a URL,
+/// because a consumer comparing licences should not be doing string surgery on
+/// `http://` vs `https://` and a trailing slash.
+pub fn licence_from_url(url: &str) -> Option<String> {
+    let u = url.trim().trim_end_matches('/').to_ascii_lowercase();
+    if u.is_empty() {
+        return None;
+    }
+    if u.contains("publicdomain/zero") {
+        return Some("CC0-1.0".to_string());
+    }
+    if u.contains("publicdomain/mark") {
+        return Some("PublicDomain".to_string());
+    }
+    // .../licenses/<variant>/<version>[/<jurisdiction>]
+    let rest = u.split("/licenses/").nth(1)?;
+    let mut parts = rest.split('/');
+    let variant = parts.next()?.to_ascii_uppercase();
+    let version = parts.next().unwrap_or("4.0");
+    if variant.is_empty() {
+        return None;
+    }
+    Some(format!("CC-{variant}-{version}"))
+}