Author SHA1 Message Date
Manielle 3564f170b4 Open Library card: a named work (openlibraryid lookup) needs only a title; cover falls back to its ISBN, description optional 2026-10-15 14:36:04 +02:00
Worph 8bf94cb186 release (first build on gitea-watch / IPCR)
Publish image / publish (push) Successful in 2m11s
2026-10-13 11:03:04 +00:00
Worph 9d20e3ae15 gitea workflow: push to IPCR's direct host (Cloudflare 100 MB cap) [skip ci] 2026-10-13 10:02:33 +00:00
Worph 87df5ef104 gitea workflow: publish to IPCR (metadec.eth) instead of ghcr
Publish image / publish (push) Successful in 2m12s
2026-10-12 18:30:48 +00:00
Worph e87afa68c8 SDK v1.7.0: advertises over beacon v2 as metamesh.feeder/<upstream>
Publish image / publish (push) Failing after 14s
2026-10-04 17:35:16 +00:00
Worph aed0db3cda improved search
Publish image / publish (push) Failing after 13s
2026-09-12 16:37:22 +02:00
Worph e9bf9d3f34 rework domain field
Publish image / publish (push) Failing after 14s
2026-09-02 00:04:19 +02:00
Worph 5b03be5f82 meta data improvement 2026-09-01 11:48:32 +02:00
Worph ba3b0a5cb8 buffix 2026-08-31 21:26:25 +02:00
Worph 3ee1aec0ab bugfix 2026-08-31 20:53:24 +02:00
Worph 4c7248da58 align to metadata keys doc 2026-08-30 19:51:24 +02:00
Worph 4dc6d46c47 rework plugin 2026-08-24 16:32:04 +02:00
Worph a0227f8378 genre management 2026-08-01 20:22:20 +02:00
Worph e0ad53ce19 fix cids 2026-07-14 16:07:43 +02:00
Worph c34d4a8ab2 bugfix
Publish image / publish (push) Failing after 13s
2026-06-15 00:16:32 +02:00
Worph 815d902dda doc 2026-06-14 21:42:51 +02:00
25 changed files with 4823 additions and 3620 deletions
+6
View File
@@ -0,0 +1,6 @@
# FIX CRLF always when developer has not set
# Linux/Mac: git config --global core.autocrlf input
# Windows: git config --global core.autocrlf true
# Auto detect text files and perform LF normalization
* text=auto
* eol=lf
+85
View File
@@ -0,0 +1,85 @@
# Gitea twin of .github/workflows/publish.yml. Gitea runs only .gitea/workflows/ once this
# folder exists, and GitHub never reads it.
#
# Publishes to IPCR on watch.nsl.sh instead of ghcr.io:
# push ipcr-push-85-17-246-67.nip.io/metadec.eth/<repo>:<tag> docker login as metadec.eth with the
# IPCR_PUSH_TOKEN secret (the metadec.eth namespace token from the IPCR dashboard)
# pull ipcr-watch.nsl.sh/ipns/metadec.eth/metadec.eth/<repo>:<tag> anonymous
# The push goes to ipcr-push-85-17-246-67.nip.io, the box's own address, not ipcr-push-watch.nsl.sh: Cloudflare fronts
# *.watch.nsl.sh and refuses any request over 100 MB (413), which a large image layer is.
# That host's certificate is signed by the box's private Mesh Router CA, given to buildkit from
# the IPCR_CA_PEM repo variable. docker login would make the job's Docker daemon check the
# certificate without that CA, so the login is written straight into ~/.docker/config.json.
# The runner's job image (node:22-bookworm) has no docker CLI, so the static one is
# installed first; buildx installs itself. No `type=gha` cache: the runner has none.
name: Publish image
# Build and push the feeder image to the IPCR (ipcr-push-watch.nsl.sh):
# * every push to main -> the moving `main` tag
# * every `v*` tag -> semver tags (e.g. v1.0.0 -> 1.0.0, 1.0, 1)
# The build context is the repo root (the workspace); `meta-feeder-sdk` is a
# tag-pinned git dependency fetched during the build. The Dockerfile lives under
# feeder-plugin/book-feeder/.
on:
push:
branches: [main]
tags: ['v*']
env:
REGISTRY: ipcr-push-85-17-246-67.nip.io
IMAGE_NAME: metadec.eth/${{ github.event.repository.name }}
jobs:
publish:
runs-on: ubuntu-latest
permissions:
contents: read
packages: write
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Install Docker CLI
run: curl -fsSL https://download.docker.com/linux/static/stable/x86_64/docker-28.5.1.tgz | tar -xz -C /usr/local/bin --strip-components=1 docker/docker
- name: Trust the Mesh Router CA
run: printf '%s\n' "$IPCR_CA_PEM" > /tmp/ipcr-ca.pem
env:
IPCR_CA_PEM: ${{ vars.IPCR_CA_PEM }}
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
with:
version: latest
buildkitd-config-inline: |
[registry."${{ env.REGISTRY }}"]
ca = ["/tmp/ipcr-ca.pem"]
- name: Log in to IPCR
env:
IPCR_PUSH_TOKEN: ${{ secrets.IPCR_PUSH_TOKEN }}
run: |
mkdir -p ~/.docker
printf '{"auths":{"%s":{"auth":"%s"}}}\n' "$REGISTRY" "$(printf 'metadec.eth:%s' "$IPCR_PUSH_TOKEN" | base64 -w0)" > ~/.docker/config.json
- name: Extract image metadata
id: meta
uses: docker/metadata-action@v5
with:
images: ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
tags: |
type=ref,event=branch
type=semver,pattern={{version}}
type=semver,pattern={{major}}.{{minor}}
type=semver,pattern={{major}}
type=sha,format=short
- name: Build and push
uses: docker/build-push-action@v6
with:
context: .
file: ./feeder-plugin/book-feeder/Dockerfile
push: true
tags: ${{ steps.meta.outputs.tags }}
labels: ${{ steps.meta.outputs.labels }}
+2 -2
View File
@@ -3,8 +3,8 @@ name: Publish image
# Build and push the feeder image to the GitHub Container Registry (ghcr.io):
# * every push to main -> the moving `main` tag
# * every `v*` tag -> semver tags (e.g. v1.0.0 -> 1.0.0, 1.0, 1)
# The build context is the repo root (the workspace) so the vendored
# `meta-feeder-sdk` path dependency resolves; the Dockerfile lives under
# The build context is the repo root (the workspace); `meta-feeder-sdk` is a
# tag-pinned git dependency fetched during the build. The Dockerfile lives under
# feeder-plugin/book-feeder/.
on:
push:
Generated
+28 -3472
View File
File diff suppressed because it is too large Load Diff
+5 -1
View File
@@ -1,6 +1,10 @@
[workspace]
resolver = "2"
members = ["crates/*", "feeder-plugin/*"]
members = ["feeder-plugin/*"]
# The vendored SDK copy is no longer built: book-feeder pins meta-feeder-sdk by
# git tag. Excluded (not deleted) so a stray `cargo` run inside it does not try
# to join this workspace.
exclude = ["crates/meta-feeder-sdk"]
[workspace.package]
edition = "2021"
+96 -1
View File
@@ -1 +1,96 @@
# meta-feeder-book
# meta-feeder-book
Book **feeder sidecar** for [MetaMesh](https://github.com/worph/meta-gateway).
One binary, `book-feeder`, hosts two upstreams:
| Upstream | Serves | What it is |
|----------|--------|------------|
| `gutenberg` | `document` × `book` | Public-domain **editions** from Project Gutenberg (via the [Gutendex](https://gutendex.com) API), with an epub url-locator per record |
| `openlibrary` | `card` × `book` | **Work cards** from [Open Library](https://openlibrary.org) — the book identity tier (`TMDB : video :: AniList : comics :: Open Library : books`) |
The two meet through `openlibraryid`. A Gutenberg edition carries it only when
Open Library itself records the link: exactly one Open Library work lists the
etext's Project Gutenberg id (`search.json?q=id_project_gutenberg:<id>`). The
edition then also carries the transient trust claim `anchored=true`,
`anchorSource=openlibrary`, which the gateway stores as
`anchoredBy/openlibrary:OL…W`. There is no title or author matching.
An Open Library card is addressed `compute_card_cid("openlibrary", "OL…W")`,
the bare work id, byte-identical to meta-read's `card_cid::openlibrary`. A work
with no cover or no description gets no card.
## Role in MetaMesh
A feeder is a stateless HTTP sidecar. It **finds records and fetches bytes**; it
does *not* talk to meta-core or the libp2p blockstore. The gateway core that
calls it owns the meta-core store-back and the blockstore seeding. A gateway
registers this feeder as a `RemoteFeederPlugin` pointing at its `/` and then
drives the contract:
| Endpoint | Purpose |
|----------|---------|
| `GET /manifest` | feeder identity + capabilities |
| `GET /health` | liveness |
| `POST /query`, `POST /query_stream` | structured search against the upstream |
| `POST /compute` | enrichment / outcome compute |
| `GET /fetch/:upstream_id/:record_id` | fetch a record's bytes |
| `GET /blob/:upstream_id/:cid` | fetch a content-addressed blob |
| `GET /config`, `GET /config/schema`, `GET\|PUT /config/values` | runtime config UI + API |
## Configuration
| Env var | Default | Notes |
|---------|---------|-------|
| `META_FEEDER_HTTP_LISTEN` | `0.0.0.0:8080` | HTTP listen address |
| `META_FEEDER_STATE_DIR` | `/data/meta-feeder` | redb cache + state |
| `OPENLIBRARY_CONTACT` | — | e-mail or URL sent in the Open Library User-Agent. First-boot seed for the `contact` config field; a value saved in the config UI wins |
| `OPENLIBRARY_TOP_N` | `12` | cards per search |
| `OPENLIBRARY_DISCOVERY_N` | `20` | cards per browse row (`popular:true` → Open Library trending, `fresh:true` → recently added) |
| `RUST_LOG` | `info` | tracing filter |
No upstream API key is required. Open Library asks API clients to identify
themselves, so set a contact in production: an identified client gets a larger
request budget. Both upstreams share one Open Library rate budget. The config
page edits the `gutenberg` upstream, and the contact saved there is read by
both.
## Image
```
ghcr.io/worph/meta-feeder-book
```
Exposes `8080`. Built and pushed by CI on every push to `main` (the `main`
tag) and on `v*` tags (semver tags).
## Build locally
The build context is the **repo root** (the Cargo workspace).
`meta-feeder-sdk` is a git dependency pinned by tag, so the builder needs
network access to github.com:
```bash
docker build -f feeder-plugin/book-feeder/Dockerfile -t ghcr.io/worph/meta-feeder-book:dev .
```
Tests (`cargo` is not needed on the host; wiremock stands in for every upstream):
```bash
docker run --rm -v "$PWD":/src -w /src -e CARGO_TARGET_DIR=/target rust:1.89 \
cargo test -p book-feeder
```
## Repo layout
```
Cargo.toml # workspace: members = feeder-plugin/*
feeder-plugin/book-feeder/ # this feeder's crate + Dockerfile
src/gutenberg.rs # `gutenberg` upstream
src/openlibrary.rs # `openlibrary` upstream (cards)
src/openlibrary_client.rs # Open Library client (budget, cache, binding lookup)
crates/meta-feeder-sdk/ # old vendored SDK copy — no longer built (excluded)
```
`meta-feeder-sdk` comes from [`worph/meta-feeder-sdk`](https://github.com/worph/meta-feeder-sdk)
at the tag in `feeder-plugin/book-feeder/Cargo.toml`. An SDK change reaches
this feeder only when it is tagged and the pin is bumped.
+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
/// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel.
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
/// via `Arc`).
@@ -105,6 +112,7 @@ impl MidhashCache {
tx.open_table(TMDB_PRINCIPAL_TOPN_TABLE)?;
tx.open_table(SUBTITLES_TABLE)?;
tx.open_table(OPENSUBTITLES_TABLE)?;
tx.open_table(MISC_TABLE)?;
tx.commit()?;
}
Ok(Self { db: Arc::new(db) })
@@ -112,6 +120,14 @@ impl MidhashCache {
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> {
let tx = self.db.begin_read()?;
let table = tx.open_table(MIDHASH_TABLE)?;
+42 -19
View File
@@ -17,6 +17,7 @@
.field { margin: 0 0 16px; }
.field > label { display: block; font-weight: 600; margin: 0 0 4px; }
.help { color: #8a8a96; font-size: 12px; margin: 3px 0 0; }
.req { color: #f0a0a0; margin-left: 3px; font-weight: 700; }
input[type=text], input[type=password], input[type=number], textarea {
width: 100%; padding: 8px 10px; border-radius: 8px;
border: 1px solid #2a2a35; background: #15151c; color: #e6e6ea; font: inherit;
@@ -46,7 +47,7 @@
</head>
<body>
<h1 id="title">Plugin configuration</h1>
<p class="sub">Secrets are write-only — set values are shown as “set”, never echoed back. Saving requires a feeder restart to take effect.</p>
<p class="sub">Secrets are write-only — set values are shown as “set”, never echoed back. Saving requires a feeder restart to take effect. Fields marked <span class="req">*</span> are required.</p>
<div id="banner" class="banner"></div>
<form id="form"></form>
<div class="bar" id="bar" style="display:none">
@@ -102,15 +103,24 @@ function linkify(text) {
function helpEl(f) { return f.help ? el("p", { class: "help" }, ...linkify(f.help)) : null; }
// A "*" marker for required fields (null for optional ones).
function reqStar(f) { return f.required ? el("span", { class: "req", text: "*", title: "required" }) : null; }
// Render one scalar field. `vals` is the object holding the value/`_set` flag.
// Every returned widget exposes `_collect()` and `_key` so collect() can read it.
function fieldInput(f, vals) {
if (f.type === "bool") {
const cb = el("input", { type: "checkbox" });
cb.checked = vals[f.key] === true;
cb._get = () => cb.checked;
return el("label", {}, cb, " ", el("span", { text: " " + f.label }));
const w = el("div", { class: "field" },
el("label", {}, cb, " ", el("span", { text: " " + f.label }), reqStar(f)),
helpEl(f));
w._collect = () => cb.checked;
w._key = f.key;
return w;
}
const label = el("label", { text: f.label });
const star = reqStar(f); if (star) label.append(star);
let input;
if (f.type === "secret") {
const isSet = vals[f.key + "_set"] === true;
@@ -134,14 +144,6 @@ function fieldInput(f, vals) {
const wrap = el("div", { class: "field" }, label, input, helpEl(f));
wrap._collect = () => input._get();
wrap._key = f.key;
// bool path returns a <label>; normalize collect/key for it
if (f.type === "bool") {
const w = el("div", { class: "field" }, label, helpEl(f));
const cb = label.querySelector("input");
w._collect = () => cb.checked;
w._key = f.key;
return w;
}
return wrap;
}
@@ -175,8 +177,9 @@ function recordArray(f, vals) {
const add = el("button", { type: "button", class: "add", text: "+ add " + f.label.toLowerCase().replace(/s$/, "") });
add.onclick = () => addRecord({});
const wrap = el("div", { class: "field" },
el("label", { text: f.label }), helpEl(f), list, add);
const lbl = el("label", { text: f.label });
const star = reqStar(f); if (star) lbl.append(star);
const wrap = el("div", { class: "field" }, lbl, helpEl(f), list, add);
wrap._key = f.key;
wrap._collect = () => Array.from(list.children).map((c) => c._collect());
return wrap;
@@ -215,14 +218,34 @@ async function save() {
headers: { "Content-Type": "application/json" },
body: JSON.stringify(collect()),
});
if (!res.ok) throw new Error((await res.text()) || res.status);
if (!res.ok) {
const detail = (await res.text().catch(() => "")).trim();
throw new Error(detail ? `the feeder rejected it (HTTP ${res.status}): ${detail}`
: `the feeder returned HTTP ${res.status}`);
}
const j = await res.json().catch(() => ({}));
if (j.restart_required) banner("warn", "Saved. Restart the feeder container for changes to take effect.");
else banner("ok", "Saved.");
// Re-fetch so secret pills reflect the new stored state.
await load();
if (j.reloading) {
// The feeder is restarting itself (~2-3s) to apply the change. Poll it
// back up, then refresh the form so secret pills reflect stored state.
banner("warn", "Saved — reloading the feeder to apply (~3 s)…");
let tries = 6;
const poll = async () => {
try { await load(); banner("ok", "Reloaded — new config applied."); }
catch { if (--tries > 0) setTimeout(poll, 1500); else banner("warn", "Saved & reloading — reopen this panel shortly to confirm."); }
};
setTimeout(poll, 3000);
} else {
if (j.restart_required) banner("warn", "Saved. Restart the feeder container for changes to take effect.");
else banner("ok", "Saved.");
// Re-fetch so secret pills reflect the new stored state.
await load();
}
} catch (e) {
banner("err", "Save failed: " + e.message);
const msg = (e && e.message) ? e.message : String(e);
const friendly = /Failed to fetch|NetworkError|load failed/i.test(msg)
? "couldn't reach the feeder — it may be restarting; wait a moment and try again."
: msg;
banner("err", "Could not save: " + friendly);
} finally {
$("save").disabled = false;
}
+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());
}
}
+31 -7
View File
@@ -73,16 +73,40 @@ fn env_nonempty(key: &str) -> Option<String> {
.filter(|s| !s.is_empty())
}
/// Derive a co-bundled sidecar plugin URL from the feeder's own hostname, by the
/// app-compose convention `<prefix>-feeder` → `<prefix>-<plugin>:8080`
/// (e.g. `metafeedertorznab-feeder` → `http://metafeedertorznab-tmdb:8080`). Lets
/// a feeder reach its bundled filename-parser / tmdb sidecars with **no explicit
/// `*_PLUGIN_URL` env** — config lives in the UI, wiring is derived. Returns
/// `None` when `HOSTNAME` is unset or not in `<prefix>-feeder` form, in which case
/// the caller falls back to the env seed (then soft-skips that plugin).
fn derive_sidecar_url(plugin: &str) -> Option<String> {
let host = env_nonempty("HOSTNAME")?;
let prefix = host.strip_suffix("-feeder")?;
Some(format!("http://{prefix}-{plugin}:8080"))
}
impl EnrichmentConfig {
/// Parse from the feeder env. Returns `None` when `META_CORE_URL` is unset —
/// the feeder then falls back to the legacy "return record, gateway stores"
/// path (backward compatible; used by tests and gateway-stores deployments).
/// the **bundled-feeder** path (records returned over HTTP, the gateway
/// dispatcher stores them). Self-publishing feeders that source meta-core from
/// their own dashboard config use [`from_meta_core`](Self::from_meta_core).
pub fn from_env() -> Option<Self> {
let meta_core_url = env_nonempty("META_CORE_URL")?;
Some(EnrichmentConfig {
Some(Self::from_meta_core(env_nonempty("META_CORE_URL")?))
}
/// Build with an explicit meta-core root — sourced from the feeder's own
/// `config.json` (dashboard) when set, else the `META_CORE_URL` env seed. The
/// sidecar URLs come from their `*_PLUGIN_URL` env, falling back to the
/// hostname-derived co-bundled sidecar (see [`derive_sidecar_url`]); the rest
/// keep their env seeds / defaults. Secrets (tmdb token, language) are
/// overlaid by the caller from the per-feeder config.
pub fn from_meta_core(meta_core_url: String) -> Self {
EnrichmentConfig {
meta_core_url,
filename_parser_url: env_nonempty("FILENAME_PARSER_URL"),
tmdb_url: env_nonempty("TMDB_PLUGIN_URL"),
filename_parser_url: env_nonempty("FILENAME_PARSER_URL")
.or_else(|| derive_sidecar_url("filename-parser")),
tmdb_url: env_nonempty("TMDB_PLUGIN_URL").or_else(|| derive_sidecar_url("tmdb")),
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()),
gateway_peer: env_nonempty("META_GATEWAY_PEER_ID")
@@ -96,7 +120,7 @@ impl EnrichmentConfig {
opensubtitles_password: env_nonempty("OPENSUBTITLES_PASSWORD"),
opensubtitles_languages: env_nonempty("OPENSUBTITLES_LANGUAGES")
.unwrap_or_else(|| "en".to_string()),
})
}
}
}
+27 -23
View File
@@ -57,9 +57,9 @@ pub struct SeasonEpisode {
/// is deliberately **not** given a single `episode`.
pub is_pack: bool,
/// True when [`season`] came from an explicit season-bearing token
/// (`S01E05`, `Season 2`, `S03`), as opposed to the "episode with no
/// season → assume 1" default. Lets a per-file reparse know whether to
/// borrow the season from the pack/parent title (see
/// (`S01E05`, `Season 2`, `S03`). An episode with no such token is left
/// season-less (no "assume season 1" default), so a per-file reparse uses
/// this to know it should borrow the season from the pack/parent title (see
/// [`extract_season_episode_with_context`]).
pub season_explicit: bool,
}
@@ -550,21 +550,22 @@ pub fn extract_season_episode(title: &str) -> SeasonEpisode {
let mut out = SeasonEpisode::default();
// season/episode 0 are valid; only < 0 means "not found".
let mut sea_eff = sea;
if sea >= 0.0 {
out.season = Some(fmt_num(sea));
out.season_explicit = true;
}
if ep >= 0.0 {
out.episode = Some(fmt_num(ep));
if sea_eff < 0.0 {
// An episode with no season — assume season 1 (matches TS). The
// season is *not* explicit, so a per-file reparse may override it
// from the parent (pack) title.
sea_eff = 1.0;
out.season = Some("1".to_string());
// An episode with no season is left season-LESS on purpose: a season is
// only asserted when explicitly hinted — an S-token in the title here,
// or (via extract_season_episode_with_context) the parent/pack title.
// Absolute-numbered releases ("Show - 28") thus carry an episode and no
// season, so the UI buckets them under "Episodes" rather than a
// fabricated "Season 1". `increment` encodes a season, so it is only
// emitted when the season is known.
if sea >= 0.0 {
out.increment = Some(fmt_num(sea * 10000.0 + ep));
}
out.increment = Some(fmt_num(sea_eff * 10000.0 + ep));
} else if sea >= 0.0 {
// Season present, no episode → a whole-season pack (`Season 3`, `S03`).
out.is_pack = true;
@@ -583,8 +584,8 @@ pub fn extract_season_episode(title: &str) -> SeasonEpisode {
pub fn extract_season_episode_with_context(parent: &str, filename: &str) -> SeasonEpisode {
let mut file_se = extract_season_episode(filename);
// Only borrow the parent's season for a concrete single episode whose own
// filename carries no *explicit* season token (so its season is the assumed
// `1` default). A per-file record is an episode, not a pack, so we don't
// filename carries no *explicit* season token (so it is otherwise
// season-less). A per-file record is an episode, not a pack, so we don't
// propagate the parent's pack/range shape here.
if file_se.episode.is_some() && !file_se.season_explicit {
if let Some(s) = find_pack_season(&preclean(parent)) {
@@ -1004,12 +1005,13 @@ mod tests {
}
#[test]
fn anime_trailing_number_assumes_season_one() {
// No season marker → episode via rightmost-number fallback,
// season assumed 1. SFV id stripped, brackets ignored.
fn anime_trailing_number_is_season_less() {
// No season marker → episode via rightmost-number fallback, and NO
// season is fabricated (absolute numbering). SFV id stripped, brackets
// ignored. The UI buckets a season-less episode under "Episodes".
assert_eq!(
se("[SubsPlease] Some Anime - 117 [FC412C51]"),
(Some("1".into()), Some("117".into()))
(None, Some("117".into()))
);
}
@@ -1184,8 +1186,8 @@ mod tests {
#[test]
fn per_file_context_recovers_season_from_parent() {
// Opened pack: the inner file carries the episode, the pack title the
// season. `Frieren - 03.mkv` alone assumes S1; with the S2 pack parent
// it resolves to S2E3.
// season. `Frieren - 03.mkv` alone is season-less; with the S2 pack
// parent it resolves to S2E3 (the parent is the "season hint").
let r = extract_season_episode_with_context("Frieren S2 (01-10) [Batch]", "Frieren - 03.mkv");
assert_eq!(r.season.as_deref(), Some("2"));
assert_eq!(r.episode.as_deref(), Some("3"));
@@ -1198,12 +1200,14 @@ mod tests {
}
#[test]
fn absolute_numbered_anime_still_assumes_season_one() {
// No season keyword → the rightmost-number fallback still fires, season
// assumed 1 (Phase 3 remaps this against TMDB episode counts later).
fn absolute_numbered_anime_is_season_less() {
// No season keyword → the rightmost-number fallback still finds the
// episode, but NO season is fabricated. Season is only ever asserted
// from an explicit token (title or parent), never inferred — so an
// absolute-numbered release lands in the UI's "Episodes" bucket.
assert_eq!(
se("[Naruto-Kun.Hu] Spy x Family - 37 [1080p].mkv"),
(Some("1".into()), Some("37".into()))
(None, Some("37".into()))
);
}
+588
View File
@@ -160,6 +160,366 @@ fn bt_file_cid(codec: u64, infohash: &[u8], file_index: u64) -> String {
format!("b{}", base32_lower_no_padding(&wire))
}
/// Custom multicodec for a self-describing **Newznab release** locator
/// (`nzb-release`). MetaMesh-private, adjacent to the torrent-file codecs in the
/// `0x10xx` range. Unlike a content hash, the cid *embeds* the indexer host +
/// release id in an identity multihash, so the credentialed meta-share peer
/// decodes `{host,id}` straight from the cid and grabs the `.nzb` via `t=get`
/// only at playback — no side-table, no `.nzb` download at search time. (The
/// `0x1004` `sha256(host‖id)` opaque hash and the `0x1003` `nzb-posting`
/// variant were both removed; only this self-describing form is emitted.)
pub const NZB_RELEASE_CODEC: u64 = 0x1005;
/// Digest byte ceiling for an `nzb-release` cid, matching meta-share's
/// `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer multihash.
const NZB_RELEASE_MAX_DIGEST: usize = 64;
/// Encode a self-describing `nzb-release` CID from a Newznab indexer's host +
/// bare release id (the hex `<guid>` id). The multihash is *identity* (`0x00`)
/// and its digest embeds the locator as `varint(host_len) ‖ host ‖ id_bytes`,
/// where `id_bytes` is the hex-decoded id. meta-share's `decode_nzb_release_cid`
/// is the exact inverse. Host-namespaced: the same release on two indexers
/// yields different cids.
///
/// Returns `None` when the id isn't hex or the digest would overflow the
/// multihash budget (an oversized host) — the caller drops that row.
pub fn compute_nzb_release_cid(api_base: &str, release_id: &str) -> Option<String> {
let id_bytes = decode_hex_id(release_id)?;
let host_b = api_base.as_bytes();
// identity-multihash digest = varint(host_len) ‖ host ‖ id_bytes
let mut digest = Vec::with_capacity(2 + host_b.len() + id_bytes.len());
write_pb_varint(host_b.len() as u64, &mut digest);
digest.extend_from_slice(host_b);
digest.extend_from_slice(&id_bytes);
if digest.len() > NZB_RELEASE_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(NZB_RELEASE_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)))
}
/// 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`
/// nibble. Returns `None` on any non-hex byte (defensive).
fn decode_hex_id(id: &str) -> Option<Vec<u8>> {
let padded;
let s: &str = if id.len() % 2 == 1 {
padded = format!("0{id}");
&padded
} else {
id
};
let bytes = s.as_bytes();
let mut out = Vec::with_capacity(bytes.len() / 2);
for pair in bytes.chunks(2) {
let hi = (pair[0] as char).to_digit(16)?;
let lo = (pair[1] as char).to_digit(16)?;
out.push(((hi << 4) | lo) as u8);
}
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.
/// The `url` locator codec (`0x1006`).
///
/// ⚠ Back-ported into this vendored SDK copy because the shared crate has it and
/// this one predates it. Keep it byte-identical to
/// `meta-feeder-sdk::hash::compute_url_cid`: it is how a record gets an address
/// without the feeder ever fetching bytes, and two encoders disagreeing would
/// mint two cids for one URL.
pub const URL_LOCATOR_CODEC: u64 = 0x1006;
/// Encode an `http(s)` URL as a `0x1006` identity-multihash CIDv1.
///
/// The digest IS the URL bytes, so this costs no I/O and no lookup — which is
/// what makes it usable at search time, on every record, before anything has
/// been fetched. Resolution happens later, on whichever peer wants the bytes.
///
/// `None` for anything that is not an absolute http(s) URL: a relative or
/// malformed value would encode fine and then fail unresolvably at read time,
/// which is a far more expensive way to find out.
pub fn compute_url_cid(url: &str) -> Option<String> {
let u = url.trim();
if !(u.starts_with("http://") || u.starts_with("https://")) {
return None;
}
let digest = u.as_bytes();
let mut wire = Vec::with_capacity(1 + 3 + 2 + digest.len());
wire.push(0x01); // CIDv1
write_pb_varint(URL_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)))
}
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
/// `ipfs add --cid-version=1 --raw-leaves=true --chunker=size-262144
/// --hash=sha2-256 <file>`.
@@ -456,6 +816,82 @@ mod tests {
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]
fn ipfs_cid_single_leaf_matches_kubo() {
assert_eq!(
@@ -518,4 +954,156 @@ mod tests {
assert_eq!(out.blocks[0].1.len(), 0);
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);
}
}
+8
View File
@@ -14,9 +14,17 @@
// pure noise for a single-process store, so silence the lint crate-wide.
#![allow(clippy::result_large_err)]
/// pcs-convention default meta-core root a self-publishing feeder targets when
/// neither its dashboard config nor the `META_CORE_URL` env seed sets one. Keeps
/// the modular storage seam out of compose: a standard install just works; a
/// non-standard meta-core is set in the feeder's config UI. Matches the
/// `metacore-app` service the AppStore feeder apps ship on the shared network.
pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
pub mod cache;
pub mod common;
pub mod config;
pub mod domain;
pub mod enrich;
pub mod filename_meta;
pub mod hash;
+42
View File
@@ -184,6 +184,48 @@ impl FeederStore {
&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
/// **sparse** outcome (`record: None`) so the gateway's `(_, None)` branch
/// skips it — no double write. Byte outcomes and already-sparse outcomes are
+55
View File
@@ -57,6 +57,30 @@ pub struct HashOutcome {
/// today emits.
/// - `BtV1File` — a single file inside a BitTorrent v1 torrent, custom
/// multicodec `0x1001`. Locator, not content hash: opaque to bitswap.
/// - `NzbRelease` — a Usenet release (a Newznab listing), custom multicodec
/// `nzb-release` `0x1005`. Self-describing locator: the cid embeds `{host,
/// 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
/// 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)]
pub enum HashKind {
/// Reserved: no feeder currently emits midhash256 outcomes, but the family
@@ -65,6 +89,10 @@ pub enum HashKind {
Midhash256,
Sha2_256,
BtV1File,
NzbRelease,
CardLocator,
NzbPosting,
YtVideo,
}
/// Startup-only configuration error returned by [`FeederPlugin::configure`].
@@ -101,6 +129,33 @@ pub fn upstream_id_field(upstream_id: &str) -> String {
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
/// exactly one `Box<dyn FeederPlugin>`. Lives behind `&self` in steady state —
/// 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;
}
// `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 {
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
/// 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 {
matches!(v.trim().to_ascii_lowercase().as_str(), "true" | "1" | "yes")
}
+147 -19
View File
@@ -54,6 +54,10 @@ pub enum HashKindDto {
Midhash256,
Sha2_256,
BtV1File,
NzbRelease,
CardLocator,
NzbPosting,
YtVideo,
}
impl From<HashKind> for HashKindDto {
@@ -62,6 +66,10 @@ impl From<HashKind> for HashKindDto {
HashKind::Midhash256 => HashKindDto::Midhash256,
HashKind::Sha2_256 => HashKindDto::Sha2_256,
HashKind::BtV1File => HashKindDto::BtV1File,
HashKind::NzbRelease => HashKindDto::NzbRelease,
HashKind::CardLocator => HashKindDto::CardLocator,
HashKind::NzbPosting => HashKindDto::NzbPosting,
HashKind::YtVideo => HashKindDto::YtVideo,
}
}
}
@@ -72,6 +80,10 @@ impl From<HashKindDto> for HashKind {
HashKindDto::Midhash256 => HashKind::Midhash256,
HashKindDto::Sha2_256 => HashKind::Sha2_256,
HashKindDto::BtV1File => HashKind::BtV1File,
HashKindDto::NzbRelease => HashKind::NzbRelease,
HashKindDto::CardLocator => HashKind::CardLocator,
HashKindDto::NzbPosting => HashKind::NzbPosting,
HashKindDto::YtVideo => HashKind::YtVideo,
}
}
}
@@ -424,8 +436,20 @@ async fn config_values_put(
std::fs::write(&tmp, body)
.and_then(|_| std::fs::rename(&tmp, &path))
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("write config: {e}")).into_response())?;
info!(target: "meta-feeder", upstream_id = id, path = %path.display(), "feeder config saved (restart to apply)");
Ok(Json(serde_json::json!({ "saved": true, "restart_required": true })))
info!(target: "meta-feeder", upstream_id = id, path = %path.display(), "feeder config saved; scheduling self-restart to apply");
// Apply the new config by restarting the process. We can't hot-swap config
// in-place: plugins hold an exclusive redb cache handle for their whole life,
// so a fresh instance can't reopen the cache while the old one is alive. A
// clean process exit is the reliable apply path — Docker's restart policy
// (`unless-stopped`) brings the feeder back in ~2-3s and `configure()`
// re-reads config.json at startup. Delay briefly so THIS response flushes to
// the client before we exit.
tokio::spawn(async {
tokio::time::sleep(std::time::Duration::from_millis(800)).await;
info!(target: "meta-feeder", "restarting now to apply saved config");
std::process::exit(0);
});
Ok(Json(serde_json::json!({ "saved": true, "reloading": true })))
}
/// `POST /enrich/callback` — sink for the enrichment plugins' completion
@@ -478,10 +502,13 @@ pub fn router(
/// Configure each plugin (cache dir under `state_dir/gateway/<id>`), then serve
/// the feeder HTTP contract on `listen` until shutdown.
///
/// A plugin whose `configure()` returns [`ConfigError::MissingConfig`] is
/// **soft-skipped** (warn + dropped) — mirroring the gateway registry's
/// behaviour, so a misconfigured plugin doesn't take the whole feeder down.
/// Any other `ConfigError` aborts startup.
/// The feeder process never aborts on a config problem: a plugin whose
/// `configure()` returns [`ConfigError::MissingConfig`] is **soft-skipped**
/// (warn + dropped — an optional source the operator didn't ask for), and any
/// other failure (bad/empty key, cache-dir setup, …) registers a
/// [`DegradedPlugin`] that stays up and reports the reason via `/health` and
/// `/manifest`. The container keeps running so the error is visible in the UX
/// instead of crash-looping.
pub async fn serve_feeders(
plugins: Vec<Box<dyn FeederPlugin>>,
state_dir: impl Into<PathBuf>,
@@ -503,9 +530,72 @@ pub async fn serve_feeders(
.map_err(anyhow::Error::from)
}
/// Run each plugin's `configure()` against its per-plugin cache dir, applying
/// the soft-skip rule, and return the surviving plugins keyed by `upstream_id`.
/// Public so feeder binaries / tests can configure without serving.
/// A plugin whose `configure()` (or cache-dir setup) failed. It stays
/// **registered** so the failure is visible at `GET /health` (as
/// [`PluginHealth::Degraded`]) and `GET /manifest`; any query/compute against it
/// returns the captured reason as a permanent error. This is what keeps one
/// misconfigured feeder from crash-looping the whole container — the operator
/// sees the reason in the UX and fixes the config, no restart-loop required.
struct DegradedPlugin {
id: &'static str,
reason: String,
file_types: &'static [&'static str],
content_kinds: &'static [&'static str],
/// Preserved from the wrapped plugin so a degraded/unconfigured feeder still
/// serves its config form — otherwise the dashboard shows "no configuration
/// required" and the operator can never fix the very config that's missing.
schema: crate::config::ConfigSchema,
}
#[async_trait::async_trait]
impl FeederPlugin for DegradedPlugin {
fn upstream_id(&self) -> &'static str {
self.id
}
fn configure(&mut self, _cache_dir: &FsPath) -> Result<(), ConfigError> {
Ok(())
}
async fn handle_query(
&self,
_query: &GatewayQuery,
_max_results: usize,
) -> Result<Vec<DiscoveryRecord>, GatewayError> {
Err(GatewayError::Permanent(self.reason.clone()))
}
async fn compute_outcomes(
&self,
_record_id: &str,
) -> Result<Vec<HashOutcome>, GatewayError> {
Err(GatewayError::Permanent(self.reason.clone()))
}
fn health(&self) -> PluginHealth {
PluginHealth::Degraded {
reason: self.reason.clone(),
}
}
fn served_file_types(&self) -> &'static [&'static str] {
self.file_types
}
fn served_content_kinds(&self) -> &'static [&'static str] {
self.content_kinds
}
fn config_schema(&self) -> crate::config::ConfigSchema {
self.schema.clone()
}
}
/// Run each plugin's `configure()` against its per-plugin cache dir and return
/// the plugins keyed by `upstream_id`. Never errors on a per-plugin problem:
/// `MissingConfig` is soft-skipped, every other failure becomes a
/// [`DegradedPlugin`] (see [`serve_feeders`]). Public so feeder binaries /
/// tests can configure without serving.
pub fn configure_plugins(
plugins: Vec<Box<dyn FeederPlugin>>,
state_dir: &FsPath,
@@ -513,25 +603,63 @@ pub fn configure_plugins(
let mut out: HashMap<String, Arc<dyn FeederPlugin>> = HashMap::new();
for mut plugin in plugins {
let id = plugin.upstream_id();
// Captured up front (both are `'static`, so this borrow of `plugin`
// ends immediately) — lets us build a DegradedPlugin without holding a
// borrow across `configure(&mut self)` / the move into `out`.
let file_types = plugin.served_file_types();
let content_kinds = plugin.served_content_kinds();
// Preserve the config surface so a degraded stand-in still serves the
// real config form (see DegradedPlugin.schema).
let schema = plugin.config_schema();
let cache_dir = state_dir.join("gateway").join(id);
std::fs::create_dir_all(&cache_dir).map_err(|e| {
anyhow::anyhow!("create cache dir {} for {id}: {e}", cache_dir.display())
})?;
// Build (don't insert) a degraded stand-in: keeps the container up and
// surfaces the reason via /health instead of crash-looping.
let make_degraded = |reason: String| -> Arc<dyn FeederPlugin> {
warn!(
target: "meta-feeder",
upstream_id = id,
%reason,
"feeder plugin degraded: container stays up; error surfaced via /health"
);
Arc::new(DegradedPlugin {
id,
reason,
file_types,
content_kinds,
schema: schema.clone(),
})
};
if let Err(e) = std::fs::create_dir_all(&cache_dir) {
let reason = format!("cache dir {} setup failed: {e}", cache_dir.display());
out.insert(id.to_string(), make_degraded(reason));
continue;
}
match plugin.configure(&cache_dir) {
Ok(()) => {
let id = id.to_string();
out.insert(id, Arc::from(plugin));
out.insert(id.to_string(), Arc::from(plugin));
}
Err(ConfigError::MissingConfig { plugin: p, what }) => {
// Keep a degraded stand-in (don't drop it) so its config form
// stays reachable from the dashboard and the operator can supply
// the missing config — then restart the feeder to apply it.
warn!(
target: "meta-feeder",
upstream_id = p,
missing = what,
"feeder plugin skipped: required config not supplied; \
other plugins continue"
upstream_id = %p,
missing = %what,
"feeder plugin degraded: required config not supplied; \
config form still served so it can be fixed from the dashboard"
);
out.insert(
id.to_string(),
make_degraded(format!("required config not supplied: {what}")),
);
}
Err(other) => return Err(anyhow::anyhow!(other)),
Err(other) => {
out.insert(id.to_string(), make_degraded(other.to_string()));
}
}
}
Ok(out)
+9 -4
View File
@@ -6,9 +6,10 @@ rust-version.workspace = true
license.workspace = true
publish = false
# Book feeder sidecar: Project Gutenberg (via Gutendex). The proof-of-concept
# feeder — single upstream, no preview-seeding, no BitTorrent — that validates
# the meta-feeder-sdk + feeder HTTP contract end to end.
# Book feeder sidecar, two upstreams:
# - `gutenberg` — Project Gutenberg editions (via Gutendex), the byte tier.
# - `openlibrary` — Open Library work cards, the book identity tier
# (`TMDB : video :: AniList : comics :: Open Library : books`).
[lib]
name = "book_feeder"
@@ -19,7 +20,11 @@ name = "book-feeder"
path = "src/main.rs"
[dependencies]
meta-feeder-sdk = { path = "../../crates/meta-feeder-sdk" }
# Pinned by TAG, never by path. A path dep resolves OUTSIDE the crate, and the
# vendored copy this used to point at had already drifted behind the shared SDK
# (no `RateBudget`, a capped card cid). The rest of the fleet pins the same
# way; see meta-feeder-anilist's manifest.
meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.7.0" }
tokio = { version = "1.40", features = ["full"] }
axum = { version = "0.7", features = ["json"] }
+7 -7
View File
@@ -1,12 +1,13 @@
# syntax=docker/dockerfile:1
# =============================================================================
# book-feeder — Project Gutenberg feeder sidecar. Multi-stage build.
# book-feeder — Project Gutenberg + Open Library feeder sidecar. Multi-stage
# build; build context is this repo's root (the Cargo workspace).
#
# The build context must be `packages/meta-gateway/` (the workspace root) so
# the `meta-feeder-sdk` path dependency (../../crates/meta-feeder-sdk) resolves.
# Same BuildKit cache-mount strategy as the gateway Dockerfile: the cargo
# registry/git and `target/` persist across builds. The release binary is
# copied OUT of the target cache mount within the same RUN.
# meta-feeder-sdk arrives as a git dependency (tag-pinned), so the vendored
# `crates/` copy is not copied in — `git` must be present in the builder for
# cargo to fetch it. Same BuildKit cache-mount strategy as the gateway
# Dockerfile: the cargo registry/git and `target/` persist across builds. The
# release binary is copied OUT of the target cache mount within the same RUN.
# =============================================================================
FROM rust:1.89-slim-bookworm AS builder
@@ -20,7 +21,6 @@ RUN apt-get update && apt-get install -y --no-install-recommends \
&& rm -rf /var/lib/apt/lists/*
COPY Cargo.toml Cargo.lock rust-toolchain.toml ./
COPY crates/ crates/
COPY feeder-plugin/ feeder-plugin/
RUN --mount=type=cache,id=cargo-registry,target=/usr/local/cargo/registry \
File diff suppressed because it is too large Load Diff
+11 -3
View File
@@ -1,5 +1,13 @@
//! `book-feeder` library surface — exposes the Gutenberg plugin so integration
//! tests (and, later, other binaries) can construct it. The binary
//! (`main.rs`) is a thin wrapper over [`meta_feeder_sdk::serve_feeders`].
//! `book-feeder` library surface — exposes the plugins so integration tests
//! (and, later, other binaries) can construct them. The binary (`main.rs`) is a
//! thin wrapper over [`meta_feeder_sdk::serve_feeders`].
//!
//! Two upstreams:
//! - [`gutenberg`] — Project Gutenberg editions, the byte tier (`document` × `book`);
//! - [`openlibrary`] — Open Library work cards, the identity tier (`card` × `book`),
//! over the client in [`openlibrary_client`], which `gutenberg` also uses to
//! bind an edition to its work.
pub mod gutenberg;
pub mod openlibrary;
pub mod openlibrary_client;
+26 -5
View File
@@ -1,17 +1,27 @@
//! `book-feeder` — Project Gutenberg feeder sidecar.
//! `book-feeder` — the book tier's feeder sidecar.
//!
//! Hosts the [`gutenberg::GutenbergPlugin`] over the feeder HTTP contract via
//! [`meta_feeder_sdk::serve_feeders`]. The gateway core reaches it server-to-
//! server on the internal docker network.
//! Hosts two plugins over the feeder HTTP contract via
//! [`meta_feeder_sdk::serve_feeders`]:
//! - [`GutenbergPlugin`] — Project Gutenberg editions (`document` × `book`);
//! - [`OpenLibraryPlugin`] — Open Library work cards (`card` × `book`).
//!
//! The gateway core reaches it server-to-server on the internal docker network.
//!
//! Env:
//! - `META_FEEDER_HTTP_LISTEN` — listen addr (default `0.0.0.0:8080`)
//! - `META_FEEDER_STATE_DIR` — per-plugin cache root (default `/data/meta-feeder`)
//! - `OPENLIBRARY_CONTACT` — e-mail/URL sent in the Open Library User-Agent;
//! first-boot seed for the `contact` config field (the saved field wins)
//! - `OPENLIBRARY_TOP_N`, `OPENLIBRARY_DISCOVERY_N` — optional card result sizing
//! - `RUST_LOG` — tracing filter (default `info`)
use std::net::SocketAddr;
use std::path::Path;
use book_feeder::gutenberg::GutenbergPlugin;
use book_feeder::openlibrary::OpenLibraryPlugin;
use book_feeder::openlibrary_client;
use meta_feeder_sdk::plugin::FeederPlugin;
use meta_feeder_sdk::serve_feeders;
use tracing_subscriber::EnvFilter;
@@ -27,5 +37,16 @@ async fn main() -> anyhow::Result<()> {
let state_dir =
std::env::var("META_FEEDER_STATE_DIR").unwrap_or_else(|_| "/data/meta-feeder".to_string());
serve_feeders(vec![Box::new(GutenbergPlugin::new())], state_dir, listen).await
// ONE budget for both plugins: the Gutenberg → Open Library binding and the
// card plugin call the same upstream, which rate-limits per client, not per
// plugin. Two buckets would each think they had the whole allowance.
let contact =
openlibrary_client::load_contact(&Path::new(&state_dir).join("gateway").join("openlibrary"));
let budget = openlibrary_client::budget(contact.is_some());
let plugins: Vec<Box<dyn FeederPlugin>> = vec![
Box::new(GutenbergPlugin::new().with_openlibrary_budget(budget.clone())),
Box::new(OpenLibraryPlugin::new().with_budget(budget)),
];
serve_feeders(plugins, state_dir, listen).await
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,843 @@
//! The Open Library HTTP client: rate budget, redb read-through cache, and the
//! calls the book tier makes.
//!
//! Shared by both upstreams this binary hosts:
//! - the `openlibrary` card plugin — search, trending, "new", work lookups;
//! - the `gutenberg` plugin — the `id_project_gutenberg` binding lookup
//! (reading-model §2.4), which is what lets an edition join its work.
//!
//! Plain JSON over `reqwest`, like every feeder in the fleet. Shapes confirmed
//! against the live API on 2026-09-11: `search.json` answers `{numFound, docs}`,
//! `/trending/daily.json` answers `{works}` with the same doc shape (and honours
//! `fields=`), and a work JSON's `description` is either a bare string or a
//! `{type: "/type/text", value}` object — both occur in the catalogue.
use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use meta_feeder_sdk::budget::{Lease, RateBudget};
use meta_feeder_sdk::cache::MidhashCache;
use meta_feeder_sdk::config::ConfigField;
use meta_feeder_sdk::types::GatewayError;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tracing::{debug, warn};
/// Canonical Open Library base URL. Overridable for tests (wiremock).
pub const DEFAULT_API_BASE: &str = "https://openlibrary.org";
/// Cover image base. `-L` is the large rendition; the gateway seeds whatever
/// `poster_url` names, so one size is published, not three.
const COVERS_BASE: &str = "https://covers.openlibrary.org/b/id";
/// Cover by ISBN. Open Library often holds an *edition's* cover for a work whose
/// own `covers` list is empty.
const COVERS_ISBN_BASE: &str = "https://covers.openlibrary.org/b/isbn";
/// Config key (and env seed) for the operator contact sent in the User-Agent.
pub const CONTACT_KEY: &str = "contact";
pub const CONTACT_ENV: &str = "OPENLIBRARY_CONTACT";
/// How long a caller waits for a rate-budget lease.
///
/// Three callers, three patiences. A **search** has someone watching a spinner.
/// A **browse row** is painted by a timer and is worth less late than never
/// blocking the search behind it. A **binding lookup** rides on a Gutenberg
/// result that is already in hand, and must never make that result late: an
/// unbound edition is still a readable edition, and the next query binds it.
pub const SEARCH_DEADLINE: Duration = Duration::from_secs(20);
pub const BROWSE_DEADLINE: Duration = Duration::from_secs(5);
pub const BINDING_DEADLINE: Duration = Duration::from_secs(2);
/// HTTP timeout for a single upstream call.
const HTTP_TIMEOUT_SECS: u64 = 20;
/// Attempts per call when the request never produced a response. openlibrary.org
/// refused connections outright for a stretch while this tier was being planned;
/// one retry covers a blink without hammering an upstream that answered.
const SEND_ATTEMPTS: usize = 2;
const RETRY_BACKOFF_MS: u64 = 250;
/// The field selection for every doc-shaped request, written once so a card
/// resolved by search and the same card resolved by id are byte-identical.
const SEARCH_FIELDS: &str = "key,title,author_name,cover_i,first_publish_year,language";
/// A lookup of one named work also asks for its ISBNs, so a cover can be found by
/// ISBN. Not on the shared list: a search row would carry hundreds of ISBNs per doc.
const DOC_BY_ID_FIELDS: &str = "key,title,author_name,cover_i,first_publish_year,language,isbn";
/// Cache lifetimes.
///
/// ⚠ Unlike AniList's, these lists are not immutable. `trending/daily` changes
/// every day and `sort=new` every hour; a permanent cache would freeze the
/// Popular Books row on whatever it showed the first time. Work JSON and author
/// names are edited rarely and are cached long.
const DAY: u64 = 86_400;
const WORK_TTL_SECS: u64 = 30 * DAY;
const SEARCH_TTL_SECS: u64 = DAY;
const BROWSE_TTL_SECS: u64 = 6 * 3600;
/// A binding is a fact Open Library records; it is re-checked monthly. A miss is
/// re-checked weekly, because Open Library librarians add Gutenberg ids to
/// editions all the time and a permanent negative would never see them.
const BINDING_HIT_TTL_SECS: u64 = 30 * DAY;
const BINDING_MISS_TTL_SECS: u64 = 7 * DAY;
/// Authors resolved on the work-JSON fallback. Each is one more request.
const MAX_AUTHORS: usize = 4;
/// Build the process budget for Open Library.
///
/// Open Library asks API clients to identify themselves (app name + contact in
/// the User-Agent) and throttles anonymous traffic first, so an identified
/// client gets a larger bucket. The exact published ceiling could not be read
/// back from the developer page when this was written — an overestimate is
/// safe regardless: a 429 freezes the whole bucket for its `Retry-After` via
/// `note_throttled`, rather than each caller discovering it alone.
pub fn budget(identified: bool) -> Arc<RateBudget> {
if identified {
RateBudget::new(3.0, 6.0)
} else {
RateBudget::new(1.0, 3.0)
}
}
/// The operator contact, file-or-env.
///
/// ⚠ **Looked up in both plugins' config dirs, not just the caller's.** The
/// SDK's `/config` page edits the *lowest* hosted upstream id, which in this
/// binary is `gutenberg`, so a contact saved through the dashboard lands in
/// `gateway/gutenberg/config.json` — yet both upstreams send requests to Open
/// Library and both must identify themselves. File wins over env, as for every
/// feeder config.
pub fn load_contact(cache_dir: &Path) -> Option<String> {
let mut dirs = vec![cache_dir.to_path_buf()];
if let Some(parent) = cache_dir.parent() {
for id in ["gutenberg", "openlibrary"] {
dirs.push(parent.join(id));
}
}
for dir in dirs {
let Ok(bytes) = std::fs::read(dir.join("config.json")) else {
continue;
};
let Ok(v) = serde_json::from_slice::<Value>(&bytes) else {
continue;
};
if let Some(c) = v.get(CONTACT_KEY).and_then(Value::as_str).and_then(sanitize_contact) {
return Some(c);
}
}
env_contact()
}
/// The first-boot seed: `OPENLIBRARY_CONTACT`.
pub fn env_contact() -> Option<String> {
std::env::var(CONTACT_ENV).ok().as_deref().and_then(sanitize_contact)
}
/// The config field both plugins declare (see [`load_contact`] for why both).
pub fn contact_field() -> ConfigField {
ConfigField::text(CONTACT_KEY, "Open Library contact").with_help(
"An e-mail address or URL sent in the User-Agent of every Open Library request \
(the openlibrary cards and the gutenberg → Open Library binding). Open Library \
asks API clients to identify themselves; an identified client gets a larger \
request budget.",
)
}
/// Reduce a contact to something a header can carry.
///
/// `build_http_client` treats a client build failure as impossible and panics,
/// and an invalid header value is exactly such a failure — so a stray newline
/// in a config field must not reach it. Parentheses are dropped too: they would
/// close the User-Agent comment early.
fn sanitize_contact(raw: &str) -> Option<String> {
let c: String = raw
.chars()
.filter(|c| c.is_ascii_graphic() || *c == ' ')
.filter(|c| !matches!(c, '(' | ')'))
.take(128)
.collect();
let c = c.trim();
(!c.is_empty()).then(|| c.to_string())
}
fn user_agent(contact: Option<&str>) -> String {
let base = concat!(
"meta-feeder-book/",
env!("CARGO_PKG_VERSION"),
" (gateway:openlibrary; +https://github.com/worph/meta-feeder-book"
);
match contact {
Some(c) => format!("{base}; {c})"),
None => format!("{base})"),
}
}
/// Is `s` a bare Open Library **work** id (`OL66554W`)?
///
/// ⚠ The `W` is the point. `OL…M` is an edition and `OL…A` an author, and the
/// search index does return edition keys under `/works/` for some junk records
/// (`/works/OL12710531M`, observed live on a `first_publish_year` query) — an
/// edition minted as a card would be a second tile for every printing.
pub fn is_work_id(s: &str) -> bool {
s.len() > 3
&& s.starts_with("OL")
&& s.ends_with('W')
&& s[2..s.len() - 1].bytes().all(|b| b.is_ascii_digit())
}
/// One search/trending doc — the subset this feeder reads.
///
/// ⚠ Nothing mutable is read: no `edition_count`, `ratings_*`,
/// `want_to_read_count` or `ebook_access`. All drift without the work changing
/// (METADATA_KEYS rule #4).
#[derive(Debug, Clone, Default, Deserialize)]
pub struct SearchDoc {
#[serde(default)]
pub key: String,
#[serde(default)]
pub title: Option<String>,
#[serde(default)]
pub author_name: Vec<String>,
#[serde(default)]
pub cover_i: Option<i64>,
#[serde(default)]
pub first_publish_year: Option<i32>,
#[serde(default)]
pub id_project_gutenberg: Vec<String>,
/// ISBNs of the work's editions. Only requested for an `openlibraryid`
/// lookup (see [`DOC_BY_ID_FIELDS`]); empty everywhere else.
#[serde(default)]
pub isbn: Vec<String>,
}
impl SearchDoc {
/// The bare work id, when the key names a work.
pub fn work_id(&self) -> Option<&str> {
self.key.strip_prefix("/works/").filter(|id| is_work_id(id))
}
/// The large cover URL. Open Library uses `-1` for "no cover" in `covers`
/// arrays, so anything not positive is absent.
pub fn cover_url(&self) -> Option<String> {
self.cover_i
.filter(|c| *c > 0)
.map(|c| format!("{COVERS_BASE}/{c}-L.jpg"))
}
/// A cover looked up by one of the work's ISBNs (an ISBN-13 when there is one),
/// for a work with no cover of its own.
///
/// `?default=false` makes Open Library answer 404 when it has no image for the
/// ISBN, instead of a 1×1 placeholder that would be seeded as the poster.
pub fn isbn_cover_url(&self) -> Option<String> {
let digits_only = |s: &&String| s.chars().all(|c| c.is_ascii_digit());
let isbn = self
.isbn
.iter()
.filter(digits_only)
.find(|s| s.len() == 13)
.or_else(|| self.isbn.iter().filter(digits_only).find(|s| s.len() == 10))?;
Some(format!("{COVERS_ISBN_BASE}/{isbn}-L.jpg?default=false"))
}
}
/// What a work JSON contributes to a card.
#[derive(Debug, Clone)]
pub struct WorkText {
/// The id the work JSON actually lives under — differs from the id asked
/// for when that one was merged away (see [`OpenLibraryClient::work_json`]).
pub id: String,
pub description: Option<String>,
}
/// Cache envelope: the body plus when it was fetched, so lists can expire.
#[derive(Serialize, Deserialize)]
struct Cached {
at: u64,
body: Value,
}
fn now_secs() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
pub struct OpenLibraryClient {
http: reqwest::Client,
budget: Arc<RateBudget>,
cache: Option<MidhashCache>,
/// Overridable so tests can point the client at a wiremock server.
api_base: String,
}
impl OpenLibraryClient {
pub fn new(
budget: Arc<RateBudget>,
cache: Option<MidhashCache>,
api_base: Option<String>,
contact: Option<&str>,
) -> Self {
Self {
http: meta_feeder_sdk::common::build_http_client(
HTTP_TIMEOUT_SECS,
&user_agent(contact),
None,
),
budget,
cache,
api_base: api_base.unwrap_or_else(|| DEFAULT_API_BASE.to_string()),
}
}
fn cache_get(&self, key: &str) -> Option<Cached> {
let hit = self.cache.as_ref()?.get_misc(key).ok().flatten()?;
serde_json::from_str::<Cached>(&hit).ok()
}
fn cache_put(&self, key: &str, body: &Value) {
let Some(c) = &self.cache else { return };
let envelope = Cached {
at: now_secs(),
body: body.clone(),
};
if let Ok(s) = serde_json::to_string(&envelope) {
if let Err(e) = c.put_misc(key, &s) {
warn!(upstream = "openlibrary", key, error = ?e, "cache put failed (non-fatal)");
}
}
}
/// One GET, budget-gated, never cached.
async fn get(
&self,
path: &str,
query: &[(&str, String)],
deadline: Duration,
) -> Result<Value, GatewayError> {
if matches!(self.budget.acquire(deadline).await, Lease::DeadlineExceeded) {
// Transient, not Permanent: this upstream is busy, not broken.
return Err(GatewayError::Transient(
"openlibrary: rate budget deadline exceeded".into(),
));
}
let url = format!("{}{path}", self.api_base.trim_end_matches('/'));
let mut last = String::new();
let mut response = None;
for attempt in 1..=SEND_ATTEMPTS {
match self.http.get(&url).query(query).send().await {
Ok(r) => {
response = Some(r);
break;
}
Err(e) => {
last = e.to_string();
warn!(upstream = "openlibrary", path, attempt, error = %last, "transport failure");
if attempt < SEND_ATTEMPTS {
tokio::time::sleep(Duration::from_millis(RETRY_BACKOFF_MS)).await;
}
}
}
}
let resp = response.ok_or_else(|| {
GatewayError::Transient(format!("openlibrary: GET {path}: {last}"))
})?;
// 429 carries Retry-After; freeze the whole bucket, not just this call.
if let Err(e) = meta_feeder_sdk::common::map_status(&resp) {
if let GatewayError::RateLimited { retry_after_s } = &e {
self.budget
.note_throttled(Duration::from_secs(*retry_after_s as u64));
}
return Err(e);
}
resp.json::<Value>()
.await
.map_err(|e| GatewayError::Transient(format!("openlibrary: bad json from {path}: {e}")))
}
/// [`Self::get`] behind the read-through cache. Only successes are cached:
/// a transport blink or a 5xx must never be remembered as "nothing here".
async fn get_cached(
&self,
key: &str,
ttl_secs: u64,
path: &str,
query: &[(&str, String)],
deadline: Duration,
) -> Result<Value, GatewayError> {
if let Some(c) = self.cache_get(key) {
if now_secs().saturating_sub(c.at) <= ttl_secs {
return Ok(c.body);
}
}
let body = self.get(path, query, deadline).await?;
self.cache_put(key, &body);
Ok(body)
}
/// Free-text search.
pub async fn search(
&self,
text: &str,
limit: usize,
deadline: Duration,
) -> Result<Vec<SearchDoc>, GatewayError> {
let normalised = text.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase();
let body = self
.get_cached(
&format!("openlibrary:search:{limit}:{normalised}"),
SEARCH_TTL_SECS,
"/search.json",
&[
("q", text.trim().to_string()),
("fields", SEARCH_FIELDS.into()),
("limit", limit.to_string()),
],
deadline,
)
.await?;
Ok(parse_docs(body.get("docs")))
}
/// The `popular:true` row: what Open Library readers opened today.
///
/// The ordering is a popularity signal, but no count is read or stored.
pub async fn trending(
&self,
limit: usize,
deadline: Duration,
) -> Result<Vec<SearchDoc>, GatewayError> {
let body = self
.get_cached(
&format!("openlibrary:trending:daily:{limit}"),
BROWSE_TTL_SECS,
"/trending/daily.json",
&[("fields", SEARCH_FIELDS.into()), ("limit", limit.to_string())],
deadline,
)
.await?;
Ok(parse_docs(body.get("works")))
}
/// The `fresh:true` row: recently added works first published this year or
/// last.
///
/// ⚠ `search.json` rejects `q=*` (422, "Query too short"), and `sort=new`
/// over a broad query is dominated by coverless catalogue stubs. The year
/// window keeps the row about new books; the card gate (cover + description)
/// drops the stubs that remain, so the caller over-fetches.
pub async fn fresh(
&self,
limit: usize,
deadline: Duration,
) -> Result<Vec<SearchDoc>, GatewayError> {
let year = 1970 + now_secs() / 31_556_952;
let body = self
.get_cached(
&format!("openlibrary:fresh:{year}:{limit}"),
BROWSE_TTL_SECS,
"/search.json",
&[
("q", format!("first_publish_year:[{} TO {year}]", year - 1)),
("sort", "new".into()),
("fields", SEARCH_FIELDS.into()),
("limit", limit.to_string()),
],
deadline,
)
.await?;
Ok(parse_docs(body.get("docs")))
}
/// Resolve one work id to a doc — the anchored lookup.
///
/// Search by key first, so the record matches what a search would have
/// produced (same fields, same `first_publish_year`; the work JSON's
/// `first_publish_date` is frequently an edition's date — `1853` for a novel
/// first published in 1813). The work JSON plus its author records are the
/// fallback for a work the search index does not hold; that path carries no
/// year, because absent is better than wrong.
pub async fn doc_by_work_id(
&self,
id: &str,
deadline: Duration,
) -> Result<Option<SearchDoc>, GatewayError> {
let body = self
.get_cached(
// `key2`: entries cached before ISBNs were requested lack them.
&format!("openlibrary:key2:{id}"),
WORK_TTL_SECS,
"/search.json",
&[
("q", format!("key:/works/{id}")),
("fields", DOC_BY_ID_FIELDS.into()),
("limit", "1".into()),
],
deadline,
)
.await?;
if let Some(doc) = parse_docs(body.get("docs"))
.into_iter()
.find(|d| d.work_id() == Some(id))
{
return Ok(Some(doc));
}
let Some((work_id, work)) = self.work_json(id, deadline).await? else {
return Ok(None);
};
let author_keys: Vec<String> = work
.get("authors")
.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|x| x.pointer("/author/key").and_then(Value::as_str))
.filter_map(|k| k.strip_prefix("/authors/"))
.map(str::to_string)
.take(MAX_AUTHORS)
.collect()
})
.unwrap_or_default();
let mut author_name = Vec::new();
for author in author_keys {
match self.author_name(&author, deadline).await {
Ok(Some(name)) => author_name.push(name),
Ok(None) => {}
Err(e) => {
debug!(upstream = "openlibrary", author, error = ?e, "author lookup failed")
}
}
}
Ok(Some(SearchDoc {
key: format!("/works/{work_id}"),
title: work.get("title").and_then(Value::as_str).map(str::to_string),
author_name,
cover_i: work
.get("covers")
.and_then(Value::as_array)
.and_then(|a| a.iter().filter_map(Value::as_i64).find(|c| *c > 0)),
first_publish_year: None,
id_project_gutenberg: Vec::new(),
isbn: Vec::new(),
}))
}
async fn author_name(&self, id: &str, deadline: Duration) -> Result<Option<String>, GatewayError> {
let body = match self
.get_cached(
&format!("openlibrary:author:{id}"),
WORK_TTL_SECS,
&format!("/authors/{id}.json"),
&[],
deadline,
)
.await
{
Ok(b) => b,
Err(GatewayError::NotFound) => return Ok(None),
Err(e) => return Err(e),
};
Ok(body
.get("name")
.and_then(Value::as_str)
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string))
}
/// The work JSON and the id it actually lives under, following one merge
/// redirect.
///
/// ⚠ A merged work answers **200** with `{"type": {"key": "/type/redirect"},
/// "location": "/works/OL…W"}` (observed live: `OL45883W` → `OL45804W`) —
/// not an HTTP redirect, so an HTTP client follows nothing and a naive
/// reader finds no description.
async fn work_json(
&self,
id: &str,
deadline: Duration,
) -> Result<Option<(String, Value)>, GatewayError> {
let mut id = id.to_string();
for _ in 0..2 {
let body = match self
.get_cached(
&format!("openlibrary:work:{id}"),
WORK_TTL_SECS,
&format!("/works/{id}.json"),
&[],
deadline,
)
.await
{
Ok(b) => b,
Err(GatewayError::NotFound) => return Ok(None),
Err(e) => return Err(e),
};
if body.pointer("/type/key").and_then(Value::as_str) != Some("/type/redirect") {
return Ok(Some((id, body)));
}
match body
.get("location")
.and_then(Value::as_str)
.and_then(|l| l.strip_prefix("/works/"))
.filter(|l| is_work_id(l))
{
Some(next) => id = next.to_string(),
None => return Ok(None),
}
}
Ok(None)
}
/// A work's canonical id and plain-text description.
pub async fn work_text(&self, id: &str, deadline: Duration) -> Result<Option<WorkText>, GatewayError> {
Ok(self.work_json(id, deadline).await?.map(|(id, work)| WorkText {
id,
description: text_value(work.get("description")).and_then(|d| clean_description(&d)),
}))
}
/// The Open Library work that records Project Gutenberg etext `gutenberg_id`.
///
/// `Some` only when **exactly one** work lists it — an identifier link Open
/// Library itself records, which is what grades a binding Anchored
/// (reading-model §2.4). Zero works is no binding; two or more is an
/// unmerged duplicate in Open Library, and picking one would be a guess.
///
/// ⚠ Never a title or author search. That is a Matched-grade binding and is
/// out of scope here; emitting it under an anchored claim would promote a
/// resemblance to a proof.
pub async fn gutenberg_work(
&self,
gutenberg_id: u64,
deadline: Duration,
) -> Result<Option<String>, GatewayError> {
let key = format!("openlibrary:gutenberg:{gutenberg_id}");
if let Some(c) = self.cache_get(&key) {
let work = c.body.get("work").and_then(Value::as_str).map(str::to_string);
let ttl = if work.is_some() {
BINDING_HIT_TTL_SECS
} else {
BINDING_MISS_TTL_SECS
};
if now_secs().saturating_sub(c.at) <= ttl {
return Ok(work);
}
}
let body = self
.get(
"/search.json",
&[
("q", format!("id_project_gutenberg:{gutenberg_id}")),
("fields", "key,id_project_gutenberg".into()),
// Two, not one: `limit=1` would make "exactly one" unprovable.
("limit", "2".into()),
],
deadline,
)
.await?;
let work = single_gutenberg_work(&body, gutenberg_id);
self.cache_put(&key, &serde_json::json!({ "work": work }));
Ok(work)
}
}
/// Decide a binding from a `q=id_project_gutenberg:<id>` response.
fn single_gutenberg_work(body: &Value, gutenberg_id: u64) -> Option<String> {
let docs = parse_docs(body.get("docs"));
let total = body
.get("numFound")
.or_else(|| body.get("num_found"))
.and_then(Value::as_u64)
.unwrap_or(docs.len() as u64);
if total != 1 || docs.len() != 1 {
return None;
}
let doc = &docs[0];
// The doc must itself list the etext. The query is an exact field match, so
// this only fails on an index quirk — and a binding graded Anchored cannot
// rest on a quirk.
let wanted = gutenberg_id.to_string();
if !doc.id_project_gutenberg.iter().any(|g| g.trim() == wanted) {
return None;
}
doc.work_id().map(str::to_string)
}
fn parse_docs(v: Option<&Value>) -> Vec<SearchDoc> {
v.and_then(Value::as_array)
.map(|a| {
a.iter()
.filter_map(|d| serde_json::from_value::<SearchDoc>(d.clone()).ok())
.collect()
})
.unwrap_or_default()
}
/// A text field that is either `"…"` or `{"type": "/type/text", "value": "…"}`.
fn text_value(v: Option<&Value>) -> Option<String> {
match v? {
Value::String(s) => Some(s.clone()),
Value::Object(o) => o.get("value").and_then(Value::as_str).map(str::to_string),
_ => None,
}
}
/// Community descriptions carry Windows line endings and, often, a trailing
/// `----------` rule followed by "Also contained in:" link lists. The rule and
/// what follows are catalogue furniture, not synopsis.
fn clean_description(raw: &str) -> Option<String> {
let text = raw.replace("\r\n", "\n").replace('\r', "\n");
let body = text.split("\n----").next().unwrap_or_default().trim();
(!body.is_empty()).then(|| body.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn work_ids_are_works_only() {
assert!(is_work_id("OL66554W"));
assert!(!is_work_id("OL12710531M"), "an edition key is not a work");
assert!(!is_work_id("OL21594A"), "an author key is not a work");
assert!(!is_work_id("OLW"));
assert!(!is_work_id("/works/OL66554W"));
}
#[test]
fn description_reads_both_live_shapes() {
let plain = serde_json::json!({ "description": "A novel." });
let typed = serde_json::json!({ "description": { "type": "/type/text", "value": "A novel." } });
assert_eq!(text_value(plain.get("description")).as_deref(), Some("A novel."));
assert_eq!(text_value(typed.get("description")).as_deref(), Some("A novel."));
assert_eq!(text_value(serde_json::json!({ "description": 3 }).get("description")), None);
}
#[test]
fn description_drops_the_link_list_after_the_rule() {
let raw = "First para.\r\n\r\nSecond para.\r\n\r\n----------\r\nAlso contained in:\r\n[x](/works/OL1W)";
assert_eq!(
clean_description(raw).as_deref(),
Some("First para.\n\nSecond para.")
);
assert_eq!(clean_description(" \r\n "), None);
}
#[test]
fn cover_minus_one_is_no_cover() {
let mut d = SearchDoc {
key: "/works/OL66554W".into(),
cover_i: Some(-1),
..Default::default()
};
assert_eq!(d.cover_url(), None);
d.cover_i = Some(14348537);
assert_eq!(
d.cover_url().as_deref(),
Some("https://covers.openlibrary.org/b/id/14348537-L.jpg")
);
}
#[test]
fn a_binding_needs_exactly_one_work_that_lists_the_etext() {
let one = serde_json::json!({ "numFound": 1, "docs": [
{ "key": "/works/OL66554W", "id_project_gutenberg": ["1342", "42671"] }
]});
assert_eq!(single_gutenberg_work(&one, 1342).as_deref(), Some("OL66554W"));
let none = serde_json::json!({ "numFound": 0, "docs": [] });
assert_eq!(single_gutenberg_work(&none, 84), None);
// `limit=2` returned two docs: an unmerged duplicate, not a binding.
let two = serde_json::json!({ "numFound": 2, "docs": [
{ "key": "/works/OL1W", "id_project_gutenberg": ["11"] },
{ "key": "/works/OL2W", "id_project_gutenberg": ["11"] }
]});
assert_eq!(single_gutenberg_work(&two, 11), None);
// More found than returned also counts as ambiguous.
let more = serde_json::json!({ "numFound": 3, "docs": [
{ "key": "/works/OL1W", "id_project_gutenberg": ["11"] }
]});
assert_eq!(single_gutenberg_work(&more, 11), None);
// A doc that does not list the etext itself proves nothing.
let unlisted = serde_json::json!({ "numFound": 1, "docs": [
{ "key": "/works/OL1W", "id_project_gutenberg": ["999"] }
]});
assert_eq!(single_gutenberg_work(&unlisted, 11), None);
// An edition key is never bound as a work.
let edition = serde_json::json!({ "numFound": 1, "docs": [
{ "key": "/works/OL12710531M", "id_project_gutenberg": ["11"] }
]});
assert_eq!(single_gutenberg_work(&edition, 11), None);
}
#[test]
fn contact_is_made_header_safe() {
assert_eq!(
sanitize_contact(" [email protected]\r\n(x) ").as_deref(),
Some("[email protected]")
);
assert_eq!(sanitize_contact("\n"), None);
assert!(user_agent(Some("[email protected]")).ends_with("; [email protected])"));
// Must build without panicking.
let _ = OpenLibraryClient::new(budget(false), None, None, Some("[email protected]"));
}
#[test]
fn a_dashboard_saved_contact_is_found_from_either_plugin_dir() {
let root = tempfile::tempdir().unwrap();
let gutenberg = root.path().join("gutenberg");
let openlibrary = root.path().join("openlibrary");
std::fs::create_dir_all(&gutenberg).unwrap();
std::fs::create_dir_all(&openlibrary).unwrap();
// The SDK's config page saves under the lowest upstream id.
std::fs::write(
gutenberg.join("config.json"),
r#"{"contact":"[email protected]"}"#,
)
.unwrap();
assert_eq!(load_contact(&openlibrary).as_deref(), Some("[email protected]"));
assert_eq!(load_contact(&gutenberg).as_deref(), Some("[email protected]"));
}
/// A cached miss is not re-asked inside its window, and a failure is never
/// cached at all.
#[tokio::test]
async fn binding_lookups_cache_answers_but_never_failures() {
use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/search.json"))
.and(query_param("q", "id_project_gutenberg:84"))
.respond_with(
ResponseTemplate::new(200).set_body_json(serde_json::json!({ "numFound": 0, "docs": [] })),
)
.expect(1)
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/search.json"))
.and(query_param("q", "id_project_gutenberg:11"))
.respond_with(ResponseTemplate::new(503))
.expect(2)
.mount(&server)
.await;
let dir = tempfile::tempdir().unwrap();
let cache = MidhashCache::open(dir.path()).unwrap();
let c = OpenLibraryClient::new(RateBudget::new(100.0, 100.0), Some(cache), Some(server.uri()), None);
assert_eq!(c.gutenberg_work(84, SEARCH_DEADLINE).await.unwrap(), None);
assert_eq!(c.gutenberg_work(84, SEARCH_DEADLINE).await.unwrap(), None);
assert!(c.gutenberg_work(11, SEARCH_DEADLINE).await.is_err());
assert!(c.gutenberg_work(11, SEARCH_DEADLINE).await.is_err());
}
}
+166 -22
View File
@@ -1,24 +1,36 @@
//! End-to-end contract test for the feeder HTTP surface: boots the real
//! `serve_feeders` router (book-feeder hosting GutenbergPlugin) on an ephemeral
//! port, pointed at a wiremock Gutendex, and drives `/manifest`, `/query`, and
//! `/compute` over HTTP. Deterministic — no live Gutendex.
//! `serve_feeders` router (book-feeder hosting GutenbergPlugin AND
//! OpenLibraryPlugin, as `main.rs` does) on an ephemeral port, pointed at one
//! wiremock server that plays Gutendex, the Gutenberg RDF host and Open Library,
//! and drives `/manifest`, `/query`, and `/compute` over HTTP. Deterministic —
//! no live upstream.
use book_feeder::gutenberg::GutenbergPlugin;
use book_feeder::openlibrary::OpenLibraryPlugin;
use book_feeder::openlibrary_client;
use meta_feeder_sdk::{
configure_plugins, hash, router, ComputeRequest, ComputeResponse, GatewayQuery, HashKindDto,
ManifestResponse, QueryRequest, QueryResponse,
configure_plugins, hash, router, ComputeRequest, ComputeResponse, FeederPlugin, GatewayQuery,
HashKindDto, ManifestResponse, QueryRequest, QueryResponse,
};
use wiremock::matchers::{method, path};
use wiremock::matchers::{method, path, query_param};
use wiremock::{Mock, MockServer, ResponseTemplate};
/// Spawn the feeder router (Gutenberg pointed at `upstream`) on an ephemeral
/// port; return its base URL. The TempDir is returned so the per-plugin cache
/// outlives the test.
/// Spawn the feeder router (every upstream pointed at `upstream`) on an
/// ephemeral port; return its base URL. The TempDir is returned so the
/// per-plugin cache outlives the test.
async fn spawn_feeder(upstream: &MockServer) -> (String, tempfile::TempDir) {
let dir = tempfile::tempdir().expect("tempdir");
let plugin = GutenbergPlugin::with_base_url(upstream.uri());
let plugins =
configure_plugins(vec![Box::new(plugin)], dir.path()).expect("configure plugins");
let budget = openlibrary_client::budget(false);
let plugins: Vec<Box<dyn FeederPlugin>> = vec![
Box::new(
GutenbergPlugin::with_base_url(upstream.uri())
.with_rdf_base_url(upstream.uri())
.with_openlibrary_base_url(upstream.uri())
.with_openlibrary_budget(budget.clone()),
),
Box::new(OpenLibraryPlugin::with_api_base(upstream.uri()).with_budget(budget)),
];
let plugins = configure_plugins(plugins, dir.path()).expect("configure plugins");
let app = router(plugins, "book-feeder-test".to_string(), dir.path());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -44,7 +56,7 @@ fn books_json() -> serde_json::Value {
}
#[tokio::test]
async fn manifest_lists_book_capabilities() {
async fn manifest_lists_both_upstreams() {
let upstream = MockServer::start().await;
let (base, _dir) = spawn_feeder(&upstream).await;
@@ -57,21 +69,36 @@ async fn manifest_lists_book_capabilities() {
.await
.expect("decode manifest");
assert_eq!(manifest.plugins.len(), 1);
let p = &manifest.plugins[0];
assert_eq!(p.id, "gutenberg");
assert_eq!(p.served_file_types, vec!["document".to_string()]);
assert_eq!(p.served_content_kinds, vec!["book".to_string()]);
assert_eq!(manifest.plugins.len(), 2);
let gutenberg = manifest.plugins.iter().find(|p| p.id == "gutenberg").expect("gutenberg");
assert_eq!(gutenberg.served_file_types, vec!["document".to_string()]);
assert_eq!(gutenberg.served_content_kinds, vec!["book".to_string()]);
let openlibrary = manifest
.plugins
.iter()
.find(|p| p.id == "openlibrary")
.expect("openlibrary");
assert_eq!(openlibrary.served_file_types, vec!["card".to_string()]);
assert_eq!(openlibrary.served_content_kinds, vec!["book".to_string()]);
}
#[tokio::test]
async fn query_returns_gutenberg_records() {
async fn query_returns_gutenberg_records_with_their_open_library_claim() {
let upstream = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/books"))
.and(path("/books/"))
.respond_with(ResponseTemplate::new(200).set_body_json(books_json()))
.mount(&upstream)
.await;
Mock::given(method("GET"))
.and(path("/search.json"))
.and(query_param("q", "id_project_gutenberg:1342"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"numFound": 1,
"docs": [{ "key": "/works/OL66554W", "id_project_gutenberg": ["1342", "42671"] }]
})))
.mount(&upstream)
.await;
let (base, _dir) = spawn_feeder(&upstream).await;
let resp: QueryResponse = reqwest::Client::new()
@@ -90,9 +117,75 @@ async fn query_returns_gutenberg_records() {
assert_eq!(resp.records.len(), 1);
let r = &resp.records[0];
let f = |k: &str| r.fields.get(k).map(String::as_str);
assert_eq!(r.upstream_id, "gutenberg");
assert_eq!(r.fields.get("gutenbergid").map(String::as_str), Some("1342"));
assert_eq!(r.fields.get("contentKind").map(String::as_str), Some("book"));
assert_eq!(f("gutenbergid"), Some("1342"));
assert_eq!(f("contentKind"), Some("book"));
assert_eq!(f("authors/Jane Austen"), Some("true"));
assert_eq!(f("languages/eng"), Some("true"));
// The transient claim the gateway transcribes into
// `anchoredBy/openlibrary:OL66554W`.
assert_eq!(f("openlibraryid"), Some("OL66554W"));
assert_eq!(f("anchored"), Some("true"));
assert_eq!(f("anchorSource"), Some("openlibrary"));
assert_eq!(f("anchorMethod"), None, "an identifier link is not a title match");
}
#[tokio::test]
async fn query_returns_an_open_library_card() {
let upstream = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/search.json"))
.and(query_param("q", "pride"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"numFound": 1,
"docs": [{
"key": "/works/OL66554W",
"title": "Pride and Prejudice",
"author_name": ["Jane Austen"],
"cover_i": 14348537,
"first_publish_year": 1813
}]
})))
.mount(&upstream)
.await;
Mock::given(method("GET"))
.and(path("/works/OL66554W.json"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"key": "/works/OL66554W",
"description": "A novel of manners."
})))
.mount(&upstream)
.await;
let (base, _dir) = spawn_feeder(&upstream).await;
let mut query = GatewayQuery::from_free_text("pride");
query.filters.insert("fileType".into(), vec!["card".into()]);
let resp: QueryResponse = reqwest::Client::new()
.post(format!("{base}/query"))
.json(&QueryRequest {
upstream_id: "openlibrary".to_string(),
query,
max_results: 10,
})
.send()
.await
.expect("POST /query")
.json()
.await
.expect("decode query response");
assert_eq!(resp.records.len(), 1);
let r = &resp.records[0];
assert_eq!(r.upstream_id, "openlibrary");
assert_eq!(r.record_id, "OL66554W");
assert_eq!(r.fields.get("fileType").map(String::as_str), Some("card"));
assert_eq!(
r.fields
.get("cids/bagdsaaaubnxxazlonruwe4tboj4u6tbwgy2tkncx")
.map(String::as_str),
Some("true")
);
}
#[tokio::test]
@@ -151,6 +244,57 @@ async fn compute_returns_outcome_with_inline_epub_bytes() {
assert_eq!(hash::compute_ipfs_cid(&decoded), expected_cid);
}
#[tokio::test]
async fn compute_for_an_open_library_card_is_a_byteless_card_locator() {
let upstream = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/search.json"))
.and(query_param("q", "key:/works/OL45804W"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"numFound": 1,
"docs": [{
"key": "/works/OL45804W",
"title": "Fantastic Mr Fox",
"author_name": ["Roald Dahl"],
"cover_i": 6498519,
"first_publish_year": 1970
}]
})))
.mount(&upstream)
.await;
Mock::given(method("GET"))
.and(path("/works/OL45804W.json"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"key": "/works/OL45804W",
"description": "A clever fox outwits three farmers."
})))
.mount(&upstream)
.await;
let (base, _dir) = spawn_feeder(&upstream).await;
let resp: ComputeResponse = reqwest::Client::new()
.post(format!("{base}/compute"))
.json(&ComputeRequest {
upstream_id: "openlibrary".to_string(),
record_id: "OL45804W".to_string(),
})
.send()
.await
.expect("POST /compute")
.json()
.await
.expect("decode compute response");
assert_eq!(resp.outcomes.len(), 1);
let o = &resp.outcomes[0];
assert_eq!(o.hash, "bagdsaaaubnxxazlonruwe4tboj4u6tbugu4dancx");
assert_eq!(o.hash_kind, HashKindDto::CardLocator);
assert!(o.bytes_b64.is_none());
assert!(o.file_extension.is_none());
let record = o.record.as_ref().expect("the card record rides the outcome");
assert_eq!(record.fields.get("title").map(String::as_str), Some("Fantastic Mr Fox"));
}
#[tokio::test]
async fn query_unknown_upstream_is_404() {
let upstream = MockServer::start().await;