align with metadata standard

.gitattributes

@@ -0,0 +1,6 @@
+# FIX CRLF always when developer has not set
+# Linux/Mac: git config --global core.autocrlf input
+# Windows: git config --global core.autocrlf true
+# Auto detect text files and perform LF normalization
+* text=auto
+* eol=lf

.github/workflows/publish.yml

@@ -1,77 +1,59 @@
-name: Build and Publish Docker Image
+name: Publish image
 
+# Build and push the feeder image to the GitHub Container Registry (ghcr.io):
+#   * every push to main  -> the moving `main` tag
+#   * every `v*` tag       -> semver tags (e.g. v1.0.0 -> 1.0.0, 1.0, 1)
+# The build context is the repo root (the workspace) so the vendored
+# `meta-feeder-sdk` path dependency resolves; the Dockerfile lives under
+# feeder-plugin/tribler-feeder/.
 on:
   push:
-    branches:
-      - main
-    tags:
-      - 'v*'
-  pull_request:
-    branches:
-      - main
-  workflow_dispatch:
+    branches: [main]
+    tags: ['v*']
 
 env:
   REGISTRY: ghcr.io
   IMAGE_NAME: ${{ github.repository }}
 
 jobs:
-  build-and-push:
+  publish:
     runs-on: ubuntu-latest
     permissions:
       contents: read
       packages: write
-
     steps:
-      - name: Checkout repository
+      - name: Checkout
         uses: actions/checkout@v4
 
-      - name: Free disk space
-        uses: jlumbroso/free-disk-space@main
-        with:
-          tool-cache: false
-          android: true
-          dotnet: true
-          haskell: true
-          large-packages: true
-          docker-images: true
-          swap-storage: true
-
-      - name: Set up QEMU
-        uses: docker/setup-qemu-action@v3
-
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@v3
 
