rework domain field

crates/meta-feeder-sdk/src/cache.rs

@@ -53,6 +53,13 @@ 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");
+/// 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
+/// detected as a change on the next boot. Keys are plugin-namespaced
+/// (`prowlarr:indexer-set`). redb creates the table on open, so an older
+/// `.redb` file gains it transparently.
+const MISC_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("misc");
 
 /// Per-plugin midhash cache. Cheap to clone (the inner `Database` is shared
 /// via `Arc`).
@@ -105,6 +112,7 @@ impl MidhashCache {
             tx.open_table(TMDB_PRINCIPAL_TOPN_TABLE)?;
             tx.open_table(SUBTITLES_TABLE)?;
             tx.open_table(OPENSUBTITLES_TABLE)?;
+            tx.open_table(MISC_TABLE)?;
             tx.commit()?;
         }
         Ok(Self { db: Arc::new(db) })
@@ -112,6 +120,14 @@ impl MidhashCache {
 
     str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE);
 
+    str_table_accessors!(
+        /// Read a durable plugin scratch value (see [`MISC_TABLE`]).
+        get_misc,
+        /// Write a durable plugin scratch value (see [`MISC_TABLE`]).
+        put_misc,
+        MISC_TABLE,
+    );
+
     pub fn entry_count(&self) -> Result<u64, redb::Error> {
         let tx = self.db.begin_read()?;
         let table = tx.open_table(MIDHASH_TABLE)?;

crates/meta-feeder-sdk/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());
+    }
+}

crates/meta-feeder-sdk/src/hash.rs

@@ -183,9 +183,9 @@ const NZB_RELEASE_MAX_DIGEST: usize = 64;
 ///
 /// Returns `None` when the id isn't hex or the digest would overflow the
 /// multihash budget (an oversized host) — the caller drops that row.
