3 Commits
Author SHA1 Message Date
Worph 1c9d415a1e improvement 2026-09-13 14:30:54 +02:00
Worph 60c4e855a5 lift limit on cid 2026-09-09 16:58:10 +02:00
Worph 03aa977e58 rework domain field 2026-09-02 00:03:22 +02:00
7 changed files with 938 additions and 89 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
[package] [package]
name = "meta-feeder-sdk" name = "meta-feeder-sdk"
version = "0.1.0" version = "1.3.0"
edition = "2021" edition = "2021"
rust-version = "1.85" rust-version = "1.85"
license = "MIT" license = "MIT"
+40 -7
View File
@@ -17,7 +17,7 @@ bytes; it never joins the swarm.
```toml ```toml
[dependencies] [dependencies]
meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v0.1.0" } meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.3.0" }
``` ```
Not on crates.io yet — the contract is still moving, so consumers pin a tag. Not on crates.io yet — the contract is still moving, so consumers pin a tag.
@@ -47,8 +47,32 @@ Four required methods; everything else has a default:
| `compute_outcomes` | resolve a `record_id` to `HashOutcome`s | `bytes: None` for metadata-only sources | | `compute_outcomes` | resolve a `record_id` to `HashOutcome`s | `bytes: None` for metadata-only sources |
Defaults you can override: `handle_query_stream`, `handle_fetch`, `health`, Defaults you can override: `handle_query_stream`, `handle_fetch`, `health`,
`served_file_types`, `served_content_kinds`, `config_schema`, `config_values`, `served_file_types`, `served_content_kinds`, `package`, `redeems`,
`get_blob`. `config_schema`, `config_values`, `get_blob`.
### Redeeming locators (1.3.0)
A plugin can **redeem** a locator family — turn a locator cid into the bytes it
names, usually spending a metered quota (a Newznab `.nzb` grab, an OpenSubtitles
download):
1. Return a `RedeemClaim` from `redeems()` — `RedeemClaim::nzb_release(hosts)` or
`RedeemClaim::provider_file(sources)` — reflecting the *current* config (no
credential → no claim). Served per plugin in `GET /manifest` and, for polling,
at `GET /redeems` → `{"redeems": {"<upstream_id>": [claim…]}}`.
2. Name a stable `package()` — the `/files/plugin/<package>/` folder the gateway
stores the bytes under.
3. In `compute_outcomes`, accept the locator cid as `record_id` and return ONE
`Sha2_256` outcome: `hash = hash::compute_ipfs_cid(&bytes)`, `bytes`,
`file_extension`, and optionally a `record` whose fields the gateway merges
onto the locator record. Return `NotFound` when the cid is not yours (e.g. no
key for that host) so the gateway can try the next claimer; `RateLimited` when
the quota is spent.
The gateway calls this only on a real play, one claimer at a time, and writes
the bytes' cid onto the locator record under the claim's `field` (`manifest` /
`file`), so each file's quota is spent once. Decode helpers:
`hash::decode_nzb_release_cid`, `hash::decode_provider_file_cid`, `hash::codec_of`.
### The routing gate ### The routing gate
@@ -101,6 +125,9 @@ into the bitswap blockstore.
The last two carry an identity multihash. Anything that dispatches on the The last two carry an identity multihash. Anything that dispatches on the
multihash instead of the **codec** will silently mis-rank them. multihash instead of the **codec** will silently mis-rank them.
`hash` also encodes/decodes the redeemable locators: `nzb-release` (`0x1005`)
and `provider-file` (`0x100A`, `(source, id)` — `docs/cid-formats.md` §8).
## Layout ## Layout
| module | what | | module | what |
@@ -121,13 +148,19 @@ multihash instead of the **codec** will silently mis-rank them.
cargo build && cargo test cargo build && cargo test
``` ```
To work on the SDK from inside a feeder checkout without pushing a tag, add a To test a feeder against an SDK change **before its tag is pushed**: cargo will
gitignored `.cargo/config.toml` in the feeder: not resolve an unpushed tag, not even under a `[patch]` override ("failed to
find tag"). Build a scratch copy of the feeder (outside its git tree) with the
dependency swapped for a path, and run `cargo test` there:
```toml ```bash
paths = ["../meta-feeder-sdk"] mkdir -p /tmp/feeder && (cd meta-feeder-x && tar cf - --exclude=./target --exclude=./.git .) | (cd /tmp/feeder && tar xf -)
sed -i 's#meta-feeder-sdk = { git = .*#meta-feeder-sdk = { path = "/abs/path/to/meta-feeder-sdk" }#' /tmp/feeder/feeder-plugin/*/Cargo.toml
(cd /tmp/feeder && cargo test)
``` ```
Never commit the path form: each feeder repo is its own Docker build context.
## Context ## Context
Design rationale, the feeder roster, and the naming rules live in the meta-root: Design rationale, the feeder roster, and the naming rules live in the meta-root:
+334 -60
View File
@@ -1,23 +1,32 @@
//! `contentKind` → `domain` — the routing table from METADATA_KEYS.md §1. //! `contentKind` → `domain` / `workForm` — the routing tables from
//! METADATA_KEYS.md §1.
//! //!
//! `domain` answers **"which application is this record destined for?"**. It //! `domain` answers **"which application is this record destined for?"**. It
//! is the coarsest facet in the registry and the key every client filters its //! is the coarsest facet in the registry and the key every client filters its
//! catalogue on: //! catalogue on — **one value per app**, no alternation:
//! //!
//! ```text //! ```text
//! meta-watch → domain:film|tv //! meta-watch → domain:screen
//! meta-listen → domain:music //! meta-listen → domain:music
//! meta-read → domain:literature //! meta-read → domain:literature
//! ``` //! ```
//! //!
//! # Why it is stored and not derived on read //! `workForm` answers the *other* question the old `film`/`tv` split was
//! secretly encoding: **is this work complete in itself, or one instalment of
//! a run?** (`standalone` | `serial`). It is cross-domain — a comic issue is
//! `serial` in `literature` exactly as an episode is `serial` in `screen` —
//! which is what lets a client ask for it independently of the wall.
//!
//! # Why they are stored and not derived on read
//! //!
//! `contentKind`'s vocabulary is deliberately open — a new feeder may mint a //! `contentKind`'s vocabulary is deliberately open — a new feeder may mint a
//! kind and ship without a fleet-wide update. That only stays safe because the //! kind and ship without a fleet-wide update. That only stays safe because the
//! domain travels *with* the record: a peer that has never heard of //! domain travels *with* the record: a peer that has never heard of
//! `podcastEpisode` still routes it correctly, because it never has to //! `podcastEpisode` still routes it correctly, because it never has to
//! interpret the token. Deriving on read would silently drop every record //! interpret the token. Deriving on read would silently drop every record
//! whose kind the reader predates. //! whose kind the reader predates. The same argument applies to `workForm`,
//! plus a second one: the two kinds that most need it (`pack` and `card`) are
//! exactly the two the kind alone cannot decide.
//! //!
//! # ⚠ The binding is permanent and one-way //! # ⚠ The binding is permanent and one-way
//! //!
@@ -27,6 +36,10 @@
//! cannot heal — the wrong domain is already written into records across the //! cannot heal — the wrong domain is already written into records across the
//! mesh. If a kind genuinely has to move, mint a new kind. //! mesh. If a kind genuinely has to move, mint a new kind.
//! //!
//! The `film`+`tv` → `screen` merge (METADATA_KEYS.md §14.17) is the one
//! deliberate exception, taken once and early and paid for with a one-shot
//! meta-core corpus sweep. It is not a precedent.
//!
//! # Why `domain` and not one kind-key per medium //! # Why `domain` and not one kind-key per medium
//! //!
//! `record_matches` is AND-across-keys / OR-within-key, and a non-matching key //! `record_matches` is AND-across-keys / OR-within-key, and a non-matching key
@@ -34,7 +47,7 @@
//! `documentKind`) would force meta-listen to express "music video **or** //! `documentKind`) would force meta-listen to express "music video **or**
//! audio track" as an OR across two *different* keys — which the query //! audio track" as an OR across two *different* keys — which the query
//! language cannot represent at all. One domain key, OR-ed within itself, //! language cannot represent at all. One domain key, OR-ed within itself,
//! always can. //! always can. `workForm` is one key everywhere for the same reason.
/// The `contentKind` → `domain` table. A kind absent from this list has no /// The `contentKind` → `domain` table. A kind absent from this list has no
/// domain and is **not independently routable** — correct for sidecars /// domain and is **not independently routable** — correct for sidecars
@@ -42,14 +55,19 @@
/// else. /// else.
/// ///
/// ⚠ `pack` is deliberately absent. A season pack and an album release are the /// ⚠ `pack` is deliberately absent. A season pack and an album release are the
/// same structural thing, so the kind alone cannot tell `tv` from `music`; the /// same structural thing, so the kind alone cannot tell `screen` from `music`;
/// writer stamps `pack`'s domain from its own context (the same reason /// the writer stamps `pack`'s domain from its own context (the same reason
/// `derive_file_type` cannot have a second `pack` arm). /// `derive_file_type` cannot have a second `pack` arm).
///
/// ⚠ `podcast` / `podcastEpisode` are absent too: the `spoken` domain was
/// retired unused (METADATA_KEYS.md §14.17), and the PR that first *writes*
/// those kinds picks their domain.
const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
// film / tv // screen — everything meta-watch consumes, films and serials alike. The
("movie", "film"), // standalone/serial split lives in `workForm`, not here.
("series", "tv"), ("movie", "screen"),
("episode", "tv"), ("series", "screen"),
("episode", "screen"),
// music // music
("track", "music"), ("track", "music"),
("album", "music"), ("album", "music"),
@@ -57,9 +75,6 @@ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
("musicVideo", "music"), ("musicVideo", "music"),
("djMix", "music"), ("djMix", "music"),
("liveSet", "music"), ("liveSet", "music"),
// spoken
("podcast", "spoken"),
("podcastEpisode", "spoken"),
// literature — an audiobook's identity is the book (same author, same ISBN // literature — an audiobook's identity is the book (same author, same ISBN
// family, same series as the ebook), so meta-read groups the two editions // family, same series as the ebook), so meta-read groups the two editions
// on one page. The player does not decide the domain. // on one page. The player does not decide the domain.
@@ -72,6 +87,37 @@ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
("paper", "science"), ("paper", "science"),
]; ];
/// The `contentKind` → `workForm` table (METADATA_KEYS.md §1).
///
/// `standalone` = a work complete in itself. `serial` = an instalment of an
/// ongoing run, **or the container or headline for one** — which is why a
/// `series` card and an `episode` are both `serial`.
///
/// ⚠ `pack` is absent for the same reason it is absent above: a season pack is
/// `serial`, an album release is `standalone`, and only the writer knows which.
const WORK_FORM_BY_CONTENT_KIND: &[(&str, &str)] = &[
// screen
("movie", "standalone"),
("series", "serial"),
("episode", "serial"),
// music — an album is a closed work, not an ongoing run, so every rung of
// the music ladder is standalone.
("track", "standalone"),
("album", "standalone"),
("artist", "standalone"),
("musicVideo", "standalone"),
("djMix", "standalone"),
("liveSet", "standalone"),
// literature — the split that proves `workForm` is not a screen-only axis.
("book", "standalone"),
("audiobook", "standalone"),
("comic", "serial"),
("manga", "serial"),
("magazine", "serial"),
// science
("paper", "standalone"),
];
/// Resolve the domain a `contentKind` belongs to. /// Resolve the domain a `contentKind` belongs to.
/// ///
/// Returns `None` for `pack` (not derivable from the kind — the writer supplies /// Returns `None` for `pack` (not derivable from the kind — the writer supplies
@@ -79,13 +125,41 @@ const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
/// build has never heard of. A caller that already knows the domain from its /// build has never heard of. A caller that already knows the domain from its
/// own context must prefer that over this table. /// own context must prefer that over this table.
pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> { pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> {
let k = content_kind.trim(); lookup(DOMAIN_BY_CONTENT_KIND, content_kind)
DOMAIN_BY_CONTENT_KIND }
/// Resolve the work form a `contentKind` sits on. Same contract as
/// [`domain_for_content_kind`]: `None` means "not derivable", never a guess.
pub fn work_form_for_content_kind(content_kind: &str) -> Option<&'static str> {
lookup(WORK_FORM_BY_CONTENT_KIND, content_kind)
}
fn lookup(table: &'static [(&'static str, &'static str)], key: &str) -> Option<&'static str> {
let k = key.trim();
table
.iter() .iter()
.find(|(kind, _)| kind.eq_ignore_ascii_case(k)) .find(|(kind, _)| kind.eq_ignore_ascii_case(k))
.map(|(_, domain)| *domain) .map(|(_, value)| *value)
} }
/// Kinds whose `domain` the kind→domain table deliberately cannot decide, so
/// the **writer** supplies it (METADATA_KEYS.md §1). Today that is `pack`
/// alone: a season pack and an album release are the same structural thing, and
/// only the producer knows which — indexer-feeder's `bt.rs` reads the domain off
/// the per-file kind *before* overwriting it with `pack` precisely so the value
/// survives.
///
/// They must ride along in every `domain:` → `contentKind:` expansion. The
/// expansion is only ever a **pre-filter** — the real `domain:` term is enforced
/// afterwards by [`crate::query_eval::record_matches`] — so widening it is free,
/// while omitting these kinds would drop exactly the records whose `domain` is
/// the most authoritative one on the wire.
pub const WRITER_SUPPLIED_DOMAIN_KINDS: &[&str] = &["pack"];
/// The `workForm` twin of [`WRITER_SUPPLIED_DOMAIN_KINDS`], and for the same
/// reason: a season pack is `serial`, an album release is `standalone`.
pub const WRITER_SUPPLIED_WORK_FORM_KINDS: &[&str] = &["pack"];
/// The inverse: every `contentKind` that belongs to `domain`. /// The inverse: every `contentKind` that belongs to `domain`.
/// ///
/// Used at the gateway-routing boundary. A gateway declares the *content kinds* /// Used at the gateway-routing boundary. A gateway declares the *content kinds*
@@ -93,56 +167,134 @@ pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> {
/// matched by expanding it here rather than by adding a `served_domains` field /// matched by expanding it here rather than by adding a `served_domains` field
/// to the heartbeat — which would be a wire-protocol change requiring every /// to the heartbeat — which would be a wire-protocol change requiring every
/// feeder manifest and every peer to move at once. /// feeder manifest and every peer to move at once.
///
/// ⚠ Since the `film`+`tv` merge this returns **several** kinds for `screen`
/// (`movie`, `series`, `episode`) where the two old values were disjoint. A
/// caller that reads only the first element silently means "movie" — scan the
/// whole list, or narrow with a `workForm:` term first.
pub fn content_kinds_for_domain(domain: &str) -> Vec<&'static str> { pub fn content_kinds_for_domain(domain: &str) -> Vec<&'static str> {
let d = domain.trim(); inverse(DOMAIN_BY_CONTENT_KIND, domain)
DOMAIN_BY_CONTENT_KIND }
/// The inverse of the `workForm` table. Same routing-boundary role, same
/// scan-the-whole-list warning as [`content_kinds_for_domain`].
pub fn content_kinds_for_work_form(work_form: &str) -> Vec<&'static str> {
inverse(WORK_FORM_BY_CONTENT_KIND, work_form)
}
fn inverse(table: &'static [(&'static str, &'static str)], value: &str) -> Vec<&'static str> {
let v = value.trim();
table
.iter() .iter()
.filter(|(_, dom)| dom.eq_ignore_ascii_case(d)) .filter(|(_, val)| val.eq_ignore_ascii_case(v))
.map(|(kind, _)| *kind) .map(|(kind, _)| *kind)
.collect() .collect()
} }
/// Rewrite a query so a `domain:` filter can reach a plugin that has never /// Rewrite a query so a `domain:` / `workForm:` filter can reach a plugin that
/// heard of the key. /// has never heard of either key.
/// ///
/// Returns `None` when the query carries no `domain:` filter — the common /// Returns `None` when the query carries neither — the common path, and cheaper
/// path, and cheaper than cloning. /// than cloning.
/// ///
/// # Why translate instead of teaching every plugin /// # Why translate instead of teaching every plugin
/// ///
/// `record_matches` fails a record outright when a filter key is missing from /// `record_matches` fails a record outright when a filter key is missing from
/// its fields, and `domain` is missing from every record a not-yet-updated /// its fields, and `domain` is missing from every record a not-yet-updated
/// feeder emits. Nine of the fifteen feeders pin the SDK by git tag, so they /// feeder emits (`workForm` from every record written before it existed).
/// cannot pick up a new key on our schedule. Translating `domain:literature` /// Most feeders pin the SDK by git tag, so they cannot pick up a new key on our
/// into the `contentKind:` alternation it covers means the wire query a plugin /// schedule. Translating `domain:literature` into the `contentKind:`
/// sees never mentions `domain` at all, and old plugins keep working unchanged /// alternation it covers means the wire query a plugin sees never mentions
/// — while the dispatcher still enforces the real `domain:` filter afterwards, /// either key — old plugins keep working unchanged, while the dispatcher still
/// against records it has just stamped. /// enforces the real filters afterwards, against records it has just stamped.
///
/// # The two axes intersect
///
/// `domain:screen workForm:serial` covers `series` and `episode` — the
/// **intersection**, not the union. Widening to the union would hand a serial
/// query the whole film corpus as a pre-filter; the post-stamp check would then
/// throw it away, having already paid for it upstream.
/// ///
/// An explicit `contentKind:` filter already in the query is left alone: it is /// An explicit `contentKind:` filter already in the query is left alone: it is
/// the narrower of the two, and the `domain` term is still enforced /// the narrower of the three, and the `domain`/`workForm` terms are still
/// post-stamp, so the AND semantics hold either way. /// enforced post-stamp, so the AND semantics hold either way.
pub fn expand_domain_filter(query: &crate::query::GatewayQuery) -> Option<crate::query::GatewayQuery> { pub fn expand_routing_filters(
let domains = query.filters.get("domain")?; query: &crate::query::GatewayQuery,
) -> Option<crate::query::GatewayQuery> {
let domains = query.filters.get("domain");
let work_forms = query.filters.get("workForm");
if domains.is_none() && work_forms.is_none() {
return None;
}
let by_domain = domains.map(|vs| expand_values(vs, content_kinds_for_domain));
let by_work_form = work_forms.map(|vs| expand_values(vs, content_kinds_for_work_form));
let mut out = query.clone(); let mut out = query.clone();
out.filters.remove("domain"); out.filters.remove("domain");
if !out.filters.contains_key("contentKind") { out.filters.remove("workForm");
let kinds: Vec<String> = domains if out.filters.contains_key("contentKind") {
return Some(out);
}
let mut kinds: Vec<&'static str> = match (by_domain, by_work_form) {
(Some(d), Some(w)) => d.into_iter().filter(|k| w.contains(k)).collect(),
(Some(d), None) => d,
(None, Some(w)) => w,
(None, None) => unreachable!("guarded above"),
};
// Domain-ambiguous kinds come along whenever the expansion is non-empty
// — see [`WRITER_SUPPLIED_DOMAIN_KINDS`]. Skipped when nothing mapped,
// so an unknown value still expands to nothing and leaves the filter
// out entirely (below) rather than narrowing to `contentKind:pack`.
if !kinds.is_empty() {
for k in WRITER_SUPPLIED_DOMAIN_KINDS
.iter() .iter()
.flat_map(|d| content_kinds_for_domain(d)) .chain(WRITER_SUPPLIED_WORK_FORM_KINDS.iter())
.map(|k| k.to_string()) {
.collect(); if !kinds.contains(k) {
// An unknown domain expands to nothing. Inserting an empty kinds.push(k);
// alternation would drop every record at the plugin; leaving the }
// filter out lets the plugin answer and the post-stamp `domain:` }
}
// An unknown value — or two axes with no kind in common — expands to
// nothing. Inserting an empty alternation would drop every record at the
// plugin; leaving the filter out lets the plugin answer and the post-stamp
// check do the (correct, empty) filtering. // check do the (correct, empty) filtering.
if !kinds.is_empty() { if !kinds.is_empty() {
out.filters.insert("contentKind".to_string(), kinds); out.filters.insert(
} "contentKind".to_string(),
kinds.into_iter().map(|k| k.to_string()).collect(),
);
} }
Some(out) Some(out)
} }
/// Deprecated alias for [`expand_routing_filters`], kept so a caller pinned to
/// an older SDK keeps compiling. It expands `workForm:` too — the name is the
/// only thing that is stale.
#[deprecated(note = "renamed to expand_routing_filters; it now expands workForm: as well")]
pub fn expand_domain_filter(
query: &crate::query::GatewayQuery,
) -> Option<crate::query::GatewayQuery> {
expand_routing_filters(query)
}
fn expand_values(
values: &[String],
expand: fn(&str) -> Vec<&'static str>,
) -> Vec<&'static str> {
let mut out: Vec<&'static str> = Vec::new();
for v in values {
for k in expand(v) {
if !out.contains(&k) {
out.push(k);
}
}
}
out
}
/// Stamp `domain` on a field map that already carries a `contentKind`. /// Stamp `domain` on a field map that already carries a `contentKind`.
/// ///
/// The convention from METADATA_KEYS.md §1: **whoever writes `contentKind` /// The convention from METADATA_KEYS.md §1: **whoever writes `contentKind`
@@ -150,14 +302,29 @@ pub fn expand_domain_filter(query: &crate::query::GatewayQuery) -> Option<crate:
/// overwritten — a writer that knows better than the table (a `pack`, say) has /// overwritten — a writer that knows better than the table (a `pack`, say) has
/// already set it. /// already set it.
pub fn stamp_domain(fields: &mut std::collections::BTreeMap<String, String>) { pub fn stamp_domain(fields: &mut std::collections::BTreeMap<String, String>) {
if fields.contains_key("domain") { stamp(fields, "domain", domain_for_content_kind);
}
/// Stamp `workForm` the same way, from the same `contentKind`. Call it
/// wherever [`stamp_domain`] is called — the registry treats the two as one
/// breath.
pub fn stamp_work_form(fields: &mut std::collections::BTreeMap<String, String>) {
stamp(fields, "workForm", work_form_for_content_kind);
}
fn stamp(
fields: &mut std::collections::BTreeMap<String, String>,
key: &str,
derive: fn(&str) -> Option<&'static str>,
) {
if fields.contains_key(key) {
return; return;
} }
let Some(kind) = fields.get("contentKind").cloned() else { let Some(kind) = fields.get("contentKind").cloned() else {
return; return;
}; };
if let Some(domain) = domain_for_content_kind(&kind) { if let Some(value) = derive(&kind) {
fields.insert("domain".to_string(), domain.to_string()); fields.insert(key.to_string(), value.to_string());
} }
} }
@@ -168,8 +335,8 @@ mod tests {
#[test] #[test]
fn kinds_map_to_their_domain() { fn kinds_map_to_their_domain() {
assert_eq!(domain_for_content_kind("movie"), Some("film")); assert_eq!(domain_for_content_kind("movie"), Some("screen"));
assert_eq!(domain_for_content_kind("episode"), Some("tv")); assert_eq!(domain_for_content_kind("episode"), Some("screen"));
assert_eq!(domain_for_content_kind("track"), Some("music")); assert_eq!(domain_for_content_kind("track"), Some("music"));
// The case that proves medium and domain are different axes. // The case that proves medium and domain are different axes.
assert_eq!(domain_for_content_kind("musicVideo"), Some("music")); assert_eq!(domain_for_content_kind("musicVideo"), Some("music"));
@@ -178,13 +345,45 @@ mod tests {
} }
#[test] #[test]
fn pack_has_no_derivable_domain() { fn film_and_tv_are_one_domain_now() {
// The merge (METADATA_KEYS.md §14.17): meta-watch's wall is one term.
assert_eq!(
domain_for_content_kind("movie"),
domain_for_content_kind("series")
);
// ...and the split it used to carry lives here instead.
assert_eq!(work_form_for_content_kind("movie"), Some("standalone"));
assert_eq!(work_form_for_content_kind("series"), Some("serial"));
assert_eq!(work_form_for_content_kind("episode"), Some("serial"));
}
#[test]
fn work_form_is_not_a_screen_only_axis() {
// The point of the key: it means the same thing in every domain.
assert_eq!(work_form_for_content_kind("comic"), Some("serial"));
assert_eq!(work_form_for_content_kind("book"), Some("standalone"));
assert_eq!(work_form_for_content_kind("album"), Some("standalone"));
}
#[test]
fn the_spoken_domain_is_gone() {
// Retired unused. The kinds stay reserved with no domain until the PR
// that writes them picks one.
assert_eq!(domain_for_content_kind("podcast"), None);
assert_eq!(domain_for_content_kind("podcastEpisode"), None);
assert!(content_kinds_for_domain("spoken").is_empty());
}
#[test]
fn pack_has_no_derivable_domain_or_work_form() {
assert_eq!(domain_for_content_kind("pack"), None); assert_eq!(domain_for_content_kind("pack"), None);
assert_eq!(work_form_for_content_kind("pack"), None);
} }
#[test] #[test]
fn unknown_kind_is_none_not_a_guess() { fn unknown_kind_is_none_not_a_guess() {
assert_eq!(domain_for_content_kind("hologram"), None); assert_eq!(domain_for_content_kind("hologram"), None);
assert_eq!(work_form_for_content_kind("hologram"), None);
} }
#[test] #[test]
@@ -196,6 +395,18 @@ mod tests {
for kind in content_kinds_for_domain("literature") { for kind in content_kinds_for_domain("literature") {
assert_eq!(domain_for_content_kind(kind), Some("literature")); assert_eq!(domain_for_content_kind(kind), Some("literature"));
} }
for kind in content_kinds_for_work_form("serial") {
assert_eq!(work_form_for_content_kind(kind), Some("serial"));
}
}
#[test]
fn screen_expands_to_every_rung_not_just_the_first() {
// The regression the merge creates for any caller reading `.first()`.
let screen = content_kinds_for_domain("screen");
assert!(screen.contains(&"movie"));
assert!(screen.contains(&"series"));
assert!(screen.contains(&"episode"));
} }
fn q(filters: &[(&str, &[&str])]) -> crate::query::GatewayQuery { fn q(filters: &[(&str, &[&str])]) -> crate::query::GatewayQuery {
@@ -217,13 +428,13 @@ mod tests {
} }
#[test] #[test]
fn no_domain_filter_is_left_untouched() { fn no_routing_filter_is_left_untouched() {
assert!(expand_domain_filter(&q(&[("fileType", &["video"])])).is_none()); assert!(expand_routing_filters(&q(&[("fileType", &["video"])])).is_none());
} }
#[test] #[test]
fn domain_becomes_the_content_kinds_it_covers() { fn domain_becomes_the_content_kinds_it_covers() {
let out = expand_domain_filter(&q(&[("domain", &["literature"])])).unwrap(); let out = expand_routing_filters(&q(&[("domain", &["literature"])])).unwrap();
// The plugin never sees `domain:` — that is the whole point. // The plugin never sees `domain:` — that is the whole point.
assert!(!out.filters.contains_key("domain")); assert!(!out.filters.contains_key("domain"));
let kinds = out.filters.get("contentKind").unwrap(); let kinds = out.filters.get("contentKind").unwrap();
@@ -232,9 +443,35 @@ mod tests {
assert!(!kinds.contains(&"track".to_string())); assert!(!kinds.contains(&"track".to_string()));
} }
#[test]
fn work_form_alone_also_expands() {
let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
assert!(!out.filters.contains_key("workForm"));
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"episode".to_string()));
assert!(kinds.contains(&"comic".to_string()));
assert!(!kinds.contains(&"movie".to_string()));
}
#[test]
fn the_two_axes_intersect_they_do_not_union() {
let out = expand_routing_filters(&q(&[
("domain", &["screen"][..]),
("workForm", &["serial"][..]),
]))
.unwrap();
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"series".to_string()));
assert!(kinds.contains(&"episode".to_string()));
// The union would have dragged the whole film corpus in.
assert!(!kinds.contains(&"movie".to_string()));
// A `serial` comic is in `literature`, not on meta-watch's wall.
assert!(!kinds.contains(&"comic".to_string()));
}
#[test] #[test]
fn an_explicit_content_kind_filter_wins() { fn an_explicit_content_kind_filter_wins() {
let out = expand_domain_filter(&q(&[ let out = expand_routing_filters(&q(&[
("domain", &["music"][..]), ("domain", &["music"][..]),
("contentKind", &["track"][..]), ("contentKind", &["track"][..]),
])) ]))
@@ -246,10 +483,40 @@ mod tests {
} }
#[test] #[test]
fn unknown_domain_adds_no_alternation() { fn expansion_carries_the_writer_supplied_kinds() {
// `pack` has no domain in the table (a season pack and an album are the
// same structural thing), so the writer stamps one. Leaving it out of
// the alternation would make a `domain:screen` pre-filter drop the whole
// torrent-pack corpus, which `record_matches` would then never see.
let out = expand_routing_filters(&q(&[("domain", &["screen"])])).unwrap();
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"episode".to_string()));
assert!(kinds.contains(&"pack".to_string()));
// ...and it rides along on the `workForm` axis too, exactly once.
let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
let kinds = out.filters.get("contentKind").unwrap();
assert_eq!(kinds.iter().filter(|k| *k == "pack").count(), 1);
}
#[test]
fn unknown_value_adds_no_alternation() {
// An empty alternation would drop every record at the plugin; the // An empty alternation would drop every record at the plugin; the
// post-stamp `domain:` check does the (correct, empty) filtering. // post-stamp check does the (correct, empty) filtering.
let out = expand_domain_filter(&q(&[("domain", &["hologram"])])).unwrap(); let out = expand_routing_filters(&q(&[("domain", &["hologram"])])).unwrap();
// Not even `pack`: a lone `contentKind:pack` would be a NARROWING of a
// filter that should widen to nothing.
assert!(!out.filters.contains_key("contentKind"));
}
#[test]
fn two_axes_with_nothing_in_common_widen_rather_than_narrow() {
// No music kind is `serial`. The pre-filter drops out; the post-stamp
// check still enforces both terms.
let out = expand_routing_filters(&q(&[
("domain", &["music"][..]),
("workForm", &["serial"][..]),
]))
.unwrap();
assert!(!out.filters.contains_key("contentKind")); assert!(!out.filters.contains_key("contentKind"));
} }
@@ -258,18 +525,25 @@ mod tests {
let mut f: BTreeMap<String, String> = BTreeMap::new(); let mut f: BTreeMap<String, String> = BTreeMap::new();
f.insert("contentKind".into(), "episode".into()); f.insert("contentKind".into(), "episode".into());
stamp_domain(&mut f); stamp_domain(&mut f);
assert_eq!(f.get("domain").map(String::as_str), Some("tv")); stamp_work_form(&mut f);
assert_eq!(f.get("domain").map(String::as_str), Some("screen"));
assert_eq!(f.get("workForm").map(String::as_str), Some("serial"));
// A pack writer that already knows its context keeps it. // A pack writer that already knows its context keeps it.
let mut p: BTreeMap<String, String> = BTreeMap::new(); let mut p: BTreeMap<String, String> = BTreeMap::new();
p.insert("contentKind".into(), "pack".into()); p.insert("contentKind".into(), "pack".into());
p.insert("domain".into(), "music".into()); p.insert("domain".into(), "music".into());
p.insert("workForm".into(), "standalone".into());
stamp_domain(&mut p); stamp_domain(&mut p);
stamp_work_form(&mut p);
assert_eq!(p.get("domain").map(String::as_str), Some("music")); assert_eq!(p.get("domain").map(String::as_str), Some("music"));
assert_eq!(p.get("workForm").map(String::as_str), Some("standalone"));
// No contentKind → no domain invented. // No contentKind → nothing invented.
let mut n: BTreeMap<String, String> = BTreeMap::new(); let mut n: BTreeMap<String, String> = BTreeMap::new();
stamp_domain(&mut n); stamp_domain(&mut n);
stamp_work_form(&mut n);
assert!(n.get("domain").is_none()); assert!(n.get("domain").is_none());
assert!(n.get("workForm").is_none());
} }
} }
+290 -18
View File
@@ -206,6 +206,76 @@ pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option<Strin
Some(format!("b{}", base32_lower_no_padding(&wire))) Some(format!("b{}", base32_lower_no_padding(&wire)))
} }
/// Decode a self-describing `nzb-release` cid into `(api_base, id)` — the exact
/// inverse of [`compute_nzb_release_cid`], string-level (no `cid` crate).
///
/// `api_base` is scheme-less authority + optional path (`api.nzbgeek.info/api`);
/// match credentials on its authority (split at the first `/`), grab against
/// `https://{api_base}`. `id` is lowercase hex, re-encoded from the digest — so
/// an **odd-length** id comes back left-padded with `0` (the encoder pads it),
/// which Newznab `t=get` accepts as the same release.
///
/// `None` on a wrong codec, a non-identity multihash, or a malformed digest.
pub fn decode_nzb_release_cid(cid: &str) -> Option<(String, String)> {
let digest = identity_locator_digest(cid, NZB_RELEASE_CODEC)?;
let mut pos = 0usize;
let base_len = read_varint(&digest, &mut pos)? as usize;
let rest = digest.get(pos..)?;
if rest.len() <= base_len {
return None; // need the base AND at least one id byte
}
let (base, id) = rest.split_at(base_len);
let api_base = std::str::from_utf8(base).ok()?.to_string();
let id_hex: String = id.iter().map(|b| format!("{b:02x}")).collect();
Some((api_base, id_hex))
}
/// The multicodec of a base32 (`b…`) CIDv1, or `None` for anything else
/// (CIDv0 `Qm…`, other multibases, garbage). Dispatch on this — never on the
/// multihash, which is identity (`0x00`) for every locator family.
pub fn codec_of(cid: &str) -> Option<u64> {
let wire = base32_lower_decode(cid.strip_prefix('b')?)?;
let mut pos = 0usize;
if read_varint(&wire, &mut pos)? != 1 {
return None;
}
read_varint(&wire, &mut pos)
}
/// The digest of a well-formed identity-multihash CIDv1 carrying `codec`, or
/// `None`. Shared by the locator decoders.
fn identity_locator_digest(cid: &str, codec: u64) -> Option<Vec<u8>> {
let wire = base32_lower_decode(cid.strip_prefix('b')?)?;
let mut pos = 0usize;
if read_varint(&wire, &mut pos)? != 1
|| read_varint(&wire, &mut pos)? != codec
|| read_varint(&wire, &mut pos)? != 0x00
{
return None;
}
let len = read_varint(&wire, &mut pos)? as usize;
let digest = wire.get(pos..)?;
(digest.len() == len).then(|| digest.to_vec())
}
/// Unsigned LEB128 varint at `*pos`, advancing it.
fn read_varint(bytes: &[u8], pos: &mut usize) -> Option<u64> {
let mut value = 0u64;
let mut shift = 0u32;
loop {
let byte = *bytes.get(*pos)?;
*pos += 1;
value |= u64::from(byte & 0x7f) << shift;
if byte & 0x80 == 0 {
return Some(value);
}
shift += 7;
if shift >= 64 {
return None;
}
}
}
/// Hex-decode a Newznab release id. The guid parser only keeps hex runs, so the /// Hex-decode a Newznab release id. The guid parser only keeps hex runs, so the
/// id is all-hex by construction; odd-length ids are left-padded with a `0` /// id is all-hex by construction; odd-length ids are left-padded with a `0`
/// nibble. Returns `None` on any non-hex byte (defensive). /// nibble. Returns `None` on any non-hex byte (defensive).
@@ -248,8 +318,12 @@ fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
/// merge (a TMDB and a MyAnimeList locator on one meta-core record) safe. /// merge (a TMDB and a MyAnimeList locator on one meta-core record) safe.
pub const CARD_LOCATOR_CODEC: u64 = 0x1007; pub const CARD_LOCATOR_CODEC: u64 = 0x1007;
/// Digest byte ceiling for a card locator, matching meta-share's /// Digest byte ceiling for the locators that still carry one — `yt-video`
/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash. /// (via [`kinded_locator_cid`]) and `ext-play`. Matches meta-share's
/// `MAX_MULTIHASH_SIZE`, whose `CidGeneric<64>` rejects a longer multihash.
///
/// ⚠ **[`compute_card_cid`] is deliberately NOT subject to this** — see its
/// doc. The name is kept because both remaining users were written against it.
const CARD_LOCATOR_MAX_DIGEST: usize = 64; const CARD_LOCATOR_MAX_DIGEST: usize = 64;
/// Encode a self-describing **card locator** CID from a metadata source and /// Encode a self-describing **card locator** CID from a metadata source and
@@ -270,8 +344,28 @@ const CARD_LOCATOR_MAX_DIGEST: usize = 64;
/// operation driven by an explicit cross-source id mapping — never inferred /// operation driven by an explicit cross-source id mapping — never inferred
/// here (see `meta-gateway/docs/others/card-tier-search.md` §9). /// here (see `meta-gateway/docs/others/card-tier-search.md` §9).
/// ///
/// Returns `None` when `source` is empty or the digest would overflow the /// # No length ceiling, deliberately
/// multihash budget — the caller drops that card. ///
/// A CIDv1 digest length is a varint, so the wire format imposes no limit; the
/// 64-byte figure elsewhere in this file is Rust's fixed-size `CidGeneric<64>`,
/// which meta-share needs only because beetswap is
/// `Behaviour<const MAX_MULTIHASH_SIZE, B>`. A card is never bitswap content
/// (seeding is gated on sha2-256), so that ceiling does not apply — exactly the
/// reasoning `METADATA_KEYS.md` §2.1 already records for the `url` locator,
/// which ships URLs past it today. meta-core's Go `Decode` hand-parses varints
/// with no cap and reads these correctly.
///
/// This is load-bearing for URL-shaped ids: a card keyed on a work's own
/// address (`("weebcentral.com", "/series/…/some-long-slug")`) runs to ~88
/// bytes, and capping would silently drop the card rather than mint it.
///
/// ⚠ The one consumer that still cares is meta-share's `cid_rank`, which parses
/// through `MsCid::from_str`. A locator it cannot parse scores `RANK_UNKNOWN`
/// instead of `RANK_LOCATOR`; harmless for a card (its record carries one CID,
/// so it wins the canonical election at any rank) but pinned by a golden vector
/// in `cid-rank-vectors.json` so the two implementations cannot drift unseen.
///
/// Returns `None` only when `source` or `id` is empty.
pub fn compute_card_cid(source: &str, id: &str) -> Option<String> { pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
if source.is_empty() || id.is_empty() { if source.is_empty() || id.is_empty() {
return None; return None;
@@ -284,9 +378,6 @@ pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
write_pb_varint(source_b.len() as u64, &mut digest); write_pb_varint(source_b.len() as u64, &mut digest);
digest.extend_from_slice(source_b); digest.extend_from_slice(source_b);
digest.extend_from_slice(id_b); digest.extend_from_slice(id_b);
if digest.len() > CARD_LOCATOR_MAX_DIGEST {
return None;
}
// CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest] // CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest]
let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len()); let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
@@ -378,8 +469,39 @@ pub fn compute_ext_play_cid(url: &str) -> Option<String> {
Some(identity_locator_cid(EXT_PLAY_CODEC, digest)) Some(identity_locator_cid(EXT_PLAY_CODEC, digest))
} }
/// Custom multicodec for a fetchable **`provider-file`** locator —
/// `docs/cid-formats.md` §8. One codec for every provider whose bytes are
/// fetched the same way; the provider lives inside the digest:
/// `varint(len(source)) ‖ source ‖ id`. Framing byte-identical to `card`
/// (`0x1007`) — only the codec differs — but the byte path *redeems* it (through
/// the claiming feeder's `/compute`) instead of 404ing.
pub const PROVIDER_FILE_CODEC: u64 = 0x100A;
/// Encode a `provider-file` CID from a registered source token
/// (`opensubtitles`) and that source's id (`file:7061834`). `None` on an empty
/// part or a digest past 64 bytes — the caller drops that row rather than emit
/// an address no peer can parse.
pub fn compute_provider_file_cid(source: &str, id: &str) -> Option<String> {
kinded_locator_cid(PROVIDER_FILE_CODEC, source, id)
}
/// Inverse of [`compute_provider_file_cid`]: `(source, id)`. Matches the
/// **codec slot** — a `card` with identical framing does not decode here.
pub fn decode_provider_file_cid(cid: &str) -> Option<(String, String)> {
let digest = identity_locator_digest(cid, PROVIDER_FILE_CODEC)?;
let mut pos = 0usize;
let source_len = read_varint(&digest, &mut pos)? as usize;
let source = digest.get(pos..pos.checked_add(source_len)?)?;
let id = digest.get(pos + source_len..)?;
Some((
String::from_utf8(source.to_vec()).ok()?,
String::from_utf8(id.to_vec()).ok()?,
))
}
/// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used /// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used
/// by [`compute_card_cid`] and [`compute_yt_video_cid`]. /// by [`compute_card_cid`], [`compute_yt_video_cid`] and
/// [`compute_provider_file_cid`].
fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option<String> { fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option<String> {
if kind.is_empty() || id.is_empty() { if kind.is_empty() || id.is_empty() {
return None; return None;
@@ -1067,19 +1189,88 @@ mod tests {
assert_eq!(&digest[5..], b"tv:95479"); assert_eq!(&digest[5..], b"tv:95479");
} }
/// Empty inputs are rejected rather than encoded as an ambiguous locator, /// Empty inputs are rejected rather than encoded as an ambiguous locator.
/// and an id past the 64-byte multihash budget (meta-share's
/// `CidGeneric<64>`) returns `None` so the caller drops the card instead of
/// emitting a cid that peer would refuse to parse.
#[test] #[test]
fn card_cid_rejects_empty_and_oversized() { fn card_cid_rejects_empty_halves() {
assert!(compute_card_cid("", "tv:1").is_none()); assert!(compute_card_cid("", "tv:1").is_none());
assert!(compute_card_cid("tmdb", "").is_none()); assert!(compute_card_cid("tmdb", "").is_none());
let huge = "x".repeat(CARD_LOCATOR_MAX_DIGEST); }
assert!(compute_card_cid("tmdb", &huge).is_none());
// Exactly at the ceiling still encodes. /// ⚠ A card locator has **no length ceiling**, and this is the test that
let fits = "x".repeat(CARD_LOCATOR_MAX_DIGEST - 5); // varint(4) + "tmdb" /// says so on purpose — it used to assert the opposite.
assert!(compute_card_cid("tmdb", &fits).is_some()); ///
/// The 64-byte figure is meta-share's `CidGeneric<64>` bitswap alias, and a
/// card is never bitswap content. Capping here would silently drop exactly
/// the cards a URL-shaped identity exists to mint: a `weebcentral.com` work
/// path digests to ~88 bytes. Same reasoning `METADATA_KEYS.md` §2.1
/// already records for the `url` locator.
#[test]
fn card_cid_has_no_length_ceiling() {
let long_id = "/series/01J76XYGF1RKSPWA4TQ6GGSVFV/\
Ive-Been-Killing-Slimes-for-300-Years-The-Red-Dragon-Academy-for-Girls";
let cid = compute_card_cid("weebcentral.com", long_id).expect("no ceiling");
let (source, id) = split_card_digest(&cid);
assert!(
source.len() + id.len() > CARD_LOCATOR_MAX_DIGEST,
"fixture must actually exceed the old cap, got {}",
source.len() + id.len()
);
assert_eq!(source, "weebcentral.com");
assert_eq!(id, long_id);
}
/// The host-form identity: the source slot names the site that *owns* the
/// work, never the tool that reached it. Pinned as a literal because
/// `meta-read`'s `card_cid::encode` carries the mirrored assertion — the two
/// addressing one work differently is the failure the whole scheme cannot
/// survive.
#[test]
fn host_form_card_cid_golden_vector() {
let cid = compute_card_cid("mangadex.org", "/manga/b0b721ff-c388-4486-aa0f-c2b0bb321512")
.expect("card cid");
assert_eq!(
cid,
"bagdsaabybrwwc3thmfsgk6bon5zgol3nmfxgoyjpmiygenzsgftgmlldgm4dqljugq4dmllbmeygmlldgjrdaytcgmzdcnjrgi"
);
}
/// `varint(source_len) ‖ source ‖ id` back out of a card locator, so the
/// tests above can assert on the halves without a decoder in this crate.
fn split_card_digest(cid: &str) -> (String, String) {
let raw = base32_lower_decode(&cid[1..]);
let mut i = 0usize;
let mut rd = || {
let (mut n, mut sh) = (0u64, 0u32);
loop {
let b = raw[i];
i += 1;
n |= u64::from(b & 0x7f) << sh;
if b & 0x80 == 0 {
return n;
}
sh += 7;
}
};
assert_eq!(rd(), 1, "CIDv1");
assert_eq!(rd(), CARD_LOCATOR_CODEC);
assert_eq!(rd(), 0, "identity multihash");
let len = rd() as usize;
let digest = &raw[i..i + len];
let mut j = 0usize;
let (mut sl, mut sh) = (0usize, 0u32);
loop {
let b = digest[j];
j += 1;
sl |= usize::from(b & 0x7f) << sh;
if b & 0x80 == 0 {
break;
}
sh += 7;
}
(
String::from_utf8(digest[j..j + sl].to_vec()).unwrap(),
String::from_utf8(digest[j + sl..].to_vec()).unwrap(),
)
} }
// ---- nzb-posting (0x1003) — the dedup contract ----------------------- // ---- nzb-posting (0x1003) — the dedup contract -----------------------
@@ -1299,3 +1490,84 @@ mod roundtrip_tests {
assert_eq!(decode_url_cid(&card), None); assert_eq!(decode_url_cid(&card), None);
} }
} }
#[cfg(test)]
mod redeem_locator_tests {
use super::*;
/// `docs/cid-formats.md` §8.2 golden vector — moved here from
/// meta-feeder-opensubtitles' local `locator.rs`; the vector proves the
/// move changed no bytes.
#[test]
fn provider_file_matches_the_documented_vector() {
assert_eq!(
compute_provider_file_cid("opensubtitles", "file:7061834").as_deref(),
Some("bagfcaaa2bvxxazloon2we5djorwgk43gnfwgkorxga3dcobtgq")
);
}
#[test]
fn provider_file_round_trips_and_refuses_other_codecs() {
let cid = compute_provider_file_cid("opensubtitles", "file:12658754").unwrap();
assert_eq!(
decode_provider_file_cid(&cid),
Some(("opensubtitles".to_string(), "file:12658754".to_string()))
);
// Identical framing, different codec: a card must not decode as a file.
let card = compute_card_cid("tmdb", "tv:95479").unwrap();
assert_eq!(card, "bagdsaaanar2g2zdcor3duojvgq3ts");
assert_eq!(decode_provider_file_cid(&card), None);
assert_eq!(decode_provider_file_cid("not-a-cid"), None);
assert_eq!(decode_provider_file_cid(""), None);
}
#[test]
fn provider_file_digest_ceiling_is_64_bytes() {
// varint(13) + "opensubtitles" = 14 bytes, so the id may use 50.
assert!(compute_provider_file_cid("opensubtitles", &"x".repeat(50)).is_some());
assert!(compute_provider_file_cid("opensubtitles", &"x".repeat(51)).is_none());
}
#[test]
fn nzb_release_round_trips() {
let cid = compute_nzb_release_cid("api.nzbgeek.info/api", "0123abcdef").unwrap();
assert_eq!(
decode_nzb_release_cid(&cid),
Some(("api.nzbgeek.info/api".to_string(), "0123abcdef".to_string()))
);
// A host:port authority survives too.
let cid = compute_nzb_release_cid("127.0.0.1:8080", "ff00").unwrap();
assert_eq!(
decode_nzb_release_cid(&cid),
Some(("127.0.0.1:8080".to_string(), "ff00".to_string()))
);
}
/// The encoder left-pads an odd-length id, so the decode is the padded form.
#[test]
fn nzb_release_odd_length_id_comes_back_padded() {
let cid = compute_nzb_release_cid("api.nzb.life", "abc").unwrap();
assert_eq!(
decode_nzb_release_cid(&cid),
Some(("api.nzb.life".to_string(), "0abc".to_string()))
);
}
#[test]
fn nzb_release_refuses_other_codecs() {
let card = compute_card_cid("tmdb", "tv:95479").unwrap();
assert_eq!(decode_nzb_release_cid(&card), None);
assert_eq!(decode_nzb_release_cid("bafkreigarbage"), None);
}
#[test]
fn codec_of_reads_the_codec_slot() {
let nzb = compute_nzb_release_cid("api.nzb.life", "ab").unwrap();
assert_eq!(codec_of(&nzb), Some(NZB_RELEASE_CODEC));
let file = compute_provider_file_cid("opensubtitles", "file:1").unwrap();
assert_eq!(codec_of(&file), Some(PROVIDER_FILE_CODEC));
assert_eq!(codec_of(&compute_ipfs_cid(b"hello")), Some(0x55));
assert_eq!(codec_of("QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG"), None);
assert_eq!(codec_of(""), None);
}
}
+2
View File
@@ -45,10 +45,12 @@ pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher};
pub use meta_core::FeederStore; pub use meta_core::FeederStore;
pub use plugin::{ pub use plugin::{
upstream_id_field, ConfigError, FeederPlugin, HashKind, HashOutcome, PluginRegistry, upstream_id_field, ConfigError, FeederPlugin, HashKind, HashOutcome, PluginRegistry,
RedeemClaim,
}; };
pub use query::{GatewayQuery, GatewaySearchEvent, GatewayWireError, Negation, RangeFilter}; pub use query::{GatewayQuery, GatewaySearchEvent, GatewayWireError, Negation, RangeFilter};
pub use serve::{ pub use serve::{
configure_plugins, router, serve_feeders, ComputeRequest, ComputeResponse, HashKindDto, configure_plugins, router, serve_feeders, ComputeRequest, ComputeResponse, HashKindDto,
HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse, HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse,
RedeemsResponse,
}; };
pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth}; pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth};
+94
View File
@@ -17,6 +17,7 @@ use std::path::Path;
use async_trait::async_trait; use async_trait::async_trait;
use futures::stream::BoxStream; use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
// Re-exported so plugins can `use meta_feeder_sdk::plugin::GatewayQuery` (the // Re-exported so plugins can `use meta_feeder_sdk::plugin::GatewayQuery` (the
// gateway's original `plugin.rs` re-exported it the same way). // gateway's original `plugin.rs` re-exported it the same way).
@@ -95,6 +96,66 @@ pub enum HashKind {
YtVideo, YtVideo,
} }
/// `RedeemClaim::codec` for a Newznab release locator (`0x1005`).
pub const REDEEM_CODEC_NZB_RELEASE: &str = "nzb-release";
/// `RedeemClaim::codec` for a `provider-file` locator (`0x100A`).
pub const REDEEM_CODEC_PROVIDER_FILE: &str = "provider-file";
/// `RedeemClaim::field` naming the `.nzb` manifest pointer on a release record.
pub const POINTER_FIELD_MANIFEST: &str = "manifest";
/// `RedeemClaim::field` naming the fetched-file pointer on a locator record.
pub const POINTER_FIELD_FILE: &str = "file";
/// A plugin's claim that it can **redeem** a locator family: turn a locator cid
/// into the bytes it names, through `POST /compute` with `record_id` = the
/// locator cid.
///
/// Redeeming usually spends a metered upstream quota (a Newznab `t=get` grab,
/// an OpenSubtitles `/download`), so the gateway routes a redeem only to the
/// plugins claiming the cid's codec, one at a time, and only on a real play. It
/// stores the returned bytes and writes their cid onto the locator record under
/// [`field`](Self::field), so the quota is spent once per file.
///
/// Served at `GET /manifest` (per plugin) and `GET /redeems`. Reflects the
/// plugin's *current* config: a plugin with no credential for a family claims
/// nothing for it.
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
pub struct RedeemClaim {
/// Locator family: [`REDEEM_CODEC_NZB_RELEASE`] or [`REDEEM_CODEC_PROVIDER_FILE`].
pub codec: String,
/// Pointer field the gateway writes the redeemed bytes' cid under:
/// [`POINTER_FIELD_MANIFEST`] or [`POINTER_FIELD_FILE`].
pub field: String,
/// Indexer hosts this plugin holds a key for (`nzb-release`) — bare
/// authority, lowercase, never the key itself.
#[serde(default)]
pub hosts: Vec<String>,
/// Provider sources this plugin redeems (`provider-file`), e.g. `opensubtitles`.
#[serde(default)]
pub sources: Vec<String>,
}
impl RedeemClaim {
/// Claim `nzb-release` locators for these indexer hosts.
pub fn nzb_release(hosts: Vec<String>) -> Self {
Self {
codec: REDEEM_CODEC_NZB_RELEASE.to_string(),
field: POINTER_FIELD_MANIFEST.to_string(),
hosts,
sources: Vec::new(),
}
}
/// Claim `provider-file` locators for these sources.
pub fn provider_file(sources: Vec<String>) -> Self {
Self {
codec: REDEEM_CODEC_PROVIDER_FILE.to_string(),
field: POINTER_FIELD_FILE.to_string(),
hosts: Vec::new(),
sources,
}
}
}
/// Startup-only configuration error returned by [`FeederPlugin::configure`]. /// Startup-only configuration error returned by [`FeederPlugin::configure`].
#[derive(Debug, thiserror::Error)] #[derive(Debug, thiserror::Error)]
pub enum ConfigError { pub enum ConfigError {
@@ -215,6 +276,20 @@ pub trait FeederPlugin: Send + Sync + 'static {
&[] &[]
} }
/// Folder under meta-core's `/files/plugin/` where the gateway stores bytes
/// this plugin redeems (e.g. `meta-feeder-usenet`). Stable across
/// deployments — unlike the feeder's compose service name. Default: none.
fn package(&self) -> Option<&'static str> {
None
}
/// Locator families this plugin can redeem through
/// [`compute_outcomes`](FeederPlugin::compute_outcomes) — see
/// [`RedeemClaim`]. Computed from the current config. Default: none.
fn redeems(&self) -> Vec<RedeemClaim> {
Vec::new()
}
/// Self-describing config schema. The feeder SDK serves this at /// Self-describing config schema. The feeder SDK serves this at
/// `GET /config/schema` and renders a generic form from it — the gateway and /// `GET /config/schema` and renders a generic form from it — the gateway and
/// UI carry no per-plugin field knowledge. Default: no configuration. /// UI carry no per-plugin field knowledge. Default: no configuration.
@@ -255,4 +330,23 @@ mod tests {
assert!(msg.contains("scihub")); assert!(msg.contains("scihub"));
assert!(msg.contains("mirrors")); assert!(msg.contains("mirrors"));
} }
/// The JSON keys are the gateway/meta-share contract — pin them.
#[test]
fn redeem_claim_wire_shape() {
let claim = RedeemClaim::nzb_release(vec!["api.nzb.life".into()]);
assert_eq!(
serde_json::to_value(&claim).unwrap(),
serde_json::json!({
"codec": "nzb-release",
"field": "manifest",
"hosts": ["api.nzb.life"],
"sources": []
})
);
// hosts/sources may be omitted on the wire.
let parsed: RedeemClaim =
serde_json::from_str(r#"{"codec":"provider-file","field":"file"}"#).unwrap();
assert_eq!(parsed, RedeemClaim::provider_file(vec![]));
}
} }
+176 -2
View File
@@ -10,6 +10,17 @@
//! - `POST /compute` → [`ComputeRequest`] → [`ComputeResponse`] //! - `POST /compute` → [`ComputeRequest`] → [`ComputeResponse`]
//! - `GET /fetch/{upstream_id}/{record_id}` → streamed bytes (or 404) //! - `GET /fetch/{upstream_id}/{record_id}` → streamed bytes (or 404)
//! - `GET /blob/{upstream_id}/{cid}` → bytes (or 404) //! - `GET /blob/{upstream_id}/{cid}` → bytes (or 404)
//! - `GET /redeems` → [`RedeemsResponse`] (current redeem claims)
//!
//! ## Redeeming a locator
//!
//! A plugin that advertises a [`RedeemClaim`] answers `POST /compute` with
//! `record_id` = a locator cid of that family (a Newznab `nzb-release`, an
//! OpenSubtitles `provider-file`) by fetching the bytes it names — usually
//! spending a metered quota — and returning ONE `Sha2_256` outcome carrying
//! them. `NotFound` means "not mine" (e.g. no key for that indexer host), so the
//! gateway tries the next claimer. The claims are re-read from `GET /redeems`
//! so a config change shows up without restarting the gateway.
//! //!
//! ## Bytes & the v1 simplification //! ## Bytes & the v1 simplification
//! //!
@@ -21,7 +32,7 @@
//! documented follow-up — those outcomes are metadata-only anyway //! documented follow-up — those outcomes are metadata-only anyway
//! (`bytes: None`), so they never ride this path. //! (`bytes: None`), so they never ride this path.
use std::collections::HashMap; use std::collections::{BTreeMap, HashMap};
use std::net::SocketAddr; use std::net::SocketAddr;
use std::path::{Path as FsPath, PathBuf}; use std::path::{Path as FsPath, PathBuf};
use std::sync::Arc; use std::sync::Arc;
@@ -41,7 +52,7 @@ use serde_json::Value;
use tracing::{info, warn}; use tracing::{info, warn};
use crate::config::{self, CONFIG_PAGE_HTML}; use crate::config::{self, CONFIG_PAGE_HTML};
use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome}; use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim};
use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError}; use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError};
use crate::types::{DiscoveryRecord, GatewayError, PluginHealth}; use crate::types::{DiscoveryRecord, GatewayError, PluginHealth};
@@ -94,6 +105,22 @@ pub struct PluginManifest {
pub id: String, pub id: String,
pub served_file_types: Vec<String>, pub served_file_types: Vec<String>,
pub served_content_kinds: Vec<String>, pub served_content_kinds: Vec<String>,
/// [`FeederPlugin::package`] — the `/files/plugin/<package>/` folder redeemed
/// bytes are stored under. `null` when the plugin redeems nothing.
#[serde(default)]
pub package: Option<String>,
/// [`FeederPlugin::redeems`] at the time the manifest was served.
#[serde(default)]
pub redeems: Vec<RedeemClaim>,
}
/// `GET /redeems` — every hosted plugin's current [`RedeemClaim`]s, keyed by
/// `upstream_id` (an empty list for a plugin that claims nothing). Polled by
/// the gateway, so a credential added on the config page is advertised without
/// a gateway restart.
#[derive(Serialize, Deserialize, Clone, Debug, Default)]
pub struct RedeemsResponse {
pub redeems: BTreeMap<String, Vec<RedeemClaim>>,
} }
/// `GET /manifest` — the feeder's hosted plugins + their served types. The /// `GET /manifest` — the feeder's hosted plugins + their served types. The
@@ -224,6 +251,17 @@ impl AppState {
/// status semantics the gateway's `RemoteFeederPlugin` will translate back /// status semantics the gateway's `RemoteFeederPlugin` will translate back
/// into a `GatewayError` on the core side. /// into a `GatewayError` on the core side.
fn gateway_error_response(e: GatewayError) -> Response { fn gateway_error_response(e: GatewayError) -> Response {
// A 429 carries the wait as a `Retry-After` header too, so the gateway can
// forward it to the client (a redeem blocked on a spent quota) instead of
// parsing the message.
if let GatewayError::RateLimited { retry_after_s } = &e {
return (
StatusCode::TOO_MANY_REQUESTS,
[(axum::http::header::RETRY_AFTER, retry_after_s.to_string())],
format!("rate limited; retry in {retry_after_s}s"),
)
.into_response();
}
let (status, msg) = match e { let (status, msg) = match e {
GatewayError::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()), GatewayError::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()),
GatewayError::RateLimited { retry_after_s } => ( GatewayError::RateLimited { retry_after_s } => (
@@ -251,6 +289,8 @@ async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
.iter() .iter()
.map(|s| s.to_string()) .map(|s| s.to_string())
.collect(), .collect(),
package: p.package().map(str::to_string),
redeems: p.redeems(),
}) })
.collect(); .collect();
plugins.sort_by(|a, b| a.id.cmp(&b.id)); plugins.sort_by(|a, b| a.id.cmp(&b.id));
@@ -260,6 +300,16 @@ async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
}) })
} }
async fn redeems(State(state): State<AppState>) -> Json<RedeemsResponse> {
Json(RedeemsResponse {
redeems: state
.plugins
.values()
.map(|p| (p.upstream_id().to_string(), p.redeems()))
.collect(),
})
}
async fn health(State(state): State<AppState>) -> Json<HealthResponse> { async fn health(State(state): State<AppState>) -> Json<HealthResponse> {
let mut entries: Vec<PluginHealthEntry> = state let mut entries: Vec<PluginHealthEntry> = state
.plugins .plugins
@@ -492,6 +542,7 @@ pub fn router(
.route("/compute", post(compute)) .route("/compute", post(compute))
.route("/fetch/:upstream_id/:record_id", get(fetch)) .route("/fetch/:upstream_id/:record_id", get(fetch))
.route("/blob/:upstream_id/:cid", get(blob)) .route("/blob/:upstream_id/:cid", get(blob))
.route("/redeems", get(redeems))
.route("/config", get(config_page)) .route("/config", get(config_page))
.route("/config/schema", get(config_schema)) .route("/config/schema", get(config_schema))
.route("/config/values", get(config_values_get).put(config_values_put)) .route("/config/values", get(config_values_get).put(config_values_put))
@@ -541,6 +592,9 @@ struct DegradedPlugin {
reason: String, reason: String,
file_types: &'static [&'static str], file_types: &'static [&'static str],
content_kinds: &'static [&'static str], content_kinds: &'static [&'static str],
/// Preserved so the manifest keeps naming the plugin's folder. A degraded
/// plugin claims no redeems (it cannot redeem anything).
package: Option<&'static str>,
/// Preserved from the wrapped plugin so a degraded/unconfigured feeder still /// Preserved from the wrapped plugin so a degraded/unconfigured feeder still
/// serves its config form — otherwise the dashboard shows "no configuration /// serves its config form — otherwise the dashboard shows "no configuration
/// required" and the operator can never fix the very config that's missing. /// required" and the operator can never fix the very config that's missing.
@@ -586,6 +640,10 @@ impl FeederPlugin for DegradedPlugin {
self.content_kinds self.content_kinds
} }
fn package(&self) -> Option<&'static str> {
self.package
}
fn config_schema(&self) -> crate::config::ConfigSchema { fn config_schema(&self) -> crate::config::ConfigSchema {
self.schema.clone() self.schema.clone()
} }
@@ -608,6 +666,7 @@ pub fn configure_plugins(
// borrow across `configure(&mut self)` / the move into `out`. // borrow across `configure(&mut self)` / the move into `out`.
let file_types = plugin.served_file_types(); let file_types = plugin.served_file_types();
let content_kinds = plugin.served_content_kinds(); let content_kinds = plugin.served_content_kinds();
let package = plugin.package();
// Preserve the config surface so a degraded stand-in still serves the // Preserve the config surface so a degraded stand-in still serves the
// real config form (see DegradedPlugin.schema). // real config form (see DegradedPlugin.schema).
let schema = plugin.config_schema(); let schema = plugin.config_schema();
@@ -627,6 +686,7 @@ pub fn configure_plugins(
reason, reason,
file_types, file_types,
content_kinds, content_kinds,
package,
schema: schema.clone(), schema: schema.clone(),
}) })
}; };
@@ -759,4 +819,118 @@ mod tests {
assert!(matches!(events[1], GatewaySearchEvent::EnrichPatch { .. })); assert!(matches!(events[1], GatewaySearchEvent::EnrichPatch { .. }));
assert!(matches!(events[2], GatewaySearchEvent::Done)); assert!(matches!(events[2], GatewaySearchEvent::Done));
} }
/// A plugin that redeems `nzb-release` locators for one host.
struct StubRedeemer;
#[async_trait]
impl FeederPlugin for StubRedeemer {
fn upstream_id(&self) -> &'static str {
"redeemer"
}
fn configure(&mut self, _cache_dir: &FsPath) -> Result<(), ConfigError> {
Ok(())
}
async fn handle_query(
&self,
_q: &GatewayQuery,
_n: usize,
) -> Result<Vec<DiscoveryRecord>, GatewayError> {
Ok(vec![])
}
async fn compute_outcomes(
&self,
_record_id: &str,
) -> Result<Vec<HashOutcome>, GatewayError> {
Ok(vec![])
}
fn package(&self) -> Option<&'static str> {
Some("meta-feeder-stub")
}
fn redeems(&self) -> Vec<RedeemClaim> {
vec![RedeemClaim::nzb_release(vec!["api.example".into()])]
}
}
async fn serve_stubs() -> String {
let mut plugins: HashMap<String, Arc<dyn FeederPlugin>> = HashMap::new();
plugins.insert("redeemer".to_string(), Arc::new(StubRedeemer));
plugins.insert("stub".to_string(), Arc::new(StubStreaming));
let app = router(plugins, "test".to_string(), std::env::temp_dir());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
format!("http://{addr}")
}
/// The manifest carries `package` + `redeems` per plugin; a plugin that
/// claims nothing serializes `null` + `[]`.
#[tokio::test]
async fn manifest_carries_package_and_redeems() {
let base = serve_stubs().await;
let m: serde_json::Value = reqwest::get(format!("{base}/manifest"))
.await
.unwrap()
.json()
.await
.unwrap();
let redeemer = &m["plugins"][0];
assert_eq!(redeemer["id"], "redeemer");
assert_eq!(redeemer["package"], "meta-feeder-stub");
assert_eq!(
redeemer["redeems"],
serde_json::json!([{
"codec": "nzb-release", "field": "manifest",
"hosts": ["api.example"], "sources": []
}])
);
let stub = &m["plugins"][1];
assert_eq!(stub["package"], serde_json::Value::Null);
assert_eq!(stub["redeems"], serde_json::json!([]));
}
#[tokio::test]
async fn redeems_route_keys_claims_by_upstream_id() {
let base = serve_stubs().await;
let r: RedeemsResponse = reqwest::get(format!("{base}/redeems"))
.await
.unwrap()
.json()
.await
.unwrap();
assert_eq!(
r.redeems.get("redeemer"),
Some(&vec![RedeemClaim::nzb_release(vec!["api.example".into()])])
);
assert_eq!(r.redeems.get("stub"), Some(&vec![]));
}
/// The gateway forwards a feeder 429's `Retry-After` — pin that it is set.
#[test]
fn rate_limited_sets_retry_after() {
let resp = gateway_error_response(GatewayError::RateLimited { retry_after_s: 3600 });
assert_eq!(resp.status(), StatusCode::TOO_MANY_REQUESTS);
assert_eq!(
resp.headers()
.get(axum::http::header::RETRY_AFTER)
.and_then(|v| v.to_str().ok()),
Some("3600")
);
// Other errors carry no Retry-After.
let resp = gateway_error_response(GatewayError::NotFound);
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
assert!(resp.headers().get(axum::http::header::RETRY_AFTER).is_none());
}
/// A manifest from a pre-1.3 feeder (no `package` / `redeems`) still parses.
#[test]
fn old_manifest_without_redeem_fields_parses() {
let m: PluginManifest = serde_json::from_str(
r#"{"id":"x","served_file_types":[],"served_content_kinds":[]}"#,
)
.unwrap();
assert!(m.package.is_none() && m.redeems.is_empty());
}
} }