-      - name: Log in to Container Registry
+      - name: Log in to GitHub Container Registry
         uses: docker/login-action@v3
         with:
           registry: ${{ env.REGISTRY }}
           username: ${{ github.actor }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: Extract metadata (tags, labels)
+      - name: Extract image metadata
         id: meta
         uses: docker/metadata-action@v5
         with:
           images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
           tags: |
             type=ref,event=branch
-            type=ref,event=pr
-            type=sha
             type=semver,pattern={{version}}
             type=semver,pattern={{major}}.{{minor}}
             type=semver,pattern={{major}}
-            type=raw,value=latest,enable={{is_default_branch}}
+            type=sha,format=short
 
-      - name: Build and push Docker image
+      - name: Build and push
         uses: docker/build-push-action@v6
         with:
           context: .
-          platforms: linux/amd64,linux/arm64
+          file: ./feeder-plugin/tribler-feeder/Dockerfile
           push: true
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
-          provenance: false
           cache-from: type=gha
           cache-to: type=gha,mode=max

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/enrich.rs

@@ -107,7 +107,7 @@ impl EnrichmentConfig {
             filename_parser_url: env_nonempty("FILENAME_PARSER_URL")
                 .or_else(|| derive_sidecar_url("filename-parser")),
             tmdb_url: env_nonempty("TMDB_PLUGIN_URL").or_else(|| derive_sidecar_url("tmdb")),
-            tmdb_token: env_nonempty("TMDB_TOKEN").or_else(|| env_nonempty("PROWLARR_TMDB_TOKEN")),
+            tmdb_token: env_nonempty("TMDB_TOKEN").or_else(|| env_nonempty("TORZNAB_TMDB_TOKEN")),
             tmdb_language: env_nonempty("TMDB_LANGUAGE").unwrap_or_else(|| "en-US".to_string()),
             gateway_peer: env_nonempty("META_GATEWAY_PEER_ID")
                 .or_else(|| env_nonempty("HOSTNAME"))

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

@@ -160,31 +160,329 @@ fn bt_file_cid(codec: u64, infohash: &[u8], file_index: u64) -> String {
     format!("b{}", base32_lower_no_padding(&wire))
 }
 
-/// Custom multicodec for a generic **Newznab release** locator (`nzb-release`).
-/// MetaMesh-private, adjacent to the torrent-file codecs (`0x1002` is taken by
-/// [`BTIH_V2_FILE_CODEC`]). It identifies an indexer *listing* (host + release
-/// id), so it is derivable at search time with **no `.nzb` download**. See
-/// `USENET-GATEWAY-STUDY.md` §13. (The v1 content-stable `nzb-posting` `0x1003`
-/// variant — `sha256(message-ids)` — was removed; only this one is emitted.)
-pub const NZB_RELEASE_CODEC: u64 = 0x1004;
+/// Custom multicodec for a self-describing **Newznab release** locator
+/// (`nzb-release`). MetaMesh-private, adjacent to the torrent-file codecs in the
+/// `0x10xx` range. Unlike a content hash, the cid *embeds* the indexer host +
+/// release id in an identity multihash, so the credentialed meta-share peer
+/// decodes `{host,id}` straight from the cid and grabs the `.nzb` via `t=get`
+/// only at playback — no side-table, no `.nzb` download at search time. (The
+/// `0x1004` `sha256(host‖id)` opaque hash and the `0x1003` `nzb-posting`
+/// variant were both removed; only this self-describing form is emitted.)
+pub const NZB_RELEASE_CODEC: u64 = 0x1005;
 
-/// Encode an `nzb-release` CID from a Newznab indexer's host + bare release id
-/// (the 32-hex `<guid>` id). `sha2-256(host ‖ "\n" ‖ id)` — host-namespaced so
-/// the same movie on two indexers (different postings) yields different cids.
-/// Deterministic + opaque, same wire house-style as the other locator codecs.
-pub fn compute_nzb_release_cid(host: &str, release_id: &str) -> String {
-    let mut hasher = Sha256::new();
-    hasher.update(host.as_bytes());
-    hasher.update(b"\n");
-    hasher.update(release_id.as_bytes());
-    let digest: [u8; 32] = hasher.finalize().into();
+/// Digest byte ceiling for an `nzb-release` cid, matching meta-share's
+/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash.
+const NZB_RELEASE_MAX_DIGEST: usize = 64;
 
-    // CIDv1: [version=0x01][codec varint][multihash code varint][len=0x20][digest]
-    let mut wire = Vec::with_capacity(1 + 4 + 1 + 32);
+/// Encode a self-describing `nzb-release` CID from a Newznab indexer's host +
+/// bare release id (the hex `<guid>` id). The multihash is *identity* (`0x00`)
+/// and its digest embeds the locator as `varint(host_len) ‖ host ‖ id_bytes`,
+/// where `id_bytes` is the hex-decoded id. meta-share's `decode_nzb_release_cid`
+/// is the exact inverse. Host-namespaced: the same release on two indexers
+/// yields different cids.
+///
+/// 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(api_base: &str, release_id: &str) -> Option<String> {
+    let id_bytes = decode_hex_id(release_id)?;
+    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());
+    write_pb_varint(host_b.len() as u64, &mut digest);
+    digest.extend_from_slice(host_b);
+    digest.extend_from_slice(&id_bytes);
+    if digest.len() > NZB_RELEASE_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());
     wire.push(0x01);
     write_pb_varint(NZB_RELEASE_CODEC, &mut wire);
-    write_pb_varint(NZB_RELEASE_CODEC, &mut wire);
-    wire.push(0x20);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(&digest);
+    Some(format!("b{}", base32_lower_no_padding(&wire)))
+}
+
+/// Hex-decode a Newznab release id. The guid parser only keeps hex runs, so the
+/// id is all-hex by construction; odd-length ids are left-padded with a `0`
+/// nibble. Returns `None` on any non-hex byte (defensive).
+fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
+    let padded;
+    let s: &str = if id.len() % 2 == 1 {
+        padded = format!("0{id}");
+        &padded
+    } else {
+        id
+    };
+    let bytes = s.as_bytes();
+    let mut out = Vec::with_capacity(bytes.len() / 2);
+    for pair in bytes.chunks(2) {
+        let hi = (pair[0] as char).to_digit(16)?;
+        let lo = (pair[1] as char).to_digit(16)?;
+        out.push(((hi << 4) | lo) as u8);
+    }
+    Some(out)
+}
+
+/// Custom multicodec for a **card locator** — the identity of a *work* (a
+/// series, a film) as published by a metadata bridge, rather than of any
+/// bytes. MetaMesh-private, the next free slot after `0x1006` url. (`0x1004`
+/// nzb-release-hash stays **retired — do not reuse**. `0x1003` was retired too,
+/// but has since been **revived** for the self-scanned Usenet posting identity —
+/// see [`NZB_POSTING_CODEC`] and `METADATA_KEYS.md` §2.)
+///
+/// This is the first CID family in the platform where **no bytes exist
+/// anywhere, ever** — `0x1005`/`0x1006` are locators too, but both eventually
+/// resolve to bytes (NNTP, HTTP), and `0x1000` midhash256 is a digest over
+/// real file bytes. A card is pure identity: the record *is* the payload, and
+/// it lives in meta-core. Consequences, all already handled:
+///
+/// * Bitswap seeding is gated on `HashKind::Sha2_256` (gateway invariant 6),
+///   so a card is never offered to the swarm — nothing to seed.
+/// * meta-share's byte path 404s for it, which is correct.
+/// * It ranks `RANK_LOCATOR` (5), so it can never outrank a real digest for
+///   `canonical_cid` — which is what makes the future "one card, two CIDs"
+///   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.
+const CARD_LOCATOR_MAX_DIGEST: usize = 64;
+
+/// Encode a self-describing **card locator** CID from a metadata source and
+/// that source's own id for the work (`("tmdb", "tv:95479")`,
+/// `("mal", "52991")`). The multihash is *identity* (`0x00`) and its digest
+/// embeds the locator as `varint(source_len) ‖ source ‖ id`, both UTF-8 —
+/// exactly the shape [`compute_nzb_release_cid`] uses, minus the hex decode
+/// (a card id is arbitrary text, not a hex release id).
+///
+/// **The CID is a pure function of `(source, id)`**, which is the whole point:
+/// any peer that knows a tmdb id can derive the card's address offline and
+/// look it up locally, with no discovery round-trip. It also means the same
+/// card discovered by two different peers converges on one meta-core record
+/// for free, via the `cids/<bareCid>` reverse index.
+///
+/// Source-namespaced: the same work on TMDB and on MyAnimeList yields
+/// different cids. Merging those two into one record is a meta-core alias
+/// 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.
+pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
+    if source.is_empty() || id.is_empty() {
+        return None;
+    }
+    let source_b = source.as_bytes();
+    let id_b = id.as_bytes();
+
+    // identity-multihash digest = varint(source_len) ‖ source ‖ id
+    let mut digest = Vec::with_capacity(2 + source_b.len() + id_b.len());
+    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());
+    wire.push(0x01);
+    write_pb_varint(CARD_LOCATOR_CODEC, &mut wire);
+    wire.push(0x00); // multihash code: identity
+    write_pb_varint(digest.len() as u64, &mut wire);
+    wire.extend_from_slice(&digest);
+    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.
+///
+/// Unlike [`NZB_RELEASE_CODEC`] (`0x1005`), this is **not** a locator: it
+/// embeds no indexer host, so any peer with a plain NNTP provider can fetch the
+/// articles — no `IndexerCred`, no `t=get` grab. That portability is the whole
+/// point of running our own header scanner.
+///
+/// It is also **not** fetchable by cid: the digest is over the Message-ID set,
+/// not over any block's bytes, so bitswap can never satisfy a want for it (the
+/// receiver derives a block's cid by hashing the block — see
+/// `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.
+pub const NZB_POSTING_CODEC: u64 = 0x1003;
+
+/// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs.
+///
+/// # The normalisation rule IS the dedup contract
+///
+/// Mirrored verbatim in meta-share (`nzb/manifest.rs`), which re-derives this
+/// digest from the fetched manifest and rejects a mismatch. **Any change here
+/// must land on both sides together** — a drift silently makes every posting
+/// unplayable. The rule:
+///
+/// 1. Trim each id, and strip a surrounding `<…>` pair if present (NZB
+///    `<segment>` bodies carry the bare form, but be defensive — NNTP `BODY`
+///    re-adds the brackets).
+/// 2. Drop empties.
+/// 3. Deduplicate, then **sort** lexicographically by bytes.
+/// 4. Join with `\n` and SHA-256 the result.
+///
+/// Case is **preserved**: a Message-ID's local part is case-sensitive per RFC
+/// 5322, so lowercasing could collide two genuinely distinct articles.
+///
+/// Sorting is what makes the cid survive segment/file reordering between two
+/// NZBs of the same posting — the reason we hash the *id set* and never the raw
+/// `.nzb` bytes (the retired `0x1004` did the latter and broke dedup; see
+/// meta-share `nzb/mod.rs`).
+///
+/// **v1 scope:** every segment counts, par2 volumes included. Two NZBs for one
+/// posting that disagree about whether to bundle par2 therefore mint different
+/// cids. Acceptable while we are the only producer of these manifests; if a
+/// second producer ever appears, the fix is to filter par2 subjects here *and*
+/// in meta-share's mirror — do not do it on one side only.
+pub fn compute_nzb_posting_cid<S: AsRef<str>>(message_ids: &[S]) -> String {
+    let mut ids: Vec<&str> = message_ids
+        .iter()
+        .map(|s| {
+            let t = s.as_ref().trim();
+            t.strip_prefix('<')
+                .and_then(|r| r.strip_suffix('>'))
+                .unwrap_or(t)
+        })
+        .filter(|s| !s.is_empty())
+        .collect();
+    ids.sort_unstable();
+    ids.dedup();
+
+    let mut hasher = Sha256::new();
+    for (i, id) in ids.iter().enumerate() {
+        if i > 0 {
+            hasher.update(b"\n");
+        }
+        hasher.update(id.as_bytes());
+    }
+    let digest: [u8; 32] = hasher.finalize().into();
+
+    // CIDv1: [version=0x01][codec varint][mh code=0x12 sha2-256][len=32][digest]
+    let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
+    wire.push(0x01);
+    write_pb_varint(NZB_POSTING_CODEC, &mut wire);
+    wire.push(0x12); // multihash code: sha2-256
+    write_pb_varint(digest.len() as u64, &mut wire);
     wire.extend_from_slice(&digest);
     format!("b{}", base32_lower_no_padding(&wire))
 }
