5 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
Worph fd9da12d93 align to metadata keys doc 2026-08-30 19:49:49 +02:00
Worph 4baf81e2bf feeder rework 2026-08-24 19:03:09 +02:00
11 changed files with 1848 additions and 29 deletions
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "meta-feeder-sdk"
version = "0.1.0"
version = "1.3.0"
edition = "2021"
rust-version = "1.85"
license = "MIT"
+40 -7
View File
@@ -17,7 +17,7 @@ bytes; it never joins the swarm.
```toml
[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.
@@ -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 |
Defaults you can override: `handle_query_stream`, `handle_fetch`, `health`,
`served_file_types`, `served_content_kinds`, `config_schema`, `config_values`,
`get_blob`.
`served_file_types`, `served_content_kinds`, `package`, `redeems`,
`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
@@ -101,6 +125,9 @@ into the bitswap blockstore.
The last two carry an identity multihash. Anything that dispatches on the
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
| module | what |
@@ -121,13 +148,19 @@ multihash instead of the **codec** will silently mis-rank them.
cargo build && cargo test
```
To work on the SDK from inside a feeder checkout without pushing a tag, add a
gitignored `.cargo/config.toml` in the feeder:
To test a feeder against an SDK change **before its tag is pushed**: cargo will
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
paths = ["../meta-feeder-sdk"]
```bash
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
Design rationale, the feeder roster, and the naming rules live in the meta-root:
+241
View File
@@ -0,0 +1,241 @@
//! Per-upstream request budget — a shared token bucket.
//!
//! Lifted from the card feeder's `tmdb_budget.rs` (its `TmdbBudget` half; the
//! per-indexer registry has no counterpart here) and renamed, because the
//! mechanism is upstream-agnostic and the music feeder needs three independent
//! instances rather than one global.
//!
//! **Why three, not one.** MusicBrainz's 1 req/s is a hard published ceiling
//! enforced per source IP; ListenBrainz and the Internet Archive are far more
//! generous. A single shared bucket would either throttle all three to 1/s or
//! let MusicBrainz calls burst past a limit that gets the whole box blocked.
//!
//! Design notes, unchanged from the original:
//! - **Lazy refill.** Tokens accrue on each `acquire` from elapsed wall-clock,
//! so there is no background timer task and no idle cost.
//! - **Cancel-safe.** The token is decremented synchronously under the lock
//! right before `acquire` returns [`Lease::Granted`]; there is no `.await`
//! between the decision to grant and the return, so a dropped `acquire`
//! future (consumer disconnected) never spends a token.
//! - **Global pause on throttle.** One upstream 429/503 freezes *all* grants
//! on that bucket until the Retry-After window elapses, via
//! [`RateBudget::note_throttled`].
//!
//! The critical sections hold a plain `std::sync::Mutex` (no awaits inside),
//! while the waiting happens outside the lock on a `tokio::time::sleep` raced
//! against a `tokio::sync::Notify`.
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::Notify;
/// Outcome of a [`RateBudget::acquire`] call.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Lease {
/// A token was granted; the caller may make one upstream call.
Granted,
/// The wait deadline elapsed before a token became available. The caller
/// degrades best-effort — it does **not** call the upstream anyway.
DeadlineExceeded,
}
struct BucketState {
tokens: f64,
capacity: f64,
refill_per_sec: f64,
last_refill: Instant,
/// When set and in the future, all grants are frozen.
paused_until: Option<Instant>,
}
impl BucketState {
/// Accrue tokens for elapsed time since the last refill. Called under the
/// lock before inspecting `tokens`.
fn refill(&mut self, now: Instant) {
let elapsed = now.saturating_duration_since(self.last_refill).as_secs_f64();
if elapsed > 0.0 {
self.tokens = (self.tokens + elapsed * self.refill_per_sec).min(self.capacity);
self.last_refill = now;
}
}
}
/// Shared token bucket gating one upstream's API calls across the process.
pub struct RateBudget {
state: Mutex<BucketState>,
/// Woken whenever a grant *might* now be possible (a token accrued, or a
/// pause was cleared). Waiters re-check under the lock after waking; a
/// missed wake only costs an extra bounded sleep.
wake: Notify,
}
impl RateBudget {
/// Build a budget sustaining `refill_per_sec` grants/second with a burst
/// ceiling of `capacity` tokens. The bucket starts full, so the first
/// burst is served immediately.
pub fn new(refill_per_sec: f64, capacity: f64) -> Arc<Self> {
Arc::new(Self {
state: Mutex::new(BucketState {
tokens: capacity,
capacity,
refill_per_sec,
last_refill: Instant::now(),
paused_until: None,
}),
wake: Notify::new(),
})
}
/// Wait for one token, giving up after `deadline`. Cancel-safe: dropping
/// the returned future before it resolves [`Lease::Granted`] spends no
/// token.
pub async fn acquire(&self, deadline: Duration) -> Lease {
let give_up_at = Instant::now() + deadline;
loop {
// Decide-or-compute-wait, all under the lock (no awaits inside).
let sleep_for = {
let mut st = self.state.lock().expect("rate budget mutex poisoned");
let now = Instant::now();
if let Some(until) = st.paused_until {
if now >= until {
st.paused_until = None;
}
}
let paused = st.paused_until;
if paused.is_none() {
st.refill(now);
if st.tokens >= 1.0 {
// Atomic grant: decrement and return with no await between.
st.tokens -= 1.0;
return Lease::Granted;
}
}
if now >= give_up_at {
return Lease::DeadlineExceeded;
}
let until_deadline = give_up_at.saturating_duration_since(now);
let wait = match paused {
// Frozen: wake when the pause is scheduled to lift.
Some(until) => until.saturating_duration_since(now),
// Throttled: wake when the next whole token should accrue.
None => {
let needed = 1.0 - st.tokens;
let secs = if st.refill_per_sec > 0.0 {
needed / st.refill_per_sec
} else {
until_deadline.as_secs_f64()
};
Duration::from_secs_f64(secs.max(0.0))
}
};
wait.min(until_deadline)
};
tokio::select! {
_ = tokio::time::sleep(sleep_for) => {}
_ = self.wake.notified() => {}
}
if Instant::now() >= give_up_at {
return Lease::DeadlineExceeded;
}
}
}
/// Feed back an upstream throttle response (429, or MusicBrainz's 503):
/// pause all grants on this bucket until `now + retry_after`. Extends an
/// existing pause but never shortens it.
pub fn note_throttled(&self, retry_after: Duration) {
let until = Instant::now() + retry_after;
{
let mut st = self.state.lock().expect("rate budget mutex poisoned");
st.paused_until = Some(match st.paused_until {
Some(existing) if existing > until => existing,
_ => until,
});
}
// Wake waiters so they recompute their sleep against the new pause.
self.wake.notify_waiters();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test(start_paused = true)]
async fn grants_up_to_capacity_immediately() {
let b = RateBudget::new(3.0, 3.0);
for _ in 0..3 {
assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
}
}
#[tokio::test(start_paused = true)]
async fn next_token_waits_for_refill() {
let b = RateBudget::new(2.0, 2.0); // 2/s sustained
for _ in 0..2 {
assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
}
// Bucket empty; next grant must wait ~0.5 s for one token at 2/s.
let start = Instant::now();
assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
assert!(
start.elapsed() >= Duration::from_millis(450),
"expected to wait ~0.5s for refill, waited {:?}",
start.elapsed()
);
}
/// The MusicBrainz shape specifically: 1/s with a burst of 2 must not let
/// three calls through inside one second. This is the property that keeps
/// the box off their block-list.
#[tokio::test(start_paused = true)]
async fn musicbrainz_shape_holds_one_per_second() {
let b = RateBudget::new(1.0, 1.0);
let start = Instant::now();
for _ in 0..4 {
assert_eq!(b.acquire(Duration::from_secs(30)).await, Lease::Granted);
}
// 4 grants from a 2-token bucket refilling at 1/s ⇒ at least 2 s.
assert!(
start.elapsed() >= Duration::from_millis(1_900),
"4 calls at 1/s (burst 2) must take ~2s, took {:?}",
start.elapsed()
);
}
#[tokio::test(start_paused = true)]
async fn deadline_exceeded_when_drained() {
let b = RateBudget::new(0.001, 1.0); // effectively no refill within the test
assert_eq!(b.acquire(Duration::from_secs(1)).await, Lease::Granted);
assert_eq!(
b.acquire(Duration::from_millis(200)).await,
Lease::DeadlineExceeded
);
}
#[tokio::test(start_paused = true)]
async fn note_throttled_pauses_grants_even_with_tokens_available() {
let b = RateBudget::new(100.0, 100.0);
b.note_throttled(Duration::from_secs(2));
assert_eq!(
b.acquire(Duration::from_millis(500)).await,
Lease::DeadlineExceeded
);
// After the pause window, grants resume.
assert_eq!(b.acquire(Duration::from_secs(5)).await, Lease::Granted);
}
/// A pause is extended, never shortened — otherwise a late short retry
/// would cancel an earlier long one and walk straight back into the block.
#[tokio::test(start_paused = true)]
async fn a_shorter_retry_after_does_not_shorten_an_existing_pause() {
let b = RateBudget::new(100.0, 100.0);
b.note_throttled(Duration::from_secs(10));
b.note_throttled(Duration::from_secs(1));
assert_eq!(
b.acquire(Duration::from_secs(3)).await,
Lease::DeadlineExceeded
);
}
}
+154 -1
View File
@@ -13,7 +13,7 @@
use std::path::Path;
use std::sync::Arc;
use redb::{Database, ReadableTableMetadata, TableDefinition};
use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};
pub const CACHE_FILENAME: &str = "cache.redb";
@@ -53,6 +53,44 @@ const SUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("s
/// OpenSubtitles search results keyed by `"<tmdb_id>\x01<lang3>"`. Value is the
/// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel.
const OPENSUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("opensubtitles");
// ── music-family tables (MusicBrainz / Internet Archive / YouTube / Jamendo) ──
/// Ranked MusicBrainz search results keyed by `"<entity>\x01<normalised text>"`.
const MB_SEARCH_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_search");
/// Release-group detail keyed by its MBID.
const MB_RELEASE_GROUP_TABLE: TableDefinition<'_, &str, &str> =
TableDefinition::new("mb_release_group");
/// Artist detail keyed by its MBID.
const MB_ARTIST_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_artist");
/// A release-group's canonical track list keyed by the release-group MBID.
const MB_TRACKLIST_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_tracklist");
/// Internet Archive per-item file listing keyed by the item identifier.
const IA_ITEM_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("ia_item");
/// An artist's YouTube channel keys, keyed by artist MBID. See
/// `youtube::ChannelKeys`.
///
/// ⚠ **Only ever holds a POSITIVE result.** MusicBrainz answers `503` when
/// crowded and the client reports every failure as `None`, so caching a
/// negative would freeze a transport hiccup into "this artist has no channel"
/// — the exact fabrication the match probe hit (study §7.5).
const YT_CHANNEL_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("yt_channel");
/// Piped `music_songs` results keyed by the normalised search text.
const YT_SEARCH_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("yt_search");
/// Jamendo per-track detail keyed by the Jamendo track id. Permanent: a track
/// id names an immutable recording, and its licence, artist credit and
/// download URL do not change under it.
const JAMENDO_TRACK_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("jamendo_track");
/// Jamendo API calls spent, keyed by UTC `YYYY-MM`.
///
/// ⚠ **This table is not a cache — it is an accounting ledger, and losing it
/// loses money.** Every other table here can be deleted to reclaim disk with
/// no consequence beyond a slower next request. Deleting this one resets the
/// month's spend to zero while Jamendo's own counter keeps climbing, so the
/// feeder cheerfully spends a quota it has already exhausted and every call
/// fails for the rest of the month. See `consts::JAMENDO_MONTHLY_QUOTA`.
const JAMENDO_QUOTA_TABLE: TableDefinition<'_, &str, u64> = TableDefinition::new("jamendo_quota");
/// Small durable plugin scratch values that must survive a restart but aren't
/// content-derived — today, the torznab plugin's Prowlarr indexer-set
/// fingerprint, so a set that changed *while the feeder was down* is still
@@ -113,6 +151,15 @@ impl MidhashCache {
tx.open_table(SUBTITLES_TABLE)?;
tx.open_table(OPENSUBTITLES_TABLE)?;
tx.open_table(MISC_TABLE)?;
tx.open_table(MB_SEARCH_TABLE)?;
tx.open_table(MB_RELEASE_GROUP_TABLE)?;
tx.open_table(MB_ARTIST_TABLE)?;
tx.open_table(MB_TRACKLIST_TABLE)?;
tx.open_table(IA_ITEM_TABLE)?;
tx.open_table(YT_CHANNEL_TABLE)?;
tx.open_table(YT_SEARCH_TABLE)?;
tx.open_table(JAMENDO_TRACK_TABLE)?;
tx.open_table(JAMENDO_QUOTA_TABLE)?;
tx.commit()?;
}
Ok(Self { db: Arc::new(db) })
@@ -120,6 +167,47 @@ impl MidhashCache {
str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE);
// ── music-family accessors ──
str_table_accessors!(get_mb_search, put_mb_search, MB_SEARCH_TABLE);
str_table_accessors!(get_mb_release_group, put_mb_release_group, MB_RELEASE_GROUP_TABLE);
str_table_accessors!(get_mb_artist, put_mb_artist, MB_ARTIST_TABLE);
str_table_accessors!(get_mb_tracklist, put_mb_tracklist, MB_TRACKLIST_TABLE);
str_table_accessors!(get_ia_item, put_ia_item, IA_ITEM_TABLE);
str_table_accessors!(get_yt_channel, put_yt_channel, YT_CHANNEL_TABLE);
str_table_accessors!(get_yt_search, put_yt_search, YT_SEARCH_TABLE);
str_table_accessors!(get_jamendo_track, put_jamendo_track, JAMENDO_TRACK_TABLE);
pub fn jamendo_quota_spend(
&self,
month: &str,
budget: u64,
) -> Result<Option<u64>, redb::Error> {
let tx = self.db.begin_write()?;
let spent_after;
{
let mut table = tx.open_table(JAMENDO_QUOTA_TABLE)?;
let spent = table.get(month)?.map(|v| v.value()).unwrap_or(0);
if spent >= budget {
drop(table);
// Nothing was written; abort rather than commit an empty txn.
tx.abort()?;
return Ok(None);
}
spent_after = spent + 1;
table.insert(month, spent_after)?;
}
tx.commit()?;
Ok(Some(spent_after))
}
/// Calls already spent in `month`. Read-only — for health and the config
/// surface, never on the request path.
pub fn jamendo_quota_used(&self, month: &str) -> Result<u64, redb::Error> {
let tx = self.db.begin_read()?;
let table = tx.open_table(JAMENDO_QUOTA_TABLE)?;
Ok(table.get(month)?.map(|v| v.value()).unwrap_or(0))
}
str_table_accessors!(
/// Read a durable plugin scratch value (see [`MISC_TABLE`]).
get_misc,
@@ -351,3 +439,68 @@ mod tests {
assert_eq!(cache.entry_count().unwrap(), 2);
}
}
/// Cache key for a search: entity type plus the normalised query text.
///
/// Normalisation is lowercase + whitespace-collapse only. Deliberately not
/// more aggressive: MusicBrainz treats punctuation as significant (`Re:ZERO`,
/// `M!LK`, `!!!` are all real artist names), so stripping it would collapse
/// distinct queries onto one cache entry and serve the wrong answer.
pub fn search_key(entity: &str, text: &str) -> String {
let normalised = text.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase();
format!("{entity}\u{1}{normalised}")
}
#[cfg(test)]
mod music_table_tests {
use super::*;
use tempfile::tempdir;
#[test]
fn round_trips_each_table_independently() {
let dir = tempdir().expect("tempdir");
let c = MidhashCache::open(dir.path()).expect("open");
c.put_mb_search("k", "v-search").unwrap();
c.put_mb_release_group("k", "v-rg").unwrap();
c.put_mb_artist("k", "v-artist").unwrap();
c.put_mb_tracklist("k", "v-tracks").unwrap();
c.put_ia_item("k", "v-ia").unwrap();
// Same key in five tables must not collide.
assert_eq!(c.get_mb_search("k").unwrap().as_deref(), Some("v-search"));
assert_eq!(c.get_mb_release_group("k").unwrap().as_deref(), Some("v-rg"));
assert_eq!(c.get_mb_artist("k").unwrap().as_deref(), Some("v-artist"));
assert_eq!(c.get_mb_tracklist("k").unwrap().as_deref(), Some("v-tracks"));
assert_eq!(c.get_ia_item("k").unwrap().as_deref(), Some("v-ia"));
assert!(c.get_mb_search("absent").unwrap().is_none());
}
#[test]
fn search_key_normalises_case_and_whitespace_only() {
assert_eq!(
search_key("release-group", " Kind OF Blue "),
search_key("release-group", "kind of blue")
);
// Entity namespacing keeps an artist search off a release-group entry.
assert_ne!(
search_key("artist", "miles davis"),
search_key("release-group", "miles davis")
);
}
/// Punctuation is significant to MusicBrainz, so it must survive
/// normalisation — otherwise two genuinely different works share a cache
/// entry and one of them is served the other's answer.
#[test]
fn search_key_preserves_punctuation() {
assert_ne!(
search_key("artist", "m!lk"),
search_key("artist", "mlk"),
);
assert_ne!(
search_key("release-group", "re:zero"),
search_key("release-group", "rezero"),
);
}
}
+40
View File
@@ -286,3 +286,43 @@ mod tests {
assert!(decode_infohash("").is_none());
}
}
/// Percent-encode for a URL **path or query value**, escaping space as `%20`.
///
/// ⚠ **Not interchangeable with [`urlencode`]**, which encodes space as `+`.
/// The `+` form is correct for `application/x-www-form-urlencoded` bodies and
/// wrong for a Lucene-backed search API — MusicBrainz's `/ws/2` treats `+` as
/// the "required term" metacharacter, so a `+`-encoded space silently changes
/// the query's meaning rather than failing loudly. Use this one for anything
/// going into a query *value*; use `urlencode` for form bodies.
pub fn percent_encode(s: &str) -> String {
let mut out = String::with_capacity(s.len() * 2);
for b in s.bytes() {
match b {
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
out.push(b as char)
}
b' ' => out.push_str("%20"),
_ => out.push_str(&format!("%{b:02X}")),
}
}
out
}
/// Normalise a variable-precision date (`YYYY`, `YYYY-MM`, `YYYY-MM-DD`) to a
/// full `YYYY-MM-DD`, or `None` if it is not one.
///
/// ⚠ Validate the shape, never infer it from the length: a first cut matched on
/// `len()` alone, which turned the seven-character string "garbage" into the
/// `releasedate` "garbage-01" — a malformed value written to the shared hash
/// where nothing downstream would question it.
pub fn normalise_date(raw: &str) -> Option<String> {
let digits = |s: &str, n: usize| s.len() == n && s.chars().all(|c| c.is_ascii_digit());
let parts: Vec<&str> = raw.trim().split('-').collect();
match parts.as_slice() {
[y] if digits(y, 4) => Some(format!("{y}-01-01")),
[y, m] if digits(y, 4) && digits(m, 2) => Some(format!("{y}-{m}-01")),
[y, m, d] if digits(y, 4) && digits(m, 2) && digits(d, 2) => Some(format!("{y}-{m}-{d}")),
_ => None,
}
}
+549
View File
@@ -0,0 +1,549 @@
//! `contentKind` → `domain` / `workForm` — the routing tables from
//! METADATA_KEYS.md §1.
//!
//! `domain` answers **"which application is this record destined for?"**. It
//! is the coarsest facet in the registry and the key every client filters its
//! catalogue on — **one value per app**, no alternation:
//!
//! ```text
//! meta-watch → domain:screen
//! meta-listen → domain:music
//! meta-read → domain:literature
//! ```
//!
//! `workForm` answers the *other* question the old `film`/`tv` split was
//! secretly encoding: **is this work complete in itself, or one instalment of
//! a run?** (`standalone` | `serial`). It is cross-domain — a comic issue is
//! `serial` in `literature` exactly as an episode is `serial` in `screen` —
//! which is what lets a client ask for it independently of the wall.
//!
//! # Why they are stored and not derived on read
//!
//! `contentKind`'s vocabulary is deliberately open — a new feeder may mint a
//! kind and ship without a fleet-wide update. That only stays safe because the
//! domain travels *with* the record: a peer that has never heard of
//! `podcastEpisode` still routes it correctly, because it never has to
//! interpret the token. Deriving on read would silently drop every record
//! whose kind the reader predates. The same argument applies to `workForm`,
//! plus a second one: the two kinds that most need it (`pack` and `card`) are
//! exactly the two the kind alone cannot decide.
//!
//! # ⚠ The binding is permanent and one-way
//!
//! Add kinds freely; **never re-map an existing kind to a different domain**.
//! Because the value is stored, two peers on different versions that disagree
//! about where `audiobook` belongs produce a split-brain that redeploying
//! cannot heal — the wrong domain is already written into records across the
//! mesh. If a kind genuinely has to move, mint a new kind.
//!
//! The `film`+`tv` → `screen` merge (METADATA_KEYS.md §14.17) is the one
//! deliberate exception, taken once and early and paid for with a one-shot
//! meta-core corpus sweep. It is not a precedent.
//!
//! # Why `domain` and not one kind-key per medium
//!
//! `record_matches` is AND-across-keys / OR-within-key, and a non-matching key
//! fails the record outright. A per-medium split (`videoKind` / `audioKind` /
//! `documentKind`) would force meta-listen to express "music video **or**
//! audio track" as an OR across two *different* keys — which the query
//! language cannot represent at all. One domain key, OR-ed within itself,
//! always can. `workForm` is one key everywhere for the same reason.
/// The `contentKind` → `domain` table. A kind absent from this list has no
/// domain and is **not independently routable** — correct for sidecars
/// (subtitles, artwork) and for un-anchored hits, a producer bug for anything
/// else.
///
/// ⚠ `pack` is deliberately absent. A season pack and an album release are the
/// same structural thing, so the kind alone cannot tell `screen` from `music`;
/// the writer stamps `pack`'s domain from its own context (the same reason
/// `derive_file_type` cannot have a second `pack` arm).
///
/// ⚠ `podcast` / `podcastEpisode` are absent too: the `spoken` domain was
/// retired unused (METADATA_KEYS.md §14.17), and the PR that first *writes*
/// those kinds picks their domain.
const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
// screen — everything meta-watch consumes, films and serials alike. The
// standalone/serial split lives in `workForm`, not here.
("movie", "screen"),
("series", "screen"),
("episode", "screen"),
// music
("track", "music"),
("album", "music"),
("artist", "music"),
("musicVideo", "music"),
("djMix", "music"),
("liveSet", "music"),
// literature — an audiobook's identity is the book (same author, same ISBN
// family, same series as the ebook), so meta-read groups the two editions
// on one page. The player does not decide the domain.
("book", "literature"),
("audiobook", "literature"),
("comic", "literature"),
("manga", "literature"),
("magazine", "literature"),
// science
("paper", "science"),
];
/// The `contentKind` → `workForm` table (METADATA_KEYS.md §1).
///
/// `standalone` = a work complete in itself. `serial` = an instalment of an
/// ongoing run, **or the container or headline for one** — which is why a
/// `series` card and an `episode` are both `serial`.
///
/// ⚠ `pack` is absent for the same reason it is absent above: a season pack is
/// `serial`, an album release is `standalone`, and only the writer knows which.
const WORK_FORM_BY_CONTENT_KIND: &[(&str, &str)] = &[
// screen
("movie", "standalone"),
("series", "serial"),
("episode", "serial"),
// music — an album is a closed work, not an ongoing run, so every rung of
// the music ladder is standalone.
("track", "standalone"),
("album", "standalone"),
("artist", "standalone"),
("musicVideo", "standalone"),
("djMix", "standalone"),
("liveSet", "standalone"),
// literature — the split that proves `workForm` is not a screen-only axis.
("book", "standalone"),
("audiobook", "standalone"),
("comic", "serial"),
("manga", "serial"),
("magazine", "serial"),
// science
("paper", "standalone"),
];
/// Resolve the domain a `contentKind` belongs to.
///
/// Returns `None` for `pack` (not derivable from the kind — the writer supplies
/// it), for format/sidecar kinds that route nowhere, and for any kind this
/// build has never heard of. A caller that already knows the domain from its
/// own context must prefer that over this table.
pub fn domain_for_content_kind(content_kind: &str) -> Option<&'static str> {
lookup(DOMAIN_BY_CONTENT_KIND, content_kind)
}
/// Resolve the work form a `contentKind` sits on. Same contract as
/// [`domain_for_content_kind`]: `None` means "not derivable", never a guess.
pub fn work_form_for_content_kind(content_kind: &str) -> Option<&'static str> {
lookup(WORK_FORM_BY_CONTENT_KIND, content_kind)
}
fn lookup(table: &'static [(&'static str, &'static str)], key: &str) -> Option<&'static str> {
let k = key.trim();
table
.iter()
.find(|(kind, _)| kind.eq_ignore_ascii_case(k))
.map(|(_, value)| *value)
}
/// Kinds whose `domain` the kind→domain table deliberately cannot decide, so
/// the **writer** supplies it (METADATA_KEYS.md §1). Today that is `pack`
/// alone: a season pack and an album release are the same structural thing, and
/// only the producer knows which — indexer-feeder's `bt.rs` reads the domain off
/// the per-file kind *before* overwriting it with `pack` precisely so the value
/// survives.
///
/// They must ride along in every `domain:` → `contentKind:` expansion. The
/// expansion is only ever a **pre-filter** — the real `domain:` term is enforced
/// afterwards by [`crate::query_eval::record_matches`] — so widening it is free,
/// while omitting these kinds would drop exactly the records whose `domain` is
/// the most authoritative one on the wire.
pub const WRITER_SUPPLIED_DOMAIN_KINDS: &[&str] = &["pack"];
/// The `workForm` twin of [`WRITER_SUPPLIED_DOMAIN_KINDS`], and for the same
/// reason: a season pack is `serial`, an album release is `standalone`.
pub const WRITER_SUPPLIED_WORK_FORM_KINDS: &[&str] = &["pack"];
/// The inverse: every `contentKind` that belongs to `domain`.
///
/// Used at the gateway-routing boundary. A gateway declares the *content kinds*
/// it serves (`served_content_kinds`), not domains, so a `domain:` filter is
/// matched by expanding it here rather than by adding a `served_domains` field
/// to the heartbeat — which would be a wire-protocol change requiring every
/// feeder manifest and every peer to move at once.
///
/// ⚠ Since the `film`+`tv` merge this returns **several** kinds for `screen`
/// (`movie`, `series`, `episode`) where the two old values were disjoint. A
/// caller that reads only the first element silently means "movie" — scan the
/// whole list, or narrow with a `workForm:` term first.
pub fn content_kinds_for_domain(domain: &str) -> Vec<&'static str> {
inverse(DOMAIN_BY_CONTENT_KIND, domain)
}
/// The inverse of the `workForm` table. Same routing-boundary role, same
/// scan-the-whole-list warning as [`content_kinds_for_domain`].
pub fn content_kinds_for_work_form(work_form: &str) -> Vec<&'static str> {
inverse(WORK_FORM_BY_CONTENT_KIND, work_form)
}
fn inverse(table: &'static [(&'static str, &'static str)], value: &str) -> Vec<&'static str> {
let v = value.trim();
table
.iter()
.filter(|(_, val)| val.eq_ignore_ascii_case(v))
.map(|(kind, _)| *kind)
.collect()
}
/// Rewrite a query so a `domain:` / `workForm:` filter can reach a plugin that
/// has never heard of either key.
///
/// Returns `None` when the query carries neither — the common path, and cheaper
/// than cloning.
///
/// # Why translate instead of teaching every plugin
///
/// `record_matches` fails a record outright when a filter key is missing from
/// its fields, and `domain` is missing from every record a not-yet-updated
/// feeder emits (`workForm` from every record written before it existed).
/// Most feeders pin the SDK by git tag, so they cannot pick up a new key on our
/// schedule. Translating `domain:literature` into the `contentKind:`
/// alternation it covers means the wire query a plugin sees never mentions
/// either key — old plugins keep working unchanged, while the dispatcher still
/// enforces the real filters afterwards, against records it has just stamped.
///
/// # The two axes intersect
///
/// `domain:screen workForm:serial` covers `series` and `episode` — the
/// **intersection**, not the union. Widening to the union would hand a serial
/// query the whole film corpus as a pre-filter; the post-stamp check would then
/// throw it away, having already paid for it upstream.
///
/// An explicit `contentKind:` filter already in the query is left alone: it is
/// the narrower of the three, and the `domain`/`workForm` terms are still
/// enforced post-stamp, so the AND semantics hold either way.
pub fn expand_routing_filters(
query: &crate::query::GatewayQuery,
) -> Option<crate::query::GatewayQuery> {
let domains = query.filters.get("domain");
let work_forms = query.filters.get("workForm");
if domains.is_none() && work_forms.is_none() {
return None;
}
let by_domain = domains.map(|vs| expand_values(vs, content_kinds_for_domain));
let by_work_form = work_forms.map(|vs| expand_values(vs, content_kinds_for_work_form));
let mut out = query.clone();
out.filters.remove("domain");
out.filters.remove("workForm");
if out.filters.contains_key("contentKind") {
return Some(out);
}
let mut kinds: Vec<&'static str> = match (by_domain, by_work_form) {
(Some(d), Some(w)) => d.into_iter().filter(|k| w.contains(k)).collect(),
(Some(d), None) => d,
(None, Some(w)) => w,
(None, None) => unreachable!("guarded above"),
};
// Domain-ambiguous kinds come along whenever the expansion is non-empty
// — see [`WRITER_SUPPLIED_DOMAIN_KINDS`]. Skipped when nothing mapped,
// so an unknown value still expands to nothing and leaves the filter
// out entirely (below) rather than narrowing to `contentKind:pack`.
if !kinds.is_empty() {
for k in WRITER_SUPPLIED_DOMAIN_KINDS
.iter()
.chain(WRITER_SUPPLIED_WORK_FORM_KINDS.iter())
{
if !kinds.contains(k) {
kinds.push(k);
}
}
}
// An unknown value — or two axes with no kind in common — expands to
// nothing. Inserting an empty alternation would drop every record at the
// plugin; leaving the filter out lets the plugin answer and the post-stamp
// check do the (correct, empty) filtering.
if !kinds.is_empty() {
out.filters.insert(
"contentKind".to_string(),
kinds.into_iter().map(|k| k.to_string()).collect(),
);
}
Some(out)
}
/// Deprecated alias for [`expand_routing_filters`], kept so a caller pinned to
/// an older SDK keeps compiling. It expands `workForm:` too — the name is the
/// only thing that is stale.
#[deprecated(note = "renamed to expand_routing_filters; it now expands workForm: as well")]
pub fn expand_domain_filter(
query: &crate::query::GatewayQuery,
) -> Option<crate::query::GatewayQuery> {
expand_routing_filters(query)
}
fn expand_values(
values: &[String],
expand: fn(&str) -> Vec<&'static str>,
) -> Vec<&'static str> {
let mut out: Vec<&'static str> = Vec::new();
for v in values {
for k in expand(v) {
if !out.contains(&k) {
out.push(k);
}
}
}
out
}
/// Stamp `domain` on a field map that already carries a `contentKind`.
///
/// The convention from METADATA_KEYS.md §1: **whoever writes `contentKind`
/// writes `domain` in the same breath**. An existing `domain` is never
/// overwritten — a writer that knows better than the table (a `pack`, say) has
/// already set it.
pub fn stamp_domain(fields: &mut std::collections::BTreeMap<String, String>) {
stamp(fields, "domain", domain_for_content_kind);
}
/// Stamp `workForm` the same way, from the same `contentKind`. Call it
/// wherever [`stamp_domain`] is called — the registry treats the two as one
/// breath.
pub fn stamp_work_form(fields: &mut std::collections::BTreeMap<String, String>) {
stamp(fields, "workForm", work_form_for_content_kind);
}
fn stamp(
fields: &mut std::collections::BTreeMap<String, String>,
key: &str,
derive: fn(&str) -> Option<&'static str>,
) {
if fields.contains_key(key) {
return;
}
let Some(kind) = fields.get("contentKind").cloned() else {
return;
};
if let Some(value) = derive(&kind) {
fields.insert(key.to_string(), value.to_string());
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::BTreeMap;
#[test]
fn kinds_map_to_their_domain() {
assert_eq!(domain_for_content_kind("movie"), Some("screen"));
assert_eq!(domain_for_content_kind("episode"), Some("screen"));
assert_eq!(domain_for_content_kind("track"), Some("music"));
// The case that proves medium and domain are different axes.
assert_eq!(domain_for_content_kind("musicVideo"), Some("music"));
// The case that proves the player does not decide the domain.
assert_eq!(domain_for_content_kind("audiobook"), Some("literature"));
}
#[test]
fn film_and_tv_are_one_domain_now() {
// The merge (METADATA_KEYS.md §14.17): meta-watch's wall is one term.
assert_eq!(
domain_for_content_kind("movie"),
domain_for_content_kind("series")
);
// ...and the split it used to carry lives here instead.
assert_eq!(work_form_for_content_kind("movie"), Some("standalone"));
assert_eq!(work_form_for_content_kind("series"), Some("serial"));
assert_eq!(work_form_for_content_kind("episode"), Some("serial"));
}
#[test]
fn work_form_is_not_a_screen_only_axis() {
// The point of the key: it means the same thing in every domain.
assert_eq!(work_form_for_content_kind("comic"), Some("serial"));
assert_eq!(work_form_for_content_kind("book"), Some("standalone"));
assert_eq!(work_form_for_content_kind("album"), Some("standalone"));
}
#[test]
fn the_spoken_domain_is_gone() {
// Retired unused. The kinds stay reserved with no domain until the PR
// that writes them picks one.
assert_eq!(domain_for_content_kind("podcast"), None);
assert_eq!(domain_for_content_kind("podcastEpisode"), None);
assert!(content_kinds_for_domain("spoken").is_empty());
}
#[test]
fn pack_has_no_derivable_domain_or_work_form() {
assert_eq!(domain_for_content_kind("pack"), None);
assert_eq!(work_form_for_content_kind("pack"), None);
}
#[test]
fn unknown_kind_is_none_not_a_guess() {
assert_eq!(domain_for_content_kind("hologram"), None);
assert_eq!(work_form_for_content_kind("hologram"), None);
}
#[test]
fn expansion_is_the_inverse() {
let music = content_kinds_for_domain("music");
assert!(music.contains(&"track"));
assert!(music.contains(&"musicVideo"));
assert!(!music.contains(&"audiobook"));
for kind in content_kinds_for_domain("literature") {
assert_eq!(domain_for_content_kind(kind), Some("literature"));
}
for kind in content_kinds_for_work_form("serial") {
assert_eq!(work_form_for_content_kind(kind), Some("serial"));
}
}
#[test]
fn screen_expands_to_every_rung_not_just_the_first() {
// The regression the merge creates for any caller reading `.first()`.
let screen = content_kinds_for_domain("screen");
assert!(screen.contains(&"movie"));
assert!(screen.contains(&"series"));
assert!(screen.contains(&"episode"));
}
fn q(filters: &[(&str, &[&str])]) -> crate::query::GatewayQuery {
crate::query::GatewayQuery {
raw_text: String::new(),
free_text: String::new(),
filters: filters
.iter()
.map(|(k, vs)| {
(
k.to_string(),
vs.iter().map(|v| v.to_string()).collect::<Vec<_>>(),
)
})
.collect(),
ranges: Vec::new(),
negations: Vec::new(),
}
}
#[test]
fn no_routing_filter_is_left_untouched() {
assert!(expand_routing_filters(&q(&[("fileType", &["video"])])).is_none());
}
#[test]
fn domain_becomes_the_content_kinds_it_covers() {
let out = expand_routing_filters(&q(&[("domain", &["literature"])])).unwrap();
// The plugin never sees `domain:` — that is the whole point.
assert!(!out.filters.contains_key("domain"));
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"book".to_string()));
assert!(kinds.contains(&"audiobook".to_string()));
assert!(!kinds.contains(&"track".to_string()));
}
#[test]
fn work_form_alone_also_expands() {
let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
assert!(!out.filters.contains_key("workForm"));
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"episode".to_string()));
assert!(kinds.contains(&"comic".to_string()));
assert!(!kinds.contains(&"movie".to_string()));
}
#[test]
fn the_two_axes_intersect_they_do_not_union() {
let out = expand_routing_filters(&q(&[
("domain", &["screen"][..]),
("workForm", &["serial"][..]),
]))
.unwrap();
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"series".to_string()));
assert!(kinds.contains(&"episode".to_string()));
// The union would have dragged the whole film corpus in.
assert!(!kinds.contains(&"movie".to_string()));
// A `serial` comic is in `literature`, not on meta-watch's wall.
assert!(!kinds.contains(&"comic".to_string()));
}
#[test]
fn an_explicit_content_kind_filter_wins() {
let out = expand_routing_filters(&q(&[
("domain", &["music"][..]),
("contentKind", &["track"][..]),
]))
.unwrap();
assert_eq!(
out.filters.get("contentKind").map(Vec::as_slice),
Some(&["track".to_string()][..])
);
}
#[test]
fn expansion_carries_the_writer_supplied_kinds() {
// `pack` has no domain in the table (a season pack and an album are the
// same structural thing), so the writer stamps one. Leaving it out of
// the alternation would make a `domain:screen` pre-filter drop the whole
// torrent-pack corpus, which `record_matches` would then never see.
let out = expand_routing_filters(&q(&[("domain", &["screen"])])).unwrap();
let kinds = out.filters.get("contentKind").unwrap();
assert!(kinds.contains(&"episode".to_string()));
assert!(kinds.contains(&"pack".to_string()));
// ...and it rides along on the `workForm` axis too, exactly once.
let out = expand_routing_filters(&q(&[("workForm", &["serial"])])).unwrap();
let kinds = out.filters.get("contentKind").unwrap();
assert_eq!(kinds.iter().filter(|k| *k == "pack").count(), 1);
}
#[test]
fn unknown_value_adds_no_alternation() {
// An empty alternation would drop every record at the plugin; the
// post-stamp check does the (correct, empty) filtering.
let out = expand_routing_filters(&q(&[("domain", &["hologram"])])).unwrap();
// Not even `pack`: a lone `contentKind:pack` would be a NARROWING of a
// filter that should widen to nothing.
assert!(!out.filters.contains_key("contentKind"));
}
#[test]
fn two_axes_with_nothing_in_common_widen_rather_than_narrow() {
// No music kind is `serial`. The pre-filter drops out; the post-stamp
// check still enforces both terms.
let out = expand_routing_filters(&q(&[
("domain", &["music"][..]),
("workForm", &["serial"][..]),
]))
.unwrap();
assert!(!out.filters.contains_key("contentKind"));
}
#[test]
fn stamp_fills_in_and_never_overwrites() {
let mut f: BTreeMap<String, String> = BTreeMap::new();
f.insert("contentKind".into(), "episode".into());
stamp_domain(&mut f);
stamp_work_form(&mut f);
assert_eq!(f.get("domain").map(String::as_str), Some("screen"));
assert_eq!(f.get("workForm").map(String::as_str), Some("serial"));
// A pack writer that already knows its context keeps it.
let mut p: BTreeMap<String, String> = BTreeMap::new();
p.insert("contentKind".into(), "pack".into());
p.insert("domain".into(), "music".into());
p.insert("workForm".into(), "standalone".into());
stamp_domain(&mut p);
stamp_work_form(&mut p);
assert_eq!(p.get("domain").map(String::as_str), Some("music"));
assert_eq!(p.get("workForm").map(String::as_str), Some("standalone"));
// No contentKind → nothing invented.
let mut n: BTreeMap<String, String> = BTreeMap::new();
stamp_domain(&mut n);
stamp_work_form(&mut n);
assert!(n.get("domain").is_none());
assert!(n.get("workForm").is_none());
}
}
+515 -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)))
}
/// 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
/// id is all-hex by construction; odd-length ids are left-padded with a `0`
/// 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.
pub const CARD_LOCATOR_CODEC: u64 = 0x1007;
/// Digest byte ceiling for a card locator, matching meta-share's
/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash.
/// Digest byte ceiling for the locators that still carry one — `yt-video`
/// (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;
/// 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
/// here (see `meta-gateway/docs/others/card-tier-search.md` §9).
///
/// Returns `None` when `source` is empty or the digest would overflow the
/// multihash budget — the caller drops that card.
/// # No length ceiling, deliberately
///
/// 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> {
if source.is_empty() || id.is_empty() {
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);
digest.extend_from_slice(source_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]
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))
}
/// 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
/// 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> {
if kind.is_empty() || id.is_empty() {
return None;
@@ -423,6 +545,86 @@ fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
/// `beetswap::incoming_stream`). The `.nzb` manifest travels separately, as an
/// ordinary sha2-256 IPFS cid in the record's `manifest` field. See
/// `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md` §5.
/// Custom multicodec for the **url-locator** family.
///
/// A CID whose *bytes are a URL string* rather than a hash of the file. It lets
/// a record reference externally-hosted bytes without storing a raw URL field:
/// the URL is wrapped in a CID, so it flows through every cid-shaped code path
/// (`cids/*` membership, `/api/file/{cid}/raw`, the reverse index) exactly like
/// a content hash — honouring `METADATA_KEYS.md` rule #5, "every byte reference
/// is a `cid`, never a URL".
///
/// Contrast `ext-play` (`0x1009`), which is wire-identical apart from the
/// codec: `0x1006` means **"fetch these bytes once and seed them"**, while
/// `0x1009` means "open this; there are no bytes for us, ever".
///
/// ⚠ **Never route a url-CID through a fixed-64-byte multihash type.** An
/// archive download URL runs to ~75 bytes, past the 62-ish ceiling of
/// meta-share's `CidGeneric<64>` bitswap alias. That is fine — it is never
/// bitswap *content*, and the `/raw` dispatch decodes it off the raw string
/// before the fixed-size codec dispatch runs — but a consumer that parses it as
/// an `MsCid` will reject a perfectly good locator.
pub const URL_LOCATOR_CODEC: u64 = 0x1006;
/// Encode an `http(s)` URL as a `0x1006` identity-multihash CIDv1.
///
/// The identity multihash (`0x00`) means "the digest *is* the payload", so
/// [`decode_url_cid`] recovers the URL verbatim with zero I/O.
///
/// `None` for anything that is not an absolute http(s) URL — a relative or
/// malformed value would encode fine and then fail unresolvably at play time,
/// which is a far more expensive way to find out.
pub fn compute_url_cid(url: &str) -> Option<String> {
let url = url.trim();
if !(url.starts_with("http://") || url.starts_with("https://")) {
return None;
}
Some(identity_locator_cid(URL_LOCATOR_CODEC, url.as_bytes()))
}
/// The artwork locator for a `cover` / `photo` field.
///
/// A thin alias over [`compute_url_cid`] that names the *intent* at the call
/// sites: artwork is referenced by a locator cid, never by a raw `*_url` field
/// (METADATA_KEYS rule #5). `None` for a non-http(s) URL, in which case the
/// caller writes no artwork field at all — better a card the gate hides than
/// one carrying an unresolvable reference.
pub fn artwork_locator(url: &str) -> Option<String> {
compute_url_cid(url)
}
/// Recover the URL from a `url` locator cid — the exact inverse of
/// [`compute_url_cid`], with no I/O.
///
/// `None` if `cid` is not a well-formed `0x1006` identity-multihash CIDv1 over
/// an http(s) URL.
pub fn decode_url_cid(cid: &str) -> Option<String> {
let bytes = base32_lower_decode(cid.strip_prefix('b')?)?;
let mut i = 0usize;
let mut next = || -> Option<u64> {
let mut n = 0u64;
let mut shift = 0u32;
loop {
let b = *bytes.get(i)?;
i += 1;
n |= u64::from(b & 0x7f) << shift;
if b & 0x80 == 0 {
return Some(n);
}
shift += 7;
if shift > 63 {
return None;
}
}
};
if next()? != 1 || next()? != URL_LOCATOR_CODEC || next()? != 0x00 {
return None;
}
let len = next()? as usize;
let url = std::str::from_utf8(bytes.get(i..i + len)?).ok()?;
(url.starts_with("http://") || url.starts_with("https://")).then(|| url.to_string())
}
pub const NZB_POSTING_CODEC: u64 = 0x1003;
/// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs.
@@ -987,19 +1189,88 @@ mod tests {
assert_eq!(&digest[5..], b"tv:95479");
}
/// 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.
/// Empty inputs are rejected rather than encoded as an ambiguous locator.
#[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("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.
let fits = "x".repeat(CARD_LOCATOR_MAX_DIGEST - 5); // varint(4) + "tmdb"
assert!(compute_card_cid("tmdb", &fits).is_some());
}
/// ⚠ A card locator has **no length ceiling**, and this is the test that
/// says so on purpose — it used to assert the opposite.
///
/// 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 -----------------------
@@ -1074,3 +1345,229 @@ mod tests {
assert_eq!(&decoded[5..], &expect);
}
}
/// RFC 4648 base32 (lowercase, no padding) → bytes. Inverse of
/// [`base32_lower_no_padding`]; used by the identity-multihash decoders.
fn base32_lower_decode(s: &str) -> Option<Vec<u8>> {
let mut acc = 0u16;
let mut bits = 0u32;
let mut out = Vec::with_capacity(s.len() * 5 / 8);
for c in s.bytes() {
let v = match c {
b'a'..=b'z' => c - b'a',
b'2'..=b'7' => c - b'2' + 26,
_ => return None,
};
acc = (acc << 5) | u16::from(v);
bits += 5;
if bits >= 8 {
bits -= 8;
out.push((acc >> bits) as u8);
}
}
Some(out)
}
#[cfg(test)]
mod url_locator_wire_tests {
use super::*;
/// Decode a base32-lower multibase string back to bytes, so the tests can
/// assert the wire layout rather than an opaque string.
fn decode(cid: &str) -> Vec<u8> {
const A: &str = "abcdefghijklmnopqrstuvwxyz234567";
let mut bits = String::new();
for c in cid.strip_prefix('b').expect("multibase b").chars() {
let i = A.find(c).expect("base32 alphabet");
bits.push_str(&format!("{i:05b}"));
}
bits.as_bytes()
.chunks(8)
.filter(|c| c.len() == 8)
.map(|c| u8::from_str_radix(std::str::from_utf8(c).unwrap(), 2).unwrap())
.collect()
}
fn read_varint(data: &[u8], at: &mut usize) -> u64 {
let (mut v, mut shift) = (0u64, 0);
loop {
let b = data[*at];
*at += 1;
v |= ((b & 0x7f) as u64) << shift;
if b & 0x80 == 0 {
return v;
}
shift += 7;
}
}
/// Pins the on-the-wire layout `METADATA_KEYS.md` §2.1 specifies. meta-share
/// decodes exactly this; a change here is a change to a cross-binary
/// contract, not an implementation detail.
#[test]
fn the_wire_layout_matches_the_specified_encoding() {
let url = "https://archive.org/download/item/track.flac";
let cid = compute_url_cid(url).expect("cid");
let raw = decode(&cid);
let mut i = 0;
assert_eq!(raw[i], 0x01, "CIDv1");
i += 1;
assert_eq!(read_varint(&raw, &mut i), URL_LOCATOR_CODEC);
assert_eq!(raw[i], 0x00, "identity multihash");
i += 1;
let len = read_varint(&raw, &mut i) as usize;
assert_eq!(len, url.len());
assert_eq!(
std::str::from_utf8(&raw[i..i + len]).unwrap(),
url,
"the digest IS the payload — decoding must recover the URL verbatim"
);
}
/// ⚠ A real archive URL is ~75 bytes, past the ~62-byte ceiling of a
/// fixed-64 multihash type. That is expected and fine (the locator is never
/// bitswap content), but it is the thing that trips a consumer parsing it
/// as an `MsCid`, so the case is pinned deliberately rather than avoided.
#[test]
fn a_long_archive_url_still_encodes() {
let url = "https://archive.org/download/gd1977-05-08eaton/gd77-05-08eaton-d1t01.ogg";
assert!(url.len() > 62, "fixture must exercise the over-64 case");
let cid = compute_url_cid(url).expect("cid");
let raw = decode(&cid);
let mut i = 1;
read_varint(&raw, &mut i);
i += 1;
let len = read_varint(&raw, &mut i) as usize;
assert_eq!(std::str::from_utf8(&raw[i..i + len]).unwrap(), url);
}
/// Two URLs are two locators — a url-CID is not byte-derivable, so it is an
/// address this one record advertises, never a content identity.
#[test]
fn distinct_urls_yield_distinct_locators() {
assert_ne!(
compute_url_cid("https://example.com/a.flac"),
compute_url_cid("https://example.com/b.flac")
);
}
/// A relative or malformed value would encode fine and then fail
/// unresolvably at play time — the expensive place to find out.
#[test]
fn only_absolute_http_urls_are_accepted() {
assert!(compute_url_cid("/download/item/track.flac").is_none());
assert!(compute_url_cid("ftp://example.com/x").is_none());
assert!(compute_url_cid("").is_none());
assert!(compute_url_cid("archive.org/download/x").is_none());
assert!(compute_url_cid("http://example.com/x").is_some());
}
}
#[cfg(test)]
mod roundtrip_tests {
use super::*;
/// Long URLs are the whole reason this locator exists (they overflow the
/// fixed-size multihash buffer the `cid` crate uses), so the round trip is
/// pinned at a realistic Cover Art Archive length, not a toy one.
#[test]
fn encode_then_decode_is_identity() {
for url in [
"https://coverartarchive.org/release/aa997ea0-2936-40bd-884d-3af8a0e064dc/12345-500.jpg",
"https://archive.org/services/img/gd1977-05-08",
"http://example.test/a.png",
] {
let cid = compute_url_cid(url).expect("encodes");
assert!(cid.starts_with('b'));
assert_eq!(decode_url_cid(&cid).as_deref(), Some(url), "round trip: {url}");
}
}
#[test]
fn a_non_locator_cid_does_not_decode() {
// A card locator (0x1007) must not be mistaken for a url locator.
let card = crate::hash::compute_card_cid("musicbrainz", "abc").unwrap();
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);
}
}
+7
View File
@@ -21,13 +21,16 @@
/// `metacore-app` service the AppStore feeder apps ship on the shared network.
pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
pub mod budget;
pub mod cache;
pub mod common;
pub mod config;
pub mod domain;
pub mod enrich;
pub mod filename_meta;
pub mod hash;
pub mod lang;
pub mod licence;
pub mod meta_core;
pub mod plugin;
pub mod query;
@@ -35,15 +38,19 @@ pub mod query_eval;
pub mod serve;
pub mod types;
pub use budget::RateBudget;
pub use config::{ConfigField, ConfigSchema, FieldKind};
pub use licence::licence_from_url;
pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher};
pub use meta_core::FeederStore;
pub use plugin::{
upstream_id_field, ConfigError, FeederPlugin, HashKind, HashOutcome, PluginRegistry,
RedeemClaim,
};
pub use query::{GatewayQuery, GatewaySearchEvent, GatewayWireError, Negation, RangeFilter};
pub use serve::{
configure_plugins, router, serve_feeders, ComputeRequest, ComputeResponse, HashKindDto,
HealthResponse, ManifestResponse, OutcomeDto, PluginManifest, QueryRequest, QueryResponse,
RedeemsResponse,
};
pub use types::{ByteStream, DiscoveryId, DiscoveryRecord, GatewayError, Hash, PluginHealth};
+31
View File
@@ -0,0 +1,31 @@
//! Licence-URL → SPDX-style identifier.
//!
//! Generic across upstreams: Internet Archive, Jamendo and Wikimedia Commons
//! all publish a Creative Commons *URL*, while `METADATA_KEYS.md` requires an
//! SPDX-style identifier — a consumer comparing licences should not be doing
//! string surgery on `http://` vs `https://` and a trailing slash.
/// `METADATA_KEYS.md` requires an SPDX-style identifier rather than a URL,
/// because a consumer comparing licences should not be doing string surgery on
/// `http://` vs `https://` and a trailing slash.
pub fn licence_from_url(url: &str) -> Option<String> {
let u = url.trim().trim_end_matches('/').to_ascii_lowercase();
if u.is_empty() {
return None;
}
if u.contains("publicdomain/zero") {
return Some("CC0-1.0".to_string());
}
if u.contains("publicdomain/mark") {
return Some("PublicDomain".to_string());
}
// .../licenses/<variant>/<version>[/<jurisdiction>]
let rest = u.split("/licenses/").nth(1)?;
let mut parts = rest.split('/');
let variant = parts.next()?.to_ascii_uppercase();
let version = parts.next().unwrap_or("4.0");
if variant.is_empty() {
return None;
}
Some(format!("CC-{variant}-{version}"))
}
+94
View File
@@ -17,6 +17,7 @@ use std::path::Path;
use async_trait::async_trait;
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
// Re-exported so plugins can `use meta_feeder_sdk::plugin::GatewayQuery` (the
// gateway's original `plugin.rs` re-exported it the same way).
@@ -95,6 +96,66 @@ pub enum HashKind {
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`].
#[derive(Debug, thiserror::Error)]
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
/// `GET /config/schema` and renders a generic form from it — the gateway and
/// UI carry no per-plugin field knowledge. Default: no configuration.
@@ -255,4 +330,23 @@ mod tests {
assert!(msg.contains("scihub"));
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`]
//! - `GET /fetch/{upstream_id}/{record_id}` → streamed 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
//!
@@ -21,7 +32,7 @@
//! documented follow-up — those outcomes are metadata-only anyway
//! (`bytes: None`), so they never ride this path.
use std::collections::HashMap;
use std::collections::{BTreeMap, HashMap};
use std::net::SocketAddr;
use std::path::{Path as FsPath, PathBuf};
use std::sync::Arc;
@@ -41,7 +52,7 @@ use serde_json::Value;
use tracing::{info, warn};
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::types::{DiscoveryRecord, GatewayError, PluginHealth};
@@ -94,6 +105,22 @@ pub struct PluginManifest {
pub id: String,
pub served_file_types: 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
@@ -224,6 +251,17 @@ impl AppState {
/// status semantics the gateway's `RemoteFeederPlugin` will translate back
/// into a `GatewayError` on the core side.
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 {
GatewayError::NotFound => (StatusCode::NOT_FOUND, "not found".to_string()),
GatewayError::RateLimited { retry_after_s } => (
@@ -251,6 +289,8 @@ async fn manifest(State(state): State<AppState>) -> Json<ManifestResponse> {
.iter()
.map(|s| s.to_string())
.collect(),
package: p.package().map(str::to_string),
redeems: p.redeems(),
})
.collect();
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> {
let mut entries: Vec<PluginHealthEntry> = state
.plugins
@@ -492,6 +542,7 @@ pub fn router(
.route("/compute", post(compute))
.route("/fetch/:upstream_id/:record_id", get(fetch))
.route("/blob/:upstream_id/:cid", get(blob))
.route("/redeems", get(redeems))
.route("/config", get(config_page))
.route("/config/schema", get(config_schema))
.route("/config/values", get(config_values_get).put(config_values_put))
@@ -541,6 +592,9 @@ struct DegradedPlugin {
reason: String,
file_types: &'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
/// serves its config form — otherwise the dashboard shows "no configuration
/// required" and the operator can never fix the very config that's missing.
@@ -586,6 +640,10 @@ impl FeederPlugin for DegradedPlugin {
self.content_kinds
}
fn package(&self) -> Option<&'static str> {
self.package
}
fn config_schema(&self) -> crate::config::ConfigSchema {
self.schema.clone()
}
@@ -608,6 +666,7 @@ pub fn configure_plugins(
// borrow across `configure(&mut self)` / the move into `out`.
let file_types = plugin.served_file_types();
let content_kinds = plugin.served_content_kinds();
let package = plugin.package();
// Preserve the config surface so a degraded stand-in still serves the
// real config form (see DegradedPlugin.schema).
let schema = plugin.config_schema();
@@ -627,6 +686,7 @@ pub fn configure_plugins(
reason,
file_types,
content_kinds,
package,
schema: schema.clone(),
})
};
@@ -759,4 +819,118 @@ mod tests {
assert!(matches!(events[1], GatewaySearchEvent::EnrichPatch { .. }));
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());
}
}