4 Commits
Author SHA1 Message Date
Worph 1db9704990 align to metadata keys doc
Publish image / publish (push) Failing after 13s
2026-08-30 19:50:19 +02:00
Worph 4ecafe9bee rework plugin 2026-08-24 16:33:15 +02:00
Worph dc29643e29 audio support widening 2026-08-21 18:22:10 +02:00
Worph 210e980cc8 genre management
Publish image / publish (push) Failing after 14s
2026-08-01 20:20:54 +02:00
9 changed files with 625 additions and 2 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)?;
+275
View File
@@ -0,0 +1,275 @@
//! `contentKind` → `domain` — the routing table 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:
//!
//! ```text
//! meta-watch → domain:film|tv
//! meta-listen → domain:music
//! meta-read → domain:literature
//! ```
//!
//! # Why it is 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 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.
//!
//! # 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.
/// 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 `tv` from `music`; the
/// writer stamps `pack`'s domain from its own context (the same reason
/// `derive_file_type` cannot have a second `pack` arm).
const DOMAIN_BY_CONTENT_KIND: &[(&str, &str)] = &[
// film / tv
("movie", "film"),
("series", "tv"),
("episode", "tv"),
// music
("track", "music"),
("album", "music"),
("artist", "music"),
("musicVideo", "music"),
("djMix", "music"),
("liveSet", "music"),
// spoken
("podcast", "spoken"),
("podcastEpisode", "spoken"),
// 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"),
];
/// 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> {
let k = content_kind.trim();
DOMAIN_BY_CONTENT_KIND
.iter()
.find(|(kind, _)| kind.eq_ignore_ascii_case(k))
.map(|(_, domain)| *domain)
}
/// 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.
pub fn content_kinds_for_domain(domain: &str) -> Vec<&'static str> {
let d = domain.trim();
DOMAIN_BY_CONTENT_KIND
.iter()
.filter(|(_, dom)| dom.eq_ignore_ascii_case(d))
.map(|(kind, _)| *kind)
.collect()
}
/// Rewrite a query so a `domain:` filter can reach a plugin that has never
/// heard of the key.
///
/// Returns `None` when the query carries no `domain:` filter — 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. Nine of the fifteen 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 `domain` at all, and old plugins keep working unchanged
/// — while the dispatcher still enforces the real `domain:` filter afterwards,
/// against records it has just stamped.
///
/// An explicit `contentKind:` filter already in the query is left alone: it is
/// the narrower of the two, and the `domain` term is still enforced
/// post-stamp, so the AND semantics hold either way.
pub fn expand_domain_filter(query: &crate::query::GatewayQuery) -> Option<crate::query::GatewayQuery> {
let domains = query.filters.get("domain")?;
let mut out = query.clone();
out.filters.remove("domain");
if !out.filters.contains_key("contentKind") {
let kinds: Vec<String> = domains
.iter()
.flat_map(|d| content_kinds_for_domain(d))
.map(|k| k.to_string())
.collect();
// An unknown domain 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 `domain:`
// check do the (correct, empty) filtering.
if !kinds.is_empty() {
out.filters.insert("contentKind".to_string(), kinds);
}
}
Some(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>) {
if fields.contains_key("domain") {
return;
}
let Some(kind) = fields.get("contentKind").cloned() else {
return;
};
if let Some(domain) = domain_for_content_kind(&kind) {
fields.insert("domain".to_string(), domain.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("film"));
assert_eq!(domain_for_content_kind("episode"), Some("tv"));
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 pack_has_no_derivable_domain() {
assert_eq!(domain_for_content_kind("pack"), None);
}
#[test]
fn unknown_kind_is_none_not_a_guess() {
assert_eq!(domain_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"));
}
}
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_domain_filter_is_left_untouched() {
assert!(expand_domain_filter(&q(&[("fileType", &["video"])])).is_none());
}
#[test]
fn domain_becomes_the_content_kinds_it_covers() {
let out = expand_domain_filter(&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 an_explicit_content_kind_filter_wins() {
let out = expand_domain_filter(&q(&[
("domain", &["music"][..]),
("contentKind", &["track"][..]),
]))
.unwrap();
assert_eq!(
out.filters.get("contentKind").map(Vec::as_slice),
Some(&["track".to_string()][..])
);
}
#[test]
fn unknown_domain_adds_no_alternation() {
// An empty alternation would drop every record at the plugin; the
// post-stamp `domain:` check does the (correct, empty) filtering.
let out = expand_domain_filter(&q(&[("domain", &["hologram"])])).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);
assert_eq!(f.get("domain").map(String::as_str), Some("tv"));
// 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());
stamp_domain(&mut p);
assert_eq!(p.get("domain").map(String::as_str), Some("music"));
// No contentKind → no domain invented.
let mut n: BTreeMap<String, String> = BTreeMap::new();
stamp_domain(&mut n);
assert!(n.get("domain").is_none());
}
}
+189 -2
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());
@@ -298,6 +298,117 @@ pub fn compute_card_cid(source: &str, id: &str) -> Option<String> {
Some(format!("b{}", base32_lower_no_padding(&wire))) 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`), /// Custom multicodec for a **Usenet posting** identity (`nzb-posting`),
/// minted from the article Message-IDs of a release we scanned ourselves. /// minted from the article Message-IDs of a release we scanned ourselves.
/// ///
@@ -672,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!(
+1
View File
@@ -24,6 +24,7 @@ pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
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;
+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
+35
View File
@@ -74,6 +74,13 @@ pub struct HashOutcome {
/// digest is over the id set, not over any block, so the `.nzb` manifest /// 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` /// travels separately as an ordinary sha2-256 cid in the record's `manifest`
/// field. Produced by `hash::compute_nzb_posting_cid`. /// 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
@@ -85,6 +92,7 @@ pub enum HashKind {
NzbRelease, NzbRelease,
CardLocator, CardLocator,
NzbPosting, NzbPosting,
YtVideo,
} }
/// Startup-only configuration error returned by [`FeederPlugin::configure`]. /// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -121,6 +129,33 @@ pub fn upstream_id_field(upstream_id: &str) -> String {
format!("{upstream_id}id") format!("{upstream_id}id")
} }
/// Key-set prefix for the provenance `source/<label>` field (METADATA_KEYS §5).
/// Each member is its own hash field (`source/gateway:nyaa.si = "true"`) so two
/// producers that reach the same content by different routes union without a
/// last-writer-wins clobber. The `<label>` uses `:` internally (not `/`, which
/// is the key-set path separator) — e.g. `gateway:nyaa.si`, `gateway:tribler`.
pub const SOURCE_KEYSET_PREFIX: &str = "source/";
/// Stamp the SDK's default provenance member — `source/gateway:<upstream_id>` —
/// 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
/// falls back to "unknown" for a feeder record), while a plugin that knows a
/// finer origin (torznab → the specific Prowlarr indexer) stamps its own flat
/// `source/*` facets first (`source/gateway`, `source/<protocol>`,
/// `source/<indexer-slug>`) and this default then no-ops.
pub fn stamp_default_source(
fields: &mut std::collections::BTreeMap<String, String>,
upstream_id: &str,
) {
if fields.keys().any(|k| k.starts_with(SOURCE_KEYSET_PREFIX)) {
return;
}
fields.insert(
format!("{SOURCE_KEYSET_PREFIX}gateway:{upstream_id}"),
"true".to_string(),
);
}
/// Static plugin contract. Each enabled `upstream_id` is implemented by /// Static plugin contract. Each enabled `upstream_id` is implemented by
/// exactly one `Box<dyn FeederPlugin>`. Lives behind `&self` in steady state — /// exactly one `Box<dyn FeederPlugin>`. Lives behind `&self` in steady state —
/// called concurrently, owns any interior mutability for caches. /// called concurrently, owns any interior mutability for caches.
+58
View File
@@ -27,6 +27,16 @@ pub fn record_matches(fields: &BTreeMap<String, String>, query: &GatewayQuery) -
} }
continue; continue;
} }
// `genres` is a key-set too (`genres/<Name> = "true"`, METADATA_KEYS
// §`genres/{name}`) on everything the card tier writes. Without this arm
// a `genres:` filter looks up a flat `genres` field, finds nothing on a
// card, and drops it — so a genre-scoped row would return empty.
if key == "genres" {
if !genres_filter_matches(fields, allowed) {
return false;
}
continue;
}
let Some(field_value) = fields.get(key) else { let Some(field_value) = fields.get(key) else {
return false; return false;
}; };
@@ -155,6 +165,54 @@ fn record_has_concrete_language(fields: &BTreeMap<String, String>) -> bool {
/// Key-set members are written as the literal `"true"`; tolerate a couple of /// Key-set members are written as the literal `"true"`; tolerate a couple of
/// obvious truthy spellings and treat anything else as absent. /// obvious truthy spellings and treat anything else as absent.
/// Fold a genre label to its comparison form: lowercase, alphanumerics only.
///
/// Genre display names carry spaces and ampersands (`"Action & Adventure"`,
/// `"Sci-Fi & Fantasy"`, `"Science Fiction"`), and **space is the query DSL's
/// token separator** — so a filter can never spell a genre the way the record
/// stores it. Callers pass a slug (`genres:action-adventure`); folding both
/// sides makes them one value without forcing the stored name to be
/// machine-shaped.
fn genre_fold(s: &str) -> String {
s.chars()
.filter(|c| c.is_alphanumeric())
.flat_map(|c| c.to_lowercase())
.collect()
}
/// True iff the record carries one of the requested genres, in **either**
/// encoding: a truthy `genres/<Name>` key-set member (the card tier's shape) or
/// the legacy flat comma-joined `genres` field. OR across the requested genres,
/// both sides folded — see [`genre_fold`].
///
/// **Fails closed, unlike [`languages_filter_matches`].** A language filter is a
/// viewer *preference* narrowing a set, so an unknown-language release is kept.
/// A genre filter *defines what a row is*: a record with no genre information is
/// not known to be Action, so putting it in an Action row states something
/// untrue. Upstream has already filtered by genre, so this is a re-validation.
fn genres_filter_matches(fields: &BTreeMap<String, String>, allowed: &[String]) -> bool {
let wanted: Vec<String> = allowed.iter().map(|g| genre_fold(g)).collect();
if wanted.iter().all(|g| g.is_empty()) {
return true; // a filter with no usable value constrains nothing
}
for (key, value) in fields {
let Some(name) = key.strip_prefix("genres/") else {
continue;
};
if is_truthy_member(value) && wanted.contains(&genre_fold(name)) {
return true;
}
}
if let Some(flat) = fields.get("genres") {
for member in flat.split(|c| c == ',' || c == '|') {
if wanted.contains(&genre_fold(member)) {
return true;
}
}
}
false
}
fn is_truthy_member(v: &str) -> bool { fn is_truthy_member(v: &str) -> bool {
matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes") matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes")
} }
+3
View File
@@ -57,6 +57,7 @@ pub enum HashKindDto {
NzbRelease, NzbRelease,
CardLocator, CardLocator,
NzbPosting, NzbPosting,
YtVideo,
} }
impl From<HashKind> for HashKindDto { impl From<HashKind> for HashKindDto {
@@ -68,6 +69,7 @@ impl From<HashKind> for HashKindDto {
HashKind::NzbRelease => HashKindDto::NzbRelease, HashKind::NzbRelease => HashKindDto::NzbRelease,
HashKind::CardLocator => HashKindDto::CardLocator, HashKind::CardLocator => HashKindDto::CardLocator,
HashKind::NzbPosting => HashKindDto::NzbPosting, HashKind::NzbPosting => HashKindDto::NzbPosting,
HashKind::YtVideo => HashKindDto::YtVideo,
} }
} }
} }
@@ -81,6 +83,7 @@ impl From<HashKindDto> for HashKind {
HashKindDto::NzbRelease => HashKind::NzbRelease, HashKindDto::NzbRelease => HashKind::NzbRelease,
HashKindDto::CardLocator => HashKind::CardLocator, HashKindDto::CardLocator => HashKind::CardLocator,
HashKindDto::NzbPosting => HashKind::NzbPosting, HashKindDto::NzbPosting => HashKind::NzbPosting,
HashKindDto::YtVideo => HashKind::YtVideo,
} }
} }
} }
@@ -186,6 +186,12 @@ impl UsenetPlugin {
); );
if let Some(kind) = content_kind_for_category(r.category_id) { if let Some(kind) = content_kind_for_category(r.category_id) {
fields.insert("contentKind".to_string(), kind.to_string()); fields.insert("contentKind".to_string(), kind.to_string());
// The routing axis, co-written with the kind (METADATA_KEYS.md §1):
// `fileType` says what the bytes are, `domain` says which app wants
// the record. Without it the row never reaches a client wall.
if let Some(d) = meta_feeder_sdk::domain::domain_for_content_kind(kind) {
fields.insert("domain".to_string(), d.to_string());
}
} }
if let Some(ts) = r.post_date { if let Some(ts) = r.post_date {
fields.insert("publishedAt".to_string(), ts.to_string()); fields.insert("publishedAt".to_string(), ts.to_string());