1.2.0: match show-page (tmdbid:) queries to releases by name against the mesh card, stamp anchorMethod=title; indexer source name (default nntmux); season/episode/pack from release names

feeder-plugin/usenet-feeder/src/anchor.rs

@@ -0,0 +1,425 @@
+//! Name matching for anchored queries — how a self-scanned release lands on a
+//! meta-watch show page **matched by title**, never anchored.
+//!
+//! A show page asks the gateway for `tmdbid:<id> season:<n> …` with no free
+//! text, and keeps a release only when the release itself names that `tmdbid`
+//! (and carries a poster + description). NNTmux knows release names, not TMDB
+//! ids, and this feeder makes no TMDB calls. The show's identity is already on
+//! the mesh, though: the card feeder stores every show it surfaced in the
+//! gateway's meta-core as a **card record** keyed by its locator cid
+//! (`compute_card_cid("tmdb", "tv:<id>")`, the same address meta-watch routes
+//! by). So:
+//!
+//! 1. read the card (title, AKAs, poster, description) from meta-core;
+//! 2. search NNTmux by the card's title;
+//! 3. keep the releases whose cleaned name resembles one of the card's names
+//!    and whose numbering agrees with the query;
+//! 4. stamp them with the query's id, `anchored=true` + `anchorMethod=title`,
+//!    and the card's poster/description.
+//!
+//! `anchorMethod=title` is what keeps this honest: meta-watch grades it
+//! *Matched* ("By title"), below an id-confirmed release, and the gateway files
+//! it as a durable `matchedBy/tmdb:<id>` claim rather than `anchoredBy/`.
+
+use std::collections::{BTreeMap, HashMap};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use meta_feeder_sdk::filename_meta::{clean_title, extract_season_episode};
+use meta_feeder_sdk::hash::compute_card_cid;
+use meta_feeder_sdk::query::GatewayQuery;
+use tracing::{debug, warn};
+
+/// A release name must be *about* the show, not merely mention it: the card's
+/// tokens must all appear and cover this share of the release's tokens. Same
+/// rule and threshold as the torznab feeder's anchor guard
+/// (`tokens_resemble_anchor`), which is crate-private there.
+const MIN_TITLE_COVERAGE: f64 = 0.6;
+
+/// A card found in meta-core is good for an hour; a miss is retried sooner,
+/// since the card feeder may persist it any moment.
+const CARD_TTL: Duration = Duration::from_secs(3600);
+const MISS_TTL: Duration = Duration::from_secs(300);
+
+/// What a show page's query is about.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ShowKind {
+    Tv,
+    Movie,
+}
+
+impl ShowKind {
+    fn locator_prefix(self) -> &'static str {
+        match self {
+            ShowKind::Tv => "tv",
+            ShowKind::Movie => "movie",
+        }
+    }
+}
+
+/// The anchor a query carries: the TMDB id, the kinds it may be, and the
+/// numbering it asks for.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct QueryAnchor {
+    pub tmdbid: String,
+    /// Most likely first. Both when the query does not say.
+    pub kinds: Vec<ShowKind>,
+    pub season: Option<String>,
+    pub episode: Option<String>,
+}
+
+fn first_filter<'a>(q: &'a GatewayQuery, key: &str) -> Option<&'a str> {
+    q.filters
+        .get(key)
+        .and_then(|v| v.first())
+        .map(|s| s.trim())
+        .filter(|s| !s.is_empty())
+}
+
+/// `numeric` → normalised (`"02"` → `"2"`), else `None`.
+fn number(s: &str) -> Option<String> {
+    s.trim().parse::<u32>().ok().map(|n| n.to_string())
+}
+
+impl QueryAnchor {
+    /// `None` unless the query names a numeric `tmdbid`.
+    pub fn from_query(q: &GatewayQuery) -> Option<Self> {
+        let tmdbid = first_filter(q, "tmdbid").and_then(number)?;
+        let kinds_said: Vec<&str> = q
+            .filters
+            .get("contentKind")
+            .map(|v| v.iter().map(|s| s.trim()).collect())
+            .unwrap_or_default();
+        // `pack` says nothing: a film page asks for `movie OR pack`, a show
+        // page for `series OR episode OR pack`.
+        let tv = kinds_said.iter().any(|k| matches!(*k, "episode" | "series" | "tv"));
+        let movie = kinds_said.contains(&"movie");
+        let kinds = match (tv, movie) {
+            (true, false) => vec![ShowKind::Tv],
+            (false, true) => vec![ShowKind::Movie],
+            _ => vec![ShowKind::Tv, ShowKind::Movie],
+        };
+        Some(Self {
+            tmdbid,
+            kinds,
+            season: first_filter(q, "season").and_then(number),
+            episode: first_filter(q, "episode").and_then(number),
+        })
+    }
+}
+
+/// The parts of a card record this feeder uses.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Card {
+    pub kind: ShowKind,
+    /// Display title first, then `searchTitle` and the `titles/<lang3>/<name>`
+    /// AKAs. Deduped, never empty.
+    pub names: Vec<String>,
+    /// Copied onto matched releases: poster, every `description/<lang>`, and
+    /// the cross-source ids.
+    pub stamp: BTreeMap<String, String>,
+    /// The AKA the card feeder picked for searching (`3%` → `3 percent`).
+    pub search_title: Option<String>,
+}
+
+impl Card {
+    /// `None` when the record has no usable title.
+    pub fn from_fields(kind: ShowKind, f: &BTreeMap<String, String>) -> Option<Self> {
+        let mut names: Vec<String> = Vec::new();
+        let mut push = |s: &str| {
+            let s = s.trim();
+            if !s.is_empty() && !names.iter().any(|n| n.eq_ignore_ascii_case(s)) {
+                names.push(s.to_string());
+            }
+        };
+        push(f.get("title")?);
+        if let Some(s) = f.get("searchTitle") {
+            push(s);
+        }
+        for k in f.keys() {
+            if let Some(rest) = k.strip_prefix("titles/") {
+                if let Some((_, name)) = rest.split_once('/') {
+                    push(name);
+                }
+            }
+        }
+        let stamp = f
+            .iter()
+            .filter(|(k, _)| {
+                matches!(k.as_str(), "poster" | "imdbid" | "tvdbid") || k.starts_with("description/")
+            })
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect();
+        let search_title = f.get("searchTitle").map(|s| s.trim().to_string()).filter(|s| !s.is_empty());
+        Some(Self { kind, names, stamp, search_title })
+    }
+
+    /// The name to search NNTmux with (the display title), plus `searchTitle`
+    /// when the card has one (a title NNTmux could not match as typed, `3%`).
+    pub fn search_terms(&self) -> Vec<String> {
+        let mut out = vec![self.names[0].clone()];
+        if let Some(s) = &self.search_title {
+            if !out.iter().any(|o| o.eq_ignore_ascii_case(s)) {
+                out.push(s.clone());
+            }
+        }
+        out
+    }
+}
+
+fn norm_tokens(s: &str) -> Vec<String> {
+    s.split(|c: char| !c.is_alphanumeric())
+        .filter(|t| !t.is_empty())
+        .map(|t| t.to_lowercase())
+        .collect()
+}
+
+fn tokens_resemble(release: &[String], name: &str) -> bool {
+    let want = norm_tokens(name);
+    if want.is_empty() || release.is_empty() || !want.iter().all(|t| release.contains(t)) {
+        return false;
+    }
+    let covered = release.iter().filter(|t| want.contains(t)).count();
+    covered as f64 >= release.len() as f64 * MIN_TITLE_COVERAGE
+}
+
+/// Is this release name about one of these show names?
+pub fn name_matches(release_name: &str, names: &[String]) -> bool {
+    let tokens = norm_tokens(&clean_title(release_name));
+    names.iter().any(|n| tokens_resemble(&tokens, n))
+}
+
+/// Does the release's own numbering agree with what the query asks for? A
+/// number the release does not state never disqualifies it.
+pub fn numbering_agrees(release_name: &str, anchor: &QueryAnchor) -> bool {
+    let se = extract_season_episode(release_name);
+    if let (Some(want), Some(have)) = (&anchor.season, se.season.as_deref().and_then(number)) {
+        if se.season_explicit && *want != have {
+            return false;
+        }
+    }
+    if let Some(want) = &anchor.episode {
+        if let Some(have) = se.episode.as_deref().and_then(number) {
+            return *want == have;
+        }
+        // A pack carrying an explicit range: inside it or not.
+        if let (Some(a), Some(b)) = (
+            se.episode_start.as_deref().and_then(|s| s.parse::<u32>().ok()),
+            se.episode_end.as_deref().and_then(|s| s.parse::<u32>().ok()),
+        ) {
+            let w: u32 = want.parse().unwrap_or(0);
+            return (a..=b).contains(&w);
+        }
+    }
+    true
+}
+
+/// A cached lookup: the card, or `None` for a remembered miss.
+type CardSlot = Option<Arc<Card>>;
+
+/// Reads card records out of the gateway's meta-core, with a small cache.
+pub struct CardBook {
+    http: reqwest::Client,
+    meta_core_url: Option<String>,
+    cache: Mutex<HashMap<String, (Instant, CardSlot)>>,
+    warned_unset: std::sync::atomic::AtomicBool,
+}
+
+impl CardBook {
+    pub fn new(http: reqwest::Client, meta_core_url: Option<String>) -> Self {
+        Self {
+            http,
+            meta_core_url: meta_core_url.map(|u| u.trim_end_matches('/').to_string()).filter(|u| !u.is_empty()),
+            cache: Mutex::new(HashMap::new()),
+            warned_unset: std::sync::atomic::AtomicBool::new(false),
+        }
+    }
+
+    /// The card for this anchor, trying each kind the
+    /// query allows. `None` = no card on the mesh yet: nothing to match by.
+    pub async fn card_for(&self, anchor: &QueryAnchor) -> CardSlot {
+        let Some(base) = self.meta_core_url.as_deref() else {
+            if !self.warned_unset.swap(true, std::sync::atomic::Ordering::Relaxed) {
+                warn!(target: "meta-feeder", "usenet: META_CORE_URL is unset — show-page (tmdbid:) queries get no name matches");
+            }
+            return None;
+        };
+        for kind in &anchor.kinds {
+            let id = format!("{}:{}", kind.locator_prefix(), anchor.tmdbid);
+            let Some(locator) = compute_card_cid("tmdb", &id) else {
+                continue;
+            };
+            if let Some(hit) = self.cached(&locator) {
+                match hit {
+                    Some(card) => return Some(card),
+                    None => continue,
+                }
+            }
+            let found = match meta_feeder_sdk::meta_core::get_record(&self.http, base, &locator).await {
+                Ok(Some(f)) => Card::from_fields(*kind, &f).map(Arc::new),
+                Ok(None) => None,
+                Err(e) => {
+                    // Not cached: a meta-core blip should not hide a show for minutes.
+                    debug!(target: "meta-feeder", %locator, error = %e, "usenet: card lookup failed");
+                    continue;
+                }
+            };
+            self.cache.lock().unwrap().insert(locator.clone(), (Instant::now(), found.clone()));
+            if found.is_some() {
+                return found;
+            }
+        }
+        None
+    }
+
+    fn cached(&self, locator: &str) -> Option<CardSlot> {
+        let cache = self.cache.lock().unwrap();
+        let (at, v) = cache.get(locator)?;
+        let ttl = if v.is_some() { CARD_TTL } else { MISS_TTL };
+        (at.elapsed() < ttl).then(|| v.clone())
+    }
+}
+
+/// Stamp a release the anchor matched by name.
+pub fn stamp_match(fields: &mut BTreeMap<String, String>, anchor: &QueryAnchor, card: &Card) {
+    fields.insert("tmdbid".into(), anchor.tmdbid.clone());
+    fields.insert("anchored".into(), "true".into());
+    fields.insert("anchorMethod".into(), "title".into());
+    for (k, v) in &card.stamp {
+        fields.entry(k.clone()).or_insert_with(|| v.clone());
+    }
+    // A release NNTmux left uncategorised (a season pack in "Other") still
+    // belongs to the show page once its name matched: give it the routing axes
+    // the page filters on (`domain:screen`, a TV/movie kind).
+    if !fields.contains_key("contentKind") {
+        let kind = match card.kind {
+            ShowKind::Movie => "movie",
+            ShowKind::Tv if fields.contains_key("episode") => "episode",
+            ShowKind::Tv => "pack",
+        };
+        set_kind(fields, kind);
+    }
+}
+
+/// Write `contentKind` with its routing axes. `pack` is the one kind the SDK
+/// cannot route on its own (a season pack and an album are the same shape), so
+/// the writer says it: every pack this feeder emits is a screen work.
+pub fn set_kind(fields: &mut BTreeMap<String, String>, kind: &str) {
+    fields.insert("contentKind".into(), kind.into());
+    let domain = meta_feeder_sdk::domain::domain_for_content_kind(kind).or((kind == "pack").then_some("screen"));
+    if let Some(d) = domain {
+        fields.insert("domain".into(), d.to_string());
+    }
+    let form = meta_feeder_sdk::domain::work_form_for_content_kind(kind).or((kind == "pack").then_some("serial"));
+    if let Some(wf) = form {
+        fields.insert("workForm".into(), wf.to_string());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn q(filters: &[(&str, &[&str])]) -> GatewayQuery {
+        let mut g = GatewayQuery::from_free_text("");
+        for (k, vs) in filters {
+            g.filters.insert(k.to_string(), vs.iter().map(|s| s.to_string()).collect());
+        }
+        g
+    }
+
+    fn names(ns: &[&str]) -> Vec<String> {
+        ns.iter().map(|s| s.to_string()).collect()
+    }
+
+    #[test]
+    fn anchor_reads_id_kind_and_numbering() {
+        let a = QueryAnchor::from_query(&q(&[
+            ("tmdbid", &["112442"]),
+            ("season", &["02"]),
+            ("contentKind", &["series", "episode", "pack"]),
+        ]))
+        .unwrap();
+        assert_eq!(a.tmdbid, "112442");
+        assert_eq!(a.kinds, vec![ShowKind::Tv]);
+        assert_eq!(a.season.as_deref(), Some("2"));
+        assert_eq!(a.episode, None);
+
+        let film = QueryAnchor::from_query(&q(&[("tmdbid", &["603"]), ("contentKind", &["movie", "pack"])])).unwrap();
+        assert_eq!(film.kinds, vec![ShowKind::Movie]);
+        let unsaid = QueryAnchor::from_query(&q(&[("tmdbid", &["603"])])).unwrap();
+        assert_eq!(unsaid.kinds, vec![ShowKind::Tv, ShowKind::Movie]);
+
+        assert!(QueryAnchor::from_query(&q(&[])).is_none());
+        assert!(QueryAnchor::from_query(&q(&[("tmdbid", &["tt0133093"])])).is_none());
+    }
+
+    #[test]
+    fn release_names_match_the_show_not_a_mention_of_it() {
+        let n = names(&["Trash Truck"]);
+        assert!(name_matches("Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL.DDP5.1.H.264-NTb", &n));
+        assert!(name_matches("Trash.Truck.S01E02.Slumber.Party.1080p.NF.WEB-DL.DDP5.1.x264-LAZY", &n));
+        assert!(!name_matches("A.Trash.Truck.Christmas.Special.Behind.The.Scenes.Making.Of.2020.1080p", &n));
+        assert!(!name_matches("Regular.Show.S02E19.Grave.Sights.1080p.DD2.0.VC-1.REMUX-FraMeSToR", &n));
+        // An AKA matches as well as the display title.
+        assert!(name_matches("Pikappu.S01E03.1080p.WEB.h264-X", &names(&["Trash Truck", "Pikappu"])));
+    }
+
+    #[test]
+    fn numbering_must_agree_when_the_release_states_it() {
+        let a = |season: Option<&str>, episode: Option<&str>| QueryAnchor {
+            tmdbid: "1".into(),
+            kinds: vec![ShowKind::Tv],
+            season: season.map(str::to_string),
+            episode: episode.map(str::to_string),
+        };
+        let rel = "Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL-NTb";
+        assert!(numbering_agrees(rel, &a(Some("2"), None)));
+        assert!(numbering_agrees(rel, &a(Some("2"), Some("16"))));
+        assert!(!numbering_agrees(rel, &a(Some("1"), None)));
+        assert!(!numbering_agrees(rel, &a(Some("2"), Some("5"))));
+        // Says nothing about the episode → kept.
+        assert!(numbering_agrees("Trash.Truck.S02.1080p.NF.WEB-DL-NTb", &a(Some("2"), Some("5"))));
+        assert!(numbering_agrees(rel, &a(None, None)));
+    }
+
+    #[test]
+    fn card_collects_names_and_stamp() {
+        let mut f = BTreeMap::new();
+        f.insert("title".into(), "Trash Truck".into());
+        f.insert("titles/jpn/Gomi Torakku".into(), "true".into());
+        f.insert("titles/eng/trash truck".into(), "true".into());
+        f.insert("poster".into(), "bafkpost".into());
+        f.insert("description/eng".into(), "Hank and a truck.".into());
+        f.insert("imdbid".into(), "tt9288860".into());
+        f.insert("genres/Kids".into(), "true".into());
+        let c = Card::from_fields(ShowKind::Tv, &f).unwrap();
+        assert_eq!(c.names, names(&["Trash Truck", "Gomi Torakku"]));
+        assert_eq!(c.stamp.len(), 3);
+        assert!(!c.stamp.contains_key("genres/Kids"));
+        assert!(Card::from_fields(ShowKind::Tv, &BTreeMap::new()).is_none());
+    }
+
+    #[test]
+    fn stamp_marks_a_title_match_and_fills_display_fields_only_when_absent() {
+        let mut f = BTreeMap::new();
+        f.insert("title".into(), "Trash Truck".into());
+        f.insert("poster".into(), "bafkpost".into());
+        f.insert("description/eng".into(), "Hank.".into());
+        let card = Card::from_fields(ShowKind::Tv, &f).unwrap();
+        let anchor = QueryAnchor { tmdbid: "112442".into(), kinds: vec![ShowKind::Tv], season: None, episode: None };
+
+        let mut rec = BTreeMap::new();
+        rec.insert("poster".to_string(), "own".to_string());
+        rec.insert("episode".to_string(), "16".to_string());
+        stamp_match(&mut rec, &anchor, &card);
+        assert_eq!(rec["tmdbid"], "112442");
+        assert_eq!(rec["anchored"], "true");
+        assert_eq!(rec["anchorMethod"], "title");
+        assert_eq!(rec["poster"], "own");
+        assert_eq!(rec["description/eng"], "Hank.");
+        // Uncategorised by NNTmux → routed to the show page anyway.
+        assert_eq!(rec["contentKind"], "episode");
+        assert_eq!(rec["domain"], "screen");
+    }
+}

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

@@ -34,4 +34,6 @@ pub mod nntmux {
 pub mod newznab;
 
 /// The `usenet` upstream itself.
+pub mod anchor;
+
 pub mod usenet;

feeder-plugin/usenet-feeder/src/usenet.rs

@@ -49,6 +49,7 @@ use std::time::Duration;
 use async_trait::async_trait;
 use meta_feeder_sdk::common::build_http_client;
 use meta_feeder_sdk::config::{ConfigField as F, ConfigSchema};
+use meta_feeder_sdk::filename_meta::extract_season_episode;
 use meta_feeder_sdk::hash::{
     codec_of, compute_ipfs_cid, compute_nzb_posting_cid, decode_nzb_release_cid, NZB_RELEASE_CODEC,
 };
@@ -57,6 +58,7 @@ use meta_feeder_sdk::query::GatewayQuery;
 use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth};
 use tracing::{debug, info, warn};
 
+use crate::anchor::{self, CardBook, QueryAnchor};
 use crate::newznab::{self, IndexerCred};
 use crate::nntmux::api::{is_guid, NntmuxApi, DEFAULT_NNTMUX_URL};
 use crate::nntmux::cache::{NzbCache, Posting};
@@ -76,6 +78,9 @@ const USER_AGENT: &str = concat!("meta-feeder-usenet/", env!("CARGO_PKG_VERSION"
 /// Wall-clock budget for one `.nzb` grab. The viewer is waiting on it.
 const GRAB_TIMEOUT_SECS: u64 = 60;
 
+/// The `indexer` label when the config page sets none.
+pub const DEFAULT_INDEXER_NAME: &str = "nntmux";
+
 /// How often the nntmux reachability probe (`t=caps`, free) runs.
 const PROBE_INTERVAL: Duration = Duration::from_secs(300);
 
@@ -94,6 +99,10 @@ pub struct Settings {
     /// Per-host Newznab keys for redeeming `nzb-release` locators. Config page
     /// only — no env seed.
     pub indexers: Vec<IndexerCred>,
+    /// The source name every record carries (`indexer`), shown beside
+    /// "nzbgeek.info"-style names in a client's source list. It names who
+    /// indexed the release — the bytes are Usenet's for every NZB source alike.
+    pub indexer_name: String,
 }
 
 impl Settings {
@@ -103,6 +112,7 @@ impl Settings {
             nntmux_url: env("NNTMUX_URL").unwrap_or_else(|| DEFAULT_NNTMUX_URL.to_string()),
             nntmux_api_key: env("NNTMUX_API_KEY").unwrap_or_default(),
             indexers: Vec::new(),
+            indexer_name: env("INDEXER_NAME").unwrap_or_else(|| DEFAULT_INDEXER_NAME.to_string()),
         }
     }
 
@@ -122,6 +132,9 @@ impl Settings {
         if let Some(s) = text("nntmux_api_key") {
             self.nntmux_api_key = s.to_string();
         }
+        if let Some(s) = text("indexer_name") {
+            self.indexer_name = s.to_string();
+        }
         if text("db_url").is_some() && self.nntmux_api_key.is_empty() {
             // A pre-1.1 config: the feeder read nntmux's database directly.
             warn!(
@@ -140,6 +153,20 @@ impl Settings {
     }
 }
 
+/// `nntmux (watch)` → `nntmux-watch`: a source-member leaf safe to split on `:`.
+fn source_slug(name: &str) -> String {
+    let mut out = String::new();
+    for c in name.chars() {
+        if c.is_alphanumeric() || c == '.' {
+            out.extend(c.to_lowercase());
+        } else if !out.ends_with('-') {
+            out.push('-');
+        }
+    }
+    let out = out.trim_matches('-').to_string();
+    if out.is_empty() { DEFAULT_INDEXER_NAME.to_string() } else { out }
+}
+
 pub struct UsenetPlugin {
     settings: Settings,
     /// `None` when no nntmux key is configured (grab-only deployment).
@@ -157,6 +184,9 @@ pub struct UsenetPlugin {
     /// `https` in production — a locator carries no scheme and every public
     /// Newznab API is TLS. Tests point it at a plain-http mock.
     grab_scheme: &'static str,
+    /// Show cards from the gateway's meta-core, for name-matching `tmdbid:`
+    /// queries ([`crate::anchor`]).
+    cards: Arc<CardBook>,
 }
 
 impl Default for UsenetPlugin {
@@ -174,6 +204,7 @@ impl UsenetPlugin {
             nntmux_problem: Arc::new(RwLock::new(None)),
             http: build_http_client(GRAB_TIMEOUT_SECS, USER_AGENT, None),
             grab_scheme: "https",
+            cards: Arc::new(CardBook::new(reqwest::Client::new(), None)),
         }
     }
 
@@ -284,6 +315,88 @@ impl UsenetPlugin {
             .ok_or_else(|| GatewayError::Permanent(format!("nzb of release {guid} has no segments")))
     }
 
+    /// Mint each hit's cid and build its record.
+    async fn mint_records(&self, releases: &[Release]) -> Vec<DiscoveryRecord> {
+        // Mint each hit's cid here — see the `record_for` doc for why this
+        // cannot wait for compute. A release whose `.nzb` can't be had is
+        // DROPPED rather than surfaced uncid'd: an unplayable row in the
+        // client's "raw sources" list is worse than one fewer row. Once nntmux
+        // says a daily cap is spent, stop asking for this query.
+        let mut out = Vec::with_capacity(releases.len());
+        let mut skipped = 0usize;
+        let mut capped = false;
+        for r in releases {
+            if capped {
+                skipped += 1;
+                continue;
+            }
+            match self.posting(&r.guid).await {
+                Ok(p) => out.push(self.record_for(r, Some(&p.cid))),
+                Err(GatewayError::RateLimited { .. }) => {
+                    capped = true;
+                    skipped += 1;
+                    warn!(target: "meta-feeder", "usenet: nntmux download cap reached — remaining hits skipped (raise the feeder user's role caps)");
+                }
+                Err(e) => {
+                    skipped += 1;
+                    debug!(target: "meta-feeder", guid = %r.guid, error = %e, "usenet: release skipped (no usable .nzb)");
+                }
+            }
+        }
+        if skipped > 0 {
+            debug!(target: "meta-feeder", hits = out.len(), skipped, "usenet: releases skipped");
+        }
+        out
+    }
+
+    /// A `tmdbid:` query: read the show's card, search NNTmux by its title,
+    /// keep the releases whose names match and whose numbering agrees, and
+    /// stamp them as a title match. No card → nothing to match by → `[]`.
+    async fn handle_anchored(
+        &self,
+        anchor: &QueryAnchor,
+        max_results: usize,
+    ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
+        let Some(card) = self.cards.card_for(anchor).await else {
+            debug!(target: "meta-feeder", tmdbid = %anchor.tmdbid, "usenet: no card on meta-core for this id; nothing to match by name");
+            return Ok(vec![]);
+        };
+        let card = card.as_ref();
+        let (api, _) = self.nntmux()?;
+        let mut seen = std::collections::HashSet::new();
+        let mut matched: Vec<Release> = Vec::new();
+        let mut looked = 0usize;
+        for term in card.search_terms() {
+            let releases = api.search(&term, max_results).await.inspect_err(|e| self.note_problem(e))?;
+            *self.nntmux_problem.write().unwrap() = None;
+            looked += releases.len();
+            // Filter BEFORE minting: minting downloads the `.nzb`, and a
+            // release that is not this show must cost nothing.
+            for r in releases {
+                let kind_ok = !matches!(
+                    (card.kind, content_kind_for_category(r.category_id)),
+                    (anchor::ShowKind::Tv, Some("movie")) | (anchor::ShowKind::Movie, Some("episode"))
+                );
+                if kind_ok
+                    && anchor::name_matches(&r.search_name, &card.names)
+                    && anchor::numbering_agrees(&r.search_name, anchor)
+                    && seen.insert(r.guid.clone())
+                {
+                    matched.push(r);
+                }
+            }
+        }
+        let mut out = self.mint_records(&matched).await;
+        for rec in &mut out {
+            anchor::stamp_match(&mut rec.fields, anchor, card);
+        }
+        debug!(
+            target: "meta-feeder", tmdbid = %anchor.tmdbid, show = %card.names[0], looked, matched = out.len(),
+            "usenet: name-matched search"
+        );
+        Ok(out)
+    }
+
     /// Build the discovery record for one release. Metadata only — the bytes
     /// live on Usenet and are fetched by meta-share at playback, never by us.
     ///
@@ -334,27 +447,43 @@ impl UsenetPlugin {
             "fileType".to_string(),
             file_type_for_category(r.category_id).to_string(),
         );
-        if let Some(kind) = content_kind_for_category(r.category_id) {
-            fields.insert("contentKind".to_string(), kind.to_string());
+        let kind = content_kind_for_category(r.category_id);
+        // Numbering from the release name, for TV releases (and uncategorised
+        // ones that plainly carry an `SxxEyy`). A season pack gets the season
+        // only and becomes `pack`, as the torznab feeder files them.
+        let se = extract_season_episode(&r.search_name);
+        let tv = kind == Some("episode") || (kind.is_none() && se.season_explicit);
+        let kind = if tv && se.is_pack { Some("pack") } else { kind };
+        if tv {
+            if let Some(s) = &se.season {
+                fields.insert("season".to_string(), s.clone());
+            }
+            if !se.is_pack {
+                if let Some(e) = &se.episode {
+                    fields.insert("episode".to_string(), e.clone());
+                }
+            }
+        }
+        if let Some(kind) = kind {
             // The routing axes, co-written with the kind (METADATA_KEYS.md §1):
             // `fileType` says what the bytes are, `domain` says which app wants
             // the record, `workForm` whether it stands alone or is one
             // instalment. Without the domain the row never reaches a wall.
-            if let Some(d) = meta_feeder_sdk::domain::domain_for_content_kind(kind) {
-                fields.insert("domain".to_string(), d.to_string());
-            }
-            if let Some(wf) = meta_feeder_sdk::domain::work_form_for_content_kind(kind) {
-                fields.insert("workForm".to_string(), wf.to_string());
-            }
+            anchor::set_kind(&mut fields, kind);
         }
         if let Some(ts) = r.post_date {
             fields.insert("publishedAt".to_string(), ts.to_string());
         }
-        // Provenance: this catalog is ours, not an external indexer's. Kept
-        // distinct from indexer-feeder's `indexer` field on purpose — an
-        // operator reading a record should be able to tell at a glance whether
-        // its metadata came from a third party or from our own scan.
+        // Provenance: who indexed it. `indexer` is the name a client shows
+        // beside "nzbgeek.info", and a plain `source/<name>` member carries it
+        // into the source facets the way the torznab feeder files its indexer
+        // (`source/nzb.life`). `source/gateway:usenet-scan` stays as the
+        // selector for everything this feeder published.
+        let name = self.settings.indexer_name.trim();
+        let name = if name.is_empty() { DEFAULT_INDEXER_NAME } else { name };
+        fields.insert("indexer".to_string(), name.to_string());
         fields.insert("source/gateway:usenet-scan".to_string(), "true".to_string());
+        fields.insert(format!("source/{}", source_slug(name)), "true".to_string());
         DiscoveryRecord {
             upstream_id: UPSTREAM_ID.to_string(),
             record_id: r.guid.clone(),
@@ -372,6 +501,10 @@ impl FeederPlugin for UsenetPlugin {
     fn configure(&mut self, cache_dir: &Path) -> Result<(), ConfigError> {
         let mut settings = Settings::from_env();
         settings.merge_file(cache_dir);
+        // The gateway's meta-core, where show cards live (`crate::anchor`).
+        // Unset → show-page queries simply get no name matches.
+        let meta_core_url = std::env::var("META_CORE_URL").ok();
+        self.cards = Arc::new(CardBook::new(build_http_client(10, USER_AGENT, None), meta_core_url));
 
         if settings.nntmux_api_key.is_empty() {
             // Soft-skip, not a hard failure: serve /health, report Degraded.
@@ -424,6 +557,15 @@ impl FeederPlugin for UsenetPlugin {
         query: &GatewayQuery,
         max_results: usize,
     ) -> Result<Vec<DiscoveryRecord>, GatewayError> {
+        if self.nntmux.is_none() {
+            // Grab-only deployment (indexer keys, no nntmux): nothing to search.
+            return Ok(vec![]);
+        }
+        // A show page's query (`tmdbid:<id> season:<n> …`, no free text): match
+        // releases to the show BY NAME, against its card — see `crate::anchor`.
+        if let Some(anchor) = QueryAnchor::from_query(query) {
+            return self.handle_anchored(&anchor, max_results).await;
+        }
         // `free_text` is the bare-word leaves only — the structured filters
         // (contentKind:, fileType:) are the gateway's routing gate, not a search
         // term, and passing them to a LIKE would match nothing.
@@ -431,47 +573,14 @@ impl FeederPlugin for UsenetPlugin {
         if term.trim().is_empty() {
             return Ok(vec![]);
         }
-        if self.nntmux.is_none() {
-            // Grab-only deployment (indexer keys, no nntmux): nothing to search.
-            return Ok(vec![]);
-        }
         let (api, _) = self.nntmux()?;
         let releases = api
             .search(term.trim(), max_results)
             .await
             .inspect_err(|e| self.note_problem(e))?;
         *self.nntmux_problem.write().unwrap() = None;
-
-        // Mint each hit's cid here — see the `record_for` doc for why this
-        // cannot wait for compute. A release whose `.nzb` can't be had is
-        // DROPPED rather than surfaced uncid'd: an unplayable row in the
-        // client's "raw sources" list is worse than one fewer row. Once nntmux
-        // says a daily cap is spent, stop asking for this query.
-        let mut out = Vec::with_capacity(releases.len());
-        let mut skipped = 0usize;
-        let mut capped = false;
-        for r in &releases {
-            if capped {
-                skipped += 1;
-                continue;
-            }
-            match self.posting(&r.guid).await {
-                Ok(p) => out.push(self.record_for(r, Some(&p.cid))),
-                Err(GatewayError::RateLimited { .. }) => {
-                    capped = true;
-                    skipped += 1;
-                    warn!(target: "meta-feeder", "usenet: nntmux download cap reached — remaining hits skipped (raise the feeder user's role caps)");
-                }
-                Err(e) => {
-                    skipped += 1;
-                    debug!(target: "meta-feeder", guid = %r.guid, error = %e, "usenet: release skipped (no usable .nzb)");
-                }
-            }
-        }
-        debug!(
-            target: "meta-feeder", %term, hits = out.len(), skipped,
-            "usenet: nntmux search"
-        );
+        let out = self.mint_records(&releases).await;
+        debug!(target: "meta-feeder", %term, hits = out.len(), "usenet: nntmux search");
         Ok(out)
     }
 
@@ -588,11 +697,17 @@ impl FeederPlugin for UsenetPlugin {
                     ),
                 F::secret("nntmux_api_key", "nntmux API key")
                     .with_help(
-                        "API key of the nntmux user this feeder searches as — the NNTmux app's \
-                         `metamesh` user, whose role has raised daily caps (see the NNTmux app's \
-                         install tips for how to read it). Blank → the self-scan search is off \
+                        "API key of the nntmux user this feeder searches as. Create one with \
+                         `docker exec nntmux-scanner nntmux-setup api-user metamesh` (raised daily \
+                         caps) and paste the key it prints. Blank → the self-scan search is off \
                          (indexer-key redeems still work).",
                     ),
+                F::text("indexer_name", "Source name")
+                    .with_help(
+                        "How this feeder's releases are labelled in a client's source list, \
+                         beside names like nzbgeek.info (default: nntmux). Takes effect on the \
+                         next feeder restart.",
+                    ),
                 F::record_array(
                     "indexers",
                     "Indexer keys (redeem nzb-release)",
@@ -619,7 +734,54 @@ impl FeederPlugin for UsenetPlugin {
         serde_json::json!({
             "nntmux_url": self.settings.nntmux_url,
             "nntmux_api_key": self.settings.nntmux_api_key,
+            "indexer_name": self.settings.indexer_name,
             "indexers": self.settings.indexers,
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn source_slug_keeps_a_leaf_safe_name() {
+        assert_eq!(source_slug("nntmux"), "nntmux");
+        assert_eq!(source_slug("NNTmux (watch)"), "nntmux-watch");
+        assert_eq!(source_slug("my:indexer"), "my-indexer");
+        assert_eq!(source_slug("  "), "nntmux");
+    }
+
+    fn release(name: &str, category: i32) -> Release {
+        Release {
+            guid: "0f0e5b2c-1111-4c5e-9d6b-2a3b4c5d6e7f".into(),
+            search_name: name.into(),
+            category_id: category,
+            size: 1,
+            post_date: None,
+        }
+    }
+
+    #[test]
+    fn records_carry_the_indexer_label_and_numbering() {
+        let mut p = UsenetPlugin::new();
+        p.settings.indexer_name = "nntmux".into();
+        let f = p.record_for(&release("Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL-NTb", 5040), None).fields;
+        assert_eq!(f["indexer"], "nntmux");
+        assert_eq!(f["source/gateway:usenet-scan"], "true");
+        assert_eq!(f["source/nntmux"], "true");
+        assert_eq!(f["season"], "2");
+        assert_eq!(f["episode"], "16");
+        assert_eq!(f["contentKind"], "episode");
+
+        let pack = p.record_for(&release("Wonkas.The.Golden.Ticket.S01.1080p.WEB.H264-MIXED", 10), None).fields;
+        assert_eq!(pack["contentKind"], "pack");
+        assert_eq!(pack["domain"], "screen");
+        assert_eq!(pack["season"], "1");
+        assert!(!pack.contains_key("episode"));
+
+        let film = p.record_for(&release("Some.Film.2020.1080p.BluRay.x264-GRP", 2040), None).fields;
+        assert!(!film.contains_key("season"));
+        assert_eq!(film["contentKind"], "movie");
+    }
+}

feeder-plugin/usenet-feeder/tests/anchored.rs

@@ -0,0 +1,164 @@
+//! Show-page queries (`tmdbid:<id> season:<n>`, no free text): releases are
+//! matched to the show BY NAME against its card in meta-core, and stamped as a
+//! title match. Own binary because the card source is read from
+//! `META_CORE_URL`, set once here.
+
+use std::collections::BTreeMap;
+
+use meta_feeder_sdk::hash::compute_card_cid;
+use meta_feeder_sdk::serve::{configure_plugins, router};
+use meta_feeder_sdk::{GatewayQuery, QueryRequest, QueryResponse};
+use usenet_feeder::usenet::UsenetPlugin;
+use wiremock::matchers::{method, path, query_param};
+use wiremock::{Mock, MockServer, ResponseTemplate};
+
+const KEY: &str = "metamesh-key";
+
+fn item(guid: &str, title: &str, cat: u32) -> String {
+    format!(
+        r#"<item><title>{title}</title><guid isPermaLink="true">http://localhost/details/{guid}</guid>
+<newznab:attr name="category" value="{cat}"/><newznab:attr name="size" value="443121252"/></item>"#
+    )
+}
+
+fn rss(items: &[String]) -> String {
+    format!(
+        r#"<?xml version="1.0" encoding="UTF-8"?><rss version="2.0" xmlns:newznab="http://www.newznab.com/DTD/2010/feeds/attributes/"><channel>{}</channel></rss>"#,
+        items.join("")
+    )
+}
+
+const NZB: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<nzb xmlns="http://www.newzbin.com/DTD/2003/nzb">
+  <file poster="[email protected]" date="1" subject="TT [1/1] &quot;a.mkv&quot; yEnc">
+    <groups><group>alt.binaries.teevee</group></groups>
+    <segments><segment bytes="100" number="1">[email protected]</segment></segments>
+  </file>
+</nzb>"#;
+
+fn anchored(filters: &[(&str, &[&str])]) -> GatewayQuery {
+    let mut q = GatewayQuery::from_free_text("");
+    q.filters = filters
+        .iter()
+        .map(|(k, vs)| (k.to_string(), vs.iter().map(|s| s.to_string()).collect()))
+        .collect::<BTreeMap<_, _>>();
+    q
+}
+
+async fn records(base: &str, q: GatewayQuery) -> Vec<meta_feeder_sdk::DiscoveryRecord> {
+    let r = reqwest::Client::new()
+        .post(format!("{base}/query"))
+        .json(&QueryRequest { upstream_id: "usenet".into(), query: q, max_results: 50 })
+        .send()
+        .await
+        .expect("POST /query");
+    assert!(r.status().is_success(), "query status {}", r.status());
+    r.json::<QueryResponse>().await.expect("query json").records
+}
+
+#[tokio::test]
+async fn a_show_page_query_matches_releases_by_name_against_the_card() {
+    let nntmux = MockServer::start().await;
+    let core = MockServer::start().await;
+    std::env::set_var("META_CORE_URL", core.uri());
+
+    // The card the card feeder stored for tv 112442.
+    let locator = compute_card_cid("tmdb", "tv:112442").unwrap();
+    Mock::given(method("GET"))
+        .and(path(format!("/meta/{locator}")))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "hashId": locator,
+            "metadata": {
+                "title": "Trash Truck",
+                "fileType": "card",
+                "poster": "bafkreiposter",
+                "description/eng": "Hank and his best pal, a giant trash truck.",
+                "imdbid": "tt9288860",
+                "genres/Kids": "true"
+            }
+        })))
+        .mount(&core)
+        .await;
+    // No card for any other id.
+    Mock::given(method("GET"))
+        .respond_with(ResponseTemplate::new(404))
+        .with_priority(10)
+        .mount(&core)
+        .await;
+
+    Mock::given(method("GET"))
+        .and(path("/api/v1/api"))
+        .and(query_param("t", "caps"))
+        .respond_with(ResponseTemplate::new(200).set_body_string("<caps/>"))
+        .mount(&nntmux)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/api/v1/api"))
+        .and(query_param("t", "search"))
+        .and(query_param("q", "Trash Truck"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(rss(&[
+            item("aaa111", "Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL.DDP5.1.H.264-NTb", 5040),
+            item("bbb222", "Trash.Truck.S01E02.Slumber.Party.1080p.NF.WEB-DL.DDP5.1.x264-LAZY", 5040),
+            item("ccc333", "Trash.Truck.Fan.Made.Toy.Unboxing.Compilation.Vol.3.1080p", 5040),
+        ])))
+        .mount(&nntmux)
+        .await;
+    // Only the release that matches name AND season is ever downloaded.
+    Mock::given(method("GET"))
+        .and(path("/getnzb"))
+        .and(query_param("id", "aaa111.nzb"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(NZB))
+        .expect(1)
+        .mount(&nntmux)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/getnzb"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(NZB))
+        .with_priority(10)
+        .expect(0)
+        .mount(&nntmux)
+        .await;
+
+    let dir = tempfile::tempdir().expect("tempdir");
+    let cache = dir.path().join("gateway").join("usenet");
+    std::fs::create_dir_all(&cache).expect("mkdir");
+    std::fs::write(
+        cache.join("config.json"),
+        serde_json::json!({ "nntmux_url": nntmux.uri(), "nntmux_api_key": KEY }).to_string(),
+    )
+    .expect("write config");
+    let configured = configure_plugins(vec![Box::new(UsenetPlugin::new())], dir.path()).expect("configure");
+    let app = router(configured, "test".to_string(), dir.path());
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind");
+    let base = format!("http://{}", listener.local_addr().expect("addr"));
+    tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") });
+
+    // The season walk's shape: id + season + the TV kinds, no free text.
+    let recs = records(
+        &base,
+        anchored(&[
+            ("tmdbid", &["112442"]),
+            ("season", &["2"]),
+            ("contentKind", &["series", "episode", "pack"]),
+        ]),
+    )
+    .await;
+    assert_eq!(recs.len(), 1, "only the S02 release whose name is the show, got {recs:?}");
+    let f = &recs[0].fields;
+    assert_eq!(recs[0].record_id, "aaa111");
+    assert_eq!(f["tmdbid"], "112442");
+    assert_eq!(f["anchored"], "true");
+    assert_eq!(f["anchorMethod"], "title", "a name match, never an id-confirmed anchor");
+    assert_eq!(f["imdbid"], "tt9288860");
+    assert_eq!(f["poster"], "bafkreiposter");
+    assert_eq!(f["description/eng"], "Hank and his best pal, a giant trash truck.");
+    assert!(!f.contains_key("genres/Kids"));
+    assert_eq!(f["season"], "2");
+    assert_eq!(f["episode"], "16");
+    assert_eq!(f["indexer"], "nntmux");
+    assert_eq!(f["source/nntmux"], "true");
+    assert_eq!(f["fileName"], "Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL.DDP5.1.H.264-NTb");
+
+    // No card on the mesh for this id → nothing to match by name.
+    assert!(records(&base, anchored(&[("tmdbid", &["999"])])).await.is_empty());
+}