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] [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:
+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::path::Path;
use std::sync::Arc; use std::sync::Arc;
use redb::{Database, ReadableTableMetadata, TableDefinition}; use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};
pub const CACHE_FILENAME: &str = "cache.redb"; 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 /// OpenSubtitles search results keyed by `"<tmdb_id>\x01<lang3>"`. Value is the
/// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel. /// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel.
const OPENSUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("opensubtitles"); 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 /// Small durable plugin scratch values that must survive a restart but aren't
/// content-derived — today, the torznab plugin's Prowlarr indexer-set /// content-derived — today, the torznab plugin's Prowlarr indexer-set
/// fingerprint, so a set that changed *while the feeder was down* is still /// 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(SUBTITLES_TABLE)?;
tx.open_table(OPENSUBTITLES_TABLE)?; tx.open_table(OPENSUBTITLES_TABLE)?;
tx.open_table(MISC_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()?; tx.commit()?;
} }
Ok(Self { db: Arc::new(db) }) Ok(Self { db: Arc::new(db) })
@@ -120,6 +167,47 @@ impl MidhashCache {
str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE); 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!( str_table_accessors!(
/// Read a durable plugin scratch value (see [`MISC_TABLE`]). /// Read a durable plugin scratch value (see [`MISC_TABLE`]).
get_misc, get_misc,
@@ -351,3 +439,68 @@ mod tests {
assert_eq!(cache.entry_count().unwrap(), 2); 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()); 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))) 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;
@@ -423,6 +545,86 @@ fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
/// `beetswap::incoming_stream`). The `.nzb` manifest travels separately, as an /// `beetswap::incoming_stream`). The `.nzb` manifest travels separately, as an
/// ordinary sha2-256 IPFS cid in the record's `manifest` field. See /// ordinary sha2-256 IPFS cid in the record's `manifest` field. See
/// `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md` §5. /// `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; pub const NZB_POSTING_CODEC: u64 = 0x1003;
/// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs. /// 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"); 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 -----------------------
@@ -1074,3 +1345,229 @@ mod tests {
assert_eq!(&decoded[5..], &expect); 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. /// `metacore-app` service the AppStore feeder apps ship on the shared network.
pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000"; pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
pub mod budget;
pub mod cache; pub mod cache;
pub mod common; pub mod common;
pub mod config; pub mod config;
pub mod domain;
pub mod enrich; pub mod enrich;
pub mod filename_meta; pub mod filename_meta;
pub mod hash; pub mod hash;
pub mod lang; pub mod lang;
pub mod licence;
pub mod meta_core; pub mod meta_core;
pub mod plugin; pub mod plugin;
pub mod query; pub mod query;
@@ -35,15 +38,19 @@ pub mod query_eval;
pub mod serve; pub mod serve;
pub mod types; pub mod types;
pub use budget::RateBudget;
pub use config::{ConfigField, ConfigSchema, FieldKind}; pub use config::{ConfigField, ConfigSchema, FieldKind};
pub use licence::licence_from_url;
pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher}; 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};
+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 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());
}
} }