improve compute API

2026-09-13 14:31:17 +02:00
parent deeba3857b
commit 2f3ca57322
10 changed files with 845 additions and 221 deletions
@@ -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<String> {
    v.get("body")?.as_str().map(str::to_string)
 }

+// --- download shapes --------------------------------------------------------
+
+#[derive(Debug, Clone, Default, Deserialize)]
+struct LoginResponse {
+    #[serde(default)]
+    token: Option<String>,
+}
+
+/// `POST /download` answer (also the body of its `406` quota refusal).
+#[derive(Debug, Clone, Default, Deserialize)]
+#[serde(default)]
+pub struct DownloadResponse {
+    pub link: Option<String>,
+    pub file_name: Option<String>,
+    #[serde(deserialize_with = "lenient_u64")]
+    pub remaining: Option<u64>,
+    /// Human text, e.g. `23 hours and 59 minutes`.
+    pub reset_time: Option<String>,
+    /// e.g. `2022-04-08T13:03:16.000Z`.
+    pub reset_time_utc: Option<String>,
+    pub message: Option<String>,
+}
+
+/// One redeemed subtitle file.
+#[derive(Debug, Clone)]
+pub struct Downloaded {
+    pub bytes: Vec<u8>,
+    pub file_name: Option<String>,
+    pub remaining: Option<u64>,
+}
+
+/// 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<u64> {
+    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::<i64>());
+    let (y, mo, da) = (d.next()?.ok()?, d.next()?.ok()?, d.next()?.ok()?);
+    let mut t = time.split(':').map(|p| p.parse::<i64>());
+    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<u64> {
+    let mut total = 0u64;
+    let mut pending: Option<u64> = 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::<u64>() {
+            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<RateBudget>,
    cache: Option<MidhashCache>,
+    /// 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<Option<String>>,
+    /// Unix time until which the download quota is known to be spent — no
+    /// `/download` call is made before it.
+    quota_until: StdMutex<Option<u64>>,
 }

 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<u32> {
+        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<String, GatewayError> {
+        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<String, GatewayError> {
+        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<reqwest::Response, GatewayError> {
+        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<Downloaded, GatewayError> {
+        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<Vec<OsSubtitle>, 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(