@@ -475,30 +773,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn nzb_release_cid_wire_and_properties() {
-        let cid = compute_nzb_release_cid("api.nzb.life", "485c52f078b580667a01ac34a1cef6c2");
-        assert!(cid.starts_with('b'));
-        // Deterministic.
-        assert_eq!(
-            cid,
-            compute_nzb_release_cid("api.nzb.life", "485c52f078b580667a01ac34a1cef6c2")
-        );
-        // Host-namespaced: same id, different host → different cid.
-        assert_ne!(
-            cid,
-            compute_nzb_release_cid("api.nzbgeek.info", "485c52f078b580667a01ac34a1cef6c2")
-        );
-        // Explicit wire form: [01][84 20][84 20][20][sha256("host\nid")].
-        let digest: [u8; 32] =
-            Sha256::digest(b"api.nzb.life\n485c52f078b580667a01ac34a1cef6c2").into();
-        let mut want = vec![0x01, 0x84, 0x20, 0x84, 0x20, 0x20];
-        want.extend_from_slice(&digest);
-        assert_eq!(cid, format!("b{}", base32_lower_no_padding(&want)));
-        // Distinct codec from the bt-file codecs in the 0x100x neighborhood.
-        assert_ne!(cid, compute_bt_v1_file_cid(&[0xAB; 20], 0));
-    }
-
     #[test]
     fn base32_known_vectors() {
         assert_eq!(base32_lower_no_padding(b""), "");
@@ -509,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!(
@@ -571,4 +921,156 @@ mod tests {
         assert_eq!(out.blocks[0].1.len(), 0);
         assert_eq!(out.blocks[0].0, out.root);
     }
+
+    // ---- card locator (0x1007) --------------------------------------------
+
+    /// Inverse of [`base32_lower_no_padding`], test-only. Lets the card tests
+    /// assert the actual CIDv1 bytes (codec, multihash code, digest) rather
+    /// than just round-tripping the encoder against itself.
+    fn base32_lower_decode(s: &str) -> Vec<u8> {
+        const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz234567";
+        let mut out = Vec::new();
+        let (mut acc, mut bits) = (0u32, 0u32);
+        for c in s.bytes() {
+            let v = ALPHABET.iter().position(|&a| a == c).expect("base32 char") as u32;
+            acc = (acc << 5) | v;
+            bits += 5;
+            if bits >= 8 {
+                bits -= 8;
+                out.push((acc >> bits) as u8);
+                acc &= (1 << bits) - 1;
+            }
+        }
+        out
+    }
+
+    /// The property the whole card design leans on: the CID is a pure function
+    /// of `(source, id)`, so any peer can derive a card's address offline. If
+    /// this literal ever changes, every already-published card silently forks
+    /// into a second identity — treat a failure here as a wire break.
+    #[test]
+    fn card_cid_is_deterministic() {
+        let cid = compute_card_cid("tmdb", "tv:95479").expect("card cid");
+        assert_eq!(cid, compute_card_cid("tmdb", "tv:95479").unwrap());
+        assert_eq!(cid, "bagdsaaanar2g2zdcor3duojvgq3ts");
+    }
+
+    /// Source- and kind-namespaced: the same numeric id on a different bridge,
+    /// or the same id under a different media kind, must not collide.
+    #[test]
+    fn card_cid_is_source_and_kind_namespaced() {
+        let tmdb = compute_card_cid("tmdb", "tv:95479").unwrap();
+        let mal = compute_card_cid("mal", "95479").unwrap();
+        let movie = compute_card_cid("tmdb", "movie:95479").unwrap();
+        assert_ne!(tmdb, mal);
+        assert_ne!(tmdb, movie);
+    }
+
+    /// Pins the on-the-wire CIDv1 layout: v1, codec `0x1007` (varint
+    /// `[0x87,0x20]`), **identity** multihash `0x00` (this family has no
+    /// digest — the locator itself is the payload), then
+    /// `varint(source_len) ‖ source ‖ id`.
+    #[test]
+    fn card_cid_wire_shape() {
+        let cid = compute_card_cid("tmdb", "tv:95479").unwrap();
+        assert!(cid.starts_with('b'), "multibase base32-lower prefix");
+        let wire = base32_lower_decode(&cid[1..]);
+
+        assert_eq!(wire[0], 0x01, "CIDv1");
+        assert_eq!(&wire[1..3], &[0x87, 0x20], "codec varint for 0x1007");
+        assert_eq!(wire[3], 0x00, "identity multihash — no digest");
+
+        let digest_len = wire[4] as usize;
+        let digest = &wire[5..5 + digest_len];
+        assert_eq!(digest[0], 4, "varint(len(\"tmdb\"))");
+        assert_eq!(&digest[1..5], b"tmdb");
+        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.
+    #[test]
+    fn card_cid_rejects_empty_and_oversized() {
+        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());
+    }
+
+    // ---- nzb-posting (0x1003) — the dedup contract -----------------------
+    //
+    // These pin the normalisation rule documented on `compute_nzb_posting_cid`.
+    // meta-share re-derives the same digest from the fetched manifest and
+    // rejects a mismatch, so a change that breaks one of these silently makes
+    // every self-scanned posting unplayable. Mirror any edit on both sides.
+
+    /// Segment/file **order must not matter** — that is the whole reason we
+    /// hash a sorted id set instead of the raw `.nzb` bytes (the retired
+    /// `0x1004` did the latter and broke dedup across reposts).
+    #[test]
+    fn nzb_posting_cid_is_order_independent() {
+        let a = ["[email protected]", "[email protected]", "[email protected]"];
+        let b = ["[email protected]", "[email protected]", "[email protected]"];
+        let c = ["[email protected]", "[email protected]", "[email protected]"];
+        assert_eq!(compute_nzb_posting_cid(&a), compute_nzb_posting_cid(&b));
+        assert_eq!(compute_nzb_posting_cid(&a), compute_nzb_posting_cid(&c));
+    }
+
+    /// Angle brackets, surrounding whitespace, empty entries and duplicates are
+    /// all normalised away. NZB `<segment>` bodies carry the bare form but NNTP
+    /// `BODY` re-adds the brackets, so both spellings must converge.
+    #[test]
+    fn nzb_posting_cid_normalises_brackets_blanks_and_dupes() {
+        let bare = ["[email protected]", "[email protected]"];
+        let noisy = [" <[email protected]> ", "[email protected]", "", "  ", "[email protected]"];
+        assert_eq!(
+            compute_nzb_posting_cid(&bare),
+            compute_nzb_posting_cid(&noisy)
+        );
+    }
+
+    /// Case is preserved: a Message-ID's local part is case-sensitive per RFC
+    /// 5322, so lowercasing could collide two genuinely distinct articles.
+    #[test]
+    fn nzb_posting_cid_is_case_sensitive() {
+        assert_ne!(
+            compute_nzb_posting_cid(&["[email protected]"]),
+            compute_nzb_posting_cid(&["[email protected]"])
+        );
+    }
+
+    /// A different article set is a different posting.
+    #[test]
+    fn nzb_posting_cid_differs_on_different_ids() {
+        assert_ne!(
+            compute_nzb_posting_cid(&["[email protected]", "[email protected]"]),
+            compute_nzb_posting_cid(&["[email protected]", "[email protected]"])
+        );
+    }
+
+    /// Wire shape: CIDv1, codec `0x1003` (varint `[0x83, 0x20]`), multihash
+    /// sha2-256 (`0x12`), 32-byte digest. Pinned because meta-share matches on
+    /// the **codec** slot — a drift here makes every posting undecodable there.
+    #[test]
+    fn nzb_posting_cid_wire_shape() {
+        let cid = compute_nzb_posting_cid(&["[email protected]"]);
+        assert!(cid.starts_with('b'), "multibase base32-lower prefix");
+
+        // Round-trip the base32 body back to bytes to assert the header.
+        let decoded = base32_lower_decode(&cid[1..]);
+        assert_eq!(decoded[0], 0x01, "CIDv1");
+        assert_eq!(&decoded[1..3], &[0x83, 0x20], "varint(0x1003)");
+        assert_eq!(decoded[3], 0x12, "multihash sha2-256");
+        assert_eq!(decoded[4], 0x20, "digest length 32");
+        assert_eq!(decoded.len(), 5 + 32);
+
+        // The digest is sha256 over the newline-joined sorted set.
+        let expect: [u8; 32] = Sha256::digest(b"[email protected]").into();
+        assert_eq!(&decoded[5..], &expect);
+    }
 }

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

@@ -58,9 +58,29 @@ pub struct HashOutcome {
 /// - `BtV1File` — a single file inside a BitTorrent v1 torrent, custom
 ///   multicodec `0x1001`. Locator, not content hash: opaque to bitswap.
 /// - `NzbRelease` — a Usenet release (a Newznab listing), custom multicodec
-///   `nzb-release` `0x1004`. Locator like `BtV1File`: opaque to bitswap,
-///   redeemed to bytes only by a credentialed meta-share peer, which grabs the
-///   `.nzb` via the `{base, id}` descriptor stored at `nzb:<cid>` KV.
+///   `nzb-release` `0x1005`. Self-describing locator: the cid embeds `{host,
+///   id}` in an identity multihash, opaque to bitswap and redeemed to bytes
+///   only by a credentialed meta-share peer, which decodes the locator from the
+///   cid and grabs the `.nzb` via `t=get` — no KV side-table.
+/// - `CardLocator` — the identity of a *work* (a series, a film) as published by
+///   a metadata bridge, custom multicodec `0x1007`. The only family with **no
+///   bytes behind it at all**: the record IS the payload. Always paired with
+///   `bytes: None`, so it lands in the `(None, Some)` metadata-only branch
+///   above. Opaque to bitswap, and deliberately so — there is nothing to seed.
+/// - `NzbPosting` — a Usenet posting we **scanned ourselves**, custom multicodec
+///   `nzb-posting` `0x1003`. A content-derived digest over the article
+///   Message-ID set (not a locator: no host is embedded), so any peer with a
+///   plain NNTP provider can fetch it — no `IndexerCred`. Opaque to bitswap: the
+///   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
@@ -70,6 +90,9 @@ pub enum HashKind {
     Sha2_256,
     BtV1File,
     NzbRelease,
+    CardLocator,
+    NzbPosting,
+    YtVideo,
 }
 
 /// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -117,8 +140,9 @@ pub const SOURCE_KEYSET_PREFIX: &str = "source/";
 /// 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
-/// `source/gateway:<indexer>` first and this default then no-ops.
+/// 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,

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

@@ -27,6 +27,16 @@ pub fn record_matches(fields: &BTreeMap<String, String>, query: &GatewayQuery) -
             }
             continue;
         }
