rework plugin

This commit is contained in:
Worph
2026-08-24 16:33:06 +02:00
parent fadfebbf9b
commit b823654b9a
6 changed files with 586 additions and 60 deletions
+16
View File
@@ -53,6 +53,13 @@ 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");
/// 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
/// detected as a change on the next boot. Keys are plugin-namespaced
/// (`prowlarr:indexer-set`). redb creates the table on open, so an older
/// `.redb` file gains it transparently.
const MISC_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("misc");
/// Per-plugin midhash cache. Cheap to clone (the inner `Database` is shared /// Per-plugin midhash cache. Cheap to clone (the inner `Database` is shared
/// via `Arc`). /// via `Arc`).
@@ -105,6 +112,7 @@ impl MidhashCache {
tx.open_table(TMDB_PRINCIPAL_TOPN_TABLE)?; tx.open_table(TMDB_PRINCIPAL_TOPN_TABLE)?;
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.commit()?; tx.commit()?;
} }
Ok(Self { db: Arc::new(db) }) Ok(Self { db: Arc::new(db) })
@@ -112,6 +120,14 @@ impl MidhashCache {
str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE); str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE);
str_table_accessors!(
/// Read a durable plugin scratch value (see [`MISC_TABLE`]).
get_misc,
/// Write a durable plugin scratch value (see [`MISC_TABLE`]).
put_misc,
MISC_TABLE,
);
pub fn entry_count(&self) -> Result<u64, redb::Error> { pub fn entry_count(&self) -> Result<u64, redb::Error> {
let tx = self.db.begin_read()?; let tx = self.db.begin_read()?;
let table = tx.open_table(MIDHASH_TABLE)?; let table = tx.open_table(MIDHASH_TABLE)?;
+1 -1
View File
@@ -107,7 +107,7 @@ impl EnrichmentConfig {
filename_parser_url: env_nonempty("FILENAME_PARSER_URL") filename_parser_url: env_nonempty("FILENAME_PARSER_URL")
.or_else(|| derive_sidecar_url("filename-parser")), .or_else(|| derive_sidecar_url("filename-parser")),
tmdb_url: env_nonempty("TMDB_PLUGIN_URL").or_else(|| derive_sidecar_url("tmdb")), tmdb_url: env_nonempty("TMDB_PLUGIN_URL").or_else(|| derive_sidecar_url("tmdb")),
tmdb_token: env_nonempty("TMDB_TOKEN").or_else(|| env_nonempty("PROWLARR_TMDB_TOKEN")), tmdb_token: env_nonempty("TMDB_TOKEN").or_else(|| env_nonempty("TORZNAB_TMDB_TOKEN")),
tmdb_language: env_nonempty("TMDB_LANGUAGE").unwrap_or_else(|| "en-US".to_string()), tmdb_language: env_nonempty("TMDB_LANGUAGE").unwrap_or_else(|| "en-US".to_string()),
gateway_peer: env_nonempty("META_GATEWAY_PEER_ID") gateway_peer: env_nonempty("META_GATEWAY_PEER_ID")
.or_else(|| env_nonempty("HOSTNAME")) .or_else(|| env_nonempty("HOSTNAME"))
+490 -34
View File
@@ -183,9 +183,9 @@ const NZB_RELEASE_MAX_DIGEST: usize = 64;
/// ///
/// Returns `None` when the id isn't hex or the digest would overflow the /// Returns `None` when the id isn't hex or the digest would overflow the
/// multihash budget (an oversized host) — the caller drops that row. /// multihash budget (an oversized host) — the caller drops that row.
pub fn compute_nzb_release_cid(host: &str, release_id: &str) -> Option<String> { pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option<String> {
let id_bytes = decode_hex_id(release_id)?; let id_bytes = decode_hex_id(release_id)?;
let host_b = host.as_bytes(); let host_b = api_base.as_bytes();
// identity-multihash digest = varint(host_len) ‖ host ‖ id_bytes // identity-multihash digest = varint(host_len) ‖ host ‖ id_bytes
let mut digest = Vec::with_capacity(2 + host_b.len() + id_bytes.len()); let mut digest = Vec::with_capacity(2 + host_b.len() + id_bytes.len());
@@ -227,6 +227,266 @@ fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
Some(out) Some(out)
} }
/// Custom multicodec for a **card locator** — the identity of a *work* (a
/// series, a film) as published by a metadata bridge, rather than of any
/// bytes. MetaMesh-private, the next free slot after `0x1006` url. (`0x1004`
/// nzb-release-hash stays **retired — do not reuse**. `0x1003` was retired too,
/// but has since been **revived** for the self-scanned Usenet posting identity —
/// see [`NZB_POSTING_CODEC`] and `METADATA_KEYS.md` §2.)
///
/// This is the first CID family in the platform where **no bytes exist
/// anywhere, ever** — `0x1005`/`0x1006` are locators too, but both eventually
/// resolve to bytes (NNTP, HTTP), and `0x1000` midhash256 is a digest over
/// real file bytes. A card is pure identity: the record *is* the payload, and
/// it lives in meta-core. Consequences, all already handled:
///
/// * Bitswap seeding is gated on `HashKind::Sha2_256` (gateway invariant 6),
/// so a card is never offered to the swarm — nothing to seed.
/// * meta-share's byte path 404s for it, which is correct.
/// * It ranks `RANK_LOCATOR` (5), so it can never outrank a real digest for
/// `canonical_cid` — which is what makes the future "one card, two CIDs"
/// 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.
const CARD_LOCATOR_MAX_DIGEST: usize = 64;
/// Encode a self-describing **card locator** CID from a metadata source and
/// that source's own id for the work (`("tmdb", "tv:95479")`,
/// `("mal", "52991")`). The multihash is *identity* (`0x00`) and its digest
/// embeds the locator as `varint(source_len) ‖ source ‖ id`, both UTF-8 —
/// exactly the shape [`compute_nzb_release_cid`] uses, minus the hex decode
/// (a card id is arbitrary text, not a hex release id).
///
/// **The CID is a pure function of `(source, id)`**, which is the whole point:
/// any peer that knows a tmdb id can derive the card's address offline and
/// look it up locally, with no discovery round-trip. It also means the same
/// card discovered by two different peers converges on one meta-core record
/// for free, via the `cids/<bareCid>` reverse index.
///
/// Source-namespaced: the same work on TMDB and on MyAnimeList yields
/// different cids. Merging those two into one record is a meta-core alias
/// 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.
pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
if source.is_empty() || id.is_empty() {
return None;
}
let source_b = source.as_bytes();
let id_b = id.as_bytes();
// identity-multihash digest = varint(source_len) ‖ source ‖ id
let mut digest = Vec::with_capacity(2 + source_b.len() + id_b.len());
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());
wire.push(0x01);
write_pb_varint(CARD_LOCATOR_CODEC, &mut wire);
wire.push(0x00); // multihash code: identity
write_pb_varint(digest.len() as u64, &mut wire);
wire.extend_from_slice(&digest);
Some(format!("b{}", base32_lower_no_padding(&wire)))
}
// ---------------------------------------------------------------------------
// Delegated-playback locators — `yt-video` (0x1008) and `ext-play` (0x1009)
// ---------------------------------------------------------------------------
//
// These two are the limit case *beyond* `card` (0x1007). `card` addresses a
// work with no bytes anywhere; these address a **rendition whose bytes exist
// but are permanently someone else's**. An external player renders the media,
// nothing is transported, nothing is content-addressed, nothing is re-seedable.
//
// See `docs/cid-formats.md` §7 and
// `docs/study/listenbrainz-youtube-playback-tier-2026-08-21.md`.
/// Custom multicodec for a **YouTube rendition** — delegated playback.
///
/// Its own codec rather than a `url` (`0x1006`) for three reasons, in
/// descending order of how much trouble the alternative causes:
///
/// * **`0x1006` means "fetch these bytes once and seed them".** Point a `url`
/// locator at a `watch` page and meta-share does exactly that: fetches the
/// HTML, chunks it, seeds it, and links the resulting content CID back onto
/// the record *as the file*. That is the MSR1 poisoning shape, except
/// permanent and replicated across peers. A resolved `googlevideo` URL is no
/// better — time-limited and IP-bound, so it is not a locator at all.
/// * **The id is canonical, a URL is not.** `youtube.com` / `music.youtube.com`
/// / `youtu.be` all address the same video, and query parameters are tracking
/// noise, so the same rendition would mint several different CIDs. Being a
/// pure function of `(kind, id)`, this codec makes two peers that resolve the
/// same track converge on one meta-core record for free, via the
/// `cids/<bareCid>` reverse index — the property that makes [`compute_card_cid`]
/// useful, for the same reason.
/// * **It is tiny** — 11 bytes of id plus a short kind prefix.
///
/// ⚠ **An immutable id is not durable playback.** Takedowns, geo-blocks and
/// embed-disabled uploads all kill a reference whose CID stays valid forever —
/// the same rot class as [`NZB_RELEASE_CODEC`]. The mitigation is at the
/// *record* level (carry several ranked references per recording so a client
/// can fall through), never at the CID level.
pub const YT_VIDEO_CODEC: u64 = 0x1008;
/// Custom multicodec for an **external page to open, never fetch**.
///
/// Wire-identical to `url` (`0x1006`) apart from the codec, and that difference
/// is the entire contract: `0x1006` resolves cache-through and lands in the
/// blockstore, `0x1009` **must be refused by the byte path** and only ever
/// surfaces as an "Open on …↗" affordance. Use it for anything worth linking
/// but not worth a per-provider codec (Bandcamp, Jamendo, an arbitrary stream
/// page).
pub const EXT_PLAY_CODEC: u64 = 0x1009;
/// Encode a [`YT_VIDEO_CODEC`] locator from a `kind` and a bare YouTube id.
///
/// `kind` is `"video"` | `"playlist"` | `"channel"`; `id` is the bare
/// identifier (`4D7u5KF7SP8`, `OLAK5uy_…`, `UCRr1xG_2WIDs18a6cIiCxeA`) — **not**
/// a URL, and not a `watch?v=` fragment. The framing is byte-identical to
/// [`compute_card_cid`] on purpose, so each of the seven rank/decode
/// implementations extends a parser it already wrote rather than learning a
/// novel shape.
///
/// Returns `None` on an empty field or a digest past the multihash budget —
/// the caller drops that reference.
pub fn compute_yt_video_cid(kind: &str, id: &str) -> Option<String> {
kinded_locator_cid(YT_VIDEO_CODEC, kind, id)
}
/// Encode an [`EXT_PLAY_CODEC`] locator from an `http(s)` URL.
///
/// The scheme check is not cosmetic: it is what keeps a `javascript:` or
/// `data:` payload from ever reaching a client that will hand this string to
/// an anchor or a new tab.
pub fn compute_ext_play_cid(url: &str) -> Option<String> {
if !(url.starts_with("http://") || url.starts_with("https://")) {
return None;
}
let digest = url.as_bytes();
if digest.is_empty() || digest.len() > CARD_LOCATOR_MAX_DIGEST {
return None;
}
Some(identity_locator_cid(EXT_PLAY_CODEC, digest))
}
/// Shared encoder for the `varint(len(kind)) ‖ kind ‖ id` locator framing used
/// by [`compute_card_cid`] and [`compute_yt_video_cid`].
fn kinded_locator_cid(codec: u64, kind: &str, id: &str) -> Option<String> {
if kind.is_empty() || id.is_empty() {
return None;
}
let kind_b = kind.as_bytes();
let id_b = id.as_bytes();
let mut digest = Vec::with_capacity(2 + kind_b.len() + id_b.len());
write_pb_varint(kind_b.len() as u64, &mut digest);
digest.extend_from_slice(kind_b);
digest.extend_from_slice(id_b);
if digest.len() > CARD_LOCATOR_MAX_DIGEST {
return None;
}
Some(identity_locator_cid(codec, &digest))
}
/// CIDv1 with an *identity* multihash:
/// `[version=0x01][codec varint][mh=0x00][len varint][digest]`, base32lower.
fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
wire.push(0x01);
write_pb_varint(codec, &mut wire);
wire.push(0x00); // multihash code: identity
write_pb_varint(digest.len() as u64, &mut wire);
wire.extend_from_slice(digest);
format!("b{}", base32_lower_no_padding(&wire))
}
/// Custom multicodec for a **Usenet posting** identity (`nzb-posting`),
/// minted from the article Message-IDs of a release we scanned ourselves.
///
/// Unlike [`NZB_RELEASE_CODEC`] (`0x1005`), this is **not** a locator: it
/// embeds no indexer host, so any peer with a plain NNTP provider can fetch the
/// articles — no `IndexerCred`, no `t=get` grab. That portability is the whole
/// point of running our own header scanner.
///
/// It is also **not** fetchable by cid: the digest is over the Message-ID set,
/// not over any block's bytes, so bitswap can never satisfy a want for it (the
/// receiver derives a block's cid by hashing the block — see
/// `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.
pub const NZB_POSTING_CODEC: u64 = 0x1003;
/// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs.
///
/// # The normalisation rule IS the dedup contract
///
/// Mirrored verbatim in meta-share (`nzb/manifest.rs`), which re-derives this
/// digest from the fetched manifest and rejects a mismatch. **Any change here
/// must land on both sides together** — a drift silently makes every posting
/// unplayable. The rule:
///
/// 1. Trim each id, and strip a surrounding `<…>` pair if present (NZB
/// `<segment>` bodies carry the bare form, but be defensive — NNTP `BODY`
/// re-adds the brackets).
/// 2. Drop empties.
/// 3. Deduplicate, then **sort** lexicographically by bytes.
/// 4. Join with `\n` and SHA-256 the result.
///
/// Case is **preserved**: a Message-ID's local part is case-sensitive per RFC
/// 5322, so lowercasing could collide two genuinely distinct articles.
///
/// Sorting is what makes the cid survive segment/file reordering between two
/// NZBs of the same posting — the reason we hash the *id set* and never the raw
/// `.nzb` bytes (the retired `0x1004` did the latter and broke dedup; see
/// meta-share `nzb/mod.rs`).
///
/// **v1 scope:** every segment counts, par2 volumes included. Two NZBs for one
/// posting that disagree about whether to bundle par2 therefore mint different
/// cids. Acceptable while we are the only producer of these manifests; if a
/// second producer ever appears, the fix is to filter par2 subjects here *and*
/// in meta-share's mirror — do not do it on one side only.
pub fn compute_nzb_posting_cid<S: AsRef<str>>(message_ids: &[S]) -> String {
let mut ids: Vec<&str> = message_ids
.iter()
.map(|s| {
let t = s.as_ref().trim();
t.strip_prefix('<')
.and_then(|r| r.strip_suffix('>'))
.unwrap_or(t)
})
.filter(|s| !s.is_empty())
.collect();
ids.sort_unstable();
ids.dedup();
let mut hasher = Sha256::new();
for (i, id) in ids.iter().enumerate() {
if i > 0 {
hasher.update(b"\n");
}
hasher.update(id.as_bytes());
}
let digest: [u8; 32] = hasher.finalize().into();
// CIDv1: [version=0x01][codec varint][mh code=0x12 sha2-256][len=32][digest]
let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
wire.push(0x01);
write_pb_varint(NZB_POSTING_CODEC, &mut wire);
wire.push(0x12); // multihash code: sha2-256
write_pb_varint(digest.len() as u64, &mut wire);
wire.extend_from_slice(&digest);
format!("b{}", base32_lower_no_padding(&wire))
}
/// Compute a standard IPFS CIDv1 over `bytes`. Output matches kubo's /// Compute a standard IPFS CIDv1 over `bytes`. Output matches kubo's
/// `ipfs add --cid-version=1 --raw-leaves=true --chunker=size-262144 /// `ipfs add --cid-version=1 --raw-leaves=true --chunker=size-262144
/// --hash=sha2-256 <file>`. /// --hash=sha2-256 <file>`.
@@ -513,38 +773,6 @@ mod tests {
); );
} }
#[test]
fn nzb_release_cid_wire_and_properties() {
let cid =
compute_nzb_release_cid("api.nzb.life", "485c52f078b580667a01ac34a1cef6c2").unwrap();
assert!(cid.starts_with('b'));
// Deterministic.
assert_eq!(
cid,
compute_nzb_release_cid("api.nzb.life", "485c52f078b580667a01ac34a1cef6c2").unwrap()
);
// Host-namespaced: same id, different host → different cid.
assert_ne!(
cid,
compute_nzb_release_cid("api.nzbgeek.info", "485c52f078b580667a01ac34a1cef6c2")
.unwrap()
);
// Explicit self-describing wire form:
// [01][85 20][00][len][varint(host_len) ‖ host ‖ hex-decoded id].
let mut digest = Vec::new();
write_pb_varint("api.nzb.life".len() as u64, &mut digest);
digest.extend_from_slice(b"api.nzb.life");
digest.extend_from_slice(&decode_hex_id("485c52f078b580667a01ac34a1cef6c2").unwrap());
let mut want = vec![0x01, 0x85, 0x20, 0x00];
write_pb_varint(digest.len() as u64, &mut want);
want.extend_from_slice(&digest);
assert_eq!(cid, format!("b{}", base32_lower_no_padding(&want)));
// Distinct codec from the bt-file codecs in the 0x100x neighborhood.
assert_ne!(cid, compute_bt_v1_file_cid(&[0xAB; 20], 0));
// An oversized host overflows the 64-byte digest budget → None.
assert!(compute_nzb_release_cid(&"h".repeat(80), "485c52f0").is_none());
}
#[test] #[test]
fn base32_known_vectors() { fn base32_known_vectors() {
assert_eq!(base32_lower_no_padding(b""), ""); assert_eq!(base32_lower_no_padding(b""), "");
@@ -555,6 +783,82 @@ mod tests {
assert_eq!(base32_lower_no_padding(b"foobar"), "mzxw6ytboi"); assert_eq!(base32_lower_no_padding(b"foobar"), "mzxw6ytboi");
} }
#[test]
fn yt_video_cid_matches_golden_vector() {
// Pinned by `/cid-rank-vectors.json`'s `yt-video-locator` entry. The id
// is the real resolver output for Daft Punk / "Get Lucky" measured in
// the delegated-playback study §4.3.
assert_eq!(
compute_yt_video_cid("video", "4D7u5KF7SP8").unwrap(),
"bagecaaarav3gszdfn42ein3vgvfumn2tka4a"
);
}
#[test]
fn yt_video_cid_framing_matches_card_locator() {
// The two codecs share the `varint(len(kind)) || kind || id` framing on
// purpose, so every decoder extends a parser it already wrote. If this
// ever fails, the two shapes have drifted and §7.2 of docs/cid-formats.md
// is no longer true.
let yt = compute_yt_video_cid("tmdb", "tv:95479").unwrap();
let card = compute_card_cid("tmdb", "tv:95479").unwrap();
// Same digest, different codec => same length, differing only in the
// codec varint region.
assert_ne!(yt, card);
assert_eq!(yt.len(), card.len());
}
#[test]
fn yt_video_cid_is_a_pure_function_of_kind_and_id() {
// Convergence is the reason this is a codec and not a URL: two peers
// that resolve the same video must derive the same address offline.
assert_eq!(
compute_yt_video_cid("video", "4D7u5KF7SP8"),
compute_yt_video_cid("video", "4D7u5KF7SP8")
);
// ...and the kind is part of the identity, not decoration.
assert_ne!(
compute_yt_video_cid("video", "OLAK5uy_kZ8Xq"),
compute_yt_video_cid("playlist", "OLAK5uy_kZ8Xq")
);
}
#[test]
fn yt_video_cid_rejects_empty_fields() {
assert!(compute_yt_video_cid("", "4D7u5KF7SP8").is_none());
assert!(compute_yt_video_cid("video", "").is_none());
}
#[test]
fn ext_play_cid_matches_golden_vector() {
assert_eq!(
compute_ext_play_cid("https://example.bandcamp.com/album/x").unwrap(),
"bagesaabenb2hi4dthixs6zlymfwxa3dffzrgc3temnqw24bomnxw2l3bnrrhk3jppa"
);
}
#[test]
fn ext_play_cid_rejects_non_http_schemes() {
// Load-bearing: a client hands this string to an anchor or a new tab,
// so a `javascript:` or `data:` payload must never mint a locator.
assert!(compute_ext_play_cid("javascript:alert(1)").is_none());
assert!(compute_ext_play_cid("data:text/html,<script>").is_none());
assert!(compute_ext_play_cid("ftp://example.com/x").is_none());
assert!(compute_ext_play_cid("").is_none());
}
#[test]
fn ext_play_cid_differs_from_the_url_locator_for_the_same_url() {
// The whole contract lives in the codec slot: `0x1006` means "fetch and
// seed these bytes", `0x1009` means "open this, there are no bytes for
// us". Wire-identical otherwise — which is exactly why a decoder that
// matched on *shape* would seed a Bandcamp page as the file.
let url = "https://example.bandcamp.com/album/x";
let ext = compute_ext_play_cid(url).unwrap();
let card = compute_card_cid("x", url).unwrap();
assert_ne!(ext, card);
}
#[test] #[test]
fn ipfs_cid_single_leaf_matches_kubo() { fn ipfs_cid_single_leaf_matches_kubo() {
assert_eq!( assert_eq!(
@@ -617,4 +921,156 @@ mod tests {
assert_eq!(out.blocks[0].1.len(), 0); assert_eq!(out.blocks[0].1.len(), 0);
assert_eq!(out.blocks[0].0, out.root); assert_eq!(out.blocks[0].0, out.root);
} }
// ---- card locator (0x1007) --------------------------------------------
/// Inverse of [`base32_lower_no_padding`], test-only. Lets the card tests
/// assert the actual CIDv1 bytes (codec, multihash code, digest) rather
/// than just round-tripping the encoder against itself.
fn base32_lower_decode(s: &str) -> Vec<u8> {
const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz234567";
let mut out = Vec::new();
let (mut acc, mut bits) = (0u32, 0u32);
for c in s.bytes() {
let v = ALPHABET.iter().position(|&a| a == c).expect("base32 char") as u32;
acc = (acc << 5) | v;
bits += 5;
if bits >= 8 {
bits -= 8;
out.push((acc >> bits) as u8);
acc &= (1 << bits) - 1;
}
}
out
}
/// The property the whole card design leans on: the CID is a pure function
/// of `(source, id)`, so any peer can derive a card's address offline. If
/// this literal ever changes, every already-published card silently forks
/// into a second identity — treat a failure here as a wire break.
#[test]
fn card_cid_is_deterministic() {
let cid = compute_card_cid("tmdb", "tv:95479").expect("card cid");
assert_eq!(cid, compute_card_cid("tmdb", "tv:95479").unwrap());
assert_eq!(cid, "bagdsaaanar2g2zdcor3duojvgq3ts");
}
/// Source- and kind-namespaced: the same numeric id on a different bridge,
/// or the same id under a different media kind, must not collide.
#[test]
fn card_cid_is_source_and_kind_namespaced() {
let tmdb = compute_card_cid("tmdb", "tv:95479").unwrap();
let mal = compute_card_cid("mal", "95479").unwrap();
let movie = compute_card_cid("tmdb", "movie:95479").unwrap();
assert_ne!(tmdb, mal);
assert_ne!(tmdb, movie);
}
/// Pins the on-the-wire CIDv1 layout: v1, codec `0x1007` (varint
/// `[0x87,0x20]`), **identity** multihash `0x00` (this family has no
/// digest — the locator itself is the payload), then
/// `varint(source_len) ‖ source ‖ id`.
#[test]
fn card_cid_wire_shape() {
let cid = compute_card_cid("tmdb", "tv:95479").unwrap();
assert!(cid.starts_with('b'), "multibase base32-lower prefix");
let wire = base32_lower_decode(&cid[1..]);
assert_eq!(wire[0], 0x01, "CIDv1");
assert_eq!(&wire[1..3], &[0x87, 0x20], "codec varint for 0x1007");
assert_eq!(wire[3], 0x00, "identity multihash — no digest");
let digest_len = wire[4] as usize;
let digest = &wire[5..5 + digest_len];
assert_eq!(digest[0], 4, "varint(len(\"tmdb\"))");
assert_eq!(&digest[1..5], b"tmdb");
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.
#[test]
fn card_cid_rejects_empty_and_oversized() {
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());
}
// ---- nzb-posting (0x1003) — the dedup contract -----------------------
//
// These pin the normalisation rule documented on `compute_nzb_posting_cid`.
// meta-share re-derives the same digest from the fetched manifest and
// rejects a mismatch, so a change that breaks one of these silently makes
// every self-scanned posting unplayable. Mirror any edit on both sides.
/// Segment/file **order must not matter** — that is the whole reason we
/// hash a sorted id set instead of the raw `.nzb` bytes (the retired
/// `0x1004` did the latter and broke dedup across reposts).
#[test]
fn nzb_posting_cid_is_order_independent() {
let a = ["[email protected]", "[email protected]", "[email protected]"];
let b = ["[email protected]", "[email protected]", "[email protected]"];
let c = ["[email protected]", "[email protected]", "[email protected]"];
assert_eq!(compute_nzb_posting_cid(&a), compute_nzb_posting_cid(&b));
assert_eq!(compute_nzb_posting_cid(&a), compute_nzb_posting_cid(&c));
}
/// Angle brackets, surrounding whitespace, empty entries and duplicates are
/// all normalised away. NZB `<segment>` bodies carry the bare form but NNTP
/// `BODY` re-adds the brackets, so both spellings must converge.
#[test]
fn nzb_posting_cid_normalises_brackets_blanks_and_dupes() {
let bare = ["[email protected]", "[email protected]"];
let noisy = [" <[email protected]> ", "[email protected]", "", " ", "[email protected]"];
assert_eq!(
compute_nzb_posting_cid(&bare),
compute_nzb_posting_cid(&noisy)
);
}
/// Case is preserved: a Message-ID's local part is case-sensitive per RFC
/// 5322, so lowercasing could collide two genuinely distinct articles.
#[test]
fn nzb_posting_cid_is_case_sensitive() {
assert_ne!(
compute_nzb_posting_cid(&["[email protected]"]),
compute_nzb_posting_cid(&["[email protected]"])
);
}
/// A different article set is a different posting.
#[test]
fn nzb_posting_cid_differs_on_different_ids() {
assert_ne!(
compute_nzb_posting_cid(&["[email protected]", "[email protected]"]),
compute_nzb_posting_cid(&["[email protected]", "[email protected]"])
);
}
/// Wire shape: CIDv1, codec `0x1003` (varint `[0x83, 0x20]`), multihash
/// sha2-256 (`0x12`), 32-byte digest. Pinned because meta-share matches on
/// the **codec** slot — a drift here makes every posting undecodable there.
#[test]
fn nzb_posting_cid_wire_shape() {
let cid = compute_nzb_posting_cid(&["[email protected]"]);
assert!(cid.starts_with('b'), "multibase base32-lower prefix");
// Round-trip the base32 body back to bytes to assert the header.
let decoded = base32_lower_decode(&cid[1..]);
assert_eq!(decoded[0], 0x01, "CIDv1");
assert_eq!(&decoded[1..3], &[0x83, 0x20], "varint(0x1003)");
assert_eq!(decoded[3], 0x12, "multihash sha2-256");
assert_eq!(decoded[4], 0x20, "digest length 32");
assert_eq!(decoded.len(), 5 + 32);
// The digest is sha256 over the newline-joined sorted set.
let expect: [u8; 32] = Sha256::digest(b"[email protected]").into();
assert_eq!(&decoded[5..], &expect);
}
} }
+42
View File
@@ -184,6 +184,48 @@ impl FeederStore {
&self.meta_core_url &self.meta_core_url
} }
/// Raw keys containing `contains` (`GET /api/kv/search`). Returns
/// `(keys, truncated)`; `truncated` means the server capped the result and
/// there may be more.
///
/// This is a keyspace scan, so it belongs on rare control-plane paths only —
/// today, invalidating the gateway's search-coverage markers when a feeder's
/// upstream source set changes. Never call it per search.
pub async fn search_keys(&self, contains: &str, limit: usize) -> anyhow::Result<(Vec<String>, bool)> {
let url = format!("{}/api/kv/search", self.meta_core_url);
let resp = self
.http
.get(&url)
.query(&[("contains", contains), ("limit", &limit.to_string())])
.send()
.await?;
let status = resp.status();
if !status.is_success() {
anyhow::bail!("meta-core KV search `{contains}` returned {status}");
}
#[derive(Deserialize)]
struct KvSearchResponse {
#[serde(default)]
keys: Vec<String>,
#[serde(default)]
truncated: bool,
}
let body: KvSearchResponse = resp.json().await?;
Ok((body.keys, body.truncated))
}
/// Delete one raw KV key (`DELETE /api/kv/value?key=`). A missing key is not
/// an error.
pub async fn delete_kv(&self, key: &str) -> anyhow::Result<()> {
let url = format!("{}/api/kv/value", self.meta_core_url);
let resp = self.http.delete(&url).query(&[("key", key)]).send().await?;
let status = resp.status();
if status.is_success() || status == reqwest::StatusCode::NOT_FOUND {
return Ok(());
}
anyhow::bail!("meta-core KV DELETE {key} returned {status}");
}
/// Self-publish a **metadata-only** outcome to meta-core, returning the /// Self-publish a **metadata-only** outcome to meta-core, returning the
/// **sparse** outcome (`record: None`) so the gateway's `(_, None)` branch /// **sparse** outcome (`record: None`) so the gateway's `(_, None)` branch
/// skips it — no double write. Byte outcomes and already-sparse outcomes are /// skips it — no double write. Byte outcomes and already-sparse outcomes are
+25 -2
View File
@@ -62,6 +62,25 @@ pub struct HashOutcome {
/// id}` in an identity multihash, opaque to bitswap and redeemed to bytes /// id}` in an identity multihash, opaque to bitswap and redeemed to bytes
/// only by a credentialed meta-share peer, which decodes the locator from the /// only by a credentialed meta-share peer, which decodes the locator from the
/// cid and grabs the `.nzb` via `t=get` — no KV side-table. /// cid and grabs the `.nzb` via `t=get` — no KV side-table.
/// - `CardLocator` — the identity of a *work* (a series, a film) as published by
/// a metadata bridge, custom multicodec `0x1007`. The only family with **no
/// bytes behind it at all**: the record IS the payload. Always paired with
/// `bytes: None`, so it lands in the `(None, Some)` metadata-only branch
/// above. Opaque to bitswap, and deliberately so — there is nothing to seed.
/// - `NzbPosting` — a Usenet posting we **scanned ourselves**, custom multicodec
/// `nzb-posting` `0x1003`. A content-derived digest over the article
/// Message-ID set (not a locator: no host is embedded), so any peer with a
/// plain NNTP provider can fetch it — no `IndexerCred`. Opaque to bitswap: the
/// digest is over the id set, not over any block, so the `.nzb` manifest
/// travels separately as an ordinary sha2-256 cid in the record's `manifest`
/// field. Produced by `hash::compute_nzb_posting_cid`.
/// - `YtVideo` — a **delegated-playback** reference, custom multicodec
/// `yt-video` `0x1008`. Like `CardLocator` it is always paired with
/// `bytes: None` and lands in the `(None, Some)` metadata-only branch, but it
/// means something stronger than "no bytes yet": the bytes **exist and are
/// permanently someone else's**. Nothing is fetched, seeded or
/// content-addressed, and meta-share's byte path REFUSES the cid with a `400`
/// rather than 404ing. Produced by `hash::compute_yt_video_cid`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HashKind { pub enum HashKind {
/// Reserved: no feeder currently emits midhash256 outcomes, but the family /// Reserved: no feeder currently emits midhash256 outcomes, but the family
@@ -71,6 +90,9 @@ pub enum HashKind {
Sha2_256, Sha2_256,
BtV1File, BtV1File,
NzbRelease, NzbRelease,
CardLocator,
NzbPosting,
YtVideo,
} }
/// Startup-only configuration error returned by [`FeederPlugin::configure`]. /// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -118,8 +140,9 @@ pub const SOURCE_KEYSET_PREFIX: &str = "source/";
/// onto a record's fields, but **only if it carries no `source/*` member yet**. /// onto a record's fields, but **only if it carries no `source/*` member yet**.
/// This gives every gateway hit a provenance label for free (so meta-watch never /// This gives every gateway hit a provenance label for free (so meta-watch never
/// falls back to "unknown" for a feeder record), while a plugin that knows a /// falls back to "unknown" for a feeder record), while a plugin that knows a
/// finer origin (torznab → the specific Prowlarr indexer) stamps its own /// finer origin (torznab → the specific Prowlarr indexer) stamps its own flat
/// `source/gateway:<indexer>` first and this default then no-ops. /// `source/*` facets first (`source/gateway`, `source/<protocol>`,
/// `source/<indexer-slug>`) and this default then no-ops.
pub fn stamp_default_source( pub fn stamp_default_source(
fields: &mut std::collections::BTreeMap<String, String>, fields: &mut std::collections::BTreeMap<String, String>,
upstream_id: &str, upstream_id: &str,
+12 -23
View File
@@ -41,7 +41,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::{stamp_default_source, ConfigError, FeederPlugin, HashKind, HashOutcome}; use crate::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome};
use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError}; use crate::query::{GatewayQuery, GatewaySearchEvent, GatewayWireError};
use crate::types::{DiscoveryRecord, GatewayError, PluginHealth}; use crate::types::{DiscoveryRecord, GatewayError, PluginHealth};
@@ -55,6 +55,9 @@ pub enum HashKindDto {
Sha2_256, Sha2_256,
BtV1File, BtV1File,
NzbRelease, NzbRelease,
CardLocator,
NzbPosting,
YtVideo,
} }
impl From<HashKind> for HashKindDto { impl From<HashKind> for HashKindDto {
@@ -64,6 +67,9 @@ impl From<HashKind> for HashKindDto {
HashKind::Sha2_256 => HashKindDto::Sha2_256, HashKind::Sha2_256 => HashKindDto::Sha2_256,
HashKind::BtV1File => HashKindDto::BtV1File, HashKind::BtV1File => HashKindDto::BtV1File,
HashKind::NzbRelease => HashKindDto::NzbRelease, HashKind::NzbRelease => HashKindDto::NzbRelease,
HashKind::CardLocator => HashKindDto::CardLocator,
HashKind::NzbPosting => HashKindDto::NzbPosting,
HashKind::YtVideo => HashKindDto::YtVideo,
} }
} }
} }
@@ -75,6 +81,9 @@ impl From<HashKindDto> for HashKind {
HashKindDto::Sha2_256 => HashKind::Sha2_256, HashKindDto::Sha2_256 => HashKind::Sha2_256,
HashKindDto::BtV1File => HashKind::BtV1File, HashKindDto::BtV1File => HashKind::BtV1File,
HashKindDto::NzbRelease => HashKind::NzbRelease, HashKindDto::NzbRelease => HashKind::NzbRelease,
HashKindDto::CardLocator => HashKind::CardLocator,
HashKindDto::NzbPosting => HashKind::NzbPosting,
HashKindDto::YtVideo => HashKind::YtVideo,
} }
} }
} }
@@ -275,13 +284,10 @@ async fn query(
Json(req): Json<QueryRequest>, Json(req): Json<QueryRequest>,
) -> Result<Json<QueryResponse>, Response> { ) -> Result<Json<QueryResponse>, Response> {
let plugin = state.plugin(&req.upstream_id)?; let plugin = state.plugin(&req.upstream_id)?;
let mut records = plugin let records = plugin
.handle_query(&req.query, req.max_results as usize) .handle_query(&req.query, req.max_results as usize)
.await .await
.map_err(gateway_error_response)?; .map_err(gateway_error_response)?;
for r in &mut records {
stamp_default_source(&mut r.fields, &req.upstream_id);
}
Ok(Json(QueryResponse { records })) Ok(Json(QueryResponse { records }))
} }
@@ -305,18 +311,6 @@ async fn query_stream(
.handle_query_stream(&req.query, req.max_results as usize) .handle_query_stream(&req.query, req.max_results as usize)
.await .await
.map_err(gateway_error_response)?; .map_err(gateway_error_response)?;
// Stamp the default provenance member on each streamed `Base` record (a
// no-op when the plugin already carries a finer `source/*`, e.g. torznab's
// per-indexer label). `EnrichPatch`/`Drop`/`Done` frames pass through: the
// base they refer to was already stamped.
let upstream_id = req.upstream_id.clone();
let events = events.map(move |ev| match ev {
GatewaySearchEvent::Base(mut rec) => {
stamp_default_source(&mut rec.fields, &upstream_id);
GatewaySearchEvent::Base(rec)
}
other => other,
});
let body = events.map(|ev| { let body = events.map(|ev| {
// Serialization of GatewaySearchEvent is infallible in practice; on the // Serialization of GatewaySearchEvent is infallible in practice; on the
// off chance it fails, terminate with an Error line so the consumer sees // off chance it fails, terminate with an Error line so the consumer sees
@@ -342,15 +336,10 @@ async fn compute(
Json(req): Json<ComputeRequest>, Json(req): Json<ComputeRequest>,
) -> Result<Json<ComputeResponse>, Response> { ) -> Result<Json<ComputeResponse>, Response> {
let plugin = state.plugin(&req.upstream_id)?; let plugin = state.plugin(&req.upstream_id)?;
let mut outcomes = plugin let outcomes = plugin
.compute_outcomes(&req.record_id) .compute_outcomes(&req.record_id)
.await .await
.map_err(gateway_error_response)?; .map_err(gateway_error_response)?;
for o in &mut outcomes {
if let Some(rec) = o.record.as_mut() {
stamp_default_source(&mut rec.fields, &req.upstream_id);
}
}
Ok(Json(ComputeResponse { Ok(Json(ComputeResponse {
outcomes: outcomes.into_iter().map(OutcomeDto::from_outcome).collect(), outcomes: outcomes.into_iter().map(OutcomeDto::from_outcome).collect(),
})) }))