diff --git a/README.md b/README.md index 97cf805..ab62c2d 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,15 @@ gateway queries), not a meta-sort *plugin*. The enrichment plugin that attaches subtitles to local files is the separate `metamesh-plugin-opensubtitle`. See `docs/project-architecture/plugins-vs-feeders.md` in the meta-root. -**Search only.** Nothing is downloaded, so the account's download quota is never -spent. Each result is one subtitle **file** addressed by a `provider-file` -locator (`0x100A`, `docs/cid-formats.md` §8) that embeds -`("opensubtitles", "file:")`. Retrieving the bytes is meta-share's job. +**Search is free.** Each result is one subtitle **file** addressed by a +`provider-file` locator (`0x100A`, `docs/cid-formats.md` §8) that embeds +`("opensubtitles", "file:")`; searching never spends download quota. + +**Downloads are redeems.** When a viewer plays a subtitle, the gateway asks this +feeder to redeem the locator (`POST /compute`), which calls OpenSubtitles +`/download` — one unit of the daily quota: the API key's anonymous quota, or an +account's rank quota when a login is configured. The gateway stores the file, so +each file is downloaded once. See [Downloads](#downloads). ## Queries @@ -43,9 +48,37 @@ There is no `contentKind` and no `domain`, because a subtitle is a sidecar. Resu |---|---|---| | API key (required) | `apiKey` | `OPENSUBTITLES_API_KEY` | | User-Agent (`App vX.Y`) | `userAgent` | `OPENSUBTITLES_USER_AGENT` | +| Account username (optional, opensubtitles.com) | `username` | `OPENSUBTITLES_USERNAME` | +| Account password (optional) | `password` | `OPENSUBTITLES_PASSWORD` | The config page (`/config`) wins over env. Changes apply on restart. Without a key -the feeder stays up, reports degraded on `/health`, and answers nothing. +the feeder stays up, reports degraded on `/health`, and answers nothing. Without a +login it still downloads, on the API key's anonymous quota; a login (a +**.com** account — .org accounts are a different API) switches to that account's +rank quota. + +## Downloads + +`POST /compute {upstream_id: "opensubtitles", record_id}` where `record_id` is the +`provider-file` cid (or `opensubtitles:file:`): + +1. Not an `opensubtitles` file locator → `404` (not ours). +2. `POST /download {file_id}` with the `Api-Key` header → a short-lived link, + fetched by a second client that follows redirects and sends only the + User-Agent. With a login configured the call also carries a Bearer token from + `POST /login` (cached; re-login once on a `401`/`403`). +3. Answer ONE `sha2_256` outcome: the bytes, `file_extension` from the file name + (`srt` when unknown), and a record holding only what the download knows — + `extension`, `fileName`, `sizeByte`. The gateway merges those onto the search + record (which already has title, language and ids) next to the `file` pointer. + +Quota: a `406`, or `remaining: 0` on the previous download, answers `429` with the +reset time (`reset_time_utc`, else `reset_time`, else 1 h) and makes no further +`/download` call until then. A file under 10 bytes → `422`. + +`GET /redeems` and `/manifest` advertise `{codec: provider-file, field: file, +sources: ["opensubtitles"]}` whenever an API key is configured; `package` is +`meta-feeder-opensubtitles` (`/files/plugin/meta-feeder-opensubtitles/`). Also `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`), `META_FEEDER_STATE_DIR` (default `/data/meta-feeder`), and `RUST_LOG`. @@ -70,3 +103,18 @@ docker build -f feeder-plugin/opensubtitles-feeder/Dockerfile -t meta-feeder-ope `tests/contract.rs` drives the real SDK router against a wiremock OpenSubtitles, using fixtures captured from the live API. + +The SDK is pinned by tag (`v1.3.0`). Until that tag is pushed, cargo cannot +resolve it — not even under a `[patch]` override — so the recipe above fails. To +verify against a local SDK checkout, run it on a scratch copy with the +dependency swapped for a path (never commit that form): + +```bash +SCRATCH=/d/workspace/tmp-claude/os-feeder; SDK=/d/workspace/MetaMesh/meta-root-v2/packages/plugins/meta-feeder-sdk +rm -rf $SCRATCH && mkdir -p $SCRATCH && tar cf - --exclude=./target --exclude=./.git . | (cd $SCRATCH && tar xf -) +sed -i "s#meta-feeder-sdk = { git = .*#meta-feeder-sdk = { path = \"$SDK\" }#" $SCRATCH/feeder-plugin/*/Cargo.toml +# then the docker `cargo test` above with -v /d/workspace:/d/workspace -w $SCRATCH +``` + +After the tag is pushed, run `cargo update -p meta-feeder-sdk` to move +`Cargo.lock` off the v1.2.1 revision. diff --git a/feeder-plugin/opensubtitles-feeder/Cargo.toml b/feeder-plugin/opensubtitles-feeder/Cargo.toml index 1b2f0ed..922c42a 100644 --- a/feeder-plugin/opensubtitles-feeder/Cargo.toml +++ b/feeder-plugin/opensubtitles-feeder/Cargo.toml @@ -6,9 +6,10 @@ rust-version.workspace = true license.workspace = true publish = false -# OpenSubtitles subtitle search. Answers `fileType:subtitle` queries — keyword -# and id-anchored — with `provider-file` (0x100A) locator records. Search only: -# no download, no quota spend. Retrieval is meta-share's job. +# OpenSubtitles subtitles. Answers `fileType:subtitle` queries — keyword and +# id-anchored — with `provider-file` (0x100A) locator records (free), and +# REDEEMS those locators through `/compute` with an account login (`/download`, +# which spends the account's download quota) when the gateway asks on a play. [lib] name = "opensubtitles_feeder" @@ -21,7 +22,7 @@ path = "src/main.rs" [dependencies] # Pinned by TAG, never by path. A path dep resolves OUTSIDE this repo, and this # repo is the Docker build context — see meta-feeder-anilist's manifest. -meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.2.1" } +meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.3.0" } tokio = { version = "1.40", features = ["full"] } axum = { version = "0.7", features = ["json"] } @@ -36,4 +37,5 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } [dev-dependencies] wiremock = "0.6" tempfile = "3" +base64 = "0.22" tokio = { version = "1.40", features = ["full", "test-util"] } diff --git a/feeder-plugin/opensubtitles-feeder/src/client.rs b/feeder-plugin/opensubtitles-feeder/src/client.rs index 1a89f0e..332d221 100644 --- a/feeder-plugin/opensubtitles-feeder/src/client.rs +++ b/feeder-plugin/opensubtitles-feeder/src/client.rs @@ -1,15 +1,18 @@ -//! OpenSubtitles REST v1 search client — `GET /subtitles` only. +//! OpenSubtitles REST v1 client — `GET /subtitles`, plus `POST /login` and +//! `POST /download` for redeeming a file. //! -//! Search is free: OpenSubtitles meters `/download`, which this feeder never -//! calls. What *is* limited is the request rate (5/s, measured), hence the -//! shared [`RateBudget`]. +//! Search is free. `/download` is **metered per account** (a daily quota), so +//! it runs only from `compute_outcomes`, which the gateway calls on a real play +//! and at most once per file (it stores the bytes). Every API call shares the +//! request-rate [`RateBudget`] (5/s, measured). //! //! ⚠ **OpenSubtitles 301s any query string that is not in its canonical form** -//! (keys sorted, values lowercased). Redirects are disabled on this client, so +//! (keys sorted, values lowercased). Redirects are disabled on the API client, so //! a drift in [`SearchParams::to_pairs`] surfaces as a loud `Permanent` error -//! instead of a silent extra round-trip per search. +//! instead of a silent extra round-trip per search. The short-lived download +//! *link* is fetched by a second client that does follow redirects. -use std::sync::Arc; +use std::sync::{Arc, Mutex as StdMutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use meta_feeder_sdk::budget::{Lease, RateBudget}; @@ -17,9 +20,12 @@ use meta_feeder_sdk::cache::{search_key, MidhashCache}; use meta_feeder_sdk::common::{build_http_client, map_status}; use meta_feeder_sdk::types::GatewayError; use serde::{Deserialize, Deserializer}; +use tokio::sync::Mutex; use tracing::debug; -use crate::consts::{HTTP_TIMEOUT_SECS, SEARCH_CACHE_TTL_SECS, SEARCH_DEADLINE_SECS}; +use crate::consts::{ + DEFAULT_QUOTA_RETRY_SECS, HTTP_TIMEOUT_SECS, SEARCH_CACHE_TTL_SECS, SEARCH_DEADLINE_SECS, +}; /// One search, already resolved from a gateway query. #[derive(Debug, Clone, Default, PartialEq, Eq)] @@ -228,14 +234,129 @@ fn fresh_body(entry: &str, now: u64, ttl_secs: u64) -> Option { v.get("body")?.as_str().map(str::to_string) } +// --- download shapes -------------------------------------------------------- + +#[derive(Debug, Clone, Default, Deserialize)] +struct LoginResponse { + #[serde(default)] + token: Option, +} + +/// `POST /download` answer (also the body of its `406` quota refusal). +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(default)] +pub struct DownloadResponse { + pub link: Option, + pub file_name: Option, + #[serde(deserialize_with = "lenient_u64")] + pub remaining: Option, + /// Human text, e.g. `23 hours and 59 minutes`. + pub reset_time: Option, + /// e.g. `2022-04-08T13:03:16.000Z`. + pub reset_time_utc: Option, + pub message: Option, +} + +/// One redeemed subtitle file. +#[derive(Debug, Clone)] +pub struct Downloaded { + pub bytes: Vec, + pub file_name: Option, + pub remaining: Option, +} + +/// Seconds until the download quota resets: `reset_time_utc` if parseable, +/// else `reset_time` text, else [`DEFAULT_QUOTA_RETRY_SECS`]. Clamped to a day. +pub fn quota_retry_secs(reset_time_utc: Option<&str>, reset_time: Option<&str>, now: u64) -> u32 { + let secs = reset_time_utc + .and_then(parse_rfc3339_utc) + .filter(|at| *at > now) + .map(|at| at - now) + .or_else(|| reset_time.and_then(parse_duration_text).filter(|s| *s > 0)); + match secs { + Some(s) => s.clamp(1, 86_400) as u32, + None => DEFAULT_QUOTA_RETRY_SECS, + } +} + +/// `YYYY-MM-DDTHH:MM:SS[.fff](Z|+00:00)` → unix seconds. UTC only — which is +/// all OpenSubtitles sends in `reset_time_utc`. +fn parse_rfc3339_utc(s: &str) -> Option { + let s = s.trim(); + let (date, time) = s.split_once('T')?; + let time = time + .strip_suffix('Z') + .or_else(|| time.strip_suffix("+00:00"))?; + let time = time.split('.').next()?; + let mut d = date.split('-').map(|p| p.parse::()); + let (y, mo, da) = (d.next()?.ok()?, d.next()?.ok()?, d.next()?.ok()?); + let mut t = time.split(':').map(|p| p.parse::()); + let (h, mi, se) = (t.next()?.ok()?, t.next()?.ok()?, t.next()?.ok()?); + if !(1..=12).contains(&mo) || !(1..=31).contains(&da) || h > 23 || mi > 59 || se > 60 { + return None; + } + let secs = days_from_civil(y, mo, da) * 86_400 + h * 3600 + mi * 60 + se; + u64::try_from(secs).ok() +} + +/// Days since 1970-01-01 for a proleptic Gregorian date (Howard Hinnant). +fn days_from_civil(y: i64, m: i64, d: i64) -> i64 { + let y = if m <= 2 { y - 1 } else { y }; + let era = if y >= 0 { y } else { y - 399 } / 400; + let yoe = y - era * 400; + let mp = (m + 9) % 12; + let doy = (153 * mp + 2) / 5 + d - 1; + let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; + era * 146_097 + doe - 719_468 +} + +/// `23 hours and 59 minutes` → seconds. Unknown shapes → `None`. +fn parse_duration_text(s: &str) -> Option { + let mut total = 0u64; + let mut pending: Option = None; + let mut matched = false; + for word in s.split(|c: char| !c.is_ascii_alphanumeric()).filter(|w| !w.is_empty()) { + if let Ok(n) = word.parse::() { + pending = Some(n); + continue; + } + let Some(n) = pending.take() else { continue }; + let unit = word.to_ascii_lowercase(); + let mult = if unit.starts_with("hour") { + 3600 + } else if unit.starts_with("minute") { + 60 + } else if unit.starts_with("second") { + 1 + } else { + continue; + }; + total += n * mult; + matched = true; + } + matched.then_some(total) +} + // --- client ----------------------------------------------------------------- pub struct OsClient { http: reqwest::Client, + /// Fetches the pre-signed download link: redirects allowed (it may bounce + /// to a CDN), User-Agent only — no Api-Key, no Bearer. + download_http: reqwest::Client, api_base: String, api_key: String, budget: Arc, cache: Option, + /// Optional account login for `/download`. `None` → downloads run on the + /// API key alone (OpenSubtitles' anonymous quota); `Some` → a Bearer token + /// from `/login`, so the account's rank quota applies. + login: Option<(String, String)>, + /// Cached JWT from `/login`; dropped and re-fetched once on a 401/403. + token: Mutex>, + /// Unix time until which the download quota is known to be spent — no + /// `/download` call is made before it. + quota_until: StdMutex>, } impl OsClient { @@ -251,15 +372,209 @@ impl OsClient { user_agent, Some(reqwest::redirect::Policy::none()), ); + let download_http = build_http_client(HTTP_TIMEOUT_SECS, user_agent, None); Self { http, + download_http, api_base: api_base.trim_end_matches('/').to_string(), api_key, budget, cache, + login: None, + token: Mutex::new(None), + quota_until: StdMutex::new(None), } } + /// Download as an account instead of on the bare API key. Blank credentials + /// keep the anonymous (API key only) mode. + pub fn with_login(mut self, username: &str, password: &str) -> Self { + let (u, p) = (username.trim(), password.trim()); + if !u.is_empty() && !p.is_empty() { + self.login = Some((u.to_string(), p.to_string())); + } + self + } + + /// True when downloads go through an account login rather than the bare + /// API key. + pub fn has_login(&self) -> bool { + self.login.is_some() + } + + /// One token from the shared request-rate budget. Busy is Transient. + async fn api_lease(&self) -> Result<(), GatewayError> { + if matches!( + self.budget.acquire(Duration::from_secs(SEARCH_DEADLINE_SECS)).await, + Lease::DeadlineExceeded + ) { + return Err(GatewayError::Transient( + "opensubtitles: rate budget deadline exceeded".into(), + )); + } + Ok(()) + } + + /// Freeze the whole bucket on a 429, not just this caller. + fn note_error(&self, e: &GatewayError) { + if let GatewayError::RateLimited { retry_after_s } = e { + self.budget + .note_throttled(Duration::from_secs(u64::from(*retry_after_s))); + } + } + + fn quota_retry_after(&self, now: u64) -> Option { + let until = (*self.quota_until.lock().expect("quota mutex"))?; + (until > now).then(|| (until - now).min(u64::from(u32::MAX)) as u32) + } + + fn block_quota_for(&self, secs: u32, now: u64) { + *self.quota_until.lock().expect("quota mutex") = Some(now + u64::from(secs)); + } + + async fn login(&self) -> Result { + let Some((username, password)) = &self.login else { + return Err(GatewayError::Permanent( + "opensubtitles: login requested without configured credentials".into(), + )); + }; + self.api_lease().await?; + let resp = self + .http + .post(format!("{}/login", self.api_base)) + .header("Api-Key", &self.api_key) + .header(reqwest::header::ACCEPT, "application/json") + .json(&serde_json::json!({ "username": username, "password": password })) + .send() + .await + .map_err(|e| GatewayError::Transient(format!("opensubtitles login: {e}")))?; + if let Err(e) = map_status(&resp) { + self.note_error(&e); + return Err(match e { + // A missing login endpoint is a broken config, not "not mine". + GatewayError::NotFound => { + GatewayError::Permanent("opensubtitles login: endpoint not found".into()) + } + other => other, + }); + } + let body: LoginResponse = resp + .json() + .await + .map_err(|e| GatewayError::Transient(format!("opensubtitles login: bad json: {e}")))?; + body.token + .filter(|t| !t.is_empty()) + .ok_or_else(|| GatewayError::Permanent("opensubtitles login returned no token".into())) + } + + /// The cached token, logging in when there is none (or `refresh` drops it). + async fn bearer(&self, refresh: bool) -> Result { + let mut slot = self.token.lock().await; + if refresh { + *slot = None; + } + if let Some(t) = slot.as_ref() { + return Ok(t.clone()); + } + let t = self.login().await?; + *slot = Some(t.clone()); + Ok(t) + } + + async fn post_download(&self, token: Option<&str>, file_id: u64) -> Result { + self.api_lease().await?; + let mut req = self + .http + .post(format!("{}/download", self.api_base)) + .header("Api-Key", &self.api_key) + .header(reqwest::header::ACCEPT, "application/json") + .json(&serde_json::json!({ "file_id": file_id })); + if let Some(token) = token { + req = req.bearer_auth(token); + } + req.send() + .await + .map_err(|e| GatewayError::Transient(format!("opensubtitles download: {e}"))) + } + + /// Redeem one subtitle file: `/download` for a short-lived link, then fetch + /// it. ⚠ Spends one unit of the daily download quota — the API key's + /// anonymous quota, or the account's when a login is configured. + /// + /// A spent quota (`406`, or `remaining: 0` on the previous call) is + /// `RateLimited` with the reset time, and blocks further `/download` calls + /// until then. + pub async fn download(&self, file_id: u64) -> Result { + if let Some(retry_after_s) = self.quota_retry_after(now_secs()) { + return Err(GatewayError::RateLimited { retry_after_s }); + } + + let resp = if self.has_login() { + let token = self.bearer(false).await?; + let resp = self.post_download(Some(&token), file_id).await?; + if matches!(resp.status().as_u16(), 401 | 403) { + debug!(target: "opensubtitles", status = %resp.status(), "download refused the token; logging in again"); + let token = self.bearer(true).await?; + self.post_download(Some(&token), file_id).await? + } else { + resp + } + } else { + self.post_download(None, file_id).await? + }; + if resp.status().as_u16() == 406 { + let body: DownloadResponse = resp.json().await.unwrap_or_default(); + let now = now_secs(); + let retry = quota_retry_secs(body.reset_time_utc.as_deref(), body.reset_time.as_deref(), now); + self.block_quota_for(retry, now); + return Err(GatewayError::RateLimited { retry_after_s: retry }); + } + if let Err(e) = map_status(&resp) { + self.note_error(&e); + return Err(e); + } + let body: DownloadResponse = resp + .json() + .await + .map_err(|e| GatewayError::Transient(format!("opensubtitles download: bad json: {e}")))?; + let link = body + .link + .clone() + .filter(|l| !l.is_empty()) + .ok_or_else(|| GatewayError::Transient("opensubtitles download returned no link".into()))?; + if body.remaining == Some(0) { + // This file is served; the next one would be refused. + let now = now_secs(); + let retry = quota_retry_secs(body.reset_time_utc.as_deref(), body.reset_time.as_deref(), now); + self.block_quota_for(retry, now); + } + + // The link is pre-signed: keep it out of error messages. + let got = self + .download_http + .get(&link) + .send() + .await + .map_err(|e| GatewayError::Transient(format!("opensubtitles file fetch: {}", e.without_url())))?; + let status = got.status(); + if !status.is_success() { + return Err(if status.is_server_error() { + GatewayError::Transient(format!("opensubtitles file fetch: HTTP {status}")) + } else { + GatewayError::Permanent(format!("opensubtitles file fetch: HTTP {status}")) + }); + } + let bytes = got + .bytes() + .await + .map_err(|e| GatewayError::Transient(format!("opensubtitles file fetch: {}", e.without_url())))?; + Ok(Downloaded { + bytes: bytes.to_vec(), + file_name: body.file_name, + remaining: body.remaining, + }) + } + pub async fn search(&self, params: &SearchParams) -> Result, GatewayError> { let pairs = params.to_pairs(); let key = cache_key(&pairs); @@ -376,6 +691,21 @@ mod tests { assert_eq!(fresh_body("garbage", 0, 60), None); } + #[test] + fn quota_reset_prefers_the_utc_timestamp() { + // 2022-04-08T13:03:16Z = 1649422996. + assert_eq!(parse_rfc3339_utc("2022-04-08T13:03:16.000Z"), Some(1_649_422_996)); + assert_eq!(parse_rfc3339_utc("1970-01-01T00:00:00Z"), Some(0)); + assert_eq!(parse_rfc3339_utc("yesterday"), None); + let now = 1_649_422_996 - 90; + assert_eq!(quota_retry_secs(Some("2022-04-08T13:03:16.000Z"), None, now), 90); + // Falls back to the text, then to the default. + assert_eq!(quota_retry_secs(None, Some("23 hours and 59 minutes"), now), 86_340); + assert_eq!(quota_retry_secs(Some("garbage"), Some("soon"), now), DEFAULT_QUOTA_RETRY_SECS); + // A reset already in the past is not a retry time. + assert_eq!(quota_retry_secs(Some("1970-01-01T00:00:00Z"), None, now), DEFAULT_QUOTA_RETRY_SECS); + } + #[test] fn lenient_numbers_survive_odd_upstream_values() { let a: OsAttributes = serde_json::from_str( diff --git a/feeder-plugin/opensubtitles-feeder/src/consts.rs b/feeder-plugin/opensubtitles-feeder/src/consts.rs index 5caf74a..8ccafb8 100644 --- a/feeder-plugin/opensubtitles-feeder/src/consts.rs +++ b/feeder-plugin/opensubtitles-feeder/src/consts.rs @@ -32,6 +32,22 @@ pub(crate) const HTTP_TIMEOUT_SECS: u64 = 20; /// and download counts, and download count is half of the ranking. pub(crate) const SEARCH_CACHE_TTL_SECS: u64 = 24 * 60 * 60; +/// `/files/plugin//` — where the gateway stores downloaded subtitles. +pub const PACKAGE: &str = "meta-feeder-opensubtitles"; + +/// `retry_after_s` when the download quota is spent and OpenSubtitles gave no +/// parseable reset time. The quota is daily. +pub(crate) const DEFAULT_QUOTA_RETRY_SECS: u32 = 3600; + +/// A "subtitle" shorter than this is an error page or an empty file, not a +/// subtitle — refuse it rather than seed it to the mesh. +pub(crate) const MIN_SUBTITLE_BYTES: usize = 10; + +/// Extension used when the downloaded file name has no known subtitle suffix. +pub(crate) const DEFAULT_SUBTITLE_EXTENSION: &str = "srt"; + /// Env seeds; the config page's saved values win over them. pub(crate) const ENV_API_KEY: &str = "OPENSUBTITLES_API_KEY"; pub(crate) const ENV_USER_AGENT: &str = "OPENSUBTITLES_USER_AGENT"; +pub(crate) const ENV_USERNAME: &str = "OPENSUBTITLES_USERNAME"; +pub(crate) const ENV_PASSWORD: &str = "OPENSUBTITLES_PASSWORD"; diff --git a/feeder-plugin/opensubtitles-feeder/src/lib.rs b/feeder-plugin/opensubtitles-feeder/src/lib.rs index f1b0b1a..1eeb587 100644 --- a/feeder-plugin/opensubtitles-feeder/src/lib.rs +++ b/feeder-plugin/opensubtitles-feeder/src/lib.rs @@ -1,10 +1,13 @@ -//! `opensubtitles-feeder` — subtitle search over OpenSubtitles. +//! `opensubtitles-feeder` — subtitle search and download over OpenSubtitles. //! -//! A **sidecar** tier: it answers "which subtitles exist for X", never "give me -//! the bytes". Each result is one subtitle *file*, addressed by a -//! `provider-file` locator (`0x100A`, `docs/cid-formats.md` §8) that embeds -//! `("opensubtitles", "file:")`. meta-share resolves that locator later; -//! nothing here downloads, so searching never spends the account's download quota. +//! A **sidecar** tier. Search answers "which subtitles exist for X": each result +//! is one subtitle *file*, addressed by a `provider-file` locator (`0x100A`, +//! `docs/cid-formats.md` §8) that embeds `("opensubtitles", "file:")`. +//! Searching never spends the account's download quota. +//! +//! Redeeming a locator — `POST /compute` with the cid, which the gateway sends +//! on a real play — logs in and calls `/download`, spending one unit of quota; +//! the gateway stores the bytes so each file is downloaded once. //! //! ## The contracts that must not drift //! diff --git a/feeder-plugin/opensubtitles-feeder/src/locator.rs b/feeder-plugin/opensubtitles-feeder/src/locator.rs index 2fb43c1..10db86b 100644 --- a/feeder-plugin/opensubtitles-feeder/src/locator.rs +++ b/feeder-plugin/opensubtitles-feeder/src/locator.rs @@ -10,24 +10,15 @@ //! digest : varint(len(source)) ‖ source ‖ id (both UTF-8, ≤ 64 bytes) //! ``` //! -//! The framing is byte-identical to `card` (`0x1007`) — only the codec differs — -//! which is why the tests below re-derive the existing `card-locator` vector -//! with the same encoder. -//! -//! ⚠ **This is a local copy.** The SDK's framing helpers (`kinded_locator_cid`, -//! `write_pb_varint`, `base32_lower_no_padding`) are private, so it could not be -//! reused. Move it into `meta-feeder-sdk` as `encode_source_id_locator(codec, -//! source, id)` (§8 trap 7) and delete this file; the golden vectors are what -//! prove the move changed no bytes. +//! The encoder/decoder live in `meta-feeder-sdk` (`hash::compute_provider_file_cid` +//! / `decode_provider_file_cid`, with the §8.2 golden vector); this module only +//! binds them to OpenSubtitles' `file:` id form. use crate::consts::UPSTREAM_ID; -/// Custom multicodec for a fetchable `(source, id)` locator. -pub const PROVIDER_FILE_CODEC: u64 = 0x100A; - -/// meta-share's `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer -/// multihash, so a longer digest would mint a CID no peer can parse. -const MAX_DIGEST_BYTES: usize = 64; +pub use meta_feeder_sdk::hash::{ + compute_provider_file_cid as provider_file_cid, decode_provider_file_cid, PROVIDER_FILE_CODEC, +}; /// The locator for one OpenSubtitles **file** (`/download` takes a `file_id`; /// one subtitle listing can hold several files, one per CD). @@ -35,156 +26,28 @@ pub fn file_cid(file_id: u64) -> Option { provider_file_cid(UPSTREAM_ID, &format!("file:{file_id}")) } -/// Encode a `provider-file` CID. `None` when the digest would exceed 64 bytes — -/// the caller drops that row rather than emit an unparseable address. -pub fn provider_file_cid(source: &str, id: &str) -> Option { - source_id_locator(PROVIDER_FILE_CODEC, source, id) -} - -/// Inverse of [`provider_file_cid`]: `(source, id)`, or `None` for anything that -/// is not exactly a well-formed `0x100A` CID. Matches the **codec slot** — a -/// `card` CID has the same framing and must not decode here. -pub fn decode_provider_file_cid(cid: &str) -> Option<(String, String)> { - let wire = base32_lower_no_pad_decode(cid.strip_prefix('b')?)?; - let mut pos = 0; - if read_varint(&wire, &mut pos)? != 1 { +/// The OpenSubtitles `file_id` a `/compute` record id names, if it is ours. +/// +/// Accepts both shapes a caller holds: the `provider-file` cid (what the +/// gateway's redeem route sends) and this feeder's own record id +/// `opensubtitles:file:`. A cid for another source is `None` — not ours. +pub fn file_id_of(record_id: &str) -> Option { + if let Some(id) = record_id.strip_prefix(&format!("{UPSTREAM_ID}:file:")) { + return id.parse().ok(); + } + let (source, id) = decode_provider_file_cid(record_id)?; + if source != UPSTREAM_ID { return None; } - if read_varint(&wire, &mut pos)? != PROVIDER_FILE_CODEC { - return None; - } - if read_varint(&wire, &mut pos)? != 0x00 { - return None; - } - let len = read_varint(&wire, &mut pos)? as usize; - let digest = wire.get(pos..)?; - if digest.len() != len { - return None; - } - let mut dpos = 0; - let source_len = read_varint(digest, &mut dpos)? as usize; - let source = digest.get(dpos..dpos.checked_add(source_len)?)?; - let id = digest.get(dpos + source_len..)?; - Some(( - String::from_utf8(source.to_vec()).ok()?, - String::from_utf8(id.to_vec()).ok()?, - )) -} - -fn source_id_locator(codec: u64, source: &str, id: &str) -> Option { - let source = source.as_bytes(); - let id = id.as_bytes(); - - // identity-multihash digest = varint(source_len) ‖ source ‖ id - let mut digest = Vec::with_capacity(2 + source.len() + id.len()); - write_varint(source.len() as u64, &mut digest); - digest.extend_from_slice(source); - digest.extend_from_slice(id); - if digest.len() > MAX_DIGEST_BYTES { - return None; - } - - // CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest] - let mut wire = Vec::with_capacity(6 + digest.len()); - wire.push(0x01); - write_varint(codec, &mut wire); - wire.push(0x00); - write_varint(digest.len() as u64, &mut wire); - wire.extend_from_slice(&digest); - Some(format!("b{}", base32_lower_no_pad(&wire))) -} - -fn write_varint(mut value: u64, out: &mut Vec) { - loop { - let byte = (value & 0x7f) as u8; - value >>= 7; - if value == 0 { - out.push(byte); - return; - } - out.push(byte | 0x80); - } -} - -fn read_varint(bytes: &[u8], pos: &mut usize) -> Option { - let mut value = 0u64; - let mut shift = 0u32; - loop { - let byte = *bytes.get(*pos)?; - *pos += 1; - value |= u64::from(byte & 0x7f) << shift; - if byte & 0x80 == 0 { - return Some(value); - } - shift += 7; - if shift >= 64 { - return None; - } - } -} - -const BASE32_ALPHABET: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567"; - -fn base32_lower_no_pad(input: &[u8]) -> String { - let mut out = String::with_capacity((input.len() * 8).div_ceil(5)); - let mut acc = 0u64; - let mut bits = 0u32; - for &byte in input { - acc = (acc << 8) | u64::from(byte); - bits += 8; - while bits >= 5 { - bits -= 5; - out.push(BASE32_ALPHABET[((acc >> bits) & 31) as usize] as char); - } - acc &= (1u64 << bits) - 1; - } - if bits > 0 { - out.push(BASE32_ALPHABET[((acc << (5 - bits)) & 31) as usize] as char); - } - out -} - -fn base32_lower_no_pad_decode(input: &str) -> Option> { - let mut out = Vec::with_capacity(input.len() * 5 / 8); - let mut acc = 0u64; - let mut bits = 0u32; - for c in input.bytes() { - let v = match c { - b'a'..=b'z' => c - b'a', - b'2'..=b'7' => c - b'2' + 26, - _ => return None, - }; - acc = (acc << 5) | u64::from(v); - bits += 5; - if bits >= 8 { - bits -= 8; - out.push((acc >> bits) as u8); - acc &= (1u64 << bits) - 1; - } - } - Some(out) + id.strip_prefix("file:")?.parse().ok() } #[cfg(test)] mod tests { use super::*; - /// The same framing under `card`'s codec reproduces the existing - /// `card-locator` golden vector — proof the encoder is the family's framing, - /// not a look-alike. - #[test] - fn framing_reproduces_the_card_locator_vector() { - assert_eq!( - source_id_locator(0x1007, "tmdb", "tv:95479").as_deref(), - Some("bagdsaaanar2g2zdcor3duojvgq3ts") - ); - assert_eq!( - source_id_locator(0x1008, "video", "4D7u5KF7SP8").as_deref(), - Some("bagecaaarav3gszdfn42ein3vgvfumn2tka4a") - ); - } - - /// `docs/cid-formats.md` §8.2 worked example. + /// `docs/cid-formats.md` §8.2 worked example — still minted identically now + /// that the encoder lives in the SDK. #[test] fn file_cid_matches_the_documented_vector() { assert_eq!( @@ -203,13 +66,15 @@ mod tests { // Identical framing, different codec: a card must not decode as a file. assert_eq!(decode_provider_file_cid("bagdsaaanar2g2zdcor3duojvgq3ts"), None); assert_eq!(decode_provider_file_cid("not-a-cid"), None); - assert_eq!(decode_provider_file_cid(""), None); } #[test] - fn digest_ceiling_is_64_bytes() { - // varint(13) + "opensubtitles" = 14 bytes, so the id may use 50. - assert!(provider_file_cid("opensubtitles", &"x".repeat(50)).is_some()); - assert!(provider_file_cid("opensubtitles", &"x".repeat(51)).is_none()); + fn file_id_of_accepts_both_record_id_shapes() { + assert_eq!(file_id_of(&file_cid(7061834).unwrap()), Some(7061834)); + assert_eq!(file_id_of("opensubtitles:file:7061834"), Some(7061834)); + // Another provider's file, a card, and garbage are not ours. + assert_eq!(file_id_of(&provider_file_cid("other", "file:1").unwrap()), None); + assert_eq!(file_id_of("bagdsaaanar2g2zdcor3duojvgq3ts"), None); + assert_eq!(file_id_of("opensubtitles:file:abc"), None); } } diff --git a/feeder-plugin/opensubtitles-feeder/src/main.rs b/feeder-plugin/opensubtitles-feeder/src/main.rs index 8e36032..99e0946 100644 --- a/feeder-plugin/opensubtitles-feeder/src/main.rs +++ b/feeder-plugin/opensubtitles-feeder/src/main.rs @@ -2,13 +2,15 @@ //! //! Serves `fileType:subtitle`. Needs an OpenSubtitles API key (config page or //! `OPENSUBTITLES_API_KEY`); without one it stays registered but degraded and -//! answers every query with nothing. +//! answers every query with nothing. An account login additionally lets the +//! gateway redeem (download) subtitle files. //! //! 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`) //! - `OPENSUBTITLES_API_KEY` — seed API key (the config page overrides it) //! - `OPENSUBTITLES_USER_AGENT` — seed User-Agent (OpenSubtitles wants `App vX.Y`) +//! - `OPENSUBTITLES_USERNAME` / `OPENSUBTITLES_PASSWORD` — seed account login (downloads) //! - `RUST_LOG` — tracing filter (default `info`) use std::net::SocketAddr; diff --git a/feeder-plugin/opensubtitles-feeder/src/plugin.rs b/feeder-plugin/opensubtitles-feeder/src/plugin.rs index 8ce4e6b..c820c77 100644 --- a/feeder-plugin/opensubtitles-feeder/src/plugin.rs +++ b/feeder-plugin/opensubtitles-feeder/src/plugin.rs @@ -8,26 +8,30 @@ //! //! plus an optional `languages:` narrowing on either. +use std::collections::BTreeMap; use std::path::Path; use async_trait::async_trait; use meta_feeder_sdk::budget::RateBudget; use meta_feeder_sdk::common::open_midhash_cache; use meta_feeder_sdk::config::{ConfigField as F, ConfigSchema}; -use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashOutcome}; +use meta_feeder_sdk::hash::compute_ipfs_cid; +use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim}; use meta_feeder_sdk::query::GatewayQuery; use meta_feeder_sdk::query_eval::query_accepts_plugin; -use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, PluginHealth}; +use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth}; use serde::{Deserialize, Serialize}; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; use crate::client::{OsClient, SearchParams}; use crate::consts::{ - DEFAULT_USER_AGENT, ENV_API_KEY, ENV_USER_AGENT, OS_API_BASE, OS_BURST, OS_RATE_PER_SEC, + DEFAULT_SUBTITLE_EXTENSION, DEFAULT_USER_AGENT, ENV_API_KEY, ENV_PASSWORD, ENV_USERNAME, + ENV_USER_AGENT, MIN_SUBTITLE_BYTES, OS_API_BASE, OS_BURST, OS_RATE_PER_SEC, PACKAGE, UPSTREAM_ID, }; use crate::lang::to_os_codes; -use crate::record::project; +use crate::locator::file_id_of; +use crate::record::{extension_of, project}; /// Keys match [`FeederPlugin::config_schema`], which is what the SDK writes to /// `config.json` from the dashboard. @@ -37,6 +41,12 @@ struct OsConfig { api_key: String, #[serde(default, rename = "userAgent")] user_agent: String, + /// Optional account login for `/download`. Blank → downloads run on the API + /// key alone (anonymous quota). + #[serde(default)] + username: String, + #[serde(default)] + password: String, } pub struct OpenSubtitlesPlugin { @@ -54,6 +64,8 @@ impl OpenSubtitlesPlugin { config: OsConfig { api_key: std::env::var(ENV_API_KEY).unwrap_or_default(), user_agent: std::env::var(ENV_USER_AGENT).unwrap_or_default(), + username: std::env::var(ENV_USERNAME).unwrap_or_default(), + password: std::env::var(ENV_PASSWORD).unwrap_or_default(), }, api_base: OS_API_BASE.to_string(), client: None, @@ -65,13 +77,20 @@ impl OpenSubtitlesPlugin { Self { config: OsConfig { api_key: api_key.to_string(), - user_agent: String::new(), + ..OsConfig::default() }, api_base, client: None, } } + /// Test builder: an account login (downloads use the account's quota). + pub fn with_login(mut self, username: &str, password: &str) -> Self { + self.config.username = username.to_string(); + self.config.password = password.to_string(); + self + } + /// File wins over env, like every other feeder: `config.json` in the /// per-plugin cache dir, read at `configure` time. No hot reload. fn load_config(&mut self, cache_dir: &Path) { @@ -88,6 +107,12 @@ impl OpenSubtitlesPlugin { if !file_cfg.user_agent.trim().is_empty() { self.config.user_agent = file_cfg.user_agent; } + if !file_cfg.username.trim().is_empty() { + self.config.username = file_cfg.username; + } + if !file_cfg.password.trim().is_empty() { + self.config.password = file_cfg.password; + } } } @@ -212,13 +237,22 @@ impl FeederPlugin for OpenSubtitlesPlugin { "" => DEFAULT_USER_AGENT, ua => ua, }; - self.client = Some(OsClient::new( + let client = OsClient::new( self.api_base.clone(), api_key, user_agent, RateBudget::new(OS_RATE_PER_SEC, OS_BURST), Some(cache), - )); + ) + .with_login(&self.config.username, &self.config.password); + if !client.has_login() { + info!( + target: "opensubtitles", + "no OpenSubtitles account login ({ENV_USERNAME}/{ENV_PASSWORD} or config page); \ + subtitle downloads use the API key's anonymous quota" + ); + } + self.client = Some(client); Ok(()) } @@ -264,16 +298,85 @@ impl FeederPlugin for OpenSubtitlesPlugin { Ok(records) } - /// Not implemented, on purpose: a `provider-file` locator is resolved by - /// meta-share's fetch tier (`docs/cid-formats.md` §8), and the SDK's - /// `HashKind` has no variant to carry one — adding it is an SDK + gateway - /// change. Searching never needs this path. - async fn compute_outcomes(&self, _record_id: &str) -> Result, GatewayError> { - Err(GatewayError::Permanent( - "opensubtitles: provider-file locators are resolved by meta-share, not /compute \ - (docs/cid-formats.md §8)" - .into(), - )) + /// **Redeem** a `provider-file` locator: download the subtitle file it names. + /// + /// `record_id` is the `0x100A` cid (what the gateway's redeem route sends) + /// or this feeder's `opensubtitles:file:` record id. Anything else — + /// including another source's `provider-file` — is `NotFound`: not ours. + /// + /// ⚠ Spends one unit of the daily download quota (the API key's anonymous + /// quota, or the account's with a login). Returns ONE `Sha2_256` outcome + /// with the bytes and a record of what only the download knows — + /// `extension`, `fileName`, `sizeByte`. The gateway merges those onto the + /// locator record next to the `file` pointer it writes; everything else + /// (title, language, ids) is already there from the search. + async fn compute_outcomes(&self, record_id: &str) -> Result, GatewayError> { + let Some(file_id) = file_id_of(record_id) else { + return Err(GatewayError::NotFound); + }; + let Some(client) = self.client.as_ref() else { + return Err(GatewayError::Permanent( + "opensubtitles: no API key configured".into(), + )); + }; + + info!( + target: "opensubtitles", + file_id, + login = client.has_login(), + "downloading subtitle (spends download quota)" + ); + let got = client.download(file_id).await?; + if got.bytes.len() < MIN_SUBTITLE_BYTES { + return Err(GatewayError::Permanent(format!( + "opensubtitles: file {file_id} is {} bytes — not a subtitle", + got.bytes.len() + ))); + } + if let Some(remaining) = got.remaining { + debug!(target: "opensubtitles", file_id, remaining, "download quota remaining"); + } + let extension = got + .file_name + .as_deref() + .and_then(extension_of) + .unwrap_or(DEFAULT_SUBTITLE_EXTENSION) + .to_string(); + + // Facts only the download knows. The search record may lack + // `extension` (a listing whose name has no subtitle suffix) and never + // knows the real size. + let mut fields = BTreeMap::new(); + fields.insert("extension".to_string(), extension.clone()); + if let Some(name) = got.file_name.as_deref().map(str::trim).filter(|n| !n.is_empty()) { + fields.insert("fileName".to_string(), name.to_string()); + } + fields.insert("sizeByte".to_string(), got.bytes.len().to_string()); + + Ok(vec![HashOutcome { + hash: Hash(compute_ipfs_cid(&got.bytes)), + hash_kind: HashKind::Sha2_256, + bytes: Some(got.bytes.into()), + record: Some(DiscoveryRecord { + upstream_id: UPSTREAM_ID.to_string(), + record_id: format!("{UPSTREAM_ID}:file:{file_id}"), + fields, + }), + file_extension: Some(extension), + }]) + } + + fn package(&self) -> Option<&'static str> { + Some(PACKAGE) + } + + /// Claims `provider-file` for `opensubtitles` whenever an API key is + /// configured — a login only changes which quota a download spends. + fn redeems(&self) -> Vec { + if self.client.is_none() { + return Vec::new(); + } + vec![RedeemClaim::provider_file(vec![UPSTREAM_ID.to_string()])] } fn config_schema(&self) -> ConfigSchema { @@ -282,14 +385,23 @@ impl FeederPlugin for OpenSubtitlesPlugin { F::secret("apiKey", "OpenSubtitles API key") .required() .with_help( - "Consumer API key from opensubtitles.com. Search only — this feeder \ - never downloads, so no account login is needed and no download \ - quota is spent. Takes effect on the next feeder restart.", + "Consumer API key from opensubtitles.com. Enough for search (which \ + spends no quota) and for downloads, on the key's anonymous download \ + quota. Takes effect on the next feeder restart.", ), F::text("userAgent", "User-Agent").with_help( "OpenSubtitles expects the registered app name in `App vX.Y` form. \ Blank keeps the built-in default.", ), + F::text("username", "Account username (optional)").with_help( + "opensubtitles.com account (not .org) to download as, so the account's \ + rank quota applies instead of the anonymous one. Each download spends \ + one unit — once per file, since the gateway stores what it fetched. \ + Blank → downloads on the API key alone.", + ), + F::secret("password", "Account password").with_help( + "Password for the account above. Takes effect on the next feeder restart.", + ), ], } } @@ -298,6 +410,8 @@ impl FeederPlugin for OpenSubtitlesPlugin { serde_json::json!({ "apiKey": self.config.api_key, "userAgent": self.config.user_agent, + "username": self.config.username, + "password": self.config.password, }) } } diff --git a/feeder-plugin/opensubtitles-feeder/src/record.rs b/feeder-plugin/opensubtitles-feeder/src/record.rs index f10d597..b54c589 100644 --- a/feeder-plugin/opensubtitles-feeder/src/record.rs +++ b/feeder-plugin/opensubtitles-feeder/src/record.rs @@ -138,7 +138,7 @@ fn title_for(a: &OsAttributes, fd: &FeatureDetails, file: &OsFile) -> Option Option<&'static str> { +pub(crate) fn extension_of(name: &str) -> Option<&'static str> { let (_, ext) = name.rsplit_once('.')?; let ext = ext.to_ascii_lowercase(); SUBTITLE_EXTENSIONS.iter().copied().find(|e| *e == ext) diff --git a/feeder-plugin/opensubtitles-feeder/tests/contract.rs b/feeder-plugin/opensubtitles-feeder/tests/contract.rs index d7bdfa1..a41ab3a 100644 --- a/feeder-plugin/opensubtitles-feeder/tests/contract.rs +++ b/feeder-plugin/opensubtitles-feeder/tests/contract.rs @@ -12,29 +12,45 @@ //! own `record_matches` against the query that produced them; //! - the upstream sees OpenSubtitles' canonical query form (sorted keys) and the //! API key header; -//! - a query this feeder cannot answer never reaches the upstream. +//! - a query this feeder cannot answer never reaches the upstream; +//! - redeeming a `provider-file` cid through `/compute` downloads the file on +//! the API key alone, or as an account (re-logging in once on a stale token), +//! returns the download-only facts as record fields, and honours the quota. use std::collections::BTreeMap; +use base64::Engine as _; +use meta_feeder_sdk::hash::compute_ipfs_cid; use meta_feeder_sdk::query_eval::record_matches; use meta_feeder_sdk::{ - configure_plugins, router, ComputeRequest, GatewayQuery, ManifestResponse, QueryRequest, - QueryResponse, + configure_plugins, router, ComputeRequest, ComputeResponse, GatewayQuery, HashKindDto, + ManifestResponse, QueryRequest, QueryResponse, RedeemsResponse, }; -use opensubtitles_feeder::locator::decode_provider_file_cid; +use opensubtitles_feeder::locator::{decode_provider_file_cid, file_cid, provider_file_cid}; use opensubtitles_feeder::plugin::OpenSubtitlesPlugin; -use wiremock::matchers::{header, method, path, query_param}; +use wiremock::matchers::{body_partial_json, header, method, path, query_param}; use wiremock::{Match, Mock, MockServer, Request, ResponseTemplate}; const KEY: &str = "test-key"; +const USER: &str = "viewer"; +const PASS: &str = "hunter2"; +const FILE_ID: u64 = 7061834; +const SRT: &str = "1\n00:00:01,000 --> 00:00:02,000\nWake up, Neo.\n"; const KEYWORD: &str = include_str!("fixtures/search_keyword_matrix.json"); const MOVIE_FR: &str = include_str!("fixtures/search_imdb_133093_fr.json"); const EPISODE: &str = include_str!("fixtures/search_episode_1399_s01e01.json"); async fn spawn_feeder(api_key: &str, upstream: &MockServer) -> (String, tempfile::TempDir) { + spawn_plugin(OpenSubtitlesPlugin::with_api_base(api_key, upstream.uri())).await +} + +async fn spawn_feeder_with_login(upstream: &MockServer) -> (String, tempfile::TempDir) { + spawn_plugin(OpenSubtitlesPlugin::with_api_base(KEY, upstream.uri()).with_login(USER, PASS)).await +} + +async fn spawn_plugin(plugin: OpenSubtitlesPlugin) -> (String, tempfile::TempDir) { let dir = tempfile::tempdir().expect("tempdir"); - let plugin = OpenSubtitlesPlugin::with_api_base(api_key, upstream.uri()); let plugins = configure_plugins(vec![Box::new(plugin)], dir.path()).expect("configure plugins"); let app = router(plugins, "opensubtitles-feeder-test".to_string(), dir.path()); @@ -129,6 +145,14 @@ async fn manifest_advertises_subtitle_only() { assert_eq!(m.plugins[0].id, "opensubtitles"); assert_eq!(m.plugins[0].served_file_types, vec!["subtitle".to_string()]); assert!(m.plugins[0].served_content_kinds.is_empty()); + assert_eq!(m.plugins[0].package.as_deref(), Some("meta-feeder-opensubtitles")); + // An API key alone downloads (anonymous quota), so the source is claimed. + assert_eq!( + serde_json::to_value(&m.plugins[0].redeems).unwrap(), + serde_json::json!([{ + "codec": "provider-file", "field": "file", "hosts": [], "sources": ["opensubtitles"] + }]) + ); } #[tokio::test] @@ -293,19 +317,239 @@ async fn missing_api_key_degrades_and_answers_nothing() { .is_empty()); } -#[tokio::test] -async fn compute_is_refused() { - let upstream = MockServer::start().await; - let (base, _dir) = spawn_feeder(KEY, &upstream).await; +// --- redeeming a provider-file locator --------------------------------------- - let resp = reqwest::Client::new() +async fn post_compute(base: &str, record_id: &str) -> reqwest::Response { + reqwest::Client::new() .post(format!("{base}/compute")) .json(&ComputeRequest { upstream_id: "opensubtitles".into(), - record_id: "opensubtitles:file:12658754".into(), + record_id: record_id.into(), }) .send() .await + .expect("POST /compute") +} + +async fn mount_login(upstream: &MockServer, token: &str, times: Option) { + let mock = Mock::given(method("POST")) + .and(path("/login")) + .and(header("Api-Key", KEY)) + .and(body_partial_json(serde_json::json!({ "username": USER, "password": PASS }))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "token": token }))); + match times { + Some(n) => mock.up_to_n_times(n).expect(n).mount(upstream).await, + None => mock.mount(upstream).await, + } +} + +fn download_ok(upstream: &MockServer, remaining: u64) -> ResponseTemplate { + ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "link": format!("{}/files/{FILE_ID}.srt", upstream.uri()), + "file_name": "The.Matrix.1999.en.srt", + "requests": 1, + "remaining": remaining, + "reset_time_utc": "2099-01-01T00:00:00.000Z" + })) +} + +async fn mount_file(upstream: &MockServer, body: &str) { + Mock::given(method("GET")) + .and(path(format!("/files/{FILE_ID}.srt"))) + .respond_with(ResponseTemplate::new(200).set_body_string(body)) + .mount(upstream) + .await; +} + +#[tokio::test] +async fn a_provider_file_cid_redeems_to_the_subtitle_bytes() { + let upstream = MockServer::start().await; + mount_login(&upstream, "t1", None).await; + Mock::given(method("POST")) + .and(path("/download")) + .and(header("Authorization", "Bearer t1")) + .and(body_partial_json(serde_json::json!({ "file_id": FILE_ID }))) + .respond_with(download_ok(&upstream, 99)) + .expect(2) + .mount(&upstream) + .await; + mount_file(&upstream, SRT).await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + let resp = post_compute(&base, &file_cid(FILE_ID).unwrap()).await; + assert_eq!(resp.status(), 200, "{:?}", resp.text().await); + let body: ComputeResponse = resp.json().await.unwrap(); + assert_eq!(body.outcomes.len(), 1); + let o = &body.outcomes[0]; + assert_eq!(o.hash_kind, HashKindDto::Sha2_256); + assert_eq!(o.file_extension.as_deref(), Some("srt")); + let rec = o.record.as_ref().expect("download-only facts for the locator record"); + assert_eq!(rec.fields.get("extension").map(String::as_str), Some("srt")); + assert_eq!(rec.fields.get("fileName").map(String::as_str), Some("The.Matrix.1999.en.srt")); + assert_eq!(rec.fields.get("sizeByte"), Some(&SRT.len().to_string())); + assert!( + !rec.fields.keys().any(|k| k.starts_with("cids/")), + "a redeem never re-addresses the locator record" + ); + let bytes = base64::engine::general_purpose::STANDARD + .decode(o.bytes_b64.as_deref().unwrap()) .unwrap(); - assert_eq!(resp.status(), 422); + assert_eq!(bytes, SRT.as_bytes()); + assert_eq!(o.hash, compute_ipfs_cid(&bytes)); + + // The feeder's own record id form redeems the same file. + assert_eq!(post_compute(&base, &format!("opensubtitles:file:{FILE_ID}")).await.status(), 200); +} + +#[tokio::test] +async fn a_stale_token_is_refreshed_once() { + let upstream = MockServer::start().await; + // First login hands out t1, the second t2. + mount_login(&upstream, "t1", Some(1)).await; + mount_login(&upstream, "t2", Some(1)).await; + Mock::given(method("POST")) + .and(path("/download")) + .and(header("Authorization", "Bearer t1")) + .respond_with(ResponseTemplate::new(401)) + .expect(1) + .mount(&upstream) + .await; + Mock::given(method("POST")) + .and(path("/download")) + .and(header("Authorization", "Bearer t2")) + .respond_with(download_ok(&upstream, 99)) + .expect(1) + .mount(&upstream) + .await; + mount_file(&upstream, SRT).await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + assert_eq!(post_compute(&base, &file_cid(FILE_ID).unwrap()).await.status(), 200); +} + +#[tokio::test] +async fn a_spent_quota_is_rate_limited_and_stops_calling_download() { + let upstream = MockServer::start().await; + mount_login(&upstream, "t1", None).await; + Mock::given(method("POST")) + .and(path("/download")) + .respond_with(ResponseTemplate::new(406).set_body_json(serde_json::json!({ + "message": "You have downloaded your allowed 20 subtitles for 24h.", + "reset_time": "23 hours and 59 minutes", + "reset_time_utc": "2099-01-01T00:00:00.000Z" + }))) + .expect(1) + .mount(&upstream) + .await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + let cid = file_cid(FILE_ID).unwrap(); + assert_eq!(post_compute(&base, &cid).await.status(), 429); + // Blocked locally until the reset: the upstream is not asked again. + assert_eq!(post_compute(&base, &cid).await.status(), 429); +} + +#[tokio::test] +async fn the_last_allowed_download_blocks_the_next() { + let upstream = MockServer::start().await; + mount_login(&upstream, "t1", None).await; + Mock::given(method("POST")) + .and(path("/download")) + .respond_with(download_ok(&upstream, 0)) + .expect(1) + .mount(&upstream) + .await; + mount_file(&upstream, SRT).await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + let cid = file_cid(FILE_ID).unwrap(); + assert_eq!(post_compute(&base, &cid).await.status(), 200); + assert_eq!(post_compute(&base, &cid).await.status(), 429); +} + +#[tokio::test] +async fn not_our_locator_is_not_found_and_spends_nothing() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&upstream) + .await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + let other = provider_file_cid("otherprovider", "file:1").unwrap(); + assert_eq!(post_compute(&base, &other).await.status(), 404); + // A card locator shares the framing but not the codec. + assert_eq!(post_compute(&base, "bagdsaaanar2g2zdcor3duojvgq3ts").await.status(), 404); +} + +#[tokio::test] +async fn an_empty_file_is_refused() { + let upstream = MockServer::start().await; + mount_login(&upstream, "t1", None).await; + Mock::given(method("POST")) + .and(path("/download")) + .respond_with(download_ok(&upstream, 99)) + .mount(&upstream) + .await; + mount_file(&upstream, "oops").await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + assert_eq!(post_compute(&base, &file_cid(FILE_ID).unwrap()).await.status(), 422); +} + +/// Anonymous mode sends the API key and no Bearer token. +struct NoAuthorization; + +impl Match for NoAuthorization { + fn matches(&self, request: &Request) -> bool { + !request.headers.contains_key("authorization") + } +} + +#[tokio::test] +async fn an_api_key_alone_downloads_without_logging_in() { + let upstream = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/login")) + .respond_with(ResponseTemplate::new(200)) + .expect(0) + .mount(&upstream) + .await; + Mock::given(method("POST")) + .and(path("/download")) + .and(header("Api-Key", KEY)) + .and(NoAuthorization) + .and(body_partial_json(serde_json::json!({ "file_id": FILE_ID }))) + .respond_with(download_ok(&upstream, 4)) + .expect(1) + .mount(&upstream) + .await; + mount_file(&upstream, SRT).await; + let (base, _dir) = spawn_feeder(KEY, &upstream).await; + + let resp = post_compute(&base, &file_cid(FILE_ID).unwrap()).await; + assert_eq!(resp.status(), 200, "{:?}", resp.text().await); + let body: ComputeResponse = resp.json().await.unwrap(); + assert_eq!(body.outcomes.len(), 1); + assert_eq!(body.outcomes[0].hash, compute_ipfs_cid(SRT.as_bytes())); + + let r: RedeemsResponse = reqwest::get(format!("{base}/redeems")).await.unwrap().json().await.unwrap(); + assert_eq!(r.redeems["opensubtitles"].len(), 1); +} + +#[tokio::test] +async fn with_a_login_provider_file_is_claimed() { + let upstream = MockServer::start().await; + let (base, _dir) = spawn_feeder_with_login(&upstream).await; + + let r: RedeemsResponse = reqwest::get(format!("{base}/redeems")).await.unwrap().json().await.unwrap(); + assert_eq!( + serde_json::to_value(&r.redeems["opensubtitles"]).unwrap(), + serde_json::json!([{ + "codec": "provider-file", "field": "file", "hosts": [], "sources": ["opensubtitles"] + }]) + ); + let m: ManifestResponse = reqwest::get(format!("{base}/manifest")).await.unwrap().json().await.unwrap(); + assert_eq!(m.plugins[0].redeems, r.redeems["opensubtitles"]); }