improve compute API

feeder-plugin/usenet-feeder/Cargo.toml

@@ -14,8 +14,10 @@ publish = false
 # redeem it without an indexer credential.
 #
 # It ADDS a Usenet metadata source alongside meta-feeder-indexer; it does not
-# replace it. Like every feeder it is meta-core-free + blockstore-free (the
-# gateway core owns hashing/seeding/store-back — see the gateway's invariant 10).
+# replace it. It also REDEEMS external `nzb-release` (0x1005) locators through
+# `/compute`, grabbing the .nzb with per-host indexer keys set on its config
+# page. Like every feeder it is meta-core-free + blockstore-free (the gateway
+# core owns hashing/seeding/store-back — see the gateway's invariant 10).
 #
 # See `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md`.
 
@@ -28,7 +30,9 @@ name = "usenet-feeder"
 path = "src/main.rs"
 
 [dependencies]
-meta-feeder-sdk = { path = "../../crates/meta-feeder-sdk" }
+# Pinned by TAG, never by path. A path dep resolves OUTSIDE this repo, and this
+# repo is the Docker build context. (Until 1.3.0 this repo vendored a copy.)
+meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.3.0" }
 
 tokio = { version = "1.40", features = ["full"] }
 axum = { version = "0.7", features = ["json"] }
@@ -62,4 +66,5 @@ quick-xml = { version = "0.36", features = ["serialize"] }
 [dev-dependencies]
 wiremock = "0.6"
 tempfile = "3"
+base64 = "0.22"
 tokio = { version = "1.40", features = ["full", "test-util"] }

feeder-plugin/usenet-feeder/Dockerfile

@@ -3,9 +3,10 @@
 # usenet-feeder — Usenet feeder sidecar over an nntmux header-scan catalog.
 # Multi-stage build.
 #
-# Build context must be this repo root (packages/plugins/meta-feeder-usenet) so
-# the `meta-feeder-sdk` path dependency resolves. Same BuildKit cache-mount
-# strategy as the other feeders.
+# Build context must be this repo root (packages/plugins/meta-feeder-usenet).
+# `meta-feeder-sdk` is a git dependency pinned by tag, so the builder needs
+# network for `cargo fetch`. Same BuildKit cache-mount strategy as the other
+# feeders.
 #
 # This feeder talks to MariaDB (sqlx, rustls — no system MySQL client needed)
 # and reads nntmux's NZB store off a shared volume. It holds no media bytes.
