rework plugin

This commit is contained in:
Worph
2026-08-24 16:33:15 +02:00
parent dc29643e29
commit 4ecafe9bee
4 changed files with 87 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)?;
+2 -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());
+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
+27
View File
@@ -129,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.