-pub fn compute_nzb_release_cid(host: &str, release_id: &str) -> Option<String> {
+pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option<String> {
     let id_bytes = decode_hex_id(release_id)?;
-    let host_b = host.as_bytes();
+    let host_b = api_base.as_bytes();
 
     // identity-multihash digest = varint(host_len) ‖ host ‖ id_bytes
     let mut digest = Vec::with_capacity(2 + host_b.len() + id_bytes.len());
@@ -298,6 +298,117 @@ pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
     Some(format!("b{}", base32_lower_no_padding(&wire)))
 }
 
+// ---------------------------------------------------------------------------
+// Delegated-playback locators — `yt-video` (0x1008) and `ext-play` (0x1009)
+// ---------------------------------------------------------------------------
+//
+// These two are the limit case *beyond* `card` (0x1007). `card` addresses a
+// work with no bytes anywhere; these address a **rendition whose bytes exist
+// but are permanently someone else's**. An external player renders the media,
+// nothing is transported, nothing is content-addressed, nothing is re-seedable.
+//
+// See `docs/cid-formats.md` §7 and
+// `docs/study/listenbrainz-youtube-playback-tier-2026-08-21.md`.
+
+/// Custom multicodec for a **YouTube rendition** — delegated playback.
+///
+/// Its own codec rather than a `url` (`0x1006`) for three reasons, in
+/// descending order of how much trouble the alternative causes:
+///
+/// * **`0x1006` means "fetch these bytes once and seed them".** Point a `url`
+///   locator at a `watch` page and meta-share does exactly that: fetches the
+///   HTML, chunks it, seeds it, and links the resulting content CID back onto
+///   the record *as the file*. That is the MSR1 poisoning shape, except
+///   permanent and replicated across peers. A resolved `googlevideo` URL is no
+///   better — time-limited and IP-bound, so it is not a locator at all.
+/// * **The id is canonical, a URL is not.** `youtube.com` / `music.youtube.com`
+///   / `youtu.be` all address the same video, and query parameters are tracking
+///   noise, so the same rendition would mint several different CIDs. Being a
+///   pure function of `(kind, id)`, this codec makes two peers that resolve the
+///   same track converge on one meta-core record for free, via the
+///   `cids/<bareCid>` reverse index — the property that makes [`compute_card_cid`]
+///   useful, for the same reason.
+/// * **It is tiny** — 11 bytes of id plus a short kind prefix.
+///
+/// ⚠ **An immutable id is not durable playback.** Takedowns, geo-blocks and
+/// embed-disabled uploads all kill a reference whose CID stays valid forever —
+/// the same rot class as [`NZB_RELEASE_CODEC`]. The mitigation is at the
+/// *record* level (carry several ranked references per recording so a client
+/// can fall through), never at the CID level.
+pub const YT_VIDEO_CODEC: u64 = 0x1008;
+
+/// Custom multicodec for an **external page to open, never fetch**.
+///
+/// Wire-identical to `url` (`0x1006`) apart from the codec, and that difference
+/// is the entire contract: `0x1006` resolves cache-through and lands in the
+/// blockstore, `0x1009` **must be refused by the byte path** and only ever
+/// surfaces as an "Open on …↗" affordance. Use it for anything worth linking
+/// but not worth a per-provider codec (Bandcamp, Jamendo, an arbitrary stream
+/// page).
+pub const EXT_PLAY_CODEC: u64 = 0x1009;
+
+/// Encode a [`YT_VIDEO_CODEC`] locator from a `kind` and a bare YouTube id.
+///
+/// `kind` is `"video"` | `"playlist"` | `"channel"`; `id` is the bare
+/// identifier (`4D7u5KF7SP8`, `OLAK5uy_…`, `UCRr1xG_2WIDs18a6cIiCxeA`) — **not**
+/// a URL, and not a `watch?v=` fragment. The framing is byte-identical to
+/// [`compute_card_cid`] on purpose, so each of the seven rank/decode
+/// implementations extends a parser it already wrote rather than learning a
+/// novel shape.
+///
+/// Returns `None` on an empty field or a digest past the multihash budget —
+/// the caller drops that reference.
+pub fn compute_yt_video_cid(kind: &str, id: &str) -> Option<String> {
+    kinded_locator_cid(YT_VIDEO_CODEC, kind, id)
+}
+
+/// Encode an [`EXT_PLAY_CODEC`] locator from an `http(s)` URL.
+///
+/// The scheme check is not cosmetic: it is what keeps a `javascript:` or
+/// `data:` payload from ever reaching a client that will hand this string to
+/// an anchor or a new tab.
+pub fn compute_ext_play_cid(url: &str) -> Option<String> {
+    if !(url.starts_with("http://") || url.starts_with("https://")) {
+        return None;
+    }
+    let digest = url.as_bytes();
+    if digest.is_empty() || digest.len() > CARD_LOCATOR_MAX_DIGEST {
+        return None;
+    }
+    Some(identity_locator_cid(EXT_PLAY_CODEC, digest))
+}
+
+/// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used
+/// by [`compute_card_cid`] and [`compute_yt_video_cid`].
+fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option<String> {
+    if kind.is_empty() || id.is_empty() {
+        return None;
+    }
+    let kind_b = kind.as_bytes();
+    let id_b = id.as_bytes();
+
+    let mut digest = Vec::with_capacity(2 + kind_b.len() + id_b.len());
+    write_pb_varint(kind_b.len() as u64, &mut digest);
+    digest.extend_from_slice(kind_b);
+    digest.extend_from_slice(id_b);
+    if digest.len() > CARD_LOCATOR_MAX_DIGEST {
+        return None;
+    }
+    Some(identity_locator_cid(codec, &digest))
+}
+
+/// CIDv1 with an *identity* multihash:
+/// `[version=0x01][codec varint][mh=0x00][len varint][digest]`, base32lower.
+fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01);
+    write_pb_varint(codec, &mut wire);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(digest);
+    format!("b{}", base32_lower_no_padding(&wire))
+}
+
 /// Custom multicodec for a **Usenet posting** identity (`nzb-posting`),
 /// minted from the article Message-IDs of a release we scanned ourselves.
 ///