+        // `genres` is a key-set too (`genres/<Name> = "true"`, METADATA_KEYS
+        // §`genres/{name}`) on everything the card tier writes. Without this arm
+        // a `genres:` filter looks up a flat `genres` field, finds nothing on a
+        // card, and drops it — so a genre-scoped row would return empty.
+        if key == "genres" {
+            if !genres_filter_matches(fields, allowed) {
+                return false;
+            }
+            continue;
+        }
         let Some(field_value) = fields.get(key) else {
             return false;
         };
@@ -155,6 +165,54 @@ fn record_has_concrete_language(fields: &BTreeMap<String, String>) -> bool {
 
 /// Key-set members are written as the literal `"true"`; tolerate a couple of
 /// obvious truthy spellings and treat anything else as absent.
+/// Fold a genre label to its comparison form: lowercase, alphanumerics only.
+///
+/// Genre display names carry spaces and ampersands (`"Action & Adventure"`,
+/// `"Sci-Fi & Fantasy"`, `"Science Fiction"`), and **space is the query DSL's
+/// token separator** — so a filter can never spell a genre the way the record
+/// stores it. Callers pass a slug (`genres:action-adventure`); folding both
+/// sides makes them one value without forcing the stored name to be
+/// machine-shaped.
+fn genre_fold(s: &str) -> String {
+    s.chars()
+        .filter(|c| c.is_alphanumeric())
+        .flat_map(|c| c.to_lowercase())
+        .collect()
+}
+
+/// True iff the record carries one of the requested genres, in **either**
+/// encoding: a truthy `genres/<Name>` key-set member (the card tier's shape) or
+/// the legacy flat comma-joined `genres` field. OR across the requested genres,
+/// both sides folded — see [`genre_fold`].
+///
+/// **Fails closed, unlike [`languages_filter_matches`].** A language filter is a
+/// viewer *preference* narrowing a set, so an unknown-language release is kept.
+/// A genre filter *defines what a row is*: a record with no genre information is
+/// not known to be Action, so putting it in an Action row states something
+/// untrue. Upstream has already filtered by genre, so this is a re-validation.
+fn genres_filter_matches(fields: &BTreeMap<String, String>, allowed: &[String]) -> bool {
+    let wanted: Vec<String> = allowed.iter().map(|g| genre_fold(g)).collect();
+    if wanted.iter().all(|g| g.is_empty()) {
+        return true; // a filter with no usable value constrains nothing
+    }
+    for (key, value) in fields {
+        let Some(name) = key.strip_prefix("genres/") else {
+            continue;
+        };
+        if is_truthy_member(value) && wanted.contains(&genre_fold(name)) {
+            return true;
+        }
+    }
+    if let Some(flat) = fields.get("genres") {
+        for member in flat.split(|c| c == ',' || c == '|') {
+            if wanted.contains(&genre_fold(member)) {
+                return true;
+            }
+        }
+    }
+    false
+}
+
 fn is_truthy_member(v: &str) -> bool {
     matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes")
 }

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

@@ -41,7 +41,7 @@ use serde_json::Value;
 use tracing::{info, warn};
 
 use crate::config::{self, CONFIG_PAGE_HTML};
-use crate::plugin::{stamp_default_source, ConfigError, FeederPlugin, HashKind, HashOutcome};
+use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome};
 use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError};
 use crate::types::{DiscoveryRecord, GatewayError, PluginHealth};
 
