rework plugin
This commit is contained in:
@@ -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)?;
|
||||||
|
|||||||
@@ -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());
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
Reference in New Issue
Block a user