@@ -672,6 +783,82 @@ mod tests {
         assert_eq!(base32_lower_no_padding(b"foobar"), "mzxw6ytboi");
     }
 
+    #[test]
+    fn yt_video_cid_matches_golden_vector() {
+        // Pinned by `/cid-rank-vectors.json`'s `yt-video-locator` entry. The id
+        // is the real resolver output for Daft Punk / "Get Lucky" measured in
+        // the delegated-playback study §4.3.
+        assert_eq!(
+            compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap(),
+            "bagecaaarav3gszdfn42ein3vgvfumn2tka4a"
+        );
+    }
+
+    #[test]
+    fn yt_video_cid_framing_matches_card_locator() {
+        // The two codecs share the `varint(len(kind)) || kind || id` framing on
+        // purpose, so every decoder extends a parser it already wrote. If this
+        // ever fails, the two shapes have drifted and §7.2 of docs/cid-formats.md
+        // is no longer true.
+        let yt = compute_yt_video_cid("tmdb", "tv:95479").unwrap();
+        let card = compute_card_cid("tmdb", "tv:95479").unwrap();
+        // Same digest, different codec => same length, differing only in the
+        // codec varint region.
+        assert_ne!(yt, card);
+        assert_eq!(yt.len(), card.len());
+    }
+
+    #[test]
+    fn yt_video_cid_is_a_pure_function_of_kind_and_id() {
+        // Convergence is the reason this is a codec and not a URL: two peers
+        // that resolve the same video must derive the same address offline.
+        assert_eq!(
+            compute_yt_video_cid("video", "4D7u5KF7SP8"),
+            compute_yt_video_cid("video", "4D7u5KF7SP8")
+        );
+        // ...and the kind is part of the identity, not decoration.
+        assert_ne!(
+            compute_yt_video_cid("video", "OLAK5uy_kZ8Xq"),
+            compute_yt_video_cid("playlist", "OLAK5uy_kZ8Xq")
+        );
+    }
+
+    #[test]
+    fn yt_video_cid_rejects_empty_fields() {
+        assert!(compute_yt_video_cid("", "4D7u5KF7SP8").is_none());
+        assert!(compute_yt_video_cid("video", "").is_none());
+    }
+
+    #[test]
+    fn ext_play_cid_matches_golden_vector() {
+        assert_eq!(
+            compute_ext_play_cid("https://example.bandcamp.com/album/x").unwrap(),
+            "bagesaabenb2hi4dthixs6zlymfwxa3dffzrgc3temnqw24bomnxw2l3bnrrhk3jppa"
+        );
+    }
+
+    #[test]
+    fn ext_play_cid_rejects_non_http_schemes() {
+        // Load-bearing: a client hands this string to an anchor or a new tab,
+        // so a `javascript:` or `data:` payload must never mint a locator.
+        assert!(compute_ext_play_cid("javascript:alert(1)").is_none());
+        assert!(compute_ext_play_cid("data:text/html,<script>").is_none());
+        assert!(compute_ext_play_cid("ftp://example.com/x").is_none());
+        assert!(compute_ext_play_cid("").is_none());
+    }
+
+    #[test]
+    fn ext_play_cid_differs_from_the_url_locator_for_the_same_url() {
+        // The whole contract lives in the codec slot: `0x1006` means "fetch and
+        // seed these bytes", `0x1009` means "open this, there are no bytes for
+        // us". Wire-identical otherwise — which is exactly why a decoder that
+        // matched on *shape* would seed a Bandcamp page as the file.
+        let url = "https://example.bandcamp.com/album/x";
+        let ext = compute_ext_play_cid(url).unwrap();
+        let card = compute_card_cid("x", url).unwrap();
+        assert_ne!(ext, card);
+    }
+
     #[test]
     fn ipfs_cid_single_leaf_matches_kubo() {
         assert_eq!(

crates/meta-feeder-sdk/src/lib.rs

@@ -24,6 +24,7 @@ pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
 pub mod cache;
 pub mod common;
 pub mod config;
+pub mod domain;
 pub mod enrich;
 pub mod filename_meta;
 pub mod hash;

crates/meta-feeder-sdk/src/meta_core.rs

@@ -184,6 +184,48 @@ impl FeederStore {
         &self.meta_core_url
     }
 
+    /// Raw keys containing `contains` (`GET /api/kv/search`). Returns
+    /// `(keys, truncated)`; `truncated` means the server capped the result and
+    /// there may be more.
+    ///
+    /// This is a keyspace scan, so it belongs on rare control-plane paths only —
+    /// today, invalidating the gateway's search-coverage markers when a feeder's
+    /// upstream source set changes. Never call it per search.
+    pub async fn search_keys(&self, contains: &str, limit: usize) -> anyhow::Result<(Vec<String>, bool)> {
+        let url = format!("{}/api/kv/search", self.meta_core_url);
+        let resp = self
+            .http
+            .get(&url)
+            .query(&[("contains", contains), ("limit", &limit.to_string())])
+            .send()
+            .await?;
+        let status = resp.status();
+        if !status.is_success() {
+            anyhow::bail!("meta-core KV search `{contains}` returned {status}");
+        }
+        #[derive(Deserialize)]
+        struct KvSearchResponse {
+            #[serde(default)]
+            keys: Vec<String>,
+            #[serde(default)]
+            truncated: bool,
+        }
+        let body: KvSearchResponse = resp.json().await?;
+        Ok((body.keys, body.truncated))
+    }
+
+    /// Delete one raw KV key (`DELETE /api/kv/value?key=`). A missing key is not
+    /// an error.
+    pub async fn delete_kv(&self, key: &str) -> anyhow::Result<()> {
+        let url = format!("{}/api/kv/value", self.meta_core_url);
+        let resp = self.http.delete(&url).query(&[("key", key)]).send().await?;
+        let status = resp.status();
+        if status.is_success() || status == reqwest::StatusCode::NOT_FOUND {
+            return Ok(());
+        }
+        anyhow::bail!("meta-core KV DELETE {key} returned {status}");
+    }
+
     /// Self-publish a **metadata-only** outcome to meta-core, returning the
     /// **sparse** outcome (`record: None`) so the gateway's `(_, None)` branch
     /// skips it — no double write. Byte outcomes and already-sparse outcomes are

crates/meta-feeder-sdk/src/plugin.rs

@@ -74,6 +74,13 @@ pub struct HashOutcome {
 ///   digest is over the id set, not over any block, so the `.nzb` manifest
 ///   travels separately as an ordinary sha2-256 cid in the record's `manifest`
 ///   field. Produced by `hash::compute_nzb_posting_cid`.
+/// - `YtVideo` — a **delegated-playback** reference, custom multicodec
+///   `yt-video` `0x1008`. Like `CardLocator` it is always paired with
+///   `bytes: None` and lands in the `(None, Some)` metadata-only branch, but it
+///   means something stronger than "no bytes yet": the bytes **exist and are
+///   permanently someone else's**. Nothing is fetched, seeded or
+///   content-addressed, and meta-share's byte path REFUSES the cid with a `400`
+///   rather than 404ing. Produced by `hash::compute_yt_video_cid`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum HashKind {
     /// Reserved: no feeder currently emits midhash256 outcomes, but the family
@@ -85,6 +92,7 @@ pub enum HashKind {
     NzbRelease,
     CardLocator,
     NzbPosting,
+    YtVideo,
 }
 
 /// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -121,6 +129,33 @@ pub fn upstream_id_field(upstream_id: &str) -> String {
     format!("{upstream_id}id")
 }
 
+/// Key-set prefix for the provenance `source/<label>` field (METADATA_KEYS §5).
+/// Each member is its own hash field (`source/gateway:nyaa.si = "true"`) so two
+/// producers that reach the same content by different routes union without a
+/// last-writer-wins clobber. The `<label>` uses `:` internally (not `/`, which
+/// is the key-set path separator) — e.g. `gateway:nyaa.si`, `gateway:tribler`.
+pub const SOURCE_KEYSET_PREFIX: &str = "source/";
+
+/// Stamp the SDK's default provenance member — `source/gateway:<upstream_id>` —
+/// onto a record's fields, but **only if it carries no `source/*` member yet**.
+/// This gives every gateway hit a provenance label for free (so meta-watch never
+/// falls back to "unknown" for a feeder record), while a plugin that knows a
+/// finer origin (torznab → the specific Prowlarr indexer) stamps its own flat
+/// `source/*` facets first (`source/gateway`, `source/<protocol>`,
+/// `source/<indexer-slug>`) and this default then no-ops.
+pub fn stamp_default_source(
+    fields: &mut std::collections::BTreeMap<String, String>,
+    upstream_id: &str,
+) {
+    if fields.keys().any(|k| k.starts_with(SOURCE_KEYSET_PREFIX)) {
+        return;
+    }
+    fields.insert(
+        format!("{SOURCE_KEYSET_PREFIX}gateway:{upstream_id}"),
+        "true".to_string(),
+    );
+}
+
 /// Static plugin contract. Each enabled `upstream_id` is implemented by
 /// exactly one `Box<dyn FeederPlugin>`. Lives behind `&self` in steady state —
 /// called concurrently, owns any interior mutability for caches.

crates/meta-feeder-sdk/src/serve.rs

@@ -57,6 +57,7 @@ pub enum HashKindDto {
     NzbRelease,
     CardLocator,
     NzbPosting,
+    YtVideo,
 }
 
 impl From<HashKind> for HashKindDto {
@@ -68,6 +69,7 @@ impl From<HashKind> for HashKindDto {
             HashKind::NzbRelease => HashKindDto::NzbRelease,
             HashKind::CardLocator => HashKindDto::CardLocator,
             HashKind::NzbPosting => HashKindDto::NzbPosting,
+            HashKind::YtVideo => HashKindDto::YtVideo,
         }
     }
 }
@@ -81,6 +83,7 @@ impl From<HashKindDto> for HashKind {
             HashKindDto::NzbRelease => HashKind::NzbRelease,
             HashKindDto::CardLocator => HashKind::CardLocator,
             HashKindDto::NzbPosting => HashKind::NzbPosting,
+            HashKindDto::YtVideo => HashKind::YtVideo,
         }
     }
 }

