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/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

@@ -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

@@ -75,6 +75,17 @@ pub(crate) struct Card {
     pub(crate) guard_titles: Arc<Vec<String>>,
     pub(crate) overview: Option<String>,
     pub(crate) poster_path: Option<String>,
+    /// Editorial genre names (`"Animation"`, `"Drama"`, …) — METADATA_KEYS
+    /// §`genres`. The *work's* categorisation, distinct from `categories/*`,
+    /// which is the Newznab/Prowlarr distribution taxonomy a release is filed
+    /// under and which a card (having no release) never carries.
+    ///
+    /// Free on both resolution paths: a details fetch returns named genre
+    /// objects, and a list hit's bare `genre_ids` are mapped through
+    /// `Resolver::genre_names`. May be empty — TMDB genre coverage is good but
+    /// not universal, and consumers must treat "no genres" as unknown rather
+    /// than as "none".
+    pub(crate) genres: Vec<String>,
     pub(crate) year: Option<u16>,
 
     // -- structure (TV only) ------------------------------------------------
@@ -123,6 +134,27 @@ impl Card {
         n.clamp(1, MAX_CARD_SEASONS)
     }
 
+    /// The card's synopsis, trimmed and non-empty.
+    pub(crate) fn overview_text(&self) -> Option<&str> {
+        self.overview.as_deref().map(str::trim).filter(|s| !s.is_empty())
+    }
+
+    /// The card's poster path, non-empty.
+    pub(crate) fn poster(&self) -> Option<&str> {
+        self.poster_path.as_deref().filter(|s| !s.is_empty())
+    }
+
+    /// Would this card actually render? Exactly [`Card::to_record`]'s emit
+    /// condition, hoisted so a caller can predict the drop *before* projecting.
+    ///
+    /// [`crate::discovery`] needs this: it walks catalog pages until it has N
+    /// cards, and counting raw TMDB hits would overcount — a hit with no poster
+    /// or no overview is silently declined at projection time, leaving a short
+    /// row. The two must not drift, hence one predicate rather than two.
+    pub(crate) fn is_displayable(&self) -> bool {
+        self.overview_text().is_some() && self.poster().is_some()
+    }
+
     /// Project to the wire record.
     ///
     /// Returns `None` when the card lacks a poster or a description — it would
@@ -139,15 +171,32 @@ impl Card {
         tmdb: &TmdbClient,
         query_filters: &BTreeMap<String, Vec<String>>,
     ) -> Option<DiscoveryRecord> {
-        let overview = self.overview.as_deref().map(str::trim).filter(|s| !s.is_empty())?;
+        // Same two conditions as `Card::is_displayable`, which the discovery page
-        let poster_path = self.poster_path.as_deref().filter(|s| !s.is_empty())?;
+        // walk uses to predict this drop — keep them in one place.
+        let overview = self.overview_text()?;
+        let poster_path = self.poster()?;
 
         let mut fields: BTreeMap<String, String> = BTreeMap::new();
         // Type axes. `card` on the fileType axis and the work kind on the
         // 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.
@@ -171,6 +220,24 @@ impl Card {
         // `overview`).
         fields.insert("description/eng".to_string(), overview.to_string());
 
+        // Genres as a KEY-SET (`genres/<Name> = "true"`), not the legacy
+        // comma-joined `genres` value the tmdb/jellyfin plugins still write.
+        //
+        // METADATA_KEYS §14.12 is explicit that `csv-set` is the shape being
+        // migrated *away* from and that "new writers must not introduce more
+        // csv-set fields — reach for key-set, even if the surrounding family is
+        // still on the legacy shape". This is a new writer, and the reason bites
+        // here specifically: a card is the one record type multiple peers derive
+        // independently at the same CID (§9.4), so two peers resolving different
+        // genre sets must union by key-merge rather than diverge into two
+        // comma-joined strings that need string-diffing to reconcile.
+        for g in &self.genres {
+            let g = g.trim();
+            if !g.is_empty() {
+                fields.insert(format!("genres/{g}"), "true".to_string());
+            }
+        }
+
         // The bare-CID key-set member, exactly as every other feeder stamps on
         // its search records (`indexer-feeder`'s `torznab/xml.rs` does the same
         // with the release's infohash).
@@ -206,7 +273,19 @@ impl Card {
         );
 
         for (key, allowed) in query_filters {
-            if key == "languages" || fields.contains_key(key) || allowed.is_empty() {
+            // `genres` joins `languages` as an echo exclusion, for the same
+            // reason: it is stored as a key-set, so the flat field this loop
+            // would write is not the shape a reader looks for. Worse, the filter
+            // value is a *slug* (`action-adventure` — the query DSL can't spell
+            // "Action & Adventure", space being its token separator), so echoing
+            // it would persist a slug masquerading as a genre name alongside the
+            // real `genres/<Name>` members. Both re-validating tiers read the
+            // key-set, so nothing needs the echo.
+            if key == "languages"
+                || key == "genres"
+                || fields.contains_key(key)
+                || allowed.is_empty()
+            {
                 continue;
             }
             fields.insert(key.clone(), allowed.join(","));
@@ -243,6 +322,7 @@ mod tests {
             guard_titles: Arc::new(vec!["Sousou no Frieren".to_string()]),
             overview: Some("The story follows the elf mage Frieren.".to_string()),
             poster_path: Some("/dqZENchTd7lp5zit1Q7Bkjzcxpi.jpg".to_string()),
+            genres: vec!["Animation".to_string(), "Sci-Fi & Fantasy".to_string()],
             year: Some(2023),
             seasons: 1,
             season_summaries: Arc::new(Vec::new()),
@@ -326,17 +406,29 @@ mod tests {
     }
 
     /// Query filters are echoed so the record survives `record_matches` on the
-    /// way out, but `languages` is deliberately skipped (it fails open).
+    /// way out — except the two key-set fields, which a reader resolves from
+    /// their `<prefix>/<member>` members rather than from a flat value.
     #[test]
-    fn query_filters_are_echoed_except_languages() {
+    fn query_filters_are_echoed_except_the_key_set_fields() {
         let tmdb = TmdbClient::new("token".to_string());
         let mut filters = BTreeMap::new();
         filters.insert("fileType".to_string(), vec!["card".to_string()]);
-        filters.insert("genres".to_string(), vec!["fantasy".to_string()]);
+        filters.insert("popular".to_string(), vec!["true".to_string()]);
+        filters.insert("genres".to_string(), vec!["action-adventure".to_string()]);
         filters.insert("languages".to_string(), vec!["jpn".to_string()]);
         let rec = tv_card().to_record(&tmdb, &filters).expect("record");
-        assert_eq!(rec.fields["genres"], "fantasy");
+
+        assert_eq!(rec.fields["popular"], "true", "ordinary filters echo");
         assert_eq!(rec.fields["fileType"], "card", "own value wins the echo");
         assert!(!rec.fields.contains_key("languages"));
+        // The genre filter carries a *slug* the DSL can spell; echoing it would
+        // persist "action-adventure" as if it were a genre name, beside the real
+        // `genres/Action & Adventure` members.
+        assert!(
+            !rec.fields.contains_key("genres"),
+            "the genre slug must not be echoed as a flat field"
+        );
+        assert_eq!(rec.fields["genres/Animation"], "true");
+        assert_eq!(rec.fields["genres/Sci-Fi & Fantasy"], "true");
     }
 }

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

@@ -30,6 +30,70 @@ pub(crate) const TMDB_RATE_LIMIT_DEFAULT_RETRY_SECS: u64 = 5;
 /// Maximum time to wait for a TMDB budget permit before giving up (seconds).
 pub(crate) const TMDB_WAIT_DEADLINE_SECS: u64 = 3000;
 
+/// Budget-permit deadline for a **catalog page** ([`crate::discovery`]),
+/// deliberately far shorter than [`TMDB_WAIT_DEADLINE_SECS`].
+///
+/// That 3000 s ceiling is sized for a query someone is *waiting on* — a user
+/// typed a title and wants that specific answer, so out-waiting a 429 pause
+/// beats returning nothing. A browse row is the opposite: nobody asked for
+/// *this* row in particular, it refreshes on a timer, and its client keeps the
+/// previously-filled row on an empty pass (meta-watch's sticky per-row merge).
+/// Inheriting 3000 s would let a single 429 pin a row's search stream open for
+/// most of an hour to eventually produce what a 5 s bail produces now.
+pub(crate) const TMDB_DISCOVERY_WAIT_DEADLINE_SECS: u64 = 5;
+
+/// Hard ceiling on catalog pages walked for one discovery row, whatever the
+/// requested card count. Bounds a misconfigured `card_discovery_n` (or a client
+/// asking for 500) to a predictable per-row TMDB cost.
+pub(crate) const DISCOVERY_MAX_PAGES: u32 = 5;
+
+/// TMDB keyword ids excluded from an `anime:true` discovery row, as
+/// `without_keywords`: `hentai`, `softcore`, `erotica`, `pornography`,
+/// `adult animation`.
+///
+/// **Why this is needed at all.** `include_adult=false` does *not* cover it:
+/// TMDB's `adult` boolean is reserved for actual pornography, and every one of
+/// the titles this excludes comes back `adult: false`. Measured on the live
+/// endpoint, the flag removed 84 of 4465 anime series and none of the offenders.
+///
+/// **Why they surface in the first place.** `sort_by=popularity.desc` sorts on
+/// TMDB's *traffic* metric (page views, API hits, list adds), not on audience
+/// size or rating. Adult titles attract disproportionate scraping and search
+/// traffic, so they score far above their audience: measured, one carried
+/// popularity 313 on **26 votes** and another 242 on **7 votes**, against
+/// Bleach's 92 on **2197**. Narrowing the pool to the anime keyword removes the
+/// mainstream ballast that keeps them off the front of `/tv/popular`, which is
+/// why only the anime rows show it.
+///
+/// **Why these five and not `erotic`/`ecchi`.** Those two are *content
+/// descriptors* a mainstream show can legitimately carry — Mushoku Tensei
+/// (tmdb 94664, 1626 votes) carries both — so excluding them costs real titles.
+/// The five here tag adult *productions*, and no false positive was observed.
+/// Keyword ids are stable; names resolved via `search/keyword`.
+pub(crate) const DISCOVERY_EXCLUDED_KEYWORDS: &str = "198385,155477,325693,445,161919";
+
+/// Minimum TMDB vote count for a discovery row entry.
+///
+/// The residue filter behind [`DISCOVERY_EXCLUDED_KEYWORDS`]: a handful of
+/// erotic-tagged titles carry none of the five excluded keywords, and they are
+/// uniformly low-signal (7–37 votes). A vote requires a real logged-in account,
+/// which is exactly the signal the traffic-based popularity score lacks, so a
+/// floor separates them from genuinely obscure-but-real titles far better than
+/// popularity can.
+///
+/// 50 is deliberately low. It keeps Ranma½ (169) and Doraemon (237) — it is a
+/// noise gate, not a popularity contest. Raising it much further starts costing
+/// legitimate niche titles (Akane-banashi 25, Gravitation 17), which is why the
+/// keyword exclusion carries the main load and this only sweeps up behind it.
+///
+/// Applies to `/discover` rows, which is now every mode except non-anime
+/// **trending** — the one signal with no `/discover` equivalent, and therefore
+/// the one row that cannot carry this gate. The popular/top-rated modes were
+/// moved off `/{kind}/popular` and `/{kind}/top_rated` precisely so they could:
+/// three hentai titles were measured at positions 4, 5 and 17 of the
+/// *non-anime* `tv/popular` row.
+pub(crate) const DISCOVERY_MIN_VOTES: u32 = 50;
+
 /// Default number of cards returned for a free-text query (top-N by TMDB
 /// popularity).
 ///
@@ -47,6 +111,15 @@ pub(crate) const DEFAULT_CARD_TOP_N: usize = 10;
 /// never needs a cache wipe — reads slice the cached list to the configured N.
 pub(crate) const CACHED_PRINCIPAL_DEPTH: usize = 30;
 
+/// Default number of cards a **keyword-less catalog** query returns (a home
+/// row: `popular:true contentKind:series fileType:card`).
+///
+/// 20 is TMDB's own catalog page size, so the default row costs exactly one
+/// TMDB call. Raising it past 20 is legal — [`crate::discovery`] walks further
+/// pages — but each page is another call against the shared budget, paid on
+/// every warm pass of every row.
+pub(crate) const DEFAULT_CARD_DISCOVERY_N: usize = 20;
+
 /// Defensive ceiling on a card's reported season count (guards a corrupt
 /// `number_of_seasons`).
 pub(crate) const MAX_CARD_SEASONS: u32 = 50;

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

@@ -0,0 +1,708 @@
+//! Keyword-less catalog discovery: TMDB's popular / trending / top-rated lists
+//! → [`Card`]s.
+//!
+//! The card tier's *browse* surface, as opposed to its search surface. A client
+//! with a home page (meta-watch's wall) cannot ask "what should I show?" through
+//! a text query — there is no text. This branch answers a query whose intent is
+//! carried entirely by structured filters, by calling TMDB's catalog endpoints.
+//!
+//! ## Query dialect
+//!
+//! Deliberately identical to the indexer feeder's older `discovery.rs`, which
+//! answers the same dialect with metadata-only *seeds*. Same words, so a client
+//! switches tiers by swapping `fileType:video` for `fileType:card` and nothing
+//! else:
+//!
+//! - **mode** (one of) — `popular:true` and `top_rated:true` are served by
+//!   `/discover/{movie,tv}` sorted by popularity / rating (rather than the
+//!   `/{movie,tv}/popular` list endpoints, which cannot carry the adult gate —
+//!   see `DISCOVERY_EXCLUDED_KEYWORDS`); `trending:true` → the real
+//!   `/trending/{movie,tv}/week`, the one mode with no `/discover` equivalent
+//!   and therefore the one row that stays ungated.
+//! - **kind** (required) — `contentKind:movie` → films,
+//!   `contentKind:series` (or `episode`/`tvshow`/`tv`) → shows. Also what routes
+//!   the query here at all, alongside `fileType:card`.
+//! - **anime** (optional) — `anime:true` narrows `/discover` to TMDB's *anime*
+//!   keyword, sorted to match the mode. Anime trending has no keyword-filtered
+//!   endpoint either, so it approximates with popularity — and stays gated.
+//!
+//! ## One call, complete cards
+//!
+//! A TMDB catalog list already carries `title`, `overview`, `poster_path` and
+//! `year` per entry — everything [`Card::to_record`] needs to emit a renderable
+//! card. So a row costs **one** TMDB call, not one plus twenty by-id lookups.
+//!
+//! The price is a thinner id bag: `tvdbid`/`imdbid` come from `external_ids`,
+//! which a list response has no room for, so a discovery card carries only its
+//! `tmdbid`. That is the id consumers actually key on (it is what the detail
+//! page's phase-2 `tmdbid:` query is built from), and the full bag is resolved
+//! anyway the moment someone clicks the card — `card_by_id` is a different code
+//! path and still fetches it. Paying twenty extra calls per row up front to
+//! pre-fill ids nobody reads until a click would defeat the branch's purpose.
+//!
+//! ## Why nothing is cached here
+//!
+//! A catalog list is **mutable by nature** — "popular this week" is the one
+//! thing that must not be frozen. The SDK's redb tables are permanent-hit caches
+//! sized for immutable data (`external_ids` is explicitly "cached forever"), so
+//! persisting a popularity list there would pin the wall to whatever it looked
+//! like on first boot. The layer above already de-duplicates the repeat cost:
+//! the gateway persists each emitted card to meta-core and its search-coverage
+//! gate then skips a repeated `(upstream, query)` for an hour, serving meta-core
+//! instead. Every call here is budget-gated, so a burst cannot run away.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use meta_feeder_sdk::plugin::GatewayQuery;
+use tracing::debug;
+
+use crate::card::{Card, CardSource};
+use crate::consts::{DISCOVERY_EXCLUDED_KEYWORDS, DISCOVERY_MAX_PAGES, DISCOVERY_MIN_VOTES};
+use crate::resolve::{dedup_titles, Resolver};
+use crate::tmdb_client::{TmdbHit, TmdbKind};
+
+/// TMDB keyword id for "anime". `/discover` is the only catalog endpoint that
+/// takes a keyword filter, which is why `anime:true` switches endpoints rather
+/// than adding a parameter.
+const TMDB_ANIME_KEYWORD: &str = "210024";
+
+/// Results TMDB returns per catalog page — used to translate a desired card
+/// count into a page count.
+const TMDB_PAGE_SIZE: usize = 20;
+
+/// The catalog-list flavour a discovery query asks for.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+pub(crate) enum DiscoveryMode {
+    Popular,
+    Trending,
+    TopRated,
+}
+
+impl DiscoveryMode {
+    /// The query-filter key that selects this mode.
+    ///
+    /// It is also echoed onto every emitted record — not here, but by
+    /// [`Card::to_record`]'s trailing filter echo, which stamps back every
+    /// filter the query carried that the card did not already set. That echo is
+    /// load-bearing: both the gateway dispatcher and meta-search's consumer-side
+    /// `record_matches` re-apply the query's filters to each record, and a
+    /// *missing* field fails the match, so an un-echoed `popular` would see
+    /// every card dropped at one of the two tiers.
+    fn marker(self) -> &'static str {
+        match self {
+            DiscoveryMode::Popular => "popular",
+            DiscoveryMode::Trending => "trending",
+            DiscoveryMode::TopRated => "top_rated",
+        }
+    }
+
+    /// `sort_by` value for the `/discover` (anime) path. `/discover` has no
+    /// "trending", so trending approximates to popularity there — the non-anime
+    /// path still uses the real `/trending` endpoint.
+    fn discover_sort(self) -> &'static str {
+        match self {
+            DiscoveryMode::TopRated => "vote_average.desc",
+            DiscoveryMode::Popular | DiscoveryMode::Trending => "popularity.desc",
+        }
+    }
+}
+
+/// True when this is a keyword-less catalog query this branch should answer.
+///
+/// Three conditions, and each excludes a query that belongs to a *different*
+/// branch of `handle_query`: no free text (that is the principal search), no
+/// `tmdbid:` filter (that is the direct by-id lookup a deep link uses), and a
+/// truthy mode marker. Note `contentKind` alone deliberately does **not** select
+/// this branch — a text query carries it too.
+pub(crate) fn is_discovery_query(query: &GatewayQuery) -> bool {
+    query.free_text.trim().is_empty()
+        && !query.filters.contains_key("tmdbid")
+        && discovery_mode(query).is_some()
+}
+
+/// First truthy mode marker on the query, in precedence order.
+fn discovery_mode(query: &GatewayQuery) -> Option<DiscoveryMode> {
+    for mode in [
+        DiscoveryMode::Trending,
+        DiscoveryMode::Popular,
+        DiscoveryMode::TopRated,
+    ] {
+        if filter_is_true(query, mode.marker()) {
+            return Some(mode);
+        }
+    }
+    None
+}
+
+/// True iff `query.filters[key]` carries a truthy value.
+fn filter_is_true(query: &GatewayQuery, key: &str) -> bool {
+    query
+        .filters
+        .get(key)
+        .is_some_and(|v| v.iter().any(|s| s.eq_ignore_ascii_case("true")))
+}
+
+/// Map the query's `contentKind` filter to a TMDB media kind.
+///
+/// `series` is the canonical spelling for a card (it describes a whole work),
+/// but `episode`/`tvshow`/`tv` are accepted so the dialect stays a superset of
+/// the indexer feeder's — a client that has not yet switched its row queries
+/// still resolves the right endpoint here, even though its records will then be
+/// dropped downstream for the mismatch.
+fn discovery_kind(query: &GatewayQuery) -> Option<TmdbKind> {
+    let values = query.filters.get("contentKind")?;
+    for v in values {
+        match v.trim().to_ascii_lowercase().as_str() {
+            "movie" => return Some(TmdbKind::Movie),
+            "series" | "episode" | "tvshow" | "tv" => return Some(TmdbKind::Tv),
+            _ => {}
+        }
+    }
+    None
+}
+
+/// The genre slugs a `genres:` filter requested, folded for comparison.
+///
+/// Multi-valued = OR, matching the filter semantics on both re-validating tiers.
+fn requested_genres(query: &GatewayQuery) -> Vec<String> {
+    query
+        .filters
+        .get("genres")
+        .map(|vs| {
+            vs.iter()
+                .map(|v| genre_fold(v))
+                .filter(|v| !v.is_empty())
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+/// Fold a genre label for comparison: lowercase, alphanumerics only.
+///
+/// Must stay identical to `genres_filter_matches`'s fold in the SDK and in
+/// meta-search — those two re-apply the filter to every record this emits, so a
+/// divergence here means the row's own results get dropped downstream.
+fn genre_fold(s: &str) -> String {
+    s.chars()
+        .filter(|c| c.is_alphanumeric())
+        .flat_map(|c| c.to_lowercase())
+        .collect()
+}
+
+/// Resolve the requested genre slugs to TMDB genre ids via the id → name table.
+/// Unknown slugs are dropped; if none resolve the caller omits `with_genres`
+/// rather than sending an empty filter that TMDB would reject.
+fn with_genres_ids(wanted: &[String], names: &HashMap<u32, String>) -> Vec<u32> {
+    let mut ids: Vec<u32> = names
+        .iter()
+        .filter(|(_, name)| wanted.contains(&genre_fold(name)))
+        .map(|(id, _)| *id)
+        .collect();
+    ids.sort_unstable();
+    ids
+}
+
+/// Build the `path_and_query` for `TmdbClient::discovery_list` from the mode,
+/// kind, anime flag, requested genre ids, and 1-based `page`.
+fn build_path_and_query(
+    mode: DiscoveryMode,
+    kind: TmdbKind,
+    anime: bool,
+    genre_ids: &[u32],
+    page: u32,
+) -> String {
+    let seg = match kind {
+        TmdbKind::Movie => "movie",
+        TmdbKind::Tv => "tv",
+    };
+    // Trending is the one mode with no `/discover` equivalent: `/trending` is a
+    // distinct, TMDB-computed "this week" signal, not a sort order. So it keeps
+    // the list endpoint — and, being a list endpoint, **cannot carry the adult
+    // gate below**. The anime variant still routes through `/discover` (there is
+    // no keyword-filtered trending at all), approximating it with popularity.
+    // …and a genre-scoped row can't use it either: `/trending` takes no
+    // `with_genres`, and returning an unfiltered trending page for an "Action"
+    // row would fill it with whatever is trending. `/discover` sorted by
+    // popularity is the closest filterable approximation.
+    if matches!(mode, DiscoveryMode::Trending) && !anime && genre_ids.is_empty() {
+        return format!("trending/{seg}/week?page={page}");
+    }
+
+    // Everything else goes through `/discover`, including the non-anime rows
+    // that used to use `/{seg}/popular` and `/{seg}/top_rated`.
+    //
+    // Those fixed list endpoints accept neither `without_keywords` nor
+    // `vote_count.gte`, and measurement showed they genuinely need them: three
+    // hentai titles reached positions 4, 5 and 17 of the *non-anime* `tv/popular`
+    // row for the same traffic-metric reason described on
+    // [`DISCOVERY_EXCLUDED_KEYWORDS`]. `/discover` with `sort_by=popularity.desc`
+    // is the filterable equivalent, so one code path now carries the gate for
+    // every mode that can hold it.
+    //
+    // Top-rated floors higher than the shared gate: sorting by rating puts a
+    // single 10.0 vote on top, which a 50-vote gate doesn't stop.
+    let votes = match mode {
+        DiscoveryMode::TopRated => 200,
+        _ => DISCOVERY_MIN_VOTES,
+    };
+    let keyword = if anime {
+        format!("with_keywords={TMDB_ANIME_KEYWORD}&")
+    } else {
+        String::new()
+    };
+    // `with_genres` is comma-joined = AND on TMDB's side, pipe-joined = OR. A
+    // multi-valued filter means OR everywhere else in the DSL, so join with `|`
+    // to keep the wire semantics and the upstream semantics saying the same
+    // thing. URL-encoded, since a bare `|` is not valid in a query string.
+    let genres = if genre_ids.is_empty() {
+        String::new()
+    } else {
+        let joined = genre_ids
+            .iter()
+            .map(|id| id.to_string())
+            .collect::<Vec<_>>()
+            .join("%7C");
+        format!("with_genres={joined}&")
+    };
+    format!(
+        "discover/{seg}?{keyword}{genres}sort_by={}&include_adult=false\
+         &without_keywords={DISCOVERY_EXCLUDED_KEYWORDS}&vote_count.gte={votes}&page={page}",
+        mode.discover_sort(),
+    )
+}
+
+/// Project one catalog hit into a [`Card`], with no further network calls.
+///
+/// `source_id` uses the same `"tv:{id}"` / `"movie:{id}"` shape
+/// `Resolver::card_by_id` produces, so a discovery card's `record_id` — and
+/// therefore its `0x1007` locator CID — is byte-identical to the one the by-id
+/// path derives for the same work. That is what lets a card the wall surfaced
+/// and a card the search box surfaced converge on one record instead of two.
+fn card_from_hit(hit: &TmdbHit, kind: TmdbKind, genres: &HashMap<u32, String>) -> Card {
+    let seg = match kind {
+        TmdbKind::Movie => "movie",
+        TmdbKind::Tv => "tv",
+    };
+    let mut titles = vec![hit.title.clone()];
+    if let Some(o) = &hit.original_title {
+        titles.push(o.clone());
+    }
+    titles.extend(hit.alt_titles.iter().cloned());
+    Card {
+        source: CardSource::Tmdb,
+        kind,
+        source_id: format!("{seg}:{}", hit.tmdbid),
+        tmdbid: Some(hit.tmdbid),
+        // Not available on a list response — see the module doc. Resolved on
+        // the click-time `card_by_id` path instead.
+        tvdb_id: None,
+        imdb_id: None,
+        title: hit.title.clone(),
+        guard_titles: Arc::new(dedup_titles(titles)),
+        overview: hit.overview.clone().filter(|s| !s.trim().is_empty()),
+        poster_path: hit.poster_path.clone().filter(|s| !s.trim().is_empty()),
+        // A list hit carries bare genre ids; the table names them.
+        genres: hit.genre_ids.iter().filter_map(|id| genres.get(id).cloned()).collect(),
+        year: hit.year,
+        // A list response carries no season structure. `seasonCount` is simply
+        // omitted from the record (`Card::to_record` skips it at `seasons == 0`).
+        seasons: 0,
+        season_summaries: Arc::new(Vec::new()),
+    }
+}
+
+/// Resolve a keyword-less catalog query to up to `max_results` cards, walking
+/// TMDB pages until the cap is reached or a page comes back empty.
+///
+/// Best-effort throughout: no mode or no `contentKind` yields `[]`, and a
+/// rate-limit or miss breaks the walk and returns what was already gathered.
+/// The branch never errors — an empty row is the failure mode, matching
+/// `Resolver::cards_for_text`.
+pub(crate) async fn cards_for_discovery(
+    resolver: &Resolver,
+    query: &GatewayQuery,
+    max_results: usize,
+) -> Vec<Card> {
+    let (Some(mode), Some(kind)) = (discovery_mode(query), discovery_kind(query)) else {
+        return Vec::new();
+    };
+    let anime = filter_is_true(query, "anime");
+    // Fetched once per process, not per row (see `Resolver::genre_names`).
+    let gmap = resolver.genre_names(kind).await;
+    // A `genres:` filter narrows the row upstream via `with_genres`, so the row
+    // is a full page OF that genre rather than a page filtered down to it.
+    let wanted = requested_genres(query);
+    let genre_ids = with_genres_ids(&wanted, &gmap);
+    if !wanted.is_empty() && genre_ids.is_empty() {
+        // Asked for a genre this kind doesn't have (e.g. `genres:kids` on the
+        // movie side, where TMDB files it as "Family"). Returning an unfiltered
+        // page would silently fill the row with the wrong thing.
+        debug!(
+            target: "meta-share::gateway", upstream = "tmdb",
+            ?wanted, "discovery: no TMDB genre id matched; row left empty"
+        );
+        return Vec::new();
+    }
+    // One page of headroom over the arithmetic minimum: `out` counts *displayable*
+    // cards, and a catalog page reliably contains a few entries with no overview
+    // or no poster (newly-added or non-English works), which are declined at
+    // projection. Without the spare page a 20-card row comes back at 17.
+    let pages = (max_results.div_ceil(TMDB_PAGE_SIZE).max(1) as u32 + 1).min(DISCOVERY_MAX_PAGES);
+    let mut out: Vec<Card> = Vec::new();
+    for page in 1..=pages {
+        if out.len() >= max_results {
+            break;
+        }
+        let path_and_query = build_path_and_query(mode, kind, anime, &genre_ids, page);
+        let Some(hits) = resolver.discovery_page(kind, &path_and_query).await else {
+            break; // budget exhausted, 429, or miss — keep what we have
+        };
+        if hits.is_empty() {
+            break; // walked past the last catalog page
+        }
+        out.extend(
+            hits.iter()
+                .map(|h| card_from_hit(h, kind, &gmap))
+                .filter(Card::is_displayable),
+        );
+    }
+    out.truncate(max_results);
+    debug!(
+        target: "meta-share::gateway",
+        upstream = "tmdb",
+        mode = mode.marker(),
+        anime,
+        count = out.len(),
+        "resolved discovery cards"
+    );
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::BTreeMap;
+
+    fn query(filters: &[(&str, &str)]) -> GatewayQuery {
+        let mut map: BTreeMap<String, Vec<String>> = BTreeMap::new();
+        for (k, v) in filters {
+            map.entry(k.to_string()).or_default().push(v.to_string());
+        }
+        GatewayQuery {
+            raw_text: String::new(),
+            free_text: String::new(),
+            filters: map,
+            ranges: Vec::new(),
+            negations: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn selects_only_keyword_less_mode_queries() {
+        assert!(is_discovery_query(&query(&[
+            ("popular", "true"),
+            ("contentKind", "series"),
+            ("fileType", "card"),
+        ])));
+        // A mode marker but with free text → the principal search owns it.
+        let mut with_text = query(&[("popular", "true"), ("contentKind", "series")]);
+        with_text.free_text = "frieren".to_string();
+        assert!(!is_discovery_query(&with_text));
+        // An explicit id → the by-id lookup owns it.
+        assert!(!is_discovery_query(&query(&[
+            ("popular", "true"),
+            ("tmdbid", "95479"),
+        ])));
+        // contentKind alone must NOT select this branch.
+        assert!(!is_discovery_query(&query(&[
+            ("contentKind", "series"),
+            ("fileType", "card"),
+        ])));
+        // A falsey marker is not a marker.
+        assert!(!is_discovery_query(&query(&[("popular", "false")])));
+    }
+
+    /// `contentKind:series` is the canonical card spelling; the indexer feeder's
+    /// older `episode` spelling still resolves the same endpoint.
+    #[test]
+    fn content_kind_maps_series_and_the_legacy_spellings() {
+        assert_eq!(
+            discovery_kind(&query(&[("contentKind", "series")])),
+            Some(TmdbKind::Tv)
+        );
+        assert_eq!(
+            discovery_kind(&query(&[("contentKind", "episode")])),
+            Some(TmdbKind::Tv)
+        );
+        assert_eq!(
+            discovery_kind(&query(&[("contentKind", "movie")])),
+            Some(TmdbKind::Movie)
+        );
+        assert_eq!(discovery_kind(&query(&[("contentKind", "audiobook")])), None);
+        assert_eq!(discovery_kind(&query(&[])), None);
+    }
+
+    #[test]
+    fn builds_the_catalog_endpoints() {
+        // Popular / top-rated route through /discover (so the gate applies);
+        // only the sort and the vote floor differ.
+        let pop = build_path_and_query(DiscoveryMode::Popular, TmdbKind::Movie, false, &[], 1);
+        assert!(pop.starts_with("discover/movie?"), "{pop}");
+        assert!(pop.contains("sort_by=popularity.desc"), "{pop}");
+        assert!(pop.contains("page=1"), "{pop}");
+
+        let paged = build_path_and_query(DiscoveryMode::Popular, TmdbKind::Tv, false, &[], 2);
+        assert!(paged.starts_with("discover/tv?"), "{paged}");
+        assert!(paged.contains("page=2"), "{paged}");
+
+        let top = build_path_and_query(DiscoveryMode::TopRated, TmdbKind::Movie, false, &[], 1);
+        assert!(top.contains("sort_by=vote_average.desc"), "{top}");
+        assert!(top.contains("vote_count.gte=200"), "{top}");
+
+        // Trending keeps the real endpoint — see the dedicated test.
+        assert_eq!(
+            build_path_and_query(DiscoveryMode::Trending, TmdbKind::Tv, false, &[], 1),
+            "trending/tv/week?page=1"
+        );
+    }
+
+    #[test]
+    fn anime_switches_to_the_keyword_filtered_discover_endpoint() {
+        let p = build_path_and_query(DiscoveryMode::Popular, TmdbKind::Tv, true, &[], 1);
+        assert!(p.starts_with("discover/tv?"), "{p}");
+        assert!(p.contains("with_keywords=210024"), "{p}");
+        assert!(p.contains("sort_by=popularity.desc"), "{p}");
+        let top = build_path_and_query(DiscoveryMode::TopRated, TmdbKind::Movie, true, &[], 1);
+        assert!(top.contains("sort_by=vote_average.desc"), "{top}");
+        // Sorting by rating puts a 1-vote 10.0 on top, so top-rated floors higher.
+        assert!(top.contains("vote_count.gte=200"), "{top}");
+    }
+
+    /// The gate is not anime-specific. Three hentai titles were measured at
+    /// positions 4, 5 and 17 of the **non-anime** `tv/popular` row, which is why
+    /// the plain popular/top-rated modes were moved off the fixed list endpoints
+    /// (which cannot carry `without_keywords`) and onto `/discover`.
+    #[test]
+    fn non_anime_popular_and_top_rated_also_gate() {
+        for mode in [DiscoveryMode::Popular, DiscoveryMode::TopRated] {
+            let p = build_path_and_query(mode, TmdbKind::Tv, false, &[], 1);
+            assert!(p.starts_with("discover/tv?"), "{p}");
+            assert!(!p.contains("with_keywords="), "no anime narrowing here: {p}");
+            assert!(p.contains("without_keywords="), "{p}");
+            assert!(p.contains("vote_count.gte="), "{p}");
+        }
+    }
+
+    /// Trending is the documented exception: `/trending` is a TMDB-computed
+    /// signal with no `/discover` sort equivalent, so the non-anime trending row
+    /// keeps the list endpoint and cannot carry the gate.
+    #[test]
+    fn non_anime_trending_keeps_the_real_trending_endpoint() {
+        let p = build_path_and_query(DiscoveryMode::Trending, TmdbKind::Tv, false, &[], 1);
+        assert_eq!(p, "trending/tv/week?page=1");
+        // …but the anime variant has no keyword-filtered trending at all, so it
+        // still routes through /discover and stays gated.
+        let a = build_path_and_query(DiscoveryMode::Trending, TmdbKind::Tv, true, &[], 1);
+        assert!(a.starts_with("discover/tv?"), "{a}");
+        assert!(a.contains("without_keywords="), "{a}");
+    }
+
+    /// A genre-scoped row narrows **upstream** via `with_genres`, so the row is a
+    /// full page of that genre rather than a page filtered down to a handful.
+    #[test]
+    fn a_genre_row_narrows_upstream() {
+        let p = build_path_and_query(DiscoveryMode::Popular, TmdbKind::Tv, false, &[10759], 1);
+        assert!(p.contains("with_genres=10759"), "{p}");
+        // Multi-valued = OR, so pipe-joined (comma would mean AND to TMDB).
+        let multi = build_path_and_query(DiscoveryMode::Popular, TmdbKind::Tv, false, &[16, 35], 1);
+        assert!(multi.contains("with_genres=16%7C35"), "{multi}");
+        // No genre filter ⇒ the parameter is absent, not empty.
+        let none = build_path_and_query(DiscoveryMode::Popular, TmdbKind::Tv, false, &[], 1);
+        assert!(!none.contains("with_genres"), "{none}");
+    }
+
+    /// A genre row can't use `/trending` (it takes no `with_genres`), so it falls
+    /// back to the filterable `/discover` rather than returning an unfiltered
+    /// trending page under a genre heading.
+    #[test]
+    fn a_genre_scoped_trending_row_falls_back_to_discover() {
+        let p = build_path_and_query(DiscoveryMode::Trending, TmdbKind::Tv, false, &[10759], 1);
+        assert!(p.starts_with("discover/tv?"), "{p}");
+        assert!(p.contains("with_genres=10759"), "{p}");
+    }
+
+    /// Slugs resolve to TMDB ids through the id → name table, folded on both
+    /// sides so the DSL never has to spell a space or an ampersand.
+    #[test]
+    fn genre_slugs_resolve_to_tmdb_ids() {
+        let names = HashMap::from([
+            (16u32, "Animation".to_string()),
+            (10759, "Action & Adventure".to_string()),
+            (10765, "Sci-Fi & Fantasy".to_string()),
+        ]);
+        let want = |s: &str| with_genres_ids(&[genre_fold(s)], &names);
+        assert_eq!(want("action-adventure"), vec![10759]);
+        assert_eq!(want("Action & Adventure"), vec![10759]);
+        assert_eq!(want("scififantasy"), vec![10765]);
+        assert_eq!(want("animation"), vec![16]);
+        assert!(want("horror").is_empty(), "unknown slug resolves to nothing");
+    }
+
+    /// The fold must stay identical to the one both re-validating tiers use, or
+    /// a row's own results get dropped downstream.
+    #[test]
+    fn genre_fold_matches_the_consumer_side_rule() {
+        assert_eq!(genre_fold("Action & Adventure"), "actionadventure");
+        assert_eq!(genre_fold("action-adventure"), "actionadventure");
+        assert_eq!(genre_fold("Sci-Fi & Fantasy"), "scififantasy");
+        assert_eq!(genre_fold("Science Fiction"), "sciencefiction");
+        assert_eq!(genre_fold("  "), "");
+    }
+
+    /// The adult-content gate on anime rows, and *why* it is two filters.
+    ///
+    /// `include_adult=false` is not enough on its own — measured live, every
+    /// offending title returns `adult: false` (TMDB reserves that flag for
+    /// pornography), so the keyword exclusion carries the real load and the vote
+    /// floor sweeps up the erotic-tagged residue behind it. Dropping either one
+    /// puts hentai back on the home page.
+    #[test]
+    fn anime_rows_gate_adult_content_three_ways() {
+        for mode in [
+            DiscoveryMode::Popular,
+            DiscoveryMode::Trending,
+            DiscoveryMode::TopRated,
+        ] {
+            for kind in [TmdbKind::Tv, TmdbKind::Movie] {
+                let p = build_path_and_query(mode, kind, true, &[], 1);
+                assert!(p.contains("include_adult=false"), "{p}");
+                assert!(
+                    p.contains(&format!("without_keywords={DISCOVERY_EXCLUDED_KEYWORDS}")),
+                    "{p}"
+                );
+                assert!(p.contains("vote_count.gte="), "{p}");
+            }
+        }
+    }
+
+    fn genre_table() -> HashMap<u32, String> {
+        HashMap::from([(16, "Animation".to_string()), (18, "Drama".to_string())])
+    }
+
+    fn hit() -> TmdbHit {
+        TmdbHit {
+            tmdbid: 95479,
+            title: "Frieren: Beyond Journey's End".to_string(),
+            original_title: Some("葬送のフリーレン".to_string()),
+            original_language: Some("ja".to_string()),
+            overview: Some("A mage reflects on her long life.".to_string()),
+            year: Some(2023),
+            poster_path: Some("/poster.jpg".to_string()),
+            genre_ids: vec![16],
+            alt_titles: Vec::new(),
+        }
+    }
+
+    /// A list hit must carry everything a renderable card needs, so the branch
+    /// costs one TMDB call per page and no by-id follow-ups.
+    #[test]
+    fn a_list_hit_becomes_a_complete_card() {
+        let c = card_from_hit(&hit(), TmdbKind::Tv, &genre_table());
+        assert_eq!(c.tmdbid, Some(95479));
+        assert!(c.overview.is_some(), "overview is what to_record requires");
+        assert!(c.poster_path.is_some(), "poster is what to_record requires");
+        assert_eq!(c.year, Some(2023));
+        assert_eq!(c.content_kind(), "series");
+        // Genres come from the id table — the category axis a card used to lack.
+        assert_eq!(c.genres, vec!["Animation".to_string()]);
+        // The id bag beyond tmdbid needs external_ids — deliberately absent.
+        assert_eq!(c.tvdb_id, None);
+        assert_eq!(c.imdb_id, None);
+    }
+
+    /// The locator CID is derived from `record_id`, so a discovery card and a
+    /// by-id card for the same work must address identically or the two tiers
+    /// would publish two records for one work.
+    #[test]
+    fn record_id_matches_the_by_id_path() {
+        assert_eq!(card_from_hit(&hit(), TmdbKind::Tv, &genre_table()).record_id(), "tmdb:tv:95479");
+        assert_eq!(
+            card_from_hit(&hit(), TmdbKind::Movie, &genre_table()).record_id(),
+            "tmdb:movie:95479"
+        );
+    }
+
+    /// The whole point of the branch: `contentKind` on the emitted record is the
+    /// *card's* kind, never an echo of the query filter. A row asking
+    /// `contentKind:episode` must still produce `series` here — which is exactly
+    /// why a client's row query has to ask for `series` (the mismatch is then
+    /// caught by `record_matches` downstream rather than silently mislabelling).
+    /// **The invariant this whole branch rests on.** The mode/anime markers are
+    /// stamped onto the record by `Card::to_record`'s trailing filter echo, not
+    /// by anything here — and both the gateway dispatcher and meta-search
+    /// re-apply the query's filters to every record, where a *missing* field
+    /// fails the match. So if that echo is ever "tidied up", every discovery
+    /// card is silently dropped in transit and the wall goes blank with no
+    /// error. This test fails that day.
+    #[test]
+    fn a_discovery_card_survives_the_consumer_re_filter() {
+        let q = query(&[
+            ("popular", "true"),
+            ("contentKind", "series"),
+            ("fileType", "card"),
+            ("anime", "true"),
+        ]);
+        let card = card_from_hit(&hit(), TmdbKind::Tv, &genre_table());
+        let tmdb = crate::tmdb_client::TmdbClient::new("token".to_string());
+        let rec = card.to_record(&tmdb, &q.filters).expect("displayable card emits");
+
+        assert_eq!(rec.fields.get("popular").map(String::as_str), Some("true"));
+        assert_eq!(rec.fields.get("anime").map(String::as_str), Some("true"));
+        assert_eq!(rec.fields.get("fileType").map(String::as_str), Some("card"));
+        assert_eq!(
+            rec.fields.get("contentKind").map(String::as_str),
+            Some("series"),
+            "the card's own kind must win over the query filter"
+        );
+        assert!(rec.fields.contains_key("description/eng"));
+        assert!(rec.fields.contains_key("poster_url"));
+        // Genres ride as a key-set, never the legacy comma-joined value
+        // (METADATA_KEYS §14.12 — new writers must not add csv-set fields).
+        assert_eq!(rec.fields.get("genres/Animation").map(String::as_str), Some("true"));
+        assert!(!rec.fields.contains_key("genres"), "must not write the legacy csv-set");
+        // The thin id bag: resolved lazily on the click-time by-id path.
+        assert!(!rec.fields.contains_key("tvdbid"));
+        assert!(!rec.fields.contains_key("imdbid"));
+        assert!(!rec.fields.contains_key("seasonCount"));
+
+        assert!(
+            meta_feeder_sdk::query_eval::record_matches(&rec.fields, &q),
+            "a discovery card must survive the same filter re-application the \
+             dispatcher and meta-search perform"
+        );
+    }
+
+    /// A hit TMDB has not written up yet is declined at projection — which is
+    /// why the page walk counts displayable cards, not raw hits.
+    #[test]
+    fn an_incomplete_hit_is_not_displayable() {
+        let mut no_overview = hit();
+        no_overview.overview = None;
+        assert!(!card_from_hit(&no_overview, TmdbKind::Tv, &genre_table()).is_displayable());
+
+        let mut no_poster = hit();
+        no_poster.poster_path = None;
+        assert!(!card_from_hit(&no_poster, TmdbKind::Tv, &genre_table()).is_displayable());
+
+        assert!(card_from_hit(&hit(), TmdbKind::Tv, &genre_table()).is_displayable());
+    }
+
+    #[test]
+    fn content_kind_comes_from_the_card_not_the_filter() {
+        let tv = card_from_hit(&hit(), TmdbKind::Tv, &genre_table());
+        assert_eq!(tv.content_kind(), "series");
+        let movie = card_from_hit(&hit(), TmdbKind::Movie, &genre_table());
+        assert_eq!(movie.content_kind(), "movie");
+    }
+}

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

@@ -12,6 +12,7 @@
 
 pub mod card;
 pub mod consts;
+pub mod discovery;
 pub mod resolve;
 pub mod tmdb;
 pub mod tmdb_budget;

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

@@ -9,6 +9,7 @@
 //! The redb table names are unchanged from the indexer feeder's cache, so a
 //! warm cache carries straight over.
 
+use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -16,7 +17,9 @@ use meta_feeder_sdk::cache::MidhashCache;
 use tracing::debug;
 
 use crate::card::{Card, CardSource};
-use crate::consts::{CACHED_PRINCIPAL_DEPTH, TMDB_WAIT_DEADLINE_SECS};
+use crate::consts::{
+    CACHED_PRINCIPAL_DEPTH, TMDB_DISCOVERY_WAIT_DEADLINE_SECS, TMDB_WAIT_DEADLINE_SECS,
+};
 use crate::tmdb_budget::{Lease, TmdbBudget};
 use crate::tmdb_client::{
     principal_top_n, TmdbCall, TmdbClient, TmdbExternalIds, TmdbHit, TmdbKind, TmdbTvDetails,
@@ -28,6 +31,14 @@ pub(crate) struct Resolver {
     pub(crate) client: Arc<TmdbClient>,
     pub(crate) cache: MidhashCache,
     pub(crate) budget: Arc<TmdbBudget>,
+    /// `genre id → name` per kind, held for the process lifetime.
+    ///
+    /// In memory rather than in the redb cache on purpose: the table is 16 TV /
+    /// 19 movie entries and effectively static, so a process-local map costs one
+    /// call per kind per boot and needs no invalidation story — whereas the redb
+    /// tables are permanent-hit and shared byte-for-byte with the indexer
+    /// feeder's, which this is not part of.
+    genre_names: Arc<std::sync::RwLock<HashMap<&'static str, Arc<HashMap<u32, String>>>>>,
 }
 
 impl Resolver {
@@ -36,20 +47,68 @@ impl Resolver {
             client,
             cache,
             budget,
+            genre_names: Arc::new(std::sync::RwLock::new(HashMap::new())),
         }
     }
 
+    /// The `genre id → name` table for `kind`, fetched once and then reused.
+    ///
+    /// Needed only by the **list** paths (discovery/search), where TMDB gives
+    /// bare `genre_ids`; a details fetch returns named genre objects directly.
+    /// Best-effort: a failed or empty fetch yields an empty map (cards simply
+    /// carry no genres) and is **not** cached, so the next row retries.
+    pub(crate) async fn genre_names(&self, kind: TmdbKind) -> Arc<HashMap<u32, String>> {
+        let key = match kind {
+            TmdbKind::Tv => "tv",
+            TmdbKind::Movie => "movie",
+        };
+        if let Some(m) = self.genre_names.read().ok().and_then(|g| g.get(key).cloned()) {
+            return m;
+        }
+        let fetched: HashMap<u32, String> = self
+            .budgeted_with_deadline(
+                TMDB_DISCOVERY_WAIT_DEADLINE_SECS,
+                self.client.genre_list(kind),
+                |gs| {
+                    gs.into_iter()
+                        .filter(|g| !g.name.trim().is_empty())
+                        .map(|g| (g.id, g.name))
+                        .collect()
+                },
+            )
+            .await
+            .unwrap_or_default();
+        let arc = Arc::new(fetched);
+        if !arc.is_empty() {
+            if let Ok(mut g) = self.genre_names.write() {
+                g.insert(key, arc.clone());
+            }
+        }
+        arc
+    }
+
     /// Acquire a budget token, run `call`, map a hit through `on_hit`. A 429
     /// pauses the shared bucket globally rather than being retried here.
     async fn budgeted<T, R>(
         &self,
         call: impl std::future::Future<Output = TmdbCall<T>>,
         on_hit: impl FnOnce(T) -> R,
+    ) -> Option<R> {
+        self.budgeted_with_deadline(TMDB_WAIT_DEADLINE_SECS, call, on_hit)
+            .await
+    }
+
+    /// [`Self::budgeted`] with a caller-chosen permit deadline, for callers whose
+    /// cost of waiting differs from a user-facing lookup's — see
+    /// [`TMDB_DISCOVERY_WAIT_DEADLINE_SECS`].
+    async fn budgeted_with_deadline<T, R>(
+        &self,
+        deadline_secs: u64,
+        call: impl std::future::Future<Output = TmdbCall<T>>,
+        on_hit: impl FnOnce(T) -> R,
     ) -> Option<R> {
         if matches!(
-            self.budget
+            self.budget.acquire(Duration::from_secs(deadline_secs)).await,
-                .acquire(Duration::from_secs(TMDB_WAIT_DEADLINE_SECS))
-                .await,
             Lease::DeadlineExceeded
         ) {
             return None;
@@ -124,6 +183,29 @@ impl Resolver {
         .await
     }
 
+    /// One page of a TMDB **catalog list** (popular / trending / top-rated /
+    /// `discover`) — the keyword-less browse primitive behind
+    /// [`crate::discovery`]. `None` on a budget timeout, a 429, or a miss, which
+    /// the caller treats as "stop walking pages and keep what you have".
+    ///
+    /// Deliberately **uncached**, unlike every other method here: a catalog list
+    /// is the one TMDB response that is *supposed* to change under you, and the
+    /// redb tables in front of the others are permanent-hit caches with no TTL.
+    /// See the module doc of [`crate::discovery`] for why the repeat cost is
+    /// already absorbed a layer up.
+    pub(crate) async fn discovery_page(
+        &self,
+        kind: TmdbKind,
+        path_and_query: &str,
+    ) -> Option<Vec<TmdbHit>> {
+        self.budgeted_with_deadline(
+            TMDB_DISCOVERY_WAIT_DEADLINE_SECS,
+            self.client.discovery_list(kind, path_and_query),
+            |hits| hits,
+        )
+        .await
+    }
+
     /// Cached principal `search/multi`: free text → up to `n` confident
     /// `(tmdbid, kind)` candidates, most popular first.
     ///
@@ -194,6 +276,7 @@ impl Resolver {
                     guard_titles: Arc::new(dedup_titles(details.guard_titles())),
                     overview: details.overview.clone().filter(|s| !s.is_empty()),
                     poster_path: details.poster_path.clone().filter(|s| !s.is_empty()),
+                    genres: genre_names_of(&details.genres),
                     year,
                     seasons: Card::clamp_seasons(details.number_of_seasons),
                     season_summaries: Arc::new(details.seasons.clone()),
@@ -201,6 +284,15 @@ impl Resolver {
             }
             TmdbKind::Movie => {
                 let hit = self.movie_hit(tmdbid).await?;
+                // The movie path round-trips through `TmdbHit` (that is what the
+                // details cache stores), which keeps genres as bare ids — so
+                // unlike the TV branch above it needs the id → name table.
+                let gmap = self.genre_names(TmdbKind::Movie).await;
+                let genres: Vec<String> = hit
+                    .genre_ids
+                    .iter()
+                    .filter_map(|id| gmap.get(id).cloned())
+                    .collect();
                 let mut titles = vec![hit.title.clone()];
                 if let Some(o) = &hit.original_title {
                     titles.push(o.clone());
@@ -217,6 +309,7 @@ impl Resolver {
                     guard_titles: Arc::new(dedup_titles(titles)),
                     overview: hit.overview.filter(|s| !s.is_empty()),
                     poster_path: hit.poster_path.filter(|s| !s.is_empty()),
+                    genres,
                     year: hit.year,
                     seasons: 0,
                     season_summaries: Arc::new(Vec::new()),
@@ -244,8 +337,19 @@ impl Resolver {
     }
 }
 
+/// Genre display names off a details payload's `{id, name}` array, trimmed and
+/// de-duplicated, order preserved.
+pub(crate) fn genre_names_of(genres: &[crate::tmdb_client::TmdbGenre]) -> Vec<String> {
+    let mut seen = std::collections::HashSet::new();
+    genres
+        .iter()
+        .map(|g| g.name.trim().to_string())
+        .filter(|n| !n.is_empty() && seen.insert(n.to_lowercase()))
+        .collect()
+}
+
 /// Case-insensitive de-dup preserving first-seen order.
-fn dedup_titles(titles: Vec<String>) -> Vec<String> {
+pub(crate) fn dedup_titles(titles: Vec<String>) -> Vec<String> {
     let mut seen = std::collections::HashSet::new();
     titles
         .into_iter()

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

@@ -22,7 +22,7 @@ use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth};
 use tracing::warn;
 
 use crate::card::{split_record_id, Card, CardSource};
-use crate::consts::DEFAULT_CARD_TOP_N;
+use crate::consts::{DEFAULT_CARD_DISCOVERY_N, DEFAULT_CARD_TOP_N};
 use crate::resolve::Resolver;
 use crate::tmdb_budget::{TmdbBudget, DEFAULT_TMDB_BURST, DEFAULT_TMDB_REFILL_PER_SEC};
 use crate::tmdb_client::TmdbClient;
@@ -39,6 +39,8 @@ pub struct TmdbConfig {
     pub tmdb_burst: Option<f64>,
     #[serde(default)]
     pub card_top_n: Option<usize>,
+    #[serde(default)]
+    pub card_discovery_n: Option<usize>,
 }
 
 impl TmdbConfig {
@@ -51,6 +53,9 @@ impl TmdbConfig {
                 .and_then(|v| v.parse().ok()),
             tmdb_burst: std::env::var("TMDB_BURST").ok().and_then(|v| v.parse().ok()),
             card_top_n: std::env::var("CARD_TOP_N").ok().and_then(|v| v.parse().ok()),
+            card_discovery_n: std::env::var("CARD_DISCOVERY_N")
+                .ok()
+                .and_then(|v| v.parse().ok()),
         }
     }
 }
@@ -62,6 +67,8 @@ pub struct TmdbCardPlugin {
     /// `depends_on` is satisfied; queries just return nothing.
     resolver: Option<Resolver>,
     top_n: usize,
+    /// Cards per keyword-less catalog row (see [`DEFAULT_CARD_DISCOVERY_N`]).
+    discovery_n: usize,
     /// Test hook: `(api_base, image_base)` overriding TMDB's real endpoints, so
     /// the contract test can point the client at a wiremock server.
     api_bases: Option<(String, String)>,
@@ -79,6 +86,7 @@ impl TmdbCardPlugin {
             config: TmdbConfig::from_env(),
             resolver: None,
             top_n: DEFAULT_CARD_TOP_N,
+            discovery_n: DEFAULT_CARD_DISCOVERY_N,
             api_bases: None,
         }
     }
@@ -111,6 +119,9 @@ impl TmdbCardPlugin {
                 if file_cfg.card_top_n.is_some() {
                     self.config.card_top_n = file_cfg.card_top_n;
                 }
+                if file_cfg.card_discovery_n.is_some() {
+                    self.config.card_discovery_n = file_cfg.card_discovery_n;
+                }
             }
         }
     }
@@ -142,6 +153,11 @@ impl FeederPlugin for TmdbCardPlugin {
         self.load_config(cache_dir);
         let cache: MidhashCache = meta_feeder_sdk::common::open_midhash_cache(cache_dir, "tmdb")?;
         self.top_n = self.config.card_top_n.filter(|n| *n > 0).unwrap_or(DEFAULT_CARD_TOP_N);
+        self.discovery_n = self
+            .config
+            .card_discovery_n
+            .filter(|n| *n > 0)
+            .unwrap_or(DEFAULT_CARD_DISCOVERY_N);
 
         let token = self.config.tmdb_token.trim().to_string();
         if token.is_empty() {
@@ -189,15 +205,33 @@ impl FeederPlugin for TmdbCardPlugin {
             self.served_file_types(),
             self.served_content_kinds(),
         ) {
+            // One misconfiguration lands here often enough to deserve a name.
+            // A browse row written against the *indexer* feeder's dialect says
+            // `contentKind:episode`, which is not in `served_content_kinds` — so
+            // the row is rejected here, before TMDB is touched, and the operator
+            // sees an empty wall with no error anywhere. A card describes a WORK,
+            // and a work is a `series`, never an `episode`.
+            if crate::discovery::is_discovery_query(query) {
+                warn!(
+                    target: "meta-share::gateway",
+                    upstream = "tmdb",
+                    content_kind = ?query.filters.get("contentKind"),
+                    "discovery row rejected by the served-kinds gate: a card is a \
+                     WORK — use contentKind:series for TV rows (not episode/tv/tvshow)"
+                );
+            }
             return Ok(Vec::new());
         }
         let Some(resolver) = self.resolver() else {
             return Ok(Vec::new()); // soft-skipped (no token)
         };
 
-        // An explicit `tmdbid:` filter resolves exactly one card — the direct
+        // Three shapes, in precedence order:
-        // "give me this work" lookup a deep link uses. Otherwise the free text
+        //   1. an explicit `tmdbid:` filter — the direct "give me this work"
-        // goes through the principal search.
+        //      lookup a deep link (or the detail page) uses;
+        //   2. a keyword-less catalog query (`popular:true contentKind:series`)
+        //      — a client's home row, answered from TMDB's catalog endpoints;
+        //   3. free text — the principal search.
         let cards: Vec<Card> = if let Some(id) = query
             .filters
             .get("tmdbid")
@@ -206,6 +240,9 @@ impl FeederPlugin for TmdbCardPlugin {
         {
             let kind = kind_hint(query, resolver, id).await;
             resolver.card_by_id(id, kind).await.into_iter().collect()
+        } else if crate::discovery::is_discovery_query(query) {
+            let n = self.discovery_n.min(max_results.max(1));
+            crate::discovery::cards_for_discovery(resolver, query, n).await
         } else {
             let free_text = query.free_text.trim();
             if free_text.is_empty() {
@@ -284,6 +321,13 @@ impl FeederPlugin for TmdbCardPlugin {
                      Blank keeps the built-in default (10). Unlike the old anchor \
                      top-N this costs no indexer requests — only TMDB.",
                 ),
+                F::number("card_discovery_n", "Cards per discovery row").with_help(
+                    "How many cards a keyword-less browse row (popular / trending / \
+                     top-rated) returns. Blank keeps the built-in default (20 — \
+                     exactly one TMDB catalog page, so a row costs ONE request). \
+                     Separate from 'Cards per search': a row is a browse surface, \
+                     not an answer to a question.",
+                ),
             ],
         }
     }
@@ -294,6 +338,7 @@ impl FeederPlugin for TmdbCardPlugin {
             "tmdb_rate_per_sec": self.config.tmdb_rate_per_sec,
             "tmdb_burst": self.config.tmdb_burst,
             "card_top_n": self.config.card_top_n,
+            "card_discovery_n": self.config.card_discovery_n,
         })
     }
 }

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

@@ -309,6 +309,27 @@ impl TmdbClient {
         }
     }
 
+    /// `GET /3/genre/{tv,movie}/list` — the id → name table for the `genre_ids`
+    /// a discovery/search list hit carries.
+    ///
+    /// TMDB returns genres as bare ids on list responses and as `{id, name}`
+    /// objects only on a details fetch, so the *browse* path has no way to name
+    /// a genre without this table. It is tiny (16 TV / 19 movie entries) and
+    /// effectively static, which is why the caller holds it for the process
+    /// lifetime instead of paying it per row.
+    pub(crate) async fn genre_list(&self, kind: TmdbKind) -> TmdbCall<Vec<TmdbGenre>> {
+        let seg = match kind {
+            TmdbKind::Movie => "movie",
+            TmdbKind::Tv => "tv",
+        };
+        let url = format!("{}/genre/{seg}/list", self.api_base.trim_end_matches('/'));
+        match self.get_json::<TmdbGenreList>(&url, "genre_list").await {
+            TmdbCall::Hit(r) => TmdbCall::Hit(r.genres),
+            TmdbCall::Miss => TmdbCall::Miss,
+            TmdbCall::RateLimited(d) => TmdbCall::RateLimited(d),
+        }
+    }
+
     /// Shared `GET <url>` → JSON helper with the standard bearer auth, search
     /// timeout, 429→`RateLimited`, non-2xx/transient→`Miss` contract. Factored
     /// out of [`tv_details`]/[`external_ids`]/[`movie_details`] (identical
@@ -393,6 +414,21 @@ pub(crate) enum TmdbKind {
     Tv,
 }
 
+/// One `{id, name}` genre entry, as returned by `genre/{kind}/list` and by the
+/// details endpoints' `genres` array.
+#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)]
+pub(crate) struct TmdbGenre {
+    pub(crate) id: u32,
+    #[serde(default)]
+    pub(crate) name: String,
+}
+
+#[derive(serde::Deserialize)]
+struct TmdbGenreList {
+    #[serde(default)]
+    genres: Vec<TmdbGenre>,
+}
+
 #[derive(serde::Deserialize)]
 pub(crate) struct TmdbSearchResponse {
     #[serde(default)]
@@ -550,6 +586,10 @@ impl TmdbHit {
 pub(crate) struct TmdbTvDetails {
     #[serde(default)]
     pub(crate) number_of_seasons: u32,
+    /// Genre objects (`{id, name}`) — present on a details fetch, unlike the
+    /// bare `genre_ids` a list response carries. Free here: same payload.
+    #[serde(default)]
+    pub(crate) genres: Vec<TmdbGenre>,
     /// One entry per season TMDB knows about, including season 0
     /// ("Specials"). `episode_count` is the released-episode total.
     #[serde(default)]
@@ -691,6 +731,9 @@ pub(crate) struct TmdbExternalIds {
 #[derive(Debug, Clone, serde::Deserialize)]
 pub(crate) struct TmdbMovieDetails {
     pub(crate) id: u64,
+    /// See [`TmdbTvDetails::genres`].
+    #[serde(default)]
+    pub(crate) genres: Vec<TmdbGenre>,
     #[serde(default)]
     pub(crate) title: Option<String>,
     #[serde(default)]