@@ -55,6 +55,9 @@ pub enum HashKindDto {
     Sha2_256,
     BtV1File,
     NzbRelease,
+    CardLocator,
+    NzbPosting,
+    YtVideo,
 }
 
 impl From<HashKind> for HashKindDto {
@@ -64,6 +67,9 @@ impl From<HashKind> for HashKindDto {
             HashKind::Sha2_256 => HashKindDto::Sha2_256,
             HashKind::BtV1File => HashKindDto::BtV1File,
             HashKind::NzbRelease => HashKindDto::NzbRelease,
+            HashKind::CardLocator => HashKindDto::CardLocator,
+            HashKind::NzbPosting => HashKindDto::NzbPosting,
+            HashKind::YtVideo => HashKindDto::YtVideo,
         }
     }
 }
@@ -75,6 +81,9 @@ impl From<HashKindDto> for HashKind {
             HashKindDto::Sha2_256 => HashKind::Sha2_256,
             HashKindDto::BtV1File => HashKind::BtV1File,
             HashKindDto::NzbRelease => HashKind::NzbRelease,
+            HashKindDto::CardLocator => HashKind::CardLocator,
+            HashKindDto::NzbPosting => HashKind::NzbPosting,
+            HashKindDto::YtVideo => HashKind::YtVideo,
         }
     }
 }
@@ -275,13 +284,10 @@ async fn query(
     Json(req): Json<QueryRequest>,
 ) -> Result<Json<QueryResponse>, Response> {
     let plugin = state.plugin(&req.upstream_id)?;
-    let mut records = plugin
+    let records = plugin
         .handle_query(&req.query, req.max_results as usize)
         .await
         .map_err(gateway_error_response)?;
-    for r in &mut records {
-        stamp_default_source(&mut r.fields, &req.upstream_id);
-    }
     Ok(Json(QueryResponse { records }))
 }
 
