2 Commits
Author SHA1 Message Date
Worph 0d9d0132de 1.1.0: reach nntmux over its Newznab API (search + getnzb, metamesh key) with a per-release .nzb cache; drop the MariaDB/storage coupling
Publish image / publish (push) Failing after 13s
2026-10-04 12:30:11 +00:00
Worph 45f88dcdfe align with metadata standard
Publish image / publish (push) Failing after 13s
2026-09-13 15:53:21 +02:00
14 changed files with 1150 additions and 1512 deletions
Generated
+20 -864
View File
File diff suppressed because it is too large Load Diff
+45 -18
View File
@@ -1,8 +1,8 @@
# meta-feeder-usenet # meta-feeder-usenet
Usenet feeder sidecar for the MetaMesh gateway tier. Bridges a Usenet header Usenet feeder sidecar for the MetaMesh gateway tier. Bridges a Usenet header
scan **we run ourselves** (an [nntmux](https://github.com/NNTmux/newznab-tmux) scan **we run ourselves** — the **NNTmux** app ([nntmux](https://github.com/NNTmux/newznab-tmux),
sidecar) into the MetaMesh network. worph/AppStore) — into the MetaMesh network, over nntmux's Newznab API.
## Why this exists ## Why this exists
@@ -33,11 +33,11 @@ Design record:
## Shape ## Shape
``` ```
[nntmux sidecar: PHP + MariaDB + Redis] [usenet-feeder: Rust, FeederPlugin] [NNTmux app: PHP + MariaDB + Redis] [usenet-feeder: Rust, FeederPlugin]
header-scan groups (few conns) ──► reads `releases` (searchname/size/guid) header-scan groups (few conns) ◄── t=search&q=… (Newznab API, `metamesh` key)
collate → releases ──► gunzips <nzb_root>/<c>/<guid>.nzb.gz collate → releases ◄── /getnzb?id=<guid> once per release → cache
writes <guid>.nzb.gz ──► mints 0x1003 from the Message-ID set writes <guid>.nzb.gz mints 0x1003 from the Message-ID set
(CBP purged; the gz is the only ──► emits the record + a relative (CBP purged; the .nzb is the only emits the record + a relative
durable home of the msg-ids) `manifest_url` onto its own /blob route durable home of the msg-ids) `manifest_url` onto its own /blob route
│ │
▼ ▼
@@ -51,9 +51,34 @@ Design record:
provider → materialise → play provider → materialise → play
``` ```
The sidecar is **persistent cache, cheap to lose** — everything in it is ### nntmux over its API, not its database (1.1)
re-derivable by re-scanning. Same pattern as the Tribler sidecar: stateful
upstream, but the gateway still sees a stateless HTTP feeder. Until 1.1 the feeder read nntmux's MariaDB with SQL and gunzipped
`<guid>.nzb.gz` off nntmux's storage volume. That made the two a single unit:
the feeder held nntmux's DB password, depended on its table layout and its NZB
shard depth, and had to mount its storage. nntmux is now its own app, and the
feeder uses the same contract any Newznab client uses (the torznab feeder does
the same with Prowlarr):
| Need | Call | nntmux counter |
|---|---|---|
| search | `GET {nntmux_url}/api/v1/api?t=search&q=…&extended=1&apikey=…` | `apirequests` |
| a release's `.nzb` (Message-IDs) | `GET {nntmux_url}/getnzb?id=<guid>.nzb&r=<key>` | `downloadrequests` |
| compute (by guid) | `t=details&id=<guid>` | `apirequests` |
| reachability (`/health`) | `t=caps` (no key) | — |
- ⚠ The API is under **`/api/v1/api`**; `/api` is a 404. `t=get` redirects to
`/getnzb` on the *public* `APP_URL` host, so the feeder calls `/getnzb` on the
internal host directly.
- Every search hit must carry its `0x1003` cid (else meta-watch shows
"Unavailable"), so a hit costs one `/getnzb` **the first time it is seen**.
Each `.nzb` is cached under `<state>/gateway/usenet/nntmux-nzb/<guid>.nzb`
(a collated release never changes) and reused by repeat searches, `/compute`
and `/blob`.
- nntmux caps every role per 24 h (Admin 1000/1000, User 10/10) and throttles
`/api` to 60/min. Use the NNTmux app's dedicated **`metamesh`** user (its role
has raised caps); the feeder paces itself to ≤ 50 requests/min, and stops a
query's downloads once nntmux reports a cap (`500`/`501`).
Like every feeder this one is **meta-core-free and blockstore-free**: it finds Like every feeder this one is **meta-core-free and blockstore-free**: it finds
records and serves bytes; the core owns hashing-into-the-blockstore, the records and serves bytes; the core owns hashing-into-the-blockstore, the
@@ -68,18 +93,20 @@ and there is no hot reload.
| Field | Env seed | What | | Field | Env seed | What |
|---|---|---| |---|---|---|
| `db_url` | `NNTMUX_DB_URL` | MariaDB DSN of nntmux's catalog, e.g. `mysql://nntmux:secret@nntmux-db:3306/nntmux`. Blank → self-scan search is off. | | `nntmux_url` | `NNTMUX_URL` | Base URL of the NNTmux app as this container reaches it (default `http://nntmux`); the API path is appended. |
| `nzb_root` | `NNTMUX_NZB_PATH` | nntmux's NZB directory **as mounted into this container**. Must be the same volume nntmux's `PATH_TO_NZBS` points at, or every release resolves to `NotFound`. | | `nntmux_api_key` | `NNTMUX_API_KEY` | API key of the nntmux user the feeder searches as (the `metamesh` user). Blank → self-scan search is off. |
| `indexers` | — (config page only) | `[{host, api_key}]` Newznab keys for redeeming `nzb-release` locators. `host` is the bare authority (`api.nzbgeek.info` — no scheme, no `/api`). | | `indexers` | — (config page only) | `[{host, api_key}]` Newznab keys for redeeming `nzb-release` locators. `host` is the bare authority (`api.nzbgeek.info` — no scheme, no `/api`). |
Both halves are optional; with neither configured the plugin reports Degraded. Both halves are optional; with neither configured the plugin reports Degraded,
A grab-only deployment (keys, no nntmux) is valid. as it does when nntmux is unreachable or refuses the key. A grab-only deployment
(keys, no nntmux) is valid. A pre-1.1 `config.json` (`db_url`/`nzb_root`) is
ignored with a warning.
Infra env: `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`), Infra env: `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`),
`META_FEEDER_STATE_DIR` (default `/data/meta-feeder`), `RUST_LOG`. `META_FEEDER_STATE_DIR` (default `/data/meta-feeder`), `RUST_LOG`.
Scan policy — which groups, backfill depth, scan interval, connection budget — Scan policy — which groups, backfill depth, scan interval, connection budget —
is the **gateway maintainer's** call and is configured on the nntmux sidecar. is the **gateway maintainer's** call and is configured in the NNTmux app.
Keep the connection budget conservative: the scanner and playback draw on the Keep the connection budget conservative: the scanner and playback draw on the
same provider slots, so a greedy value does not merely over-scan, it stalls same provider slots, so a greedy value does not merely over-scan, it stalls
playback mid-stream. playback mid-stream.
@@ -120,9 +147,9 @@ cargo clippy --workspace --all-targets -- -D warnings
docker build -f feeder-plugin/usenet-feeder/Dockerfile -t ghcr.io/worph/meta-feeder-usenet:dev . docker build -f feeder-plugin/usenet-feeder/Dockerfile -t ghcr.io/worph/meta-feeder-usenet:dev .
``` ```
`meta-feeder-sdk` is a git dependency pinned by tag (`v1.3.0`; this repo used to `meta-feeder-sdk` is a git dependency pinned by tag (`v1.2.2`; this repo used to
vendor a copy). Until that tag is pushed, cargo cannot resolve it — not even vendor a copy). An SDK change is invisible here until it is tagged and pushed,
under a `[patch]` override. To verify against a local SDK checkout, build a and cargo cannot resolve an unpushed tag — not even under a `[patch]` override. To verify against a local SDK checkout, build a
scratch copy with the dependency swapped for a path (never commit that form — scratch copy with the dependency swapped for a path (never commit that form —
this repo is the Docker build context): this repo is the Docker build context):
+2 -14
View File
@@ -9,7 +9,7 @@ publish = false
# Usenet feeder sidecar: bridges a Usenet header scan we run OURSELVES into the # Usenet feeder sidecar: bridges a Usenet header scan we run OURSELVES into the
# MetaMesh network. Unlike `meta-feeder-indexer` (which proxies an external # MetaMesh network. Unlike `meta-feeder-indexer` (which proxies an external
# Newznab/torznab indexer) this feeder owns the metadata half — it reads an # Newznab/torznab indexer) this feeder owns the metadata half — it reads an
# nntmux sidecar's release catalog and mints a PORTABLE `nzb-posting` (0x1003) # nntmux indexer (the NNTmux app, over its Newznab API) and mints a PORTABLE `nzb-posting` (0x1003)
# cid from the article Message-IDs, so any peer with a plain NNTP provider can # cid from the article Message-IDs, so any peer with a plain NNTP provider can
# redeem it without an indexer credential. # redeem it without an indexer credential.
# #
@@ -32,7 +32,7 @@ path = "src/main.rs"
[dependencies] [dependencies]
# Pinned by TAG, never by path. A path dep resolves OUTSIDE this repo, and this # 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.) # 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" } meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.2.2" }
tokio = { version = "1.40", features = ["full"] } tokio = { version = "1.40", features = ["full"] }
axum = { version = "0.7", features = ["json"] } axum = { version = "0.7", features = ["json"] }
@@ -49,18 +49,6 @@ tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
chrono = { version = "0.4", default-features = false, features = ["std", "clock"] } chrono = { version = "0.4", default-features = false, features = ["std", "clock"] }
# nntmux's release catalog lives in MariaDB. `sqlx` in offline-free runtime mode
# (no compile-time macro checking) so the build needs no live DB.
sqlx = { version = "0.8", default-features = false, features = [
"runtime-tokio-rustls",
"mysql",
"macros",
"chrono",
] }
# nntmux stores each release's manifest as a gzipped NZB on disk; the ordered
# Message-IDs survive ONLY there (it purges collections/binaries/parts after
# writing the file), so reading + gunzipping it is the load-bearing path.
flate2 = "1"
quick-xml = { version = "0.36", features = ["serialize"] } quick-xml = { version = "0.36", features = ["serialize"] }
[dev-dependencies] [dev-dependencies]
+8 -5
View File
@@ -1,8 +1,9 @@
//! `usenet-feeder` library surface. The binary (`main.rs`) wraps the plugin in //! `usenet-feeder` library surface. The binary (`main.rs`) wraps the plugin in
//! [`meta_feeder_sdk::serve_feeders`]. //! [`meta_feeder_sdk::serve_feeders`].
//! //!
//! This feeder owns the **metadata half** of Usenet: it reads an nntmux header- //! This feeder owns the **metadata half** of Usenet: it searches an nntmux
//! scan catalog we run ourselves and mints a portable `nzb-posting` (`0x1003`) //! header-scan indexer we run ourselves (the separate NNTmux app, over its
//! Newznab API) and mints a portable `nzb-posting` (`0x1003`)
//! cid from each release's article Message-IDs. Unlike `meta-feeder-indexer`'s //! cid from each release's article Message-IDs. Unlike `meta-feeder-indexer`'s
//! `0x1005` locator, that cid embeds no indexer host, so any peer with a plain //! `0x1005` locator, that cid embeds no indexer host, so any peer with a plain
//! NNTP provider can redeem it — no indexer credential anywhere in the path. //! NNTP provider can redeem it — no indexer credential anywhere in the path.
@@ -20,10 +21,12 @@
//! //!
//! See `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md`. //! See `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md`.
/// Read-only access to the nntmux sidecar: its MariaDB release catalog and the /// Our own nntmux, reached only over its Newznab HTTP API: the client, the
/// on-disk gzip NZBs that are the *only* durable home of the Message-IDs. /// per-release `.nzb` cache, the release type and the Message-ID extraction.
pub mod nntmux { pub mod nntmux {
pub mod db; pub mod api;
pub mod cache;
pub mod catalog;
pub mod nzb; pub mod nzb;
} }
+4 -4
View File
@@ -1,16 +1,16 @@
//! `usenet-feeder` — single-source Usenet feeder sidecar over an nntmux //! `usenet-feeder` — single-source Usenet feeder sidecar over our own nntmux
//! header-scan catalog. //! indexer (the NNTmux app), reached over its Newznab API.
//! //!
//! One plugin (`usenet`) served through [`meta_feeder_sdk::serve_feeders`]. //! One plugin (`usenet`) served through [`meta_feeder_sdk::serve_feeders`].
//! //!
//! Operator config — the nntmux database URL and NZB store path, and the //! Operator config — the nntmux base URL and API key, and the
//! per-host Newznab indexer keys used to redeem `nzb-release` locators — is read //! 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 //! 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 //! 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 //! is no hot reload (gateway invariant 12). Indexer keys have no env seed; set
//! them on the config page. //! them on the config page.
//! //!
//! Env (seed): `NNTMUX_DB_URL`, `NNTMUX_NZB_PATH`. //! Env (seed): `NNTMUX_URL` (default `http://nntmux`), `NNTMUX_API_KEY`.
//! Env (infra): `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`), //! Env (infra): `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`),
//! `META_FEEDER_STATE_DIR` (default `/data/meta-feeder`), `RUST_LOG`. //! `META_FEEDER_STATE_DIR` (default `/data/meta-feeder`), `RUST_LOG`.
+41 -20
View File
@@ -9,6 +9,10 @@
//! //!
//! Moved here from meta-share's `nzb/manifest_resolve.rs::grab_newznab`: the //! 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. //! indexer keys now live on this feeder, meta-share keeps only the NNTP pool.
//!
//! The error handling ([`fetch_nzb`], [`newznab_error`], [`classify`]) is shared
//! with the client for our own nntmux ([`crate::nntmux::api`]), which speaks the
//! same Newznab dialect.
use std::borrow::Cow; use std::borrow::Cow;
@@ -155,44 +159,41 @@ pub async fn grab(
) -> Result<Vec<u8>, GatewayError> { ) -> Result<Vec<u8>, GatewayError> {
let base = api_base.trim_end_matches('/'); let base = api_base.trim_end_matches('/');
let url = format!("{scheme}://{base}/api?t=get&id={id}&apikey={api_key}"); let url = format!("{scheme}://{base}/api?t=get&id={id}&apikey={api_key}");
fetch_nzb(http, &url, base).await
}
/// GET an `.nzb` and turn every way a Newznab server can refuse into the feeder
/// error contract: HTTP status, an HTML page (login wall, Cloudflare, a
/// permalink host), an oversize body, or a `<error code>` answered as `200`.
///
/// `label` names the server in error messages; `url` is never echoed (it
/// carries the API key). Shared by the external-indexer redeem ([`grab`]) and
/// the nntmux client ([`crate::nntmux::api`]).
pub async fn fetch_nzb(http: &reqwest::Client, url: &str, label: &str) -> Result<Vec<u8>, GatewayError> {
let resp = http let resp = http
.get(&url) .get(url)
.send() .send()
.await .await
.map_err(|e| GatewayError::Transient(format!("newznab grab {base}: {}", e.without_url())))?; .map_err(|e| GatewayError::Transient(format!("newznab grab {label}: {}", e.without_url())))?;
let status = resp.status(); let status = resp.status();
if !status.is_success() { if !status.is_success() {
return Err(match status.as_u16() { return Err(status_error(&resp, label));
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 let body = resp
.bytes() .bytes()
.await .await
.map_err(|e| GatewayError::Transient(format!("newznab grab {base}: {}", e.without_url())))?; .map_err(|e| GatewayError::Transient(format!("newznab grab {label}: {}", e.without_url())))?;
if body.len() > MAX_NZB_BYTES { if body.len() > MAX_NZB_BYTES {
return Err(GatewayError::Permanent(format!( return Err(GatewayError::Permanent(format!(
"newznab grab {base}: .nzb is {} bytes, over the {MAX_NZB_BYTES}-byte ceiling", "newznab grab {label}: .nzb is {} bytes, over the {MAX_NZB_BYTES}-byte ceiling",
body.len() body.len()
))); )));
} }
if looks_like_html(&body) { if looks_like_html(&body) {
return Err(GatewayError::Permanent(format!( return Err(GatewayError::Permanent(format!(
"indexer {base} answered the .nzb grab with an HTML page — check the API base and key" "indexer {label} answered the .nzb grab with an HTML page — check the API base and key"
))); )));
} }
let text: Cow<str> = String::from_utf8_lossy(&body); let text: Cow<str> = String::from_utf8_lossy(&body);
@@ -202,6 +203,26 @@ pub async fn grab(
Ok(text.into_owned().into_bytes()) Ok(text.into_owned().into_bytes())
} }
/// The feeder error for a non-2xx Newznab answer.
pub fn status_error(resp: &reqwest::Response, label: &str) -> GatewayError {
let status = resp.status();
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 {label}: HTTP {status}"))
}
_ => GatewayError::Permanent(format!("newznab {label}: HTTP {status}")),
}
}
/// Drop the NZB `<head>` except `<meta type="password">`. /// Drop the NZB `<head>` except `<meta type="password">`.
/// ///
/// The grabbed file is seeded to the whole mesh, and indexers put /// The grabbed file is seeded to the whole mesh, and indexers put
@@ -0,0 +1,386 @@
//! Client for **our own** nntmux, over its Newznab HTTP API — the only way this
//! feeder reaches it.
//!
//! nntmux is a separate app (worph/AppStore `NNTmux`). Until 1.1 this feeder
//! read nntmux's MariaDB with SQL and gunzipped `<guid>.nzb.gz` straight off
//! its storage volume; that tied it to nntmux's schema, its DB password and its
//! shard layout, across two apps. Now it uses what any Newznab client uses:
//!
//! | call | route | cost (nntmux per-user counters) |
//! |---|---|---|
//! | [`NntmuxApi::search`] | `GET {base}/api/v1/api?t=search&q=…&extended=1` | 1 `apirequests` |
//! | [`NntmuxApi::details`] | `GET {base}/api/v1/api?t=details&id=<guid>` | 1 `apirequests` |
//! | [`NntmuxApi::nzb`] | `GET {base}/getnzb?id=<guid>.nzb&r=<key>` | 1 `downloadrequests` |
//! | [`NntmuxApi::caps`] | `GET {base}/api/v1/api?t=caps` | free, no key |
//!
//! ⚠ The API lives under **`/api/v1/api`** (Laravel's `api` prefix) — plain
//! `/api` is a 404. `t=get` is not used: in this nntmux it *redirects* to
//! `/getnzb` on the **public** `APP_URL` host, which a container on `pcs` should
//! not round-trip through; `/getnzb` on the internal host is the same handler.
//!
//! ⚠ nntmux throttles the whole `/api` group to 60 requests/minute and caps
//! each role per 24 h (`apirequests`, `downloadrequests`: Admin 1000, User 10).
//! The key should belong to a dedicated user whose role has raised caps (the
//! NNTmux app provisions `metamesh`); [`Pacer`] keeps us under the per-minute
//! throttle, and the per-guid cache ([`super::cache`]) means each release's
//! `.nzb` is downloaded once, ever.
use std::collections::VecDeque;
use std::time::{Duration, Instant};
use meta_feeder_sdk::types::GatewayError;
use tokio::sync::Mutex;
use super::catalog::Release;
use crate::newznab::{classify, fetch_nzb, looks_like_html, newznab_error, status_error};
/// Default base URL: the NNTmux app's web container on the shared `pcs` network.
pub const DEFAULT_NNTMUX_URL: &str = "http://nntmux";
/// Newznab's own page-size ceiling (`<limits max="100">`).
pub const MAX_PAGE: usize = 100;
/// Requests allowed per [`PACE_WINDOW`] — under nntmux's 60/min throttle, with
/// headroom for a human browsing nntmux with the same key.
const PACE_BUDGET: usize = 50;
const PACE_WINDOW: Duration = Duration::from_secs(60);
/// Sliding-window limiter: at most [`PACE_BUDGET`] calls per [`PACE_WINDOW`].
/// A cold search (one `t=search` + one `/getnzb` per new hit) bursts freely up
/// to the budget, then waits instead of earning an HTTP 429.
#[derive(Default)]
pub struct Pacer {
sent: Mutex<VecDeque<Instant>>,
}
impl Pacer {
pub async fn wait(&self) {
loop {
let mut sent = self.sent.lock().await;
let now = Instant::now();
while sent.front().is_some_and(|t| now.duration_since(*t) >= PACE_WINDOW) {
sent.pop_front();
}
if sent.len() < PACE_BUDGET {
sent.push_back(now);
return;
}
let wait = PACE_WINDOW - now.duration_since(*sent.front().unwrap());
drop(sent);
tokio::time::sleep(wait).await;
}
}
}
pub struct NntmuxApi {
http: reqwest::Client,
/// Scheme + authority (+ optional path prefix), no trailing slash.
base: String,
key: String,
pacer: Pacer,
}
impl NntmuxApi {
pub fn new(http: reqwest::Client, base: &str, key: &str) -> Self {
let base = base.trim().trim_end_matches('/');
let base = if base.starts_with("http://") || base.starts_with("https://") {
base.to_string()
} else {
format!("http://{base}")
};
Self {
http,
base,
key: key.trim().to_string(),
pacer: Pacer::default(),
}
}
pub fn base(&self) -> &str {
&self.base
}
fn api_url(&self) -> String {
format!("{}/api/v1/api", self.base)
}
/// Free-text search, newest first. `q` is passed as typed — nntmux tokenises
/// it (Manticore full-text when present, SQL otherwise).
pub async fn search(&self, q: &str, limit: usize) -> Result<Vec<Release>, GatewayError> {
let limit = limit.clamp(1, MAX_PAGE).to_string();
let xml = self
.get_xml(&[("t", "search"), ("q", q), ("limit", &limit), ("extended", "1")])
.await?;
parse_items(&xml).map_err(|e| GatewayError::Transient(format!("nntmux search answer: {e}")))
}
/// One release's metadata, by guid (`None` = nntmux doesn't have it).
pub async fn details(&self, guid: &str) -> Result<Option<Release>, GatewayError> {
let xml = match self.get_xml(&[("t", "details"), ("id", guid), ("extended", "1")]).await {
Err(GatewayError::NotFound) => return Ok(None),
other => other?,
};
let items = parse_items(&xml)
.map_err(|e| GatewayError::Transient(format!("nntmux details answer: {e}")))?;
Ok(items.into_iter().find(|r| r.guid == guid))
}
/// The release's `.nzb` (plain XML). Spends one `downloadrequests` unit —
/// callers go through the cache.
pub async fn nzb(&self, guid: &str) -> Result<Vec<u8>, GatewayError> {
self.pacer.wait().await;
let url = format!("{}/getnzb?id={guid}.nzb&r={}", self.base, self.key);
fetch_nzb(&self.http, &url, &self.base).await
}
/// Reachability probe (`t=caps` needs no key and counts against nothing).
pub async fn caps(&self) -> Result<(), GatewayError> {
let url = format!("{}?t=caps", self.api_url());
let resp = self
.http
.get(&url)
.send()
.await
.map_err(|e| GatewayError::Transient(format!("nntmux {} unreachable: {}", self.base, e.without_url())))?;
if !resp.status().is_success() {
return Err(status_error(&resp, &self.base));
}
let body = resp
.text()
.await
.map_err(|e| GatewayError::Transient(format!("nntmux caps: {}", e.without_url())))?;
if body.contains("<caps") {
Ok(())
} else {
Err(GatewayError::Permanent(format!(
"{} did not answer t=caps like a Newznab server — is this the nntmux base URL?",
self.base
)))
}
}
/// GET the API with the key; Newznab errors (answered as HTTP 200) are mapped
/// through [`classify`]. The URL is never echoed — it carries the key.
async fn get_xml(&self, params: &[(&str, &str)]) -> Result<String, GatewayError> {
self.pacer.wait().await;
let mut query: Vec<(&str, &str)> = params.to_vec();
query.push(("apikey", &self.key));
let resp = self
.http
.get(self.api_url())
.query(&query)
.send()
.await
.map_err(|e| GatewayError::Transient(format!("nntmux {}: {}", self.base, e.without_url())))?;
if !resp.status().is_success() {
return Err(status_error(&resp, &self.base));
}
let body = resp
.bytes()
.await
.map_err(|e| GatewayError::Transient(format!("nntmux {}: {}", self.base, e.without_url())))?;
if looks_like_html(&body) {
return Err(GatewayError::Permanent(format!(
"{} answered the Newznab API with an HTML page — check the nntmux base URL",
self.base
)));
}
let text = String::from_utf8_lossy(&body).into_owned();
if let Some(err) = newznab_error(&text) {
return Err(classify(&err));
}
Ok(text)
}
}
/// Parse a Newznab RSS answer into releases. Items without a guid are skipped.
pub fn parse_items(xml: &str) -> anyhow::Result<Vec<Release>> {
use quick_xml::events::Event;
use quick_xml::Reader;
#[derive(Default)]
struct Item {
title: String,
guid: String,
size: u64,
enclosure_len: u64,
category: i32,
usenetdate: Option<i64>,
pubdate: Option<i64>,
}
let mut reader = Reader::from_str(xml);
reader.config_mut().trim_text(true);
let mut out = Vec::new();
let mut item: Option<Item> = None;
let mut field: Option<&'static str> = None;
loop {
match reader.read_event() {
Ok(Event::Start(e)) => match e.local_name().as_ref() {
b"item" => item = Some(Item::default()),
b"title" if item.is_some() => field = Some("title"),
b"guid" if item.is_some() => field = Some("guid"),
b"pubDate" if item.is_some() => field = Some("pubDate"),
_ => field = None,
},
Ok(Event::Empty(e)) => {
let Some(it) = item.as_mut() else { continue };
let name = e.local_name();
let attr = |k: &[u8]| -> Option<String> {
e.attributes()
.flatten()
.find(|a| a.key.local_name().as_ref() == k)
.and_then(|a| a.unescape_value().ok().map(|v| v.into_owned()))
};
match name.as_ref() {
b"attr" => {
let (Some(n), Some(v)) = (attr(b"name"), attr(b"value")) else { continue };
match n.as_str() {
// Several: the parent (2000) and the subcategory (2040).
"category" => {
if let Ok(c) = v.trim().parse::<i32>() {
it.category = it.category.max(c);
}
}
"size" => it.size = v.trim().parse().unwrap_or(0),
"usenetdate" => it.usenetdate = rfc2822(&v),
"guid" if it.guid.is_empty() => it.guid = v.trim().to_string(),
_ => {}
}
}
b"enclosure" => {
it.enclosure_len = attr(b"length").and_then(|l| l.trim().parse().ok()).unwrap_or(0);
}
_ => {}
}
}
Ok(Event::Text(t)) => {
let (Some(it), Some(f)) = (item.as_mut(), field) else { continue };
let v = t.unescape().map(|c| c.into_owned()).unwrap_or_default();
match f {
"title" => it.title = v,
"guid" => it.guid = guid_of(&v),
"pubDate" => it.pubdate = rfc2822(&v),
_ => {}
}
}
Ok(Event::End(e)) => {
if e.local_name().as_ref() == b"item" {
if let Some(it) = item.take() {
if !it.guid.is_empty() {
out.push(Release {
guid: it.guid,
search_name: it.title,
size: if it.size > 0 { it.size } else { it.enclosure_len },
category_id: it.category,
post_date: it.usenetdate.or(it.pubdate),
});
}
}
}
field = None;
}
Ok(Event::Eof) => break,
Err(e) => return Err(anyhow::anyhow!("newznab xml: {e}")),
_ => {}
}
}
Ok(out)
}
/// nntmux's `<guid>` is a details URL (`{server}/details/<guid>`); the release
/// guid is its last path segment. A bare guid passes through.
fn guid_of(raw: &str) -> String {
let t = raw.trim().trim_end_matches('/');
t.rsplit('/').next().unwrap_or(t).to_string()
}
fn rfc2822(s: &str) -> Option<i64> {
chrono::DateTime::parse_from_rfc2822(s.trim()).ok().map(|d| d.timestamp())
}
/// A release guid as nntmux mints it (hex/uuid): the only shape this feeder will
/// put into a URL or a cache path. Guards `get_blob`, whose id comes off the wire.
pub fn is_guid(s: &str) -> bool {
!s.is_empty() && s.len() <= 64 && s.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-')
}
#[cfg(test)]
mod tests {
use super::*;
const SEARCH: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom" xmlns:newznab="http://www.newznab.com/DTD/2010/feeds/attributes/">
<channel>
<title>NNTmux</title>
<newznab:response offset="0" total="2"/>
<item>
<title>Celebrity.Ghost.Stories.S05E08.720p.HDTV.x264-DHD</title>
<guid isPermaLink="true">http://localhost/details/97a2c0ffee1234</guid>
<link>http://localhost/getnzb?id=97a2c0ffee1234.nzb&amp;r=KEY</link>
<pubDate>Sat, 03 Oct 2026 12:00:00 +0000</pubDate>
<category>TV &gt; HD</category>
<enclosure url="http://localhost/getnzb?id=97a2c0ffee1234.nzb&amp;r=KEY" length="734003200" type="application/x-nzb"/>
<newznab:attr name="category" value="5000"/>
<newznab:attr name="category" value="5040"/>
<newznab:attr name="size" value="734003200"/>
<newznab:attr name="usenetdate" value="Fri, 02 Oct 2026 08:30:00 +0000"/>
</item>
<item>
<title>no-guid-item</title>
</item>
<item>
<title>Some.Movie.2024.1080p</title>
<guid>abcd-1234</guid>
<enclosure url="x" length="42" type="application/x-nzb"/>
<newznab:attr name="category" value="2040"/>
</item>
</channel>
</rss>"#;
#[test]
fn parses_items_guids_categories_sizes_and_dates() {
let r = parse_items(SEARCH).unwrap();
assert_eq!(r.len(), 2, "an item without a guid is skipped");
assert_eq!(r[0].guid, "97a2c0ffee1234");
assert_eq!(r[0].search_name, "Celebrity.Ghost.Stories.S05E08.720p.HDTV.x264-DHD");
assert_eq!(r[0].category_id, 5040, "the most specific category wins");
assert_eq!(r[0].size, 734_003_200);
assert_eq!(r[0].post_date, Some(1_790_929_800), "usenetdate beats pubDate");
assert_eq!(r[1].guid, "abcd-1234");
assert_eq!(r[1].size, 42, "enclosure length is the size fallback");
assert_eq!(r[1].post_date, None);
}
#[test]
fn an_empty_channel_is_no_releases() {
assert!(parse_items("<rss><channel></channel></rss>").unwrap().is_empty());
}
#[test]
fn guid_guard() {
assert!(is_guid("97a2c0ffee1234"));
assert!(is_guid("0d1b2c3d-aaaa-bbbb-cccc-000000000000"));
assert!(!is_guid("../etc/passwd"));
assert!(!is_guid("a/b"));
assert!(!is_guid(""));
}
#[test]
fn base_urls_are_normalised() {
let c = reqwest::Client::new();
assert_eq!(NntmuxApi::new(c.clone(), "nntmux", "k").base(), "http://nntmux");
assert_eq!(NntmuxApi::new(c, " http://nntmux:80/ ", "k").base(), "http://nntmux:80");
}
#[tokio::test]
async fn the_pacer_lets_a_burst_through() {
let p = Pacer::default();
let t = Instant::now();
for _ in 0..PACE_BUDGET {
p.wait().await;
}
assert!(t.elapsed() < Duration::from_secs(1));
}
}
@@ -0,0 +1,130 @@
//! Per-release `.nzb` cache, so each nntmux release is downloaded **once**.
//!
//! Every search hit must carry its portable `nzb-posting` cid (see
//! `usenet::record_for`), and that cid is a digest over the `.nzb`'s
//! Message-IDs — so a hit costs one `/getnzb` the first time it is seen. nntmux
//! counts each against the key's daily `downloadrequests`; without this cache a
//! popular term would re-download the same postings on every query. A collated
//! nntmux release never changes, so an entry never goes stale.
//!
//! Layout: `<feeder state>/nntmux-nzb/<guid>.nzb` (plain XML, written
//! atomically). The same bytes are what `get_blob` serves the gateway core for
//! seeding. An in-memory map remembers the minted cid so a cached hit is not
//! even re-parsed.
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Mutex;
use meta_feeder_sdk::hash::compute_nzb_posting_cid;
use super::api::is_guid;
use super::nzb::extract_message_ids;
/// Upper bound on the in-memory cid memo (the files themselves are unbounded).
const MEMO_MAX: usize = 50_000;
/// A release's minted identity.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Posting {
pub cid: String,
pub segment_count: usize,
}
pub struct NzbCache {
dir: PathBuf,
memo: Mutex<HashMap<String, Posting>>,
}
impl NzbCache {
pub fn new(dir: &Path) -> Self {
Self {
dir: dir.to_path_buf(),
memo: Mutex::new(HashMap::new()),
}
}
fn path(&self, guid: &str) -> Option<PathBuf> {
is_guid(guid).then(|| self.dir.join(format!("{guid}.nzb")))
}
/// The cached `.nzb`, if any.
pub fn get(&self, guid: &str) -> Option<Vec<u8>> {
std::fs::read(self.path(guid)?).ok()
}
/// Store a downloaded `.nzb` (tmp + rename, so a reader never sees half a file).
pub fn put(&self, guid: &str, xml: &[u8]) -> std::io::Result<()> {
let Some(path) = self.path(guid) else {
return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "not a release guid"));
};
std::fs::create_dir_all(&self.dir)?;
let tmp = path.with_extension("nzb.tmp");
std::fs::write(&tmp, xml)?;
std::fs::rename(&tmp, &path)
}
/// Mint (or recall) the posting identity for `guid` from its `.nzb`.
/// `None` when the document has no segments or doesn't parse.
pub fn posting(&self, guid: &str, xml: &[u8]) -> Option<Posting> {
if let Some(p) = self.memo.lock().unwrap().get(guid) {
return Some(p.clone());
}
let ids = extract_message_ids(xml).ok()?;
if ids.is_empty() {
return None;
}
let p = Posting {
cid: compute_nzb_posting_cid(&ids),
segment_count: ids.len(),
};
let mut memo = self.memo.lock().unwrap();
if memo.len() >= MEMO_MAX {
memo.clear();
}
memo.insert(guid.to_string(), p.clone());
Some(p)
}
/// The memoised identity only (no file read) — the search hot path.
pub fn known(&self, guid: &str) -> Option<Posting> {
self.memo.lock().unwrap().get(guid).cloned()
}
}
#[cfg(test)]
mod tests {
use super::*;
const NZB: &[u8] = br#"<?xml version="1.0"?><nzb><file><segments>
<segment bytes="1" number="1">a@b</segment><segment bytes="1" number="2">c@d</segment>
</segments></file></nzb>"#;
#[test]
fn put_get_and_mint() {
let dir = tempfile::tempdir().unwrap();
let c = NzbCache::new(&dir.path().join("nntmux-nzb"));
assert!(c.get("abc123").is_none());
c.put("abc123", NZB).unwrap();
assert_eq!(c.get("abc123").unwrap(), NZB);
let p = c.posting("abc123", NZB).unwrap();
assert_eq!(p.segment_count, 2);
assert_eq!(p.cid, compute_nzb_posting_cid(&["a@b".to_string(), "c@d".to_string()]));
assert_eq!(c.known("abc123"), Some(p));
}
#[test]
fn a_hostile_guid_never_becomes_a_path() {
let dir = tempfile::tempdir().unwrap();
let c = NzbCache::new(dir.path());
assert!(c.put("../escape", NZB).is_err());
assert!(c.get("../escape").is_none());
}
#[test]
fn a_segmentless_nzb_mints_nothing() {
let dir = tempfile::tempdir().unwrap();
let c = NzbCache::new(dir.path());
assert!(c.posting("g", b"<nzb></nzb>").is_none());
}
}
@@ -0,0 +1,77 @@
//! An nntmux release as the feeder sees it, and the Newznab category mapping.
//!
//! Filled from nntmux's Newznab search answer ([`super::api`]); nothing here
//! knows how it was fetched.
/// One collated nntmux release — the metadata half of a Usenet posting.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Release {
/// nntmux's public id (the last path segment of the item's `<guid>`). This
/// is our `record_id`, and the id `/getnzb` takes.
pub guid: String,
/// The collator's cleaned, scene-style name
/// (`Celebrity.Ghost.Stories.S05E08.720p.HDTV.x264-DHD`) — the item `<title>`.
pub search_name: String,
/// Total posted size in bytes (RAR + par2 included, so an over-estimate of
/// the final media file). `0` when the answer carried none.
pub size: u64,
/// nntmux category id — the most specific `newznab:attr name="category"`.
/// Coarse: 2000s = movies, 5000s = TV, 3000s = audio, 7000s = books.
pub category_id: i32,
/// Unix seconds of the original posting date (`usenetdate`), else the
/// item's `pubDate`, when either parses.
pub post_date: Option<i64>,
}
/// Map an nntmux/Newznab category id onto MetaMesh's `fileType`.
///
/// The Newznab tree is coarse and stable: 1000s console, 2000s movies, 3000s
/// audio, 4000s PC, 5000s TV, 6000s XXX, 7000s books. We only need the buckets
/// `METADATA_KEYS.md` §1 defines.
pub fn file_type_for_category(category_id: i32) -> &'static str {
match category_id / 1000 {
2 | 5 | 6 => "video",
3 => "audio",
7 => "document",
_ => "archive",
}
}
/// `contentKind` refinement of [`file_type_for_category`]. Only asserted where
/// the category tree is unambiguous — a wrong `contentKind` is worse than none,
/// because meta-watch's curated rows gate on it.
pub fn content_kind_for_category(category_id: i32) -> Option<&'static str> {
match category_id / 1000 {
2 => Some("movie"),
// ⚠ `episode`, not `tv`. `tv` is not a `contentKind` at all — it was a
// `domain` value, and it isn't even that any more (film + tv merged
// into `screen`, METADATA_KEYS.md §14.17). A kind outside the registry
// vocabulary resolves to no domain and no workForm, so the row is
// stamped with neither and is invisible to every client wall.
5 => Some("episode"),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn newznab_categories_map_to_file_types() {
assert_eq!(file_type_for_category(2040), "video"); // movies HD
assert_eq!(file_type_for_category(5040), "video"); // tv HD
assert_eq!(file_type_for_category(3010), "audio");
assert_eq!(file_type_for_category(7020), "document"); // ebook
assert_eq!(file_type_for_category(4000), "archive"); // pc
assert_eq!(file_type_for_category(10), "archive"); // nntmux "Other > Misc"
}
#[test]
fn content_kind_only_where_unambiguous() {
assert_eq!(content_kind_for_category(2040), Some("movie"));
assert_eq!(content_kind_for_category(5040), Some("episode"));
assert_eq!(content_kind_for_category(3010), None);
assert_eq!(content_kind_for_category(7020), None);
}
}
@@ -1,286 +0,0 @@
//! Read-only access to the nntmux sidecar's release catalog (MariaDB).
//!
//! We are a **consumer** of nntmux's schema, never a writer: the scanner owns
//! that database, and everything in it is re-derivable by re-scanning. Treat it
//! as persistent cache, not as a source of truth (the Tribler precedent — a
//! stateful sidecar the gateway still sees as a stateless HTTP feeder).
//!
//! Only a narrow, long-stable column set is selected (`guid`, `searchname`,
//! `name`, `size`, `postdate`, `categories_id`, `totalpart`). nntmux carries
//! ~80 columns on `releases`, most of them scanner bookkeeping; binding a wide
//! row here would make us fragile to upstream schema churn for no gain.
use anyhow::{Context, Result};
use sqlx::mysql::{MySqlPool, MySqlPoolOptions};
use sqlx::Row;
/// One collated nntmux release — the metadata half of a Usenet posting.
#[derive(Debug, Clone)]
pub struct Release {
/// nntmux's public id, and the basename of the on-disk `<guid>.nzb.gz`.
/// This is our `record_id`.
pub guid: String,
/// The collator's cleaned, scene-style name
/// (`Celebrity.Ghost.Stories.S05E08.720p.HDTV.x264-DHD`). Falls back to the
/// raw posted `name` when the collator could not improve on it.
pub search_name: String,
/// Total posted size in bytes (RAR + par2 included, so an over-estimate of
/// the final media file).
pub size: u64,
/// nntmux category id (`categories.id`). Coarse: 2000s = movies, 5000s = TV,
/// 3000s = audio, 7000s = books — the Newznab category tree.
pub category_id: i32,
/// Unix seconds of the original posting date, when nntmux recorded one.
pub post_date: Option<i64>,
}
/// Connection pool to nntmux's MariaDB. Small by design — this feeder issues a
/// handful of short SELECTs per query, and the scanner needs the headroom.
#[derive(Clone)]
pub struct Db {
pool: MySqlPool,
}
impl Db {
/// Open a pool. Does **not** verify the schema — a feeder must come up and
/// serve `/health` even when its upstream is unreachable, so the gateway's
/// `depends_on` is satisfied and discovery does not soft-skip us forever.
pub async fn connect(url: &str) -> Result<Self> {
let pool = MySqlPoolOptions::new()
.max_connections(4)
.acquire_timeout(std::time::Duration::from_secs(10))
.connect(url)
.await
.with_context(|| "connect to nntmux mariadb")?;
Ok(Self { pool })
}
/// Free-text search over the collated release names.
///
/// `LIKE %term%` on `searchname` — deliberately simple. nntmux can be built
/// with Manticore/Elasticsearch for real full-text search, but both are
/// optional in the minimum stack and we do not want to require them. If
/// search quality becomes the bottleneck, that is the upgrade path.
pub async fn search(&self, term: &str, limit: usize) -> Result<Vec<Release>> {
// ⚠ TOKENISE, don't match the phrase. Scene releases are dot-separated
// (`Breaking.Bad.S01E01.1080p...`) while users type spaces ("breaking
// bad"), so a single `LIKE %breaking bad%` matches essentially NOTHING
// — the separator never lines up. Split into words and AND them, which
// makes the separator irrelevant on both sides.
//
// Found by end-to-end search from meta-watch; the unit tests missed it
// because they only exercised the escaping, never a realistic
// scene-style name against a natural-language query.
let words = search_tokens(term);
if words.is_empty() {
return Ok(vec![]);
}
let mut sql = String::from(
"SELECT guid, searchname, name, size, categories_id, \
UNIX_TIMESTAMP(postdate) AS postdate_unix \
FROM releases WHERE ",
);
sql.push_str(
&words
.iter()
.map(|_| "searchname LIKE ?")
.collect::<Vec<_>>()
.join(" AND "),
);
sql.push_str(" ORDER BY postdate DESC LIMIT ?");
let mut q = sqlx::query(&sql);
for w in &words {
q = q.bind(format!("%{}%", escape_like(w)));
}
let rows = q
.bind(limit as i64)
.fetch_all(&self.pool)
.await
.context("query releases by name")?;
Ok(rows.iter().map(row_to_release).collect())
}
/// Look one release up by guid — the `compute_outcomes` path, where the
/// record_id came back to us from the gateway.
pub async fn by_guid(&self, guid: &str) -> Result<Option<Release>> {
let row = sqlx::query(
"SELECT guid, searchname, name, size, categories_id, \
UNIX_TIMESTAMP(postdate) AS postdate_unix \
FROM releases WHERE guid = ? LIMIT 1",
)
.bind(guid)
.fetch_optional(&self.pool)
.await
.context("query release by guid")?;
Ok(row.as_ref().map(row_to_release))
}
/// Cheap liveness probe for `health()` — confirms the catalog is reachable
/// *and* non-empty. An empty `releases` table is the visible symptom of both
/// nntmux silent-zero traps (an unset `PATH_TO_NZBS`, or groups left
/// `active=0`), so reporting it as degraded turns a silent failure into a
/// legible one.
pub async fn release_count(&self) -> Result<i64> {
let row = sqlx::query("SELECT COUNT(*) AS n FROM releases")
.fetch_one(&self.pool)
.await
.context("count releases")?;
Ok(row.try_get::<i64, _>("n").unwrap_or(0))
}
}
fn row_to_release(row: &sqlx::mysql::MySqlRow) -> Release {
let search_name: String = row
.try_get::<String, _>("searchname")
.ok()
.filter(|s| !s.is_empty())
.or_else(|| row.try_get::<String, _>("name").ok())
.unwrap_or_default();
Release {
guid: row.try_get::<String, _>("guid").unwrap_or_default(),
search_name,
// nntmux stores size as an unsigned bigint; it arrives as u64 or
// (older schemas) a decimal string.
size: row
.try_get::<u64, _>("size")
.ok()
.or_else(|| {
row.try_get::<String, _>("size")
.ok()
.and_then(|s| s.parse().ok())
})
.unwrap_or(0),
category_id: row.try_get::<i32, _>("categories_id").unwrap_or(0),
post_date: row.try_get::<i64, _>("postdate_unix").ok(),
}
}
/// Split a user query into match tokens.
///
/// Anything that is not alphanumeric is a separator — which folds the scene
/// world's `.`/`_`/`-` and the human world's space onto the same footing, so
/// "breaking bad", "Breaking.Bad" and "breaking-bad" all produce the same
/// tokens and all match `Breaking.Bad.S01E01.1080p.BluRay.x264-SCENE`.
///
/// Single characters are dropped: they match nearly every row and cost a full
/// scan for no selectivity.
fn search_tokens(term: &str) -> Vec<String> {
term.split(|c: char| !c.is_alphanumeric())
.filter(|w| w.chars().count() > 1)
.map(|w| w.to_string())
.collect()
}
/// Escape the `LIKE` wildcards so a user searching for `100%` does not match
/// everything. Backslash is MySQL's default escape character.
fn escape_like(s: &str) -> String {
s.replace('\\', "\\\\")
.replace('%', "\\%")
.replace('_', "\\_")
}
/// Map an nntmux/Newznab category id onto MetaMesh's `fileType`.
///
/// The Newznab tree is coarse and stable: 1000s console, 2000s movies, 3000s
/// audio, 4000s PC, 5000s TV, 6000s XXX, 7000s books. We only need the buckets
/// `METADATA_KEYS.md` §1 defines.
pub fn file_type_for_category(category_id: i32) -> &'static str {
match category_id / 1000 {
2 | 5 | 6 => "video",
3 => "audio",
7 => "document",
_ => "archive",
}
}
/// `contentKind` refinement of [`file_type_for_category`]. Only asserted where
/// the category tree is unambiguous — a wrong `contentKind` is worse than none,
/// because meta-watch's curated rows gate on it.
pub fn content_kind_for_category(category_id: i32) -> Option<&'static str> {
match category_id / 1000 {
2 => Some("movie"),
// ⚠ `episode`, not `tv`. `tv` is not a `contentKind` at all — it was a
// `domain` value, and it isn't even that any more (film + tv merged
// into `screen`, METADATA_KEYS.md §14.17). A kind outside the registry
// vocabulary resolves to no domain and no workForm, so the row is
// stamped with neither and is invisible to every client wall.
5 => Some("episode"),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::*;
/// ⚠ THE DOT-VS-SPACE REGRESSION. Scene releases are dot-separated and
/// users type spaces; matching the raw phrase finds nothing. Caught by an
/// end-to-end search from meta-watch that returned 0 results against a
/// catalog that plainly contained the show.
#[test]
fn a_spaced_query_tokenises_to_match_a_dotted_scene_name() {
let tokens = search_tokens("breaking bad");
assert_eq!(tokens, vec!["breaking", "bad"]);
// Every token must be a substring of the real scene name — that is
// exactly what the AND-ed LIKEs check.
let scene = "Breaking.Bad.S01E01.1080p.BluRay.x264-SCENE".to_lowercase();
assert!(
tokens.iter().all(|t| scene.contains(&t.to_lowercase())),
"tokens {tokens:?} must all appear in {scene}"
);
// The naive whole-phrase form does NOT — this is the bug, pinned.
assert!(
!scene.contains("breaking bad"),
"if this ever passes, the dotted-name assumption changed"
);
}
/// Separators are interchangeable in both directions.
#[test]
fn scene_and_human_separators_produce_the_same_tokens() {
let expect = vec!["breaking".to_string(), "bad".to_string()];
assert_eq!(search_tokens("breaking bad"), expect);
assert_eq!(search_tokens("Breaking.Bad").iter().map(|s| s.to_lowercase()).collect::<Vec<_>>(), expect);
assert_eq!(search_tokens("breaking-bad"), expect);
assert_eq!(search_tokens("breaking_bad"), expect);
assert_eq!(search_tokens(" breaking bad "), expect);
}
/// Single characters are dropped (no selectivity, full scan) and an
/// all-noise query yields nothing rather than matching the whole catalog.
#[test]
fn noise_only_queries_yield_no_tokens() {
assert!(search_tokens("").is_empty());
assert!(search_tokens(" ").is_empty());
assert!(search_tokens("- . _").is_empty());
assert_eq!(search_tokens("a bc"), vec!["bc"]);
}
#[test]
fn like_wildcards_are_escaped() {
assert_eq!(escape_like("100%"), "100\\%");
assert_eq!(escape_like("a_b"), "a\\_b");
assert_eq!(escape_like("c:\\d"), "c:\\\\d");
assert_eq!(escape_like("naruto"), "naruto");
}
#[test]
fn newznab_categories_map_to_file_types() {
assert_eq!(file_type_for_category(2040), "video"); // movies HD
assert_eq!(file_type_for_category(5040), "video"); // tv HD
assert_eq!(file_type_for_category(3010), "audio");
assert_eq!(file_type_for_category(7020), "document"); // ebook
assert_eq!(file_type_for_category(4000), "archive"); // pc
}
#[test]
fn content_kind_only_where_unambiguous() {
assert_eq!(content_kind_for_category(2040), Some("movie"));
assert_eq!(content_kind_for_category(5040), Some("episode"));
assert_eq!(content_kind_for_category(3010), None);
assert_eq!(content_kind_for_category(7020), None);
}
}
+9 -169
View File
@@ -1,106 +1,16 @@
//! Reading an nntmux release's manifest off disk and extracting its ordered //! Extracting an NZB's ordered article Message-IDs — the input of the portable
//! article Message-IDs. //! `nzb-posting` (`0x1003`) cid.
//! //!
//! # Why the file and not the database //! # Why the `.nzb` and not nntmux's metadata
//! //!
//! You **cannot** read a finished release's Message-IDs out of nntmux's //! You **cannot** read a finished release's Message-IDs out of nntmux's
//! database. `NzbService::writeNzbForReleaseId()` writes the release's NZB and //! database: `NzbService::writeNzbForReleaseId()` writes the release's NZB and
//! then *unconditionally* purges collections → binaries → parts (all FKs are //! then purges collections → binaries → parts. For any *collated* release the
//! `ON DELETE CASCADE`, and there is no retention setting). Those tables are //! ordered Message-IDs survive **only** in its `.nzb` — which this feeder now
//! transient staging for the header scan; for any *collated* release the //! downloads over nntmux's Newznab API (`/getnzb`, see [`super::api`]) and
//! ordered Message-IDs survive **only** in the on-disk gzip NZB at //! caches ([`super::cache`]), instead of gunzipping it off nntmux's storage.
//! `<nzb_root>/<split>/<guid>.nzb.gz`.
//!
//! This was confirmed on the live path during the spike (the offline
//! `import-nzbs` path never stages collections/binaries/parts at all, so it
//! cannot be used to check this). It is the single load-bearing fact behind the
//! whole feeder: lose it and you are back to grabbing `.nzb` files from an
//! external indexer, which is exactly what this design exists to avoid.
use std::io::Read;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use flate2::read::GzDecoder;
/// A parsed nntmux manifest: the raw NZB XML plus the Message-IDs in posting
/// order.
#[derive(Debug, Clone)]
pub struct Manifest {
/// The decompressed NZB XML. Handed to the gateway verbatim so it can seed
/// it as an ordinary sha2-256 cid — the consumer parses this exact document.
pub xml: Vec<u8>,
/// Every `<segment>` Message-ID across every `<file>`, in document order.
/// Bare form (no surrounding `<>`), matching how NZB stores them and how
/// meta-share's `NzbSegment.message_id` holds them.
pub message_ids: Vec<String>,
}
/// How deep we are willing to probe for a shard. nntmux caps its own loop at
/// 32, but every real deployment sits in the low single digits and a guid is a
/// 36-char uuid; 8 covers any sane `nzbsplitlevel` while keeping the probe below.
const MAX_SPLIT_LEVEL: usize = 8;
/// Build the path a release's `.nzb.gz` would occupy at an exact shard depth:
/// one directory per leading guid character, mirroring nntmux's
/// `NZB::buildNZBPath()` (`$nzbPath .= $releaseGuid[$i].'/'`).
pub fn nzb_path_at(nzb_root: &Path, guid: &str, levels: usize) -> Option<PathBuf> {
if guid.is_empty() {
return None;
}
let mut path = nzb_root.to_path_buf();
for c in guid.chars().take(levels) {
path.push(c.to_string());
}
Some(path.join(format!("{guid}.nzb.gz")))
}
/// Locate a release's `.nzb.gz` in nntmux's sharded store.
///
/// **The shard depth is an nntmux *setting* (`settings.nzbsplitlevel`), not a
/// constant.** The PHP code's own fallback is 1, but the maintainer image ships
/// a database seeded with **4** (`storage/nzb/9/7/a/2/97a2….nzb.gz`) — so a
/// hardcoded depth silently resolves to a path that does not exist, every
/// release is skipped as "no .nzb.gz on disk yet", and the feeder reports zero
/// hits while the scan is in fact working perfectly. An operator can also
/// change the setting *after* files are written, leaving a store with mixed
/// depths. So resolve by probing the candidate depths, deepest first.
///
/// Still not a glob: at most `MAX_SPLIT_LEVEL + 1` `stat` calls against exact
/// paths, no directory enumeration — cheap enough for the compute path.
pub fn nzb_path(nzb_root: &Path, guid: &str) -> Option<PathBuf> {
let deepest = guid.chars().count().min(MAX_SPLIT_LEVEL);
(0..=deepest)
.rev()
.filter_map(|levels| nzb_path_at(nzb_root, guid, levels))
.find(|path| path.exists())
}
/// Read + gunzip a release's `.nzb.gz` and pull out its Message-IDs.
///
/// Returns `Ok(None)` when the file is absent — a normal, non-fatal state: a
/// release row can exist before `createNZBs()` has written its manifest, and
/// the caller simply skips that record rather than failing the whole query.
pub fn load(nzb_root: &Path, guid: &str) -> Result<Option<Manifest>> {
let Some(path) = nzb_path(nzb_root, guid) else {
return Ok(None);
};
if !path.exists() {
return Ok(None);
}
let raw = std::fs::read(&path).with_context(|| format!("read nzb {}", path.display()))?;
let mut xml = Vec::new();
GzDecoder::new(&raw[..])
.read_to_end(&mut xml)
.with_context(|| format!("gunzip nzb {}", path.display()))?;
let message_ids = extract_message_ids(&xml)
.with_context(|| format!("parse nzb {}", path.display()))?;
if message_ids.is_empty() {
anyhow::bail!("nzb {} has no segments", path.display());
}
Ok(Some(Manifest { xml, message_ids }))
}
/// Pull every `<segment>` body out of an NZB document, in order. /// Pull every `<segment>` body out of an NZB document, in order.
/// ///
@@ -110,7 +20,7 @@ pub fn load(nzb_root: &Path, guid: &str) -> Result<Option<Manifest>> {
/// meta-share would happily play. Surrounding `<>` are stripped if a producer /// meta-share would happily play. Surrounding `<>` are stripped if a producer
/// included them — the mint normalises this too, but keeping the stored form /// included them — the mint normalises this too, but keeping the stored form
/// canonical means the two never disagree about what was hashed. /// canonical means the two never disagree about what was hashed.
pub(crate) fn extract_message_ids(xml: &[u8]) -> Result<Vec<String>> { pub fn extract_message_ids(xml: &[u8]) -> Result<Vec<String>> {
use quick_xml::events::Event; use quick_xml::events::Event;
use quick_xml::Reader; use quick_xml::Reader;
@@ -183,74 +93,4 @@ mod tests {
let ids = extract_message_ids(SAMPLE.as_bytes()).expect("parse"); let ids = extract_message_ids(SAMPLE.as_bytes()).expect("parse");
assert!(ids.iter().all(|i| !i.contains('<') && !i.contains('>'))); assert!(ids.iter().all(|i| !i.contains('<') && !i.contains('>')));
} }
/// One directory per leading guid character, exactly as nntmux's
/// `buildNZBPath()` writes them.
#[test]
fn nzb_path_at_shards_one_dir_per_leading_char() {
let p = nzb_path_at(Path::new("/s/nzb"), "abc123", 4).expect("path");
assert_eq!(p, Path::new("/s/nzb/a/b/c/1/abc123.nzb.gz"));
let p0 = nzb_path_at(Path::new("/s/nzb"), "abc123", 0).expect("path");
assert_eq!(p0, Path::new("/s/nzb/abc123.nzb.gz"));
}
/// The regression this exists for: the maintainer image seeds
/// `nzbsplitlevel = 4`, so a resolver hardcoded to depth 1 finds nothing and
/// every real release is silently skipped as "no .nzb.gz on disk yet".
#[test]
fn nzb_path_finds_a_level_4_shard() {
let dir = tempfile::tempdir().expect("tmp");
let guid = "97a2218c-d59b-4bc0-bb67-8a7d41e93fb5";
let deep = dir.path().join("9/7/a/2");
std::fs::create_dir_all(&deep).expect("mkdir");
std::fs::write(deep.join(format!("{guid}.nzb.gz")), b"x").expect("write");
let got = nzb_path(dir.path(), guid).expect("resolved");
assert_eq!(got, deep.join(format!("{guid}.nzb.gz")));
}
/// A store written at a shallower depth (or an operator who lowered
/// `nzbsplitlevel` after the fact) must still resolve.
#[test]
fn nzb_path_finds_a_level_1_shard() {
let dir = tempfile::tempdir().expect("tmp");
std::fs::create_dir_all(dir.path().join("a")).expect("mkdir");
std::fs::write(dir.path().join("a/abc123.nzb.gz"), b"x").expect("write");
let got = nzb_path(dir.path(), "abc123").expect("resolved");
assert_eq!(got, dir.path().join("a/abc123.nzb.gz"));
}
#[test]
fn nzb_path_is_none_when_absent_at_every_depth() {
let dir = tempfile::tempdir().expect("tmp");
assert!(nzb_path(dir.path(), "abc123").is_none());
}
/// A release row can exist before `createNZBs()` has written the manifest.
/// That is normal — the caller skips the record, it is not an error.
#[test]
fn missing_file_is_none_not_error() {
let dir = tempfile::tempdir().expect("tmp");
let got = load(dir.path(), "deadbeef").expect("no error");
assert!(got.is_none());
}
#[test]
fn round_trips_a_gzipped_nzb() {
use flate2::write::GzEncoder;
use flate2::Compression;
use std::io::Write;
let dir = tempfile::tempdir().expect("tmp");
std::fs::create_dir_all(dir.path().join("a")).expect("mkdir");
let mut enc = GzEncoder::new(Vec::new(), Compression::default());
enc.write_all(SAMPLE.as_bytes()).expect("gz write");
let gz = enc.finish().expect("gz finish");
std::fs::write(dir.path().join("a/abc.nzb.gz"), gz).expect("write");
let m = load(dir.path(), "abc").expect("load").expect("present");
assert_eq!(m.message_ids.len(), 3);
assert_eq!(m.xml, SAMPLE.as_bytes());
}
} }
+184 -127
View File
@@ -6,10 +6,16 @@
//! `nzb-release` (`0x1005`) cid that *embeds that indexer's host*. Only a peer //! `nzb-release` (`0x1005`) cid that *embeds that indexer's host*. Only a peer
//! holding that indexer's credential can ever redeem it. //! holding that indexer's credential can ever redeem it.
//! //!
//! This feeder scans Usenet itself, so it already holds the article Message-IDs //! This feeder fronts a Usenet scan we run ourselves (the NNTmux app), so the
//! at collation time — no grab, no scrape. That lets it mint a **portable** //! `.nzb` of every release is ours to fetch, unmetered by any third party. That
//! `nzb-posting` (`0x1003`) cid: a digest over the Message-ID set, embedding no //! lets it mint a **portable** `nzb-posting` (`0x1003`) cid: a digest over the
//! host, redeemable by any peer with a plain NNTP provider. //! Message-ID set, embedding no host, redeemable by any peer with a plain NNTP
//! provider.
//!
//! nntmux is reached **only over its Newznab API** ([`crate::nntmux::api`]) with
//! a key on this feeder's config page — the same contract the torznab feeder
//! uses with Prowlarr. Each release's `.nzb` is downloaded once and cached
//! ([`crate::nntmux::cache`]).
//! //!
//! The two coexist. This ADDS a Usenet metadata source; it removes nothing. //! The two coexist. This ADDS a Usenet metadata source; it removes nothing.
//! //!
@@ -36,8 +42,9 @@
//! Works without nntmux configured. //! Works without nntmux configured.
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::path::{Path, PathBuf}; use std::path::Path;
use std::sync::Arc; use std::sync::{Arc, RwLock};
use std::time::Duration;
use async_trait::async_trait; use async_trait::async_trait;
use meta_feeder_sdk::common::build_http_client; use meta_feeder_sdk::common::build_http_client;
@@ -48,11 +55,12 @@ use meta_feeder_sdk::hash::{
use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim}; use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim};
use meta_feeder_sdk::query::GatewayQuery; use meta_feeder_sdk::query::GatewayQuery;
use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth}; use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth};
use tokio::sync::RwLock;
use tracing::{debug, info, warn}; use tracing::{debug, info, warn};
use crate::newznab::{self, IndexerCred}; use crate::newznab::{self, IndexerCred};
use crate::nntmux::db::{content_kind_for_category, file_type_for_category, Db, Release}; use crate::nntmux::api::{is_guid, NntmuxApi, DEFAULT_NNTMUX_URL};
use crate::nntmux::cache::{NzbCache, Posting};
use crate::nntmux::catalog::{content_kind_for_category, file_type_for_category, Release};
use crate::nntmux::nzb; use crate::nntmux::nzb;
pub const UPSTREAM_ID: &str = "usenet"; pub const UPSTREAM_ID: &str = "usenet";
@@ -68,16 +76,21 @@ 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. /// Wall-clock budget for one `.nzb` grab. The viewer is waiting on it.
const GRAB_TIMEOUT_SECS: u64 = 60; const GRAB_TIMEOUT_SECS: u64 = 60;
/// How often the nntmux reachability probe (`t=caps`, free) runs.
const PROBE_INTERVAL: Duration = Duration::from_secs(300);
/// Operator-supplied settings, persisted to `<cache_dir>/config.json` by the /// 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 /// SDK's config surface. Env is a **seed only** — the file wins on the next
/// restart (invariant 12; there is no hot reload). /// restart (invariant 12; there is no hot reload).
#[derive(Debug, Clone, Default)] #[derive(Debug, Clone, Default)]
pub struct Settings { pub struct Settings {
/// `mysql://user:pass@host:3306/nntmux` /// Base URL of our nntmux (the NNTmux app's web container), e.g.
pub db_url: String, /// `http://nntmux`. The API path (`/api/v1/api`) is appended.
/// Root of nntmux's NZB store, shared with the sidecar as a volume. The pub nntmux_url: String,
/// directory nntmux's `PATH_TO_NZBS` points at. /// API key of the nntmux user this feeder searches as (the NNTmux app
pub nzb_root: String, /// provisions a `metamesh` user with raised daily caps). Blank → the
/// self-scan search is off; indexer-key redeems still work.
pub nntmux_api_key: String,
/// Per-host Newznab keys for redeeming `nzb-release` locators. Config page /// Per-host Newznab keys for redeeming `nzb-release` locators. Config page
/// only — no env seed. /// only — no env seed.
pub indexers: Vec<IndexerCred>, pub indexers: Vec<IndexerCred>,
@@ -85,10 +98,10 @@ pub struct Settings {
impl Settings { impl Settings {
fn from_env() -> Self { fn from_env() -> Self {
let env = |k: &str| std::env::var(k).ok().map(|v| v.trim().to_string()).filter(|v| !v.is_empty());
Self { Self {
db_url: std::env::var("NNTMUX_DB_URL").unwrap_or_default(), nntmux_url: env("NNTMUX_URL").unwrap_or_else(|| DEFAULT_NNTMUX_URL.to_string()),
nzb_root: std::env::var("NNTMUX_NZB_PATH") nntmux_api_key: env("NNTMUX_API_KEY").unwrap_or_default(),
.unwrap_or_else(|_| "/nntmux-nzb".to_string()),
indexers: Vec::new(), indexers: Vec::new(),
} }
} }
@@ -102,15 +115,20 @@ impl Settings {
warn!(target: "meta-feeder", path = %path.display(), "config.json is not valid json; using env seed"); warn!(target: "meta-feeder", path = %path.display(), "config.json is not valid json; using env seed");
return; return;
}; };
if let Some(s) = v.get("db_url").and_then(|x| x.as_str()) { let text = |k: &str| v.get(k).and_then(|x| x.as_str()).map(str::trim).filter(|s| !s.is_empty());
if !s.is_empty() { if let Some(s) = text("nntmux_url") {
self.db_url = s.to_string(); self.nntmux_url = s.to_string();
}
} }
if let Some(s) = v.get("nzb_root").and_then(|x| x.as_str()) { if let Some(s) = text("nntmux_api_key") {
if !s.is_empty() { self.nntmux_api_key = s.to_string();
self.nzb_root = 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!(
target: "meta-feeder",
"usenet: config.json still carries the pre-1.1 `db_url` — nntmux is now reached over \
its Newznab API; set `nntmux_api_key` (and `nntmux_url`) on the config page"
);
} }
if let Some(rows) = v.get("indexers").and_then(|x| x.as_array()) { if let Some(rows) = v.get("indexers").and_then(|x| x.as_array()) {
self.indexers = rows self.indexers = rows
@@ -124,11 +142,15 @@ impl Settings {
pub struct UsenetPlugin { pub struct UsenetPlugin {
settings: Settings, settings: Settings,
/// `None` until a successful connect. A feeder with unreachable config /// `None` when no nntmux key is configured (grab-only deployment).
/// **must still serve `/health`** so the gateway's `depends_on` is satisfied nntmux: Option<Arc<NntmuxApi>>,
/// — it soft-skips its upstream instead of failing to boot (invariant 10). /// `.nzb` per release guid, under the plugin's state dir.
db: Arc<RwLock<Option<Db>>>, cache: Option<Arc<NzbCache>>,
nzb_root: PathBuf, /// Why nntmux is unusable right now (unreachable, key refused), for
/// `health()`. A feeder with unreachable config **must still serve
/// `/health`** — it soft-skips its upstream instead of failing to boot
/// (invariant 10).
nntmux_problem: Arc<RwLock<Option<String>>>,
/// Client for `.nzb` grabs (named User-Agent, follows redirects so an HTML /// Client for `.nzb` grabs (named User-Agent, follows redirects so an HTML
/// landing page is detected rather than surfacing as a bare 302). /// landing page is detected rather than surfacing as a bare 302).
http: reqwest::Client, http: reqwest::Client,
@@ -147,8 +169,9 @@ impl UsenetPlugin {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {
settings: Settings::default(), settings: Settings::default(),
db: Arc::new(RwLock::new(None)), nntmux: None,
nzb_root: PathBuf::new(), cache: None,
nntmux_problem: Arc::new(RwLock::new(None)),
http: build_http_client(GRAB_TIMEOUT_SECS, USER_AGENT, None), http: build_http_client(GRAB_TIMEOUT_SECS, USER_AGENT, None),
grab_scheme: "https", grab_scheme: "https",
} }
@@ -222,31 +245,45 @@ impl UsenetPlugin {
}]) }])
} }
async fn db(&self) -> Result<Db, GatewayError> { fn nntmux(&self) -> Result<(&NntmuxApi, &NzbCache), GatewayError> {
self.db match (&self.nntmux, &self.cache) {
.read() (Some(api), Some(cache)) => Ok((api.as_ref(), cache.as_ref())),
.await _ => Err(GatewayError::Permanent("no nntmux API key configured".into())),
.clone() }
.ok_or_else(|| GatewayError::Permanent("nntmux database not configured".into()))
} }
/// Mint a release's `nzb-posting` cid by reading its manifest off disk. fn note_problem(&self, e: &GatewayError) {
/// if let GatewayError::Permanent(msg) = e {
/// `None` when the `.nzb.gz` is absent (a release row can exist before *self.nntmux_problem.write().unwrap() = Some(msg.clone());
/// `createNZBs()` has written it) or unreadable — the caller then emits a
/// record without a cid, which is honest: we genuinely cannot address that
/// posting yet.
fn posting_cid(&self, guid: &str) -> Option<String> {
match nzb::load(&self.nzb_root, guid) {
Ok(Some(m)) => Some(compute_nzb_posting_cid(&m.message_ids)),
Ok(None) => None,
Err(e) => {
warn!(target: "meta-feeder", %guid, error = %e, "usenet: manifest unreadable; record will carry no cid");
None
}
} }
} }
/// A release's `.nzb`: from the cache, else downloaded from nntmux once
/// (one `downloadrequests` unit) and cached.
async fn manifest(&self, guid: &str) -> Result<Vec<u8>, GatewayError> {
let (api, cache) = self.nntmux()?;
if let Some(xml) = cache.get(guid) {
return Ok(xml);
}
let xml = api.nzb(guid).await.inspect_err(|e| self.note_problem(e))?;
if let Err(e) = cache.put(guid, &xml) {
warn!(target: "meta-feeder", %guid, error = %e, "usenet: could not cache the .nzb (will re-download)");
}
Ok(xml)
}
/// Mint a release's `nzb-posting` identity from its `.nzb`.
async fn posting(&self, guid: &str) -> Result<Posting, GatewayError> {
let (_, cache) = self.nntmux()?;
if let Some(p) = cache.known(guid) {
return Ok(p);
}
let xml = self.manifest(guid).await?;
cache
.posting(guid, &xml)
.ok_or_else(|| GatewayError::Permanent(format!("nzb of release {guid} has no segments")))
}
/// Build the discovery record for one release. Metadata only — the bytes /// 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. /// live on Usenet and are fetched by meta-share at playback, never by us.
/// ///
@@ -257,9 +294,9 @@ impl UsenetPlugin {
/// title renders but is unplayable. `nzb-release` (`0x1005`) avoids this /// title renders but is unplayable. `nzb-release` (`0x1005`) avoids this
/// for free because its cid is derivable from `{host, id}` alone; ours is a /// for free because its cid is derivable from `{host, id}` alone; ours is a
/// digest over the Message-ID set, so it has to be minted from the manifest /// digest over the Message-ID set, so it has to be minted from the manifest
/// — which is a **local file read**, no network and no indexer grab, and /// — one `/getnzb` on our own nntmux the first time a release is seen, then
/// therefore fine to do per search hit. Caught by an end-to-end play from /// the cache. Caught by an end-to-end play from meta-watch that landed on
/// meta-watch that landed on "Unavailable". /// "Unavailable".
fn record_for(&self, r: &Release, cid: Option<&str>) -> DiscoveryRecord { fn record_for(&self, r: &Release, cid: Option<&str>) -> DiscoveryRecord {
let mut fields: BTreeMap<String, String> = BTreeMap::new(); let mut fields: BTreeMap<String, String> = BTreeMap::new();
// The bare-cid key-set (`cids/<cid>` = "true") — the ONLY shape // The bare-cid key-set (`cids/<cid>` = "true") — the ONLY shape
@@ -335,40 +372,49 @@ impl FeederPlugin for UsenetPlugin {
fn configure(&mut self, cache_dir: &Path) -> Result<(), ConfigError> { fn configure(&mut self, cache_dir: &Path) -> Result<(), ConfigError> {
let mut settings = Settings::from_env(); let mut settings = Settings::from_env();
settings.merge_file(cache_dir); settings.merge_file(cache_dir);
self.nzb_root = PathBuf::from(&settings.nzb_root);
if settings.db_url.is_empty() { if settings.nntmux_api_key.is_empty() {
// Soft-skip, not a hard failure: serve /health, report Degraded. // Soft-skip, not a hard failure: serve /health, report Degraded.
warn!( warn!(
target: "meta-feeder", target: "meta-feeder",
"usenet: no nntmux database url configured — upstream will soft-skip \ "usenet: no nntmux API key configured — the self-scan search is off \
(set it in the feeder's config page, or seed NNTMUX_DB_URL)" (set it on the feeder's config page: the NNTmux app's `metamesh` user key)"
); );
self.settings = settings; self.settings = settings;
return Ok(()); return Ok(());
} }
// Connect eagerly so a bad url is visible at boot rather than on the let api = Arc::new(NntmuxApi::new(self.http.clone(), &settings.nntmux_url, &settings.nntmux_api_key));
// first query, but never fatally — the sidecar may still be starting. self.cache = Some(Arc::new(NzbCache::new(&cache_dir.join("nntmux-nzb"))));
let url = settings.db_url.clone(); info!(target: "meta-feeder", nntmux = api.base(), "usenet: searching nntmux over its Newznab API");
let slot = self.db.clone();
// Reachability, now and periodically — never fatal: nntmux may still be
// starting (it takes minutes on a large store).
let probe = Arc::clone(&api);
let problem = Arc::clone(&self.nntmux_problem);
tokio::spawn(async move { tokio::spawn(async move {
match Db::connect(&url).await { loop {
Ok(db) => { match probe.caps().await {
match db.release_count().await { Ok(()) => {
Ok(0) => warn!( let mut p = problem.write().unwrap();
target: "meta-feeder", // A refused key is learned from real calls; only clear
"usenet: nntmux catalog is EMPTY — check the scan actually ran \ // what this probe can vouch for (reachability).
(PATH_TO_NZBS must be non-empty, and groups need active=1)" if p.as_deref().is_some_and(|m| m.contains("unreachable") || m.contains("t=caps")) {
), *p = None;
Ok(n) => info!(target: "meta-feeder", releases = n, "usenet: nntmux catalog connected"), }
Err(e) => warn!(target: "meta-feeder", error = %e, "usenet: catalog count failed"), }
Err(e) => {
warn!(target: "meta-feeder", nntmux = probe.base(), error = %e, "usenet: nntmux not reachable");
*problem.write().unwrap() = Some(match e {
GatewayError::Permanent(m) => m,
other => format!("nntmux {} unreachable: {other}", probe.base()),
});
} }
*slot.write().await = Some(db);
} }
Err(e) => warn!(target: "meta-feeder", error = %e, "usenet: nntmux connect failed; upstream soft-skips"), tokio::time::sleep(PROBE_INTERVAL).await;
} }
}); });
self.nntmux = Some(api);
self.settings = settings; self.settings = settings;
Ok(()) Ok(())
} }
@@ -385,39 +431,47 @@ impl FeederPlugin for UsenetPlugin {
if term.trim().is_empty() { if term.trim().is_empty() {
return Ok(vec![]); return Ok(vec![]);
} }
if self.settings.db_url.is_empty() { if self.nntmux.is_none() {
// Grab-only deployment (indexer keys, no nntmux): nothing to search. // Grab-only deployment (indexer keys, no nntmux): nothing to search.
return Ok(vec![]); return Ok(vec![]);
} }
let db = self.db().await?; let (api, _) = self.nntmux()?;
let releases = db let releases = api
.search(term.trim(), max_results) .search(term.trim(), max_results)
.await .await
.map_err(|e| GatewayError::Transient(format!("nntmux search: {e}")))?; .inspect_err(|e| self.note_problem(e))?;
*self.nntmux_problem.write().unwrap() = None;
// Mint each hit's cid here, from its on-disk manifest — see the // Mint each hit's cid here — see the `record_for` doc for why this
// `record_for` doc for why this cannot wait for compute. A release // cannot wait for compute. A release whose `.nzb` can't be had is
// whose manifest isn't written yet is DROPPED rather than surfaced // DROPPED rather than surfaced uncid'd: an unplayable row in the
// uncid'd: an unplayable row in the client's "raw sources" list is // client's "raw sources" list is worse than one fewer row. Once nntmux
// worse than one fewer row. // says a daily cap is spent, stop asking for this query.
let mut out = Vec::with_capacity(releases.len()); let mut out = Vec::with_capacity(releases.len());
let mut skipped = 0usize; let mut skipped = 0usize;
let mut capped = false;
for r in &releases { for r in &releases {
match self.posting_cid(&r.guid) { if capped {
Some(cid) => out.push(self.record_for(r, Some(&cid))), skipped += 1;
None => 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!( debug!(
target: "meta-feeder", %term, hits = out.len(), skipped, target: "meta-feeder", %term, hits = out.len(), skipped,
"usenet: catalog search" "usenet: nntmux search"
); );
if skipped > 0 {
debug!(
target: "meta-feeder", skipped,
"usenet: releases skipped — no .nzb.gz on disk yet (scan still collating?)"
);
}
Ok(out) Ok(out)
} }
@@ -427,20 +481,20 @@ impl FeederPlugin for UsenetPlugin {
if codec_of(record_id) == Some(NZB_RELEASE_CODEC) { if codec_of(record_id) == Some(NZB_RELEASE_CODEC) {
return self.redeem_release(record_id).await; return self.redeem_release(record_id).await;
} }
let db = self.db().await?; if !is_guid(record_id) {
let release = db return Err(GatewayError::NotFound);
.by_guid(record_id) }
let (api, _) = self.nntmux()?;
let release = api
.details(record_id)
.await .await
.map_err(|e| GatewayError::Transient(format!("nntmux lookup: {e}")))? .inspect_err(|e| self.note_problem(e))?
.ok_or(GatewayError::NotFound)?; .ok_or(GatewayError::NotFound)?;
// The Message-IDs survive only in the on-disk `.nzb.gz` — nntmux purges // The Message-IDs survive only in the release's `.nzb` — nntmux purges
// collections/binaries/parts once it has written the file. // collections/binaries/parts once it has written it.
let manifest = nzb::load(&self.nzb_root, record_id) let posting = self.posting(record_id).await?;
.map_err(|e| GatewayError::Permanent(format!("nzb read: {e}")))? let cid = posting.cid;
.ok_or(GatewayError::NotFound)?;
let cid = compute_nzb_posting_cid(&manifest.message_ids);
// Same cid the search path already published for this release — the // Same cid the search path already published for this release — the
// client is committing to a cid it saw in the results, so the two must // client is committing to a cid it saw in the results, so the two must
@@ -450,7 +504,7 @@ impl FeederPlugin for UsenetPlugin {
// it must be on the search path, not only here. // it must be on the search path, not only here.
record record
.fields .fields
.insert("segmentCount".to_string(), manifest.message_ids.len().to_string()); .insert("segmentCount".to_string(), posting.segment_count.to_string());
Ok(vec![HashOutcome { Ok(vec![HashOutcome {
hash: Hash(cid), hash: Hash(cid),
@@ -468,22 +522,29 @@ impl FeederPlugin for UsenetPlugin {
/// names the stored blob correctly). /// names the stored blob correctly).
async fn get_blob(&self, cid: &str) -> Option<Vec<u8>> { async fn get_blob(&self, cid: &str) -> Option<Vec<u8>> {
let guid = cid.strip_suffix(".nzb").unwrap_or(cid); let guid = cid.strip_suffix(".nzb").unwrap_or(cid);
match nzb::load(&self.nzb_root, guid) { if !is_guid(guid) {
Ok(Some(m)) => Some(m.xml), return None;
Ok(None) => None, }
// Normally a cache hit: the search that published `manifest_url` has
// already downloaded it.
match self.manifest(guid).await {
Ok(xml) => Some(xml),
Err(e) => { Err(e) => {
warn!(target: "meta-feeder", %guid, error = %e, "usenet: blob read failed"); warn!(target: "meta-feeder", %guid, error = %e, "usenet: .nzb unavailable for seeding");
None None
} }
} }
} }
fn health(&self) -> PluginHealth { fn health(&self) -> PluginHealth {
if self.settings.db_url.is_empty() && self.settings.indexers.is_empty() { if self.nntmux.is_none() && self.settings.indexers.is_empty() {
return PluginHealth::Degraded { return PluginHealth::Degraded {
reason: "neither an nntmux database nor an indexer key is configured".into(), reason: "neither an nntmux API key nor an indexer key is configured".into(),
}; };
} }
if let Some(problem) = self.nntmux_problem.read().unwrap().clone() {
return PluginHealth::Degraded { reason: problem };
}
PluginHealth::Ok PluginHealth::Ok
} }
@@ -518,23 +579,19 @@ impl FeederPlugin for UsenetPlugin {
fn config_schema(&self) -> ConfigSchema { fn config_schema(&self) -> ConfigSchema {
ConfigSchema { ConfigSchema {
fields: vec![ fields: vec![
F::text("db_url", "nntmux database URL") F::text("nntmux_url", "nntmux URL")
.with_help( .with_help(
"MariaDB DSN of the nntmux sidecar's catalog, e.g. \ "Base URL of your NNTmux app as this feeder reaches it (default \
mysql://nntmux:secret@nntmux-db:3306/nntmux. Blank → the \ http://nntmux, the app's container on the shared network). The \
self-scan search is off (indexer-key redeems still work). Takes \ Newznab API path is added automatically. Takes effect on the next \
effect on the next feeder restart (no hot reload).", feeder restart.",
), ),
F::text("nzb_root", "NZB store path") F::secret("nntmux_api_key", "nntmux API key")
.with_help( .with_help(
"Path to nntmux's NZB directory as mounted INTO this container \ "API key of the nntmux user this feeder searches as — the NNTmux app's \
— it must be the same volume nntmux's PATH_TO_NZBS points at. \ `metamesh` user, whose role has raised daily caps (see the NNTmux app's \
Each release's ordered Message-IDs survive only in the on-disk \ install tips for how to read it). Blank → the self-scan search is off \
<guid>.nzb.gz, so a wrong path here means every release \ (indexer-key redeems still work).",
resolves to NotFound. Point it at the STORE ROOT, not a shard: \
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.",
), ),
F::record_array( F::record_array(
"indexers", "indexers",
@@ -560,8 +617,8 @@ impl FeederPlugin for UsenetPlugin {
fn config_values(&self) -> serde_json::Value { fn config_values(&self) -> serde_json::Value {
serde_json::json!({ serde_json::json!({
"db_url": self.settings.db_url, "nntmux_url": self.settings.nntmux_url,
"nzb_root": self.settings.nzb_root, "nntmux_api_key": self.settings.nntmux_api_key,
"indexers": self.settings.indexers, "indexers": self.settings.indexers,
}) })
} }
@@ -7,8 +7,7 @@
//! the soft-skip invariant every feeder must honour (gateway invariant 10: a //! the soft-skip invariant every feeder must honour (gateway invariant 10: a
//! feeder with insufficient config still serves `/health` so the gateway's //! feeder with insufficient config still serves `/health` so the gateway's
//! `depends_on` is satisfied, rather than failing to boot). Live search/mint //! `depends_on` is satisfied, rather than failing to boot). Live search/mint
//! are not driven here — they need a running nntmux sidecar with real data; //! are driven against a mock nntmux in `tests/nntmux_api.rs`.
//! that path is exercised against `docker-compose.feeder-usenet.yml`.
use std::net::SocketAddr; use std::net::SocketAddr;
@@ -17,7 +16,7 @@ use usenet_feeder::usenet::UsenetPlugin;
async fn boot_feeder() -> (SocketAddr, tempfile::TempDir) { async fn boot_feeder() -> (SocketAddr, tempfile::TempDir) {
let dir = tempfile::tempdir().expect("tempdir"); let dir = tempfile::tempdir().expect("tempdir");
// Deliberately no NNTMUX_DB_URL / NNTMUX_NZB_PATH — the "operator hasn't // Deliberately no NNTMUX_API_KEY — the "operator hasn't
// configured this feeder yet" boot state. `configure()` must not error. // configured this feeder yet" boot state. `configure()` must not error.
let plugin = UsenetPlugin::new(); let plugin = UsenetPlugin::new();
let configured = configure_plugins(vec![Box::new(plugin)], dir.path()).expect("configure"); let configured = configure_plugins(vec![Box::new(plugin)], dir.path()).expect("configure");
@@ -59,7 +58,7 @@ async fn manifest_advertises_usenet() {
/// server reporting `degraded` — never a crash, never a hang, never `ok` /// server reporting `degraded` — never a crash, never a hang, never `ok`
/// (which would lie about being able to serve real content). This is what /// (which would lie about being able to serve real content). This is what
/// lets the gateway's `depends_on: condition: service_healthy` succeed even /// lets the gateway's `depends_on: condition: service_healthy` succeed even
/// before an operator has filled in nntmux's db_url on the feeder's config /// before an operator has filled in the nntmux API key on the feeder's config
/// page. /// page.
#[tokio::test] #[tokio::test]
async fn health_degrades_gracefully_with_no_nntmux_configured() { async fn health_degrades_gracefully_with_no_nntmux_configured() {
@@ -87,7 +86,7 @@ async fn health_degrades_gracefully_with_no_nntmux_configured() {
); );
assert_eq!( assert_eq!(
usenet_health["health"]["reason"], usenet_health["health"]["reason"],
"neither an nntmux database nor an indexer key is configured", "neither an nntmux API key nor an indexer key is configured",
"got {usenet_health}" "got {usenet_health}"
); );
} }
@@ -0,0 +1,240 @@
//! The self-scan path against a mock nntmux, over its Newznab API.
//!
//! Boots the real `UsenetPlugin` in the SDK router with `nntmux_url` +
//! `nntmux_api_key` in its `config.json` (the config-page shape) and a wiremock
//! nntmux answering `/api/v1/api?t=…` and `/getnzb`, then drives `/query`,
//! `/blob` and `/health` exactly as the gateway does.
use meta_feeder_sdk::hash::compute_nzb_posting_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";
const CAPS: &str = r#"<?xml version="1.0" encoding="UTF-8"?><caps><limits max="100" default="100"/></caps>"#;
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="{}"/><newznab:attr name="category" value="{cat}"/>
<newznab:attr name="size" value="734003200"/>
<newznab:attr name="usenetdate" value="Fri, 02 Oct 2026 08:30:00 +0000"/></item>"#,
cat / 1000 * 1000
)
}
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><title>NNTmux</title>{}</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="Ghost [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>"#;
fn posting() -> String {
compute_nzb_posting_cid(&["[email protected]".to_string(), "[email protected]".to_string()])
}
async fn boot(nntmux: &MockServer) -> (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!({ "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 addr = listener.local_addr().expect("addr");
tokio::spawn(async move {
axum::serve(listener, app).await.expect("serve");
});
(format!("http://{addr}"), dir)
}
async fn caps_ok(server: &MockServer) {
Mock::given(method("GET"))
.and(path("/api/v1/api"))
.and(query_param("t", "caps"))
.respond_with(ResponseTemplate::new(200).set_body_string(CAPS))
.mount(server)
.await;
}
async fn search_answers(server: &MockServer, body: String) {
Mock::given(method("GET"))
.and(path("/api/v1/api"))
.and(query_param("t", "search"))
.and(query_param("apikey", KEY))
.respond_with(ResponseTemplate::new(200).set_body_string(body))
.mount(server)
.await;
}
async fn query(base: &str, text: &str) -> reqwest::Response {
reqwest::Client::new()
.post(format!("{base}/query"))
.json(&QueryRequest {
upstream_id: "usenet".into(),
query: GatewayQuery::from_free_text(text),
max_results: 10,
})
.send()
.await
.expect("POST /query")
}
async fn records(base: &str, text: &str) -> Vec<meta_feeder_sdk::DiscoveryRecord> {
let r = query(base, text).await;
assert!(r.status().is_success(), "query status {}", r.status());
r.json::<QueryResponse>().await.expect("query json").records
}
/// Search hits carry their portable cid, minted from an `.nzb` downloaded ONCE:
/// a repeat search and the core's `/blob` fetch are served from the cache.
#[tokio::test]
async fn hits_carry_the_posting_cid_and_each_nzb_is_downloaded_once() {
let server = MockServer::start().await;
caps_ok(&server).await;
search_answers(
&server,
rss(&[
item("abc111", "Celebrity.Ghost.Stories.S05E08.720p.HDTV.x264-DHD", 5040),
item("abc222", "Ghost.Not.Collated.Yet", 2040),
]),
)
.await;
Mock::given(method("GET"))
.and(path("/getnzb"))
.and(query_param("id", "abc111.nzb"))
.and(query_param("r", KEY))
.respond_with(ResponseTemplate::new(200).set_body_string(NZB))
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/getnzb"))
.and(query_param("id", "abc222.nzb"))
.respond_with(ResponseTemplate::new(404))
.mount(&server)
.await;
let (base, _dir) = boot(&server).await;
let recs = records(&base, "ghost").await;
assert_eq!(recs.len(), 1, "a release without a usable .nzb is dropped, got {recs:?}");
let r = &recs[0];
assert_eq!(r.record_id, "abc111");
assert_eq!(r.fields.get(&format!("cids/{}", posting())).map(String::as_str), Some("true"));
assert_eq!(r.fields["manifest_url"], "/blob/usenet/abc111.nzb");
assert_eq!(r.fields["contentKind"], "episode");
assert_eq!(r.fields["fileType"], "video");
assert_eq!(r.fields["sizeByte"], "734003200");
assert_eq!(r.fields["publishedAt"], "1790929800");
// Repeat search: no second download (the `.expect(1)` above is verified on drop).
assert_eq!(records(&base, "ghost").await.len(), 1);
// The core seeds the manifest from /blob — served from the cache.
let blob = reqwest::get(format!("{base}/blob/usenet/abc111.nzb")).await.expect("blob");
assert!(blob.status().is_success());
assert_eq!(blob.text().await.unwrap(), NZB);
}
/// nntmux's daily download cap (`<error code="501">`, answered as HTTP 200):
/// stop asking for the rest of this query instead of burning a request per hit.
#[tokio::test]
async fn a_spent_download_cap_stops_the_query_early() {
let server = MockServer::start().await;
caps_ok(&server).await;
search_answers(&server, rss(&[item("aaa1", "One", 2040), item("aaa2", "Two", 2040)])).await;
Mock::given(method("GET"))
.and(path("/getnzb"))
.respond_with(
ResponseTemplate::new(200)
.set_body_string(r#"<error code="501" description="Download limit reached"/>"#),
)
.expect(1)
.mount(&server)
.await;
let (base, _dir) = boot(&server).await;
assert!(records(&base, "one").await.is_empty());
}
/// A refused key (`<error code="100">`) fails the query and shows on /health.
#[tokio::test]
async fn a_refused_key_degrades_health() {
let server = MockServer::start().await;
caps_ok(&server).await;
search_answers(
&server,
r#"<error code="100" description="Incorrect user credentials (wrong API key)"/>"#.into(),
)
.await;
let (base, _dir) = boot(&server).await;
assert!(!query(&base, "ghost").await.status().is_success());
let health: serde_json::Value = reqwest::get(format!("{base}/health")).await.unwrap().json().await.unwrap();
let usenet = health["plugins"].as_array().unwrap().iter().find(|p| p["id"] == "usenet").unwrap();
assert_eq!(usenet["health"]["state"], "degraded", "{usenet}");
assert!(usenet["health"]["reason"].as_str().unwrap().contains("API key"), "{usenet}");
}
/// A blank query never reaches nntmux (the gateway's routing gate owns structure).
#[tokio::test]
async fn a_blank_query_is_answered_locally() {
let server = MockServer::start().await;
caps_ok(&server).await;
Mock::given(method("GET"))
.and(path("/api/v1/api"))
.and(query_param("t", "search"))
.respond_with(ResponseTemplate::new(200).set_body_string(rss(&[])))
.expect(0)
.mount(&server)
.await;
let (base, _dir) = boot(&server).await;
assert!(records(&base, " ").await.is_empty());
}
/// A configured, reachable nntmux reports healthy.
#[tokio::test]
async fn a_reachable_nntmux_is_healthy() {
let server = MockServer::start().await;
caps_ok(&server).await;
let (base, _dir) = boot(&server).await;
// Let the boot-time t=caps probe land.
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
let health: serde_json::Value = reqwest::get(format!("{base}/health")).await.unwrap().json().await.unwrap();
let usenet = health["plugins"].as_array().unwrap().iter().find(|p| p["id"] == "usenet").unwrap();
assert_eq!(usenet["health"]["state"], "ok", "{usenet}");
}
/// A path-traversal id never reaches the cache or nntmux.
#[tokio::test]
async fn a_hostile_blob_id_is_refused() {
let server = MockServer::start().await;
caps_ok(&server).await;
Mock::given(method("GET")).and(path("/getnzb")).respond_with(ResponseTemplate::new(200)).expect(0).mount(&server).await;
let (base, _dir) = boot(&server).await;
let r = reqwest::get(format!("{base}/blob/usenet/..%2F..%2Fconfig.json")).await.unwrap();
assert!(!r.status().is_success());
}