@@ -22,7 +23,6 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
     && rm -rf /var/lib/apt/lists/*
 
 COPY Cargo.toml rust-toolchain.toml ./
-COPY crates/ crates/
 COPY feeder-plugin/ feeder-plugin/
 
 RUN --mount=type=cache,id=cargo-registry,target=/usr/local/cargo/registry \

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

@@ -9,6 +9,11 @@
 //!
 //! It ADDS a source alongside meta-feeder-indexer; it replaces nothing.
 //!
+//! It also **redeems** external Newznab releases: an `nzb-release` (`0x1005`)
+//! cid minted by meta-feeder-torznab is grabbed here, with this feeder's
+//! per-host indexer keys, when the gateway calls `/compute` on a real play
+//! ([`newznab`]). Indexer keys live on this feeder, not on meta-share.
+//!
 //! Like every feeder it is meta-core-free and blockstore-free: it finds records
 //! and serves bytes, and the gateway core owns hashing-into-the-blockstore, the
 //! meta-core store-back, and the libp2p wire (gateway invariant 10).
@@ -22,5 +27,8 @@ pub mod nntmux {
     pub mod nzb;
 }
 
+/// The `t=get` grab that redeems an external `nzb-release` locator.
+pub mod newznab;
+
 /// The `usenet` upstream itself.
 pub mod usenet;

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

@@ -3,10 +3,12 @@
 //!
 //! One plugin (`usenet`) served through [`meta_feeder_sdk::serve_feeders`].
 //!
-//! Operator config — the nntmux database URL and NZB store path — is read from
-//! the persisted `config.json` (dashboard-written), with the env vars below as a
-//! **first-boot seed only**: the file wins on the next restart, and there is no
-//! hot reload (gateway invariant 12).
+//! Operator config — the nntmux database URL and NZB store path, and the
+//! per-host Newznab indexer keys used to redeem `nzb-release` locators — is read
+//! from the persisted `config.json` (dashboard-written), with the env vars below
+//! as a **first-boot seed only**: the file wins on the next restart, and there
+//! is no hot reload (gateway invariant 12). Indexer keys have no env seed; set
+//! them on the config page.
 //!
 //! Env (seed): `NNTMUX_DB_URL`, `NNTMUX_NZB_PATH`.
 //! Env (infra): `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`),

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

@@ -0,0 +1,409 @@
+//! Redeeming an **external** Newznab release — an `nzb-release` (`0x1005`)
+//! locator minted by `meta-feeder-torznab` from a Prowlarr/Newznab search hit.
+//!
+//! The cid embeds the indexer's API base and the release id; redeeming it is a
+//! `t=get` grab against that indexer with *this feeder's* key for that host.
+//! The grab is **metered** by the indexer (a per-account daily download cap), so
+//! it only ever runs from `compute_outcomes`, which the gateway calls on a real
+//! play — never at search time, never speculatively.
+//!
+//! Moved here from meta-share's `nzb/manifest_resolve.rs::grab_newznab`: the
+//! indexer keys now live on this feeder, meta-share keeps only the NNTP pool.
+
+use std::borrow::Cow;
+
+use anyhow::{anyhow, Context};
+use meta_feeder_sdk::types::GatewayError;
+use serde::{Deserialize, Serialize};
+
+/// `retry_after_s` handed back when an indexer reports its request/download cap
+/// is spent. Newznab carries no reset time, and the caps are daily.
+pub const QUOTA_RETRY_SECS: u32 = 3600;
+
+/// Refuse a `.nzb` bigger than this. A 1329-segment posting is ~200 KB; 16 MiB
+/// is a ~100k-segment monster and still parseable. Same bound meta-share uses.
+pub const MAX_NZB_BYTES: usize = 16 * 1024 * 1024;
+
+/// One configured indexer key. `host` is the bare authority the key is for
+/// (`api.nzbgeek.info`); the API path comes from the cid, not from here.
+#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
+pub struct IndexerCred {
+    #[serde(default)]
+    pub host: String,
+    #[serde(default)]
+    pub api_key: String,
+}
+
+/// Normalise an indexer host for matching: drop any `http(s)://` scheme, trim a
+/// trailing slash and surrounding whitespace, lowercase. Both the configured
+/// host and the cid-decoded authority go through this before an equality
+/// compare, so an operator may enter `api.example.com` or `https://api.example.com/`.
+pub fn normalize_indexer_host(raw: &str) -> String {
+    raw.trim()
+        .trim_start_matches("https://")
+        .trim_start_matches("http://")
+        .trim_end_matches('/')
+        .to_ascii_lowercase()
+}
+
+/// The credential-matching key for anything host-shaped an operator or a
+/// locator supplies: normalised (scheme, trailing slash, case), then reduced to
+/// its authority — `https://API.nzbgeek.info/api/` and `api.nzbgeek.info/api`
+/// both become `api.nzbgeek.info`.
+pub fn host_key(raw: &str) -> String {
+    authority(&normalize_indexer_host(raw)).to_string()
+}
+
+/// The authority part of a scheme-less API base (`api.nzbgeek.info/api` →
+/// `api.nzbgeek.info`). This — never the full base — is the credential key:
+/// operators configure plain hosts, and the locator may carry a path.
+pub fn authority(api_base: &str) -> &str {
+    match api_base.split_once('/') {
+        Some((authority, _path)) => authority,
+        None => api_base,
+    }
+}
+
+/// A Newznab `<error code=".." description=".."/>` answer to `t=get`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct NewznabError {
+    pub code: Option<u32>,
+    pub description: Option<String>,
+}
+
+impl std::fmt::Display for NewznabError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match (&self.code, &self.description) {
+            (Some(c), Some(d)) => write!(f, "code {c}: {d}"),
+            (None, Some(d)) => write!(f, "{d}"),
+            (Some(c), None) => write!(f, "code {c}"),
+            (None, None) => write!(f, "unspecified error"),
+        }
+    }
+}
+
+/// Pull a Newznab `<error>` out of a `t=get` response, if that is what came
+/// back instead of an `.nzb`.
+///
+/// Deliberately a substring scrape rather than a parse: the body is tiny, the
+/// shape is fixed across indexers, and the point is to turn a silent
+/// zero-file manifest into a readable reason. Only the head is scanned, so an
+/// `<error>` deep inside a large legitimate NZB is never mistaken for a refusal.
+pub fn newznab_error(xml: &str) -> Option<NewznabError> {
+    let head = &xml[..floor_char_boundary(xml, 1024)];
+    let at = head.find("<error ")?;
+    let rest = &head[at..];
+    let end = rest.find("/>").or_else(|| rest.find('>'))? + 1;
+    let tag = &rest[..end];
+    let grab = |k: &str| -> Option<&str> {
+        let i = tag.find(k)? + k.len();
+        let t = &tag[i..];
+        let q = t.find('"')? + 1;
+        let e = t[q..].find('"')? + q;
+        Some(&t[q..e])
+    };
+    Some(NewznabError {
+        code: grab("code=").and_then(|c| c.trim().parse().ok()),
+        description: grab("description=").map(str::to_string),
+    })
+}
+
+fn floor_char_boundary(s: &str, max: usize) -> usize {
+    let mut i = s.len().min(max);
+    while !s.is_char_boundary(i) {
+        i -= 1;
+    }
+    i
+}
+
+/// Map an indexer's refusal onto the feeder error contract. The Newznab API
+/// spec's codes: `100`–`102` credentials/account, `300` no such item, `429` /
+/// `500` request limit, `501` download limit, `910` API disabled.
+pub fn classify(err: &NewznabError) -> GatewayError {
+    match err.code {
+        Some(300) => GatewayError::NotFound,
+        Some(429) | Some(500) | Some(501) => GatewayError::RateLimited {
+            retry_after_s: QUOTA_RETRY_SECS,
+        },
+        Some(100..=102) | Some(910) => GatewayError::Permanent(format!(
+            "indexer refused the .nzb grab ({err}) — check this host's API key on the feeder config page"
+        )),
+        _ => GatewayError::Permanent(format!("indexer refused the .nzb grab ({err})")),
+    }
+}
+
+/// True when an indexer answered a grab with an HTML page — a login wall, a
+/// Cloudflare challenge, or a permalink host that 302s to its website
+/// (`https://nzbgeek.info/api?t=get` does). Without this the grab dies later as
+/// an unhelpful "no segments".
+pub fn looks_like_html(body: &[u8]) -> bool {
+    let head = &body[..body.len().min(512)];
+    let text = String::from_utf8_lossy(head);
+    let t = text.trim_start_matches('\u{feff}').trim_start().to_ascii_lowercase();
+    t.starts_with("<!doctype html") || t.starts_with("<html")
+}
+
+/// Grab a release's `.nzb` from `{scheme}://{api_base}/api?t=get`.
+///
+/// Errors never echo the request URL — it carries the API key.
+pub async fn grab(
+    http: &reqwest::Client,
+    scheme: &str,
+    api_base: &str,
+    id: &str,
+    api_key: &str,
+) -> Result<Vec<u8>, GatewayError> {
+    let base = api_base.trim_end_matches('/');
+    let url = format!("{scheme}://{base}/api?t=get&id={id}&apikey={api_key}");
+    let resp = http
+        .get(&url)
+        .send()
+        .await
+        .map_err(|e| GatewayError::Transient(format!("newznab grab {base}: {}", e.without_url())))?;
+
+    let status = resp.status();
+    if !status.is_success() {
+        return Err(match status.as_u16() {
+            404 => GatewayError::NotFound,
+            429 => GatewayError::RateLimited {
+                retry_after_s: resp
+                    .headers()
+                    .get(reqwest::header::RETRY_AFTER)
+                    .and_then(|v| v.to_str().ok())
+                    .and_then(|v| v.parse().ok())
+                    .unwrap_or(QUOTA_RETRY_SECS),
+            },
+            _ if status.is_server_error() => {
+                GatewayError::Transient(format!("newznab grab {base}: HTTP {status}"))
+            }
+            _ => GatewayError::Permanent(format!("newznab grab {base}: HTTP {status}")),
+        });
+    }
+
+    let body = resp
+        .bytes()
+        .await
+        .map_err(|e| GatewayError::Transient(format!("newznab grab {base}: {}", e.without_url())))?;
+    if body.len() > MAX_NZB_BYTES {
+        return Err(GatewayError::Permanent(format!(
+            "newznab grab {base}: .nzb is {} bytes, over the {MAX_NZB_BYTES}-byte ceiling",
+            body.len()
+        )));
+    }
+    if looks_like_html(&body) {
+        return Err(GatewayError::Permanent(format!(
+            "indexer {base} answered the .nzb grab with an HTML page — check the API base and key"
+        )));
+    }
+    let text: Cow<str> = String::from_utf8_lossy(&body);
+    if let Some(err) = newznab_error(&text) {
+        return Err(classify(&err));
+    }
+    Ok(text.into_owned().into_bytes())
+}
+
+/// Drop the NZB `<head>` except `<meta type="password">`.
+///
+/// The grabbed file is seeded to the whole mesh, and indexers put
+/// account-identifying bits in `<head>` metadata. The password must survive:
+/// meta-share needs it to unpack a protected RAR set (`NzbManifest.password`).
+/// `<file>` / `<segment>` content is copied byte-for-byte, so the Message-IDs —
+/// and the `nzb-posting` cid minted from them — are unchanged. Output is UTF-8.
+pub fn strip_head_keep_password(xml: &[u8]) -> anyhow::Result<Vec<u8>> {
+    use quick_xml::events::{BytesEnd, BytesStart, Event};
+    use quick_xml::{Reader, Writer};
+
+    let text = String::from_utf8_lossy(xml);
+    let mut reader = Reader::from_str(&text);
+    let mut writer = Writer::new(Vec::with_capacity(xml.len()));
+
+    // Nesting depth inside <head> (0 = outside), and inside a kept password meta.
+    let mut head_depth = 0usize;
+    let mut password_depth = 0usize;
+    let mut head_start: Option<BytesStart<'static>> = None;
+    let mut kept: Vec<Event<'static>> = Vec::new();
+
+    loop {
+        let ev = reader.read_event().map_err(|e| anyhow!("nzb xml: {e}"))?;
+        if matches!(ev, Event::Eof) {
+            break;
+        }
+        if head_depth == 0 {
+            match &ev {
+                Event::Start(e) if e.local_name().as_ref() == b"head" => {
+                    head_depth = 1;
+                    head_start = Some(e.clone().into_owned());
+                    kept.clear();
+                }
+                Event::Empty(e) if e.local_name().as_ref() == b"head" => {}
+                _ => writer
+                    .write_event(ev)
+                    .map_err(|e| anyhow!("nzb xml write: {e}"))?,
+            }
+            continue;
+        }
+
+        match &ev {
+            Event::Start(e) => {
+                if password_depth > 0 {
+                    password_depth += 1;
+                    kept.push(ev.into_owned());
+                } else if is_password_meta(e) {
+                    password_depth = 1;
+                    kept.push(ev.into_owned());
+                } else {
+                    head_depth += 1;
+                }
+            }
+            Event::End(e) => {
+                if password_depth > 0 {
+                    password_depth -= 1;
+                    kept.push(ev.into_owned());
+                } else if head_depth == 1 && e.local_name().as_ref() == b"head" {
+                    head_depth = 0;
+                    let start = head_start.take().context("head start tag")?;
+                    if !kept.is_empty() {
+                        let name = String::from_utf8_lossy(start.name().as_ref()).into_owned();
+                        writer
+                            .write_event(Event::Start(start))
+                            .map_err(|e| anyhow!("nzb xml write: {e}"))?;
+                        for k in kept.drain(..) {
+                            writer
+                                .write_event(k)
+                                .map_err(|e| anyhow!("nzb xml write: {e}"))?;
+                        }
+                        writer
+                            .write_event(Event::End(BytesEnd::new(name)))
+                            .map_err(|e| anyhow!("nzb xml write: {e}"))?;
+                    }
+                } else {
+                    head_depth -= 1;
+                }
+            }
+            Event::Empty(e) => {
+                if password_depth > 0 || is_password_meta(e) {
+                    kept.push(ev.into_owned());
+                }
+            }
+            _ => {
+                if password_depth > 0 {
+                    kept.push(ev.into_owned());
+                }
+            }
+        }
+    }
+    Ok(writer.into_inner())
+}
+
+fn is_password_meta(e: &quick_xml::events::BytesStart<'_>) -> bool {
+    e.local_name().as_ref() == b"meta"
+        && e.attributes().flatten().any(|a| {
+            a.key.local_name().as_ref() == b"type" && a.value.eq_ignore_ascii_case(b"password")
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The exact body nzbgeek returns for an anonymous `t=get` — 200 OK, valid
+    /// XML, zero files. Reported as "nzb has no files/segments" until this
+    /// existed, which is a parser-shaped message for a credentials-shaped fault.
+    #[test]
+    fn reports_the_indexers_own_reason() {
+        let xml = r#"<?xml version="1.0" encoding="UTF-8"?>
+<error code="109" description="Invalid User Agent"/>"#;
+        let err = newznab_error(xml).expect("an error");
+        assert_eq!(err.to_string(), "code 109: Invalid User Agent");
+        assert!(matches!(classify(&err), GatewayError::Permanent(_)));
+    }
+
+    #[test]
+    fn a_real_nzb_is_not_an_error() {
+        let xml = r#"<?xml version="1.0"?>
+<nzb xmlns="http://www.newzbin.com/DTD/2003/nzb"><file subject="x"><segments>
+<segment number="1" bytes="10">abc@d</segment></segments></file></nzb>"#;
+        assert_eq!(newznab_error(xml), None);
+    }
+
+    #[test]
+    fn only_the_head_is_scanned() {
+        let xml = format!(
+            "<nzb>{}<error code=\"1\" description=\"late\"/></nzb>",
+            "x".repeat(2000)
+        );
+        assert_eq!(newznab_error(&xml), None);
+    }
+
+    #[test]
+    fn limit_codes_are_rate_limited_and_300_is_not_found() {
+        for code in [429, 500, 501] {
+            let err = NewznabError { code: Some(code), description: None };
+            assert!(
+                matches!(classify(&err), GatewayError::RateLimited { retry_after_s } if retry_after_s == QUOTA_RETRY_SECS),
+                "{code}"
+            );
+        }
+        let gone = NewznabError { code: Some(300), description: Some("No such item".into()) };
+        assert!(matches!(classify(&gone), GatewayError::NotFound));
+        let bad_key = NewznabError { code: Some(100), description: Some("Incorrect user credentials".into()) };
+        assert!(matches!(classify(&bad_key), GatewayError::Permanent(m) if m.contains("API key")));
+    }
+
+    #[test]
+    fn html_pages_are_detected() {
+        assert!(looks_like_html(b"<!DOCTYPE html><html><body>login</body></html>"));
+        assert!(looks_like_html(b"\n  <html lang=\"en\">"));
+        assert!(!looks_like_html(b"<?xml version=\"1.0\"?><nzb/>"));
+    }
+
+    #[test]
+    fn hosts_normalise_and_authority_drops_the_path() {
+        assert_eq!(normalize_indexer_host(" https://API.Example.com/ "), "api.example.com");
+        assert_eq!(authority("api.nzbgeek.info/api"), "api.nzbgeek.info");
+        assert_eq!(authority("127.0.0.1:8080"), "127.0.0.1:8080");
+        // Normalise BEFORE splitting, or a scheme's `//` is taken as the path.
+        assert_eq!(host_key("https://API.nzbgeek.info/api/"), "api.nzbgeek.info");
+        assert_eq!(host_key("api.nzbgeek.info/api"), "api.nzbgeek.info");
+    }
+
+    const WITH_HEAD: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<nzb xmlns="http://www.newzbin.com/DTD/2003/nzb">
+  <head>
+    <meta type="title">Some.Release.1080p</meta>
+    <meta type="x-grabbed-by">user-12345</meta>
+    <meta type="password">s3cr&amp;t</meta>
+  </head>
+  <file poster="[email protected]" date="1" subject="a [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>"#;
+
+    #[test]
+    fn head_strip_keeps_only_the_password() {
+        let out = String::from_utf8(strip_head_keep_password(WITH_HEAD.as_bytes()).unwrap()).unwrap();
+        assert!(out.contains(r#"<meta type="password">s3cr&amp;t</meta>"#), "{out}");
+        assert!(!out.contains("user-12345"), "{out}");
+        assert!(!out.contains("Some.Release.1080p"), "{out}");
+        assert!(out.contains("<head>") && out.contains("</head>"), "{out}");
+        // The file body is untouched.
+        assert!(out.contains(r#"<segment bytes="100" number="1">[email protected]</segment>"#), "{out}");
+        assert!(out.contains(r#"subject="a [1/1] &quot;a.mkv&quot; yEnc""#), "{out}");
+    }
+
+    #[test]
+    fn head_without_password_is_dropped_entirely() {
+        let xml = WITH_HEAD.replace(r#"<meta type="password">s3cr&amp;t</meta>"#, "");
+        let out = String::from_utf8(strip_head_keep_password(xml.as_bytes()).unwrap()).unwrap();
+        assert!(!out.contains("<head"), "{out}");
+        assert!(out.contains("[email protected]"), "{out}");
+    }
+
+    #[test]
+    fn no_head_is_a_no_op_on_content() {
+        let xml = r#"<nzb><file subject="s"><segments><segment number="1">id@x</segment></segments></file></nzb>"#;
+        let out = strip_head_keep_password(xml.as_bytes()).unwrap();
+        assert_eq!(String::from_utf8(out).unwrap(), xml);
+    }
+}

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

@@ -110,7 +110,7 @@ pub fn load(nzb_root: &Path, guid: &str) -> Result<Option<Manifest>> {
 /// meta-share would happily play. Surrounding `<>` are stripped if a producer
 /// included them — the mint normalises this too, but keeping the stored form
 /// canonical means the two never disagree about what was hashed.
-fn extract_message_ids(xml: &[u8]) -> Result<Vec<String>> {
+pub(crate) fn extract_message_ids(xml: &[u8]) -> Result<Vec<String>> {
     use quick_xml::events::Event;
     use quick_xml::Reader;
 

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

@@ -23,25 +23,51 @@
 //! sha2-256 IPFS cid, publishes it as a first-class `fileType=nzb` meta-core
 //! record, and rewrites the field to that cid. Identical to how posters are
 //! handled. See the study's §5.3.
+//!
+//! # Redeeming external releases
+//!
+//! `compute_outcomes` also accepts an `nzb-release` (`0x1005`) cid — minted by
+//! meta-feeder-torznab from a Newznab search hit — and grabs its `.nzb` with the
+//! key configured for that indexer host ([`crate::newznab`]). It returns ONE
+//! `Sha2_256` outcome carrying the `.nzb` (head stripped, password kept); the
+//! gateway stores it under `/files/plugin/meta-feeder-usenet/`, writes its cid
+//! onto the release record as `manifest`, and merges the outcome record's
+//! `cids/<nzb-posting>` so the release becomes portable to any NNTP-only peer.
+//! Works without nntmux configured.
 
 use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
 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::hash::compute_nzb_posting_cid;
-use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome};
+use meta_feeder_sdk::hash::{
+    codec_of, compute_ipfs_cid, compute_nzb_posting_cid, decode_nzb_release_cid, NZB_RELEASE_CODEC,
+};
+use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim};
 use meta_feeder_sdk::query::GatewayQuery;
 use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth};
 use tokio::sync::RwLock;
 use tracing::{debug, info, warn};
 
+use crate::newznab::{self, IndexerCred};
 use crate::nntmux::db::{content_kind_for_category, file_type_for_category, Db, Release};
 use crate::nntmux::nzb;
 
 pub const UPSTREAM_ID: &str = "usenet";
 
+/// `/files/plugin/<PACKAGE>/` — where the gateway stores the `.nzb` files this
+/// plugin redeems.
+pub const PACKAGE: &str = "meta-feeder-usenet";
+
+/// Indexers answer an anonymous `t=get` with `<error code="109" description=
+/// "Invalid User Agent"/>` (nzbgeek does), so always send a named client.
+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;
+
 /// Operator-supplied settings, persisted to `<cache_dir>/config.json` by the
 /// SDK's config surface. Env is a **seed only** — the file wins on the next
 /// restart (invariant 12; there is no hot reload).
@@ -52,6 +78,9 @@ pub struct Settings {
     /// Root of nntmux's NZB store, shared with the sidecar as a volume. The
     /// directory nntmux's `PATH_TO_NZBS` points at.
     pub nzb_root: String,
+    /// Per-host Newznab keys for redeeming `nzb-release` locators. Config page
+    /// only — no env seed.
+    pub indexers: Vec<IndexerCred>,
 }
 
 impl Settings {
@@ -60,6 +89,7 @@ impl Settings {
             db_url: std::env::var("NNTMUX_DB_URL").unwrap_or_default(),
             nzb_root: std::env::var("NNTMUX_NZB_PATH")
                 .unwrap_or_else(|_| "/nntmux-nzb".to_string()),
+            indexers: Vec::new(),
         }
     }
 
@@ -82,6 +112,13 @@ impl Settings {
                 self.nzb_root = s.to_string();
             }
         }
+        if let Some(rows) = v.get("indexers").and_then(|x| x.as_array()) {
+            self.indexers = rows
+                .iter()
+                .filter_map(|row| serde_json::from_value::<IndexerCred>(row.clone()).ok())
+                .filter(|c| !c.host.trim().is_empty() && !c.api_key.trim().is_empty())
+                .collect();
+        }
     }
 }
 
@@ -92,6 +129,12 @@ pub struct UsenetPlugin {
     /// — it soft-skips its upstream instead of failing to boot (invariant 10).
     db: Arc<RwLock<Option<Db>>>,
     nzb_root: PathBuf,
+    /// Client for `.nzb` grabs (named User-Agent, follows redirects so an HTML
+    /// landing page is detected rather than surfacing as a bare 302).
+    http: reqwest::Client,
+    /// `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,
 }
 
 impl Default for UsenetPlugin {
@@ -106,9 +149,79 @@ impl UsenetPlugin {
             settings: Settings::default(),
             db: Arc::new(RwLock::new(None)),
             nzb_root: PathBuf::new(),
+            http: build_http_client(GRAB_TIMEOUT_SECS, USER_AGENT, None),
+            grab_scheme: "https",
         }
     }
 
+    /// Grab over plain `http` instead of `https`. For tests against a local
+    /// mock indexer only — a locator carries no scheme, and a real indexer is TLS.
+    pub fn with_plain_http_grabs(mut self) -> Self {
+        self.grab_scheme = "http";
+        self
+    }
+
+    /// The configured key for an indexer authority, if any.
+    fn indexer_key(&self, authority: &str) -> Option<&str> {
+        let want = newznab::host_key(authority);
+        self.settings
+            .indexers
+            .iter()
+            .find(|c| newznab::host_key(&c.host) == want)
+            .map(|c| c.api_key.as_str())
+    }
+
+    /// Redeem an `nzb-release` locator: grab its `.nzb` with this feeder's key
+    /// for that host, strip the head (password kept), and mint the portable
+    /// `nzb-posting` cid from its Message-IDs.
+    ///
+    /// ⚠ **Spends indexer quota.** Reached only through `/compute`, which the
+    /// gateway calls on a real play and never for a release it already holds
+    /// the manifest of. `NotFound` = "not mine" (no key for that host), so the
+    /// gateway can try another claimer without anything being spent.
+    async fn redeem_release(&self, cid: &str) -> Result<Vec<HashOutcome>, GatewayError> {
+        let Some((api_base, id)) = decode_nzb_release_cid(cid) else {
+            return Err(GatewayError::NotFound);
+        };
+        let host = newznab::authority(&api_base);
+        let Some(key) = self.indexer_key(host) else {
+            debug!(target: "meta-feeder", %host, "usenet: no indexer key for this host; not ours to redeem");
+            return Err(GatewayError::NotFound);
+        };
+
+        info!(target: "meta-feeder", %host, %id, "usenet: grabbing .nzb (indexer quota)");
+        let raw = newznab::grab(&self.http, self.grab_scheme, &api_base, &id, key).await?;
+        let nzb = newznab::strip_head_keep_password(&raw)
+            .map_err(|e| GatewayError::Permanent(format!("nzb from {host} is not parseable: {e}")))?;
+        let message_ids = nzb::extract_message_ids(&nzb)
+            .map_err(|e| GatewayError::Permanent(format!("nzb from {host} is not parseable: {e}")))?;
+        if message_ids.is_empty() {
+            return Err(GatewayError::Permanent(format!(
+                "nzb from {host} has no segments"
+            )));
+        }
+
+        let posting = compute_nzb_posting_cid(&message_ids);
+        let mut fields = BTreeMap::new();
+        // Merged onto the release record by the gateway: the posting cid
+        // outranks the locator, so the release becomes redeemable by any peer
+        // with a plain NNTP provider — no indexer key anywhere after this grab.
+        fields.insert(format!("cids/{posting}"), "true".to_string());
+        fields.insert("segmentCount".to_string(), message_ids.len().to_string());
+
+        Ok(vec![HashOutcome {
+            hash: Hash(compute_ipfs_cid(&nzb)),
+            hash_kind: HashKind::Sha2_256,
+            bytes: Some(nzb.into()),
+            record: Some(DiscoveryRecord {
+                upstream_id: UPSTREAM_ID.to_string(),
+                record_id: cid.to_string(),
+                fields,
+            }),
+            file_extension: Some("nzb".to_string()),
+        }])
+    }
+
     async fn db(&self) -> Result<Db, GatewayError> {
         self.db
             .read()
@@ -272,6 +385,10 @@ impl FeederPlugin for UsenetPlugin {
         if term.trim().is_empty() {
             return Ok(vec![]);
         }
+        if self.settings.db_url.is_empty() {
+            // Grab-only deployment (indexer keys, no nntmux): nothing to search.
+            return Ok(vec![]);
+        }
         let db = self.db().await?;
         let releases = db
             .search(term.trim(), max_results)
@@ -305,6 +422,11 @@ impl FeederPlugin for UsenetPlugin {
     }
 
     async fn compute_outcomes(&self, record_id: &str) -> Result<Vec<HashOutcome>, GatewayError> {
+        // An external release locator, not one of our nntmux guids. Checked
+        // first: it needs no database.
+        if codec_of(record_id) == Some(NZB_RELEASE_CODEC) {
+            return self.redeem_release(record_id).await;
+        }
         let db = self.db().await?;
         let release = db
             .by_guid(record_id)
@@ -357,14 +479,34 @@ impl FeederPlugin for UsenetPlugin {
     }
 
     fn health(&self) -> PluginHealth {
-        if self.settings.db_url.is_empty() {
+        if self.settings.db_url.is_empty() && self.settings.indexers.is_empty() {
             return PluginHealth::Degraded {
-                reason: "nntmux database not configured".into(),
+                reason: "neither an nntmux database nor an indexer key is configured".into(),
             };
         }
         PluginHealth::Ok
     }
 
+    fn package(&self) -> Option<&'static str> {
+        Some(PACKAGE)
+    }
+
+    /// Claims `nzb-release` for every host with a key; nothing without keys.
+    fn redeems(&self) -> Vec<RedeemClaim> {
+        let mut hosts: Vec<String> = self
+            .settings
+            .indexers
+            .iter()
+            .map(|c| newznab::host_key(&c.host))
+            .collect();
+        hosts.sort();
+        hosts.dedup();
+        if hosts.is_empty() {
+            return Vec::new();
+        }
+        vec![RedeemClaim::nzb_release(hosts)]
+    }
+
     fn served_file_types(&self) -> &'static [&'static str] {
         &["video", "audio", "document", "archive"]
     }
@@ -380,10 +522,9 @@ impl FeederPlugin for UsenetPlugin {
                     .with_help(
                         "MariaDB DSN of the nntmux sidecar's catalog, e.g. \
                          mysql://nntmux:secret@nntmux-db:3306/nntmux. Blank → the \
-                         upstream soft-skips and this plugin reports Degraded. Takes \
+                         self-scan search is off (indexer-key redeems still work). Takes \
                          effect on the next feeder restart (no hot reload).",
-                    )
-                    .required(),
+                    ),
                 F::text("nzb_root", "NZB store path")
                     .with_help(
                         "Path to nntmux's NZB directory as mounted INTO this container \
@@ -394,8 +535,25 @@ impl FeederPlugin for UsenetPlugin {
                          nntmux nests each file one directory per leading guid character, \
                          `nzbsplitlevel` deep (the image ships 4 → 9/7/a/2/97a2….nzb.gz), \
                          and the depth is detected automatically.",
-                    )
-                    .required(),
+                    ),
+                F::record_array(
+                    "indexers",
+                    "Indexer keys (redeem nzb-release)",
+                    vec![
+                        F::text("host", "Host").required().with_help(
+                            "The indexer's bare host, e.g. api.nzbgeek.info — no scheme, no /api. \
+                             Matched against the host inside each nzb-release cid.",
+                        ),
+                        F::secret("api_key", "API key").required().with_help(
+                            "The indexer account's API key. Every grab spends that account's \
+                             daily download quota — once per release, on a real play only.",
+                        ),
+                    ],
+                )
+                .with_help(
+                    "Newznab keys used to grab the .nzb of releases minted by the torznab feeder \
+                     (Prowlarr search). Takes effect on the next feeder restart.",
+                ),
             ],
         }
     }
@@ -404,6 +562,7 @@ impl FeederPlugin for UsenetPlugin {
         serde_json::json!({
             "db_url": self.settings.db_url,
             "nzb_root": self.settings.nzb_root,
+            "indexers": self.settings.indexers,
         })
     }
 }

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

@@ -86,11 +86,33 @@ async fn health_degrades_gracefully_with_no_nntmux_configured() {
         "usenet plugin health entry should report Degraded, got {usenet_health}"
     );
     assert_eq!(
-        usenet_health["health"]["reason"], "nntmux database not configured",
+        usenet_health["health"]["reason"],
+        "neither an nntmux database nor an indexer key is configured",
         "got {usenet_health}"
     );
 }
 
+/// Unconfigured: the manifest still names the storage folder, but claims no
+/// redeems — there is no key to redeem anything with.
+#[tokio::test]
+async fn manifest_names_the_package_and_claims_nothing_without_keys() {
+    let (addr, _dir) = boot_feeder().await;
+    let manifest: serde_json::Value = reqwest::get(format!("http://{addr}/manifest"))
+        .await
+        .expect("manifest req")
+        .json()
+        .await
+        .expect("manifest json");
+    let usenet = manifest["plugins"]
+        .as_array()
+        .expect("plugins array")
+        .iter()
+        .find(|p| p["id"] == "usenet")
+        .expect("usenet entry");
+    assert_eq!(usenet["package"], "meta-feeder-usenet");
+    assert_eq!(usenet["redeems"], serde_json::json!([]));
+}
+
 /// Every seedable/resolvable field this feeder advertises resolving to must
 /// stay in sync with the gateway core's `SEEDABLE_FIELDS` table
 /// (`remote_feeder.rs`) — this is a documentation-level regression guard, not

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

@@ -0,0 +1,249 @@
+//! Redeeming an external `nzb-release` (`0x1005`) locator through `POST /compute`.
+//!
+//! Boots the real `UsenetPlugin` in the SDK router with indexer keys in its
+//! `config.json` (the config-page shape) and a wiremock Newznab indexer, then
+//! drives `/compute` with a locator cid exactly as the gateway's redeem route
+//! does. No nntmux: redeeming must work in a grab-only deployment.
+
+use base64::Engine as _;
+use meta_feeder_sdk::hash::{
+    compute_ipfs_cid, compute_nzb_posting_cid, compute_nzb_release_cid,
+};
+use meta_feeder_sdk::serve::{configure_plugins, router};
+use meta_feeder_sdk::{ComputeRequest, ComputeResponse, HashKindDto, RedeemsResponse};
+use usenet_feeder::usenet::UsenetPlugin;
+use wiremock::matchers::{header_exists, method, path, query_param};
+use wiremock::{Mock, MockServer, ResponseTemplate};
+
+const KEY: &str = "indexer-key-1";
+const RELEASE_ID: &str = "0123456789abcdef";
+
+const NZB: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<nzb xmlns="http://www.newzbin.com/DTD/2003/nzb">
+  <head>
+    <meta type="title">Some.Release.1080p</meta>
+    <meta type="x-grabbed-by">account-4242</meta>
+    <meta type="password">hunter2</meta>
+  </head>
+  <file poster="[email protected]" date="1" subject="Some.Release [1/1] &quot;a.mkv&quot; yEnc">
+    <groups><group>alt.binaries.teevee</group></groups>
+    <segments>
+      <segment bytes="100" number="1">[email protected]</segment>
+      <segment bytes="100" number="2">[email protected]</segment>
+    </segments>
+  </file>
+</nzb>"#;
+
+/// The indexer's scheme-less API base as it appears inside a locator.
+fn authority(server: &MockServer) -> String {
+    server.uri().trim_start_matches("http://").to_string()
+}
+
+async fn boot(indexers: serde_json::Value) -> (String, tempfile::TempDir) {
+    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!({ "indexers": indexers }).to_string(),
+    )
+    .expect("write config");
+
+    let plugin = UsenetPlugin::new().with_plain_http_grabs();
+    let configured = configure_plugins(vec![Box::new(plugin)], 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 addr = listener.local_addr().expect("addr");
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.expect("serve");
+    });
+    (format!("http://{addr}"), dir)
+}
+
+async fn boot_with_key_for(server: &MockServer) -> (String, tempfile::TempDir) {
+    boot(serde_json::json!([{ "host": authority(server), "api_key": KEY }])).await
+}
+
+async fn compute(base: &str, record_id: &str) -> reqwest::Response {
+    reqwest::Client::new()
+        .post(format!("{base}/compute"))
+        .json(&ComputeRequest {
+            upstream_id: "usenet".into(),
+            record_id: record_id.into(),
+        })
+        .send()
+        .await
+        .expect("POST /compute")
+}
+
+fn grab_mock() -> wiremock::MockBuilder {
+    Mock::given(method("GET"))
+        .and(path("/api"))
+        .and(query_param("t", "get"))
+        .and(query_param("id", RELEASE_ID))
+        .and(query_param("apikey", KEY))
+}
+
+#[tokio::test]
+async fn a_release_locator_redeems_to_the_nzb_and_its_posting_cid() {
+    let indexer = MockServer::start().await;
+    grab_mock()
+        .and(header_exists("user-agent"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(NZB))
+        .expect(1)
+        .mount(&indexer)
+        .await;
+    let (base, _dir) = boot_with_key_for(&indexer).await;
+    let cid = compute_nzb_release_cid(&authority(&indexer), RELEASE_ID).unwrap();
+
+    let resp = compute(&base, &cid).await;
+    assert_eq!(resp.status(), 200, "{:?}", resp.text().await);
+    let body: ComputeResponse = resp.json().await.expect("ComputeResponse");
+    assert_eq!(body.outcomes.len(), 1);
+    let o = &body.outcomes[0];
+    assert_eq!(o.hash_kind, HashKindDto::Sha2_256);
+    assert_eq!(o.file_extension.as_deref(), Some("nzb"));
+
+    let bytes = base64::engine::general_purpose::STANDARD
+        .decode(o.bytes_b64.as_deref().expect("bytes"))
+        .expect("base64");
+    assert_eq!(o.hash, compute_ipfs_cid(&bytes), "hash is the cid of the returned bytes");
+
+    let text = String::from_utf8(bytes).expect("utf-8");
+    assert!(text.contains(r#"<meta type="password">hunter2</meta>"#), "password kept: {text}");
+    assert!(!text.contains("account-4242"), "head metadata stripped: {text}");
+    assert!(text.contains("[email protected]") && text.contains("[email protected]"));
+
+    let record = o.record.as_ref().expect("fields to merge onto the release record");
+    let posting = compute_nzb_posting_cid(&["[email protected]", "[email protected]"]);
+    assert_eq!(record.fields.get(&format!("cids/{posting}")).map(String::as_str), Some("true"));
+    assert_eq!(record.fields.get("segmentCount").map(String::as_str), Some("2"));
+    assert!(
+        !record.fields.keys().any(|k| k == &format!("cids/{}", o.hash)),
+        "the manifest's own cid must never become a cids/ member of the release"
+    );
+}
+
+#[tokio::test]
+async fn a_host_without_a_key_is_not_ours_and_spends_nothing() {
+    let indexer = MockServer::start().await;
+    Mock::given(method("GET"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(NZB))
+        .expect(0)
+        .mount(&indexer)
+        .await;
+    let (base, _dir) =
+        boot(serde_json::json!([{ "host": "other.example", "api_key": KEY }])).await;
+    let cid = compute_nzb_release_cid(&authority(&indexer), RELEASE_ID).unwrap();
+
+    assert_eq!(compute(&base, &cid).await.status(), 404);
+}
+
+#[tokio::test]
+async fn invalid_user_agent_error_is_permanent() {
+    let indexer = MockServer::start().await;
+    grab_mock()
+        .respond_with(ResponseTemplate::new(200).set_body_string(
+            r#"<?xml version="1.0" encoding="UTF-8"?><error code="109" description="Invalid User Agent"/>"#,
+        ))
+        .mount(&indexer)
+        .await;
+    let (base, _dir) = boot_with_key_for(&indexer).await;
+    let cid = compute_nzb_release_cid(&authority(&indexer), RELEASE_ID).unwrap();
+
+    let resp = compute(&base, &cid).await;
+    assert_eq!(resp.status(), 422);
+    assert!(resp.text().await.unwrap().contains("Invalid User Agent"));
+}
+
+#[tokio::test]
+async fn http_429_is_rate_limited() {
+    let indexer = MockServer::start().await;
+    grab_mock()
+        .respond_with(ResponseTemplate::new(429).insert_header("Retry-After", "120"))
+        .mount(&indexer)
+        .await;
+    let (base, _dir) = boot_with_key_for(&indexer).await;
+    let cid = compute_nzb_release_cid(&authority(&indexer), RELEASE_ID).unwrap();
+
+    assert_eq!(compute(&base, &cid).await.status(), 429);
+}
+
+#[tokio::test]
+async fn download_limit_error_code_is_rate_limited() {
+    let indexer = MockServer::start().await;
+    grab_mock()
+        .respond_with(ResponseTemplate::new(200).set_body_string(
+            r#"<error code="501" description="Download limit reached"/>"#,
+        ))
+        .mount(&indexer)
+        .await;
+    let (base, _dir) = boot_with_key_for(&indexer).await;
+    let cid = compute_nzb_release_cid(&authority(&indexer), RELEASE_ID).unwrap();
+
+    assert_eq!(compute(&base, &cid).await.status(), 429);
+}
+
+#[tokio::test]
+async fn an_html_page_is_a_permanent_error() {
+    let indexer = MockServer::start().await;
+    grab_mock()
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_body_string("<!DOCTYPE html><html><body>Please log in</body></html>"),
+        )
+        .mount(&indexer)
+        .await;
+    let (base, _dir) = boot_with_key_for(&indexer).await;
+    let cid = compute_nzb_release_cid(&authority(&indexer), RELEASE_ID).unwrap();
+
+    let resp = compute(&base, &cid).await;
+    assert_eq!(resp.status(), 422);
+    assert!(resp.text().await.unwrap().contains("HTML"));
+}
+
+/// Keys configured → the plugin claims `nzb-release` for exactly those hosts, on
+/// both `/redeems` and `/manifest`, and reports healthy without nntmux.
+#[tokio::test]
+async fn keys_are_advertised_as_redeem_claims() {
+    let (base, _dir) = boot(serde_json::json!([
+        { "host": "https://API.nzbgeek.info/", "api_key": "k1" },
+        { "host": "api.nzb.life", "api_key": "k2" },
+        { "host": "blank-key.example", "api_key": "" }
+    ]))
+    .await;
+
+    let r: RedeemsResponse = reqwest::get(format!("{base}/redeems"))
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    let claims = serde_json::to_value(&r.redeems["usenet"]).unwrap();
+    assert_eq!(
+        claims,
+        serde_json::json!([{
+            "codec": "nzb-release",
+            "field": "manifest",
+            "hosts": ["api.nzb.life", "api.nzbgeek.info"],
+            "sources": []
+        }])
+    );
+
+    let m: serde_json::Value = reqwest::get(format!("{base}/manifest"))
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(m["plugins"][0]["package"], "meta-feeder-usenet");
+    assert_eq!(m["plugins"][0]["redeems"], claims);
+
+    let h: serde_json::Value = reqwest::get(format!("{base}/health"))
+        .await
+        .unwrap()
+        .json()
+        .await
+        .unwrap();
+    assert_eq!(h["status"], "ok", "{h}");
+}