@@ -305,18 +311,6 @@ async fn query_stream(
         .handle_query_stream(&req.query, req.max_results as usize)
         .await
         .map_err(gateway_error_response)?;
-    // Stamp the default provenance member on each streamed `Base` record (a
-    // no-op when the plugin already carries a finer `source/*`, e.g. torznab's
-    // per-indexer label). `EnrichPatch`/`Drop`/`Done` frames pass through: the
-    // base they refer to was already stamped.
-    let upstream_id = req.upstream_id.clone();
-    let events = events.map(move |ev| match ev {
-        GatewaySearchEvent::Base(mut rec) => {
-            stamp_default_source(&mut rec.fields, &upstream_id);
-            GatewaySearchEvent::Base(rec)
-        }
-        other => other,
-    });
     let body = events.map(|ev| {
         // Serialization of GatewaySearchEvent is infallible in practice; on the
         // off chance it fails, terminate with an Error line so the consumer sees
@@ -342,15 +336,10 @@ async fn compute(
     Json(req): Json<ComputeRequest>,
 ) -> Result<Json<ComputeResponse>, Response> {
     let plugin = state.plugin(&req.upstream_id)?;
-    let mut outcomes = plugin
+    let outcomes = plugin
         .compute_outcomes(&req.record_id)
         .await
         .map_err(gateway_error_response)?;
-    for o in &mut outcomes {
-        if let Some(rec) = o.record.as_mut() {
-            stamp_default_source(&mut rec.fields, &req.upstream_id);
-        }
-    }
     Ok(Json(ComputeResponse {
         outcomes: outcomes.into_iter().map(OutcomeDto::from_outcome).collect(),
     }))

feeder-plugin/tribler-feeder/src/consts.rs

@@ -81,8 +81,8 @@ pub(crate) const TORRENT_RESOLVE_MAX_PER_INDEXER: usize = 30;
 /// it arrives until the gateway sends `Done`. (The 5 s `GATEWAY_CALL_TIMEOUT`
 /// is only the one-shot fallback for gateways that don't speak the streaming
 /// protocol.) So this can be generous: the binding constraint is meta-share's
-/// `SEARCH_TIMEOUT_SECS` silence window (set to match in the dev compose), not
-/// a hard call timeout. 5 min is a safety ceiling on a pathological page;
+/// `SEARCH_TIMEOUT_SECS` silence window (300 s, the meta-search code default),
+/// not a hard call timeout. 5 min is a safety ceiling on a pathological page;
 /// a normal XXX page (≤ `TORRENT_RESOLVE_MAX_PER_INDEXER` items at ~1.9 s
 /// each, 8-wide) finishes in well under 10 s and the gateway returns then.
 pub(crate) const TORRENT_RESOLVE_BUDGET: Duration = Duration::from_secs(300);

feeder-plugin/tribler-feeder/src/discovery.rs

@@ -178,6 +178,16 @@ fn build_seed_record(
     // mode marker, contentKind, and optionally anime.
     fields.insert(mode.marker().to_string(), "true".to_string());
     fields.insert("contentKind".to_string(), content_kind.to_string());
+    // The routing axes, co-written with the kind they derive from
+    // (METADATA_KEYS.md §1): `fileType` says what the bytes are, `domain` says
+    // which app wants the record, `workForm` whether the work stands alone or
+    // is one instalment of a run.
+    if let Some(d) = meta_feeder_sdk::domain::domain_for_content_kind(content_kind) {
+        fields.insert("domain".to_string(), d.to_string());
+    }
+    if let Some(wf) = meta_feeder_sdk::domain::work_form_for_content_kind(content_kind) {
+        fields.insert("workForm".to_string(), wf.to_string());
+    }
     fields.insert("fileType".to_string(), "video".to_string());
     if anime {
         fields.insert("anime".to_string(), "true".to_string());

feeder-plugin/tribler-feeder/src/enrich.rs

@@ -13,8 +13,8 @@ use tracing::{debug, warn};
 
 use crate::title::clean_torrent_title;
 use crate::tmdb::{
-    principal_top_n, season_episode_bounds, SeasonEpisodeBounds, TmdbCall, TmdbClient,
-    TmdbExternalIds, TmdbHit, TmdbKind, TmdbTvDetails,
+    principal_top_n, season_episode_bounds, tmdb_title_member_keys, SeasonEpisodeBounds, TmdbCall,
+    TmdbClient, TmdbExternalIds, TmdbHit, TmdbKind, TmdbTvDetails,
 };
 use crate::tmdb_budget::TmdbBudget;
 use meta_feeder_sdk::cache::MidhashCache;
@@ -548,6 +548,8 @@ fn build_stub_patch(f: &BTreeMap<String, String>) -> Option<EnrichOutcome> {
         set.insert("fileType".to_string(), "video".to_string());
         set.insert("videoType".to_string(), "movie".to_string());
         set.insert("contentKind".to_string(), "movie".to_string());
+        set.insert("domain".to_string(), "screen".to_string());
+        set.insert("workForm".to_string(), "standalone".to_string());
     }
     if !has_poster {
         // Non-resolvable marker — the image 404s (broken tile), but the card
@@ -667,7 +669,7 @@ async fn compute_enrichment_inner(
         Some(resolved) => resolved,
         None => return EnrichOutcome::Noop,
     };
-    build_enrichment_patch(record, enricher, hit, matched_kind, raw_title).await
+    build_enrichment_patch(record, enricher, hit, matched_kind).await
 }
 
 /// Pick the canonical TMDB entry for `record`. Prefers the **anchored** path —
@@ -720,35 +722,32 @@ async fn resolve_enrichment_hit(
 /// content-kind reconciliation (movie clears bogus season/episode; tv normalises
 /// to `episode` and bounds-checks the parsed season/episode — an out-of-bounds
 /// verdict drops the record), and the transient `posterPath` the poster store
-/// consumes. `raw_title` is preserved as `originalTitle`.
+/// consumes. The raw release string is never copied into a title key — it
+/// stays in `fileName`.
 async fn build_enrichment_patch(
     record: &DiscoveryRecord,
     enricher: &TmdbEnricher,
     hit: TmdbHit,
     matched_kind: TmdbKind,
-    raw_title: Option<String>,
 ) -> EnrichOutcome {
     let mut set: BTreeMap<String, String> = BTreeMap::new();
     let mut remove: Vec<String> = Vec::new();
     set.insert("title".to_string(), hit.title.clone());
-    // Preserve the raw tracker release title (quality tags / release-group
-    // info) in `originalTitle`. This is the canonical home now that the
-    // bespoke `releaseTitle` key is retired — reuse-before-invent
-    // (METADATA_KEYS.md rule #1, §14.10 title sprawl).
-    if let Some(raw_title) = raw_title {
-        set.insert("originalTitle".to_string(), raw_title);
-    }
-    // TMDB's original-language title (e.g. "ナルト") is a genuine localized
-    // title — file it under the namespaced `titles/{lang3}` (METADATA_KEYS.md
-    // §3), keyed by TMDB's `original_language`, instead of clobbering
-    // `originalTitle`. Skipped when the language is outside the common 639-1
-    // set or the title is blank (originalTitle still carries the raw title).
+    // The raw tracker release string is NOT stored here — it is emitted as
+    // `fileName` at ingest (METADATA_KEYS.md §1). `originalTitle` carries TMDB's
+    // *original-language* title (e.g. "ナルト" / "Sousou no Frieren"), clean —
+    // never the release string.
     if let Some(orig) = hit.original_title.clone() {
         if !orig.trim().is_empty() {
-            if let Some(lang3) = hit.original_lang3() {
-                set.insert(format!("titles/{lang3}"), orig);
+            set.insert("originalTitle".to_string(), orig);
         }
     }
+    // Whoever writes `title`/`originalTitle` also files them as members of the
+    // language-nested `titles/{lang3}/{name}` key-set (METADATA_KEYS.md): the
+    // original under TMDB's `original_language` (`und` when unmapped), the title
+    // under `eng` when it differs from the original.
+    for key in tmdb_title_member_keys(&hit) {
+        set.insert(key, "true".to_string());
     }
     if let Some(o) = hit.overview.clone() {
         if !o.trim().is_empty() {
@@ -773,6 +772,13 @@ async fn build_enrichment_patch(
     match matched_kind {
         TmdbKind::Movie => {
             set.insert("contentKind".to_string(), "movie".to_string());
+            // The routing axes move with the kind. `domain` is `screen` for
+            // films and serials alike since the merge (METADATA_KEYS.md
+            // §14.17), so the film-vs-serial half is `workForm` — and
+            // reconciling the kind without it would leave the record shaped as
+            // a series.
+            set.insert("domain".to_string(), "screen".to_string());
+            set.insert("workForm".to_string(), "standalone".to_string());
             // Co-write the narrower `videoType` facet (METADATA_KEYS.md §14.11
             // — contentKind=movie ↔ videoType=movie). Overwrites any
             // category-derived videoType from `into_discovery_record` with the
@@ -784,6 +790,8 @@ async fn build_enrichment_patch(
         }
         TmdbKind::Tv => {
             set.insert("contentKind".to_string(), "episode".to_string());
+            set.insert("domain".to_string(), "screen".to_string());
+            set.insert("workForm".to_string(), "serial".to_string());
             // contentKind=episode ↔ videoType=tvshow (METADATA_KEYS.md §14.11).
             set.insert("videoType".to_string(), "tvshow".to_string());
             // Season/episode boundary validation. Only a parsed `season`

feeder-plugin/tribler-feeder/src/tmdb.rs

@@ -397,8 +397,8 @@ pub(crate) struct TmdbSearchItem {
     #[serde(default)]
     pub(crate) genre_ids: Vec<u32>,
     /// ISO 639-1 (2-letter) language the title was originally produced in.
-    /// Mapped to `lang3` and used to file `original_title` under
-    /// `titles/{lang3}` (METADATA_KEYS.md §3).
+    /// Mapped to `lang3` and used to file `original_title` as a
+    /// `titles/{lang3}/{name}` member (METADATA_KEYS.md §3).
     #[serde(default)]
     pub(crate) original_language: Option<String>,
 }
@@ -438,8 +438,8 @@ impl TmdbSearchItem {
 /// Map an ISO 639-1 (2-letter) code to its ISO 639-3 (`lang3`) equivalent,
 /// matching the store's convention (`eng`, `jpn`, `fra`; the 639-2/T variant
 /// where B/T differ, e.g. `deu` not `ger`). Covers the languages TMDB
-/// commonly returns; unknown codes return `None` so the caller skips the
-/// `titles/{lang3}` write rather than persisting a non-`lang3` key.
+/// commonly returns; unknown codes return `None` so the caller files the name under `und`
+/// rather than persisting a non-`lang3` key.
 pub(crate) fn iso639_1_to_3(code: &str) -> Option<&'static str> {
     Some(match code.to_ascii_lowercase().as_str() {
         "en" => "eng",
@@ -504,12 +504,53 @@ pub(crate) struct TmdbHit {
 impl TmdbHit {
     /// TMDB `original_language` (ISO 639-1) mapped to the store's `lang3`
     /// (ISO 639-3), or `None` when the language is outside the common set or
-    /// absent. Used to file `original_title` under `titles/{lang3}` (§3).
+    /// absent. Used to file `original_title` under `titles/{lang3}/{name}` (§3).
     pub(crate) fn original_lang3(&self) -> Option<&'static str> {
         iso639_1_to_3(self.original_language.as_deref()?.trim())
     }
 }
 
+/// The language TMDB's `name`/`title` is written in. Every request goes out
+/// without a `language` param, so TMDB answers in its default, en-US.
+pub(crate) const METADATA_LANG3: &str = "eng";
+
+/// The `titles/{lang3}/{name}` key-set member for `name` (METADATA_KEYS.md), or
+/// `None` when it normalises to empty: trimmed, whitespace runs collapsed,
+/// case/diacritics/`%` verbatim, and `/` (the key-set separator) written as
+/// U+2215 `∕`.
+///
+/// CROSS-BINARY CONTRACT: the indexer feeder's `tmdb.rs::title_member_key` and
+/// the card feeder's `tmdb_client.rs::title_member_key` write the same shape;
+/// two peers must produce the same key for one name.
+pub(crate) fn title_member_key(lang3: &str, name: &str) -> Option<String> {
+    let name = name.split_whitespace().collect::<Vec<_>>().join(" ");
+    if name.is_empty() {
+        return None;
+    }
+    Some(format!("titles/{lang3}/{}", name.replace('/', "\u{2215}")))
+}
+
+/// The `titles/{lang3}/{name}` members this feeder files from a TMDB hit: its
+/// title and original title (a search hit carries no AKAs). The original goes
+/// under its `original_language` (`und` when unmapped); the title under
+/// [`METADATA_LANG3`] **unless it equals the original** — TMDB falls back to
+/// the original name when it has no en-US translation, so `3%` is
+/// `titles/por/3%` only, not also an English name. Same rule as the indexer
+/// feeder's `tmdb_title_member_keys`.
+pub(crate) fn tmdb_title_member_keys(hit: &TmdbHit) -> Vec<String> {
+    let norm = |s: &str| s.split_whitespace().collect::<Vec<_>>().join(" ");
+    let mut keys = Vec::new();
+    let original = hit.original_title.as_deref().map(norm).filter(|o| !o.is_empty());
+    if let Some(o) = &original {
+        keys.extend(title_member_key(hit.original_lang3().unwrap_or("und"), o));
+    }
+    let title = norm(&hit.title);
+    if title != "(untitled)" && original.as_deref() != Some(title.as_str()) {
+        keys.extend(title_member_key(METADATA_LANG3, &title));
+    }
+    keys
+}
+
 /// Authoritative TV structure from `GET /3/tv/{id}`, used to bounds-check
 /// a title-parsed season/episode. Only the two structural fields are
 /// decoded; everything else in the (large) details payload is ignored.
@@ -849,3 +890,38 @@ mod season_bounds_tests {
         assert_eq!(season_episode_bounds(&d, Some(-1), None), SeasonEpisodeBounds::Contradiction);
     }
 }
+
+#[cfg(test)]
+mod title_member_tests {
+    use super::*;
+
+    fn hit(title: &str, orig: Option<&str>, lang: Option<&str>) -> TmdbHit {
+        TmdbHit {
+            tmdbid: 1,
+            title: title.to_string(),
+            original_title: orig.map(str::to_string),
+            original_language: lang.map(str::to_string),
+            overview: None,
+            year: None,
+            poster_path: None,
+            genre_ids: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn title_and_original_are_filed_as_language_nested_members() {
+        assert_eq!(
+            tmdb_title_member_keys(&hit("Naruto", Some("ナルト"), Some("ja"))),
+            vec!["titles/jpn/ナルト", "titles/eng/Naruto"]
+        );
+        // Title == original (no en-US translation) is filed once.
+        assert_eq!(tmdb_title_member_keys(&hit("3%", Some("3%"), Some("pt"))), vec!["titles/por/3%"]);
+        // Unmapped language → und; `/` → U+2215; TMDB's placeholder is not a name.
+        assert_eq!(
+            tmdb_title_member_keys(&hit("(untitled)", Some("Fate/Zero"), Some("xx"))),
+            vec!["titles/und/Fate\u{2215}Zero"]
+        );
+        assert_eq!(title_member_key("eng", "  3   Percent "), Some("titles/eng/3 Percent".to_string()));
+        assert_eq!(title_member_key("eng", "   "), None);
+    }
+}

feeder-plugin/tribler-feeder/src/tribler/mod.rs

@@ -530,6 +530,12 @@ impl TriblerPlugin {
                 "movie"
             };
             rec.fields.insert("contentKind".to_string(), kind.to_string());
+            if let Some(d) = meta_feeder_sdk::domain::domain_for_content_kind(kind) {
+                rec.fields.insert("domain".to_string(), d.to_string());
+            }
+            if let Some(wf) = meta_feeder_sdk::domain::work_form_for_content_kind(kind) {
+                rec.fields.insert("workForm".to_string(), wf.to_string());
+            }
         }
         // For episode records, stamp the *parsed* season/episode so the consumer
         // renders a proper series with an ordered episode list (meta-watch keys
@@ -601,10 +607,17 @@ impl TriblerPlugin {
         else {
             return vec![rec];
         };
+        // Stamp the per-file CID in both shapes: the `cids/<cid>` key-set member
+        // (what meta-share ingest and meta-watch's codec check actually read —
+        // legacy `cid_*` names are no longer tolerated) plus the named field, kept
+        // as an internal convergence key that the gateway's meta-core sink strips
+        // before the record is stored.
         let with_cid = move |mut r: DiscoveryRecord, idx: u64| -> DiscoveryRecord {
+            let cid = compute_bt_v1_file_cid(&ih20, idx);
+            r.fields.entry(format!("cids/{cid}")).or_insert_with(|| "true".to_string());
             r.fields
                 .entry("cid_btih_v1_file".to_string())
-                .or_insert_with(|| compute_bt_v1_file_cid(&ih20, idx));
+                .or_insert(cid);
             r
         };
         if is_obvious_single_file(&title) {
@@ -630,15 +643,18 @@ impl TriblerPlugin {
             .collect();
         let per_file = |f: &TorrentFile, force_episode: bool| {
             let mut r = build_file_record(&rec.fields, &rec.record_id, f);
-            r.fields.insert(
-                "cid_btih_v1_file".to_string(),
-                compute_bt_v1_file_cid(&ih20, f.index as u64),
-            );
+            let cid = compute_bt_v1_file_cid(&ih20, f.index as u64);
+            // Key-set member first — it is the shape consumers read.
+            r.fields.insert(format!("cids/{cid}"), "true".to_string());
+            r.fields.insert("cid_btih_v1_file".to_string(), cid);
             if force_episode {
                 // A multi-video pack is a TV season — force the kind so enrichment
                 // resolves the show as TV (the show title drives the TMDB search).
                 r.fields
                     .insert("contentKind".to_string(), "episode".to_string());
+                r.fields.insert("domain".to_string(), "screen".to_string());
+                r.fields
+                    .insert("workForm".to_string(), "serial".to_string());
             }
             r
         };