feeder-plugin/card-feeder/src/card.rs

@@ -181,7 +181,22 @@ impl Card {
         // contentKind axis stay orthogonal, so `fileType:card contentKind:series`
         // is a well-formed query and routing needs no special case.
         fields.insert("fileType".to_string(), "card".to_string());
-        fields.insert("contentKind".to_string(), self.content_kind().to_string());
+        let content_kind = self.content_kind();
+        fields.insert("contentKind".to_string(), content_kind.to_string());
+        // Third and fourth axes: which app the card is destined for, and what
+        // shape of work it is. `fileType=card` says "no bytes", `contentKind`
+        // says which work, `domain` says who wants it — meta-watch's wall is a
+        // `domain:screen` query — and `workForm` says film-or-serial, the split
+        // `domain` stopped carrying when film and tv merged (METADATA_KEYS.md
+        // §1, §14.17).
+        if let Some(domain) = meta_feeder_sdk::domain::domain_for_content_kind(content_kind) {
+            fields.insert("domain".to_string(), domain.to_string());
+        }
+        if let Some(work_form) =
+            meta_feeder_sdk::domain::work_form_for_content_kind(content_kind)
+        {
+            fields.insert("workForm".to_string(), work_form.to_string());
+        }
         fields.insert("title".to_string(), self.title.clone());
 
         // The id bag.