videosdk/
client.rs

1//! The HTTP client: authentication, retries, and response decoding.
2
3use std::sync::{Arc, Mutex};
4use std::time::{Duration, SystemTime, UNIX_EPOCH};
5
6use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, USER_AGENT};
7use reqwest::{Method, StatusCode};
8use serde::de::DeserializeOwned;
9use serde_json::{Map, Value};
10
11use crate::builder::{ClientBuilder, Config, TokenOptions};
12use crate::error::{normalize_api_error, Error, Result};
13use crate::token::{
14    decode_token, generate_token, verify_token, AccessTokenBuilder, GenerateTokenParams,
15    TokenClaims,
16};
17
18/// The production API endpoint.
19pub const DEFAULT_BASE_URL: &str = "https://api.videosdk.live";
20
21/// This SDK's version, reported in the `User-Agent` header of every request.
22pub const VERSION: &str = env!("CARGO_PKG_VERSION");
23
24const USER_AGENT_VALUE: &str = concat!("videosdk-rs/", env!("CARGO_PKG_VERSION"));
25
26/// Regenerate a cached token once it is within this window of expiring.
27const TOKEN_REFRESH_BUFFER: i64 = 60;
28const MAX_BACKOFF: Duration = Duration::from_secs(8);
29
30pub(crate) type BackoffFn = Arc<dyn Fn(u32) -> Duration + Send + Sync>;
31
32/// What the caller expects the response body to be.
33#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
34pub enum Expect {
35    /// A JSON body. The default.
36    #[default]
37    Json,
38    /// A raw text body, e.g. a CSV export.
39    Text,
40    /// No body; anything sent is discarded.
41    None,
42}
43
44/// A request body.
45pub(crate) enum Body {
46    Json(Value),
47    /// A pre-encoded body sent verbatim, e.g. an SDP offer. Reachable through
48    /// [`RawRequest::raw_body`].
49    Raw(Vec<u8>),
50}
51
52/// One request the SDK issues internally.
53#[derive(Default)]
54pub(crate) struct CallOptions {
55    pub(crate) query: Vec<(String, String)>,
56    pub(crate) body: Option<Body>,
57    pub(crate) headers: Vec<(String, String)>,
58    pub(crate) expect: Expect,
59}
60
61impl CallOptions {
62    pub(crate) fn new() -> Self {
63        Self::default()
64    }
65
66    pub(crate) fn json(body: impl serde::Serialize) -> Result<Self> {
67        let body = serde_json::to_value(body).map_err(|e| Error::Encode { source: e })?;
68        Ok(Self {
69            body: Some(Body::Json(body)),
70            ..Default::default()
71        })
72    }
73
74    pub(crate) fn query(mut self, query: Vec<(String, String)>) -> Self {
75        self.query = query;
76        self
77    }
78
79    pub(crate) fn expect(mut self, expect: Expect) -> Self {
80        self.expect = expect;
81        self
82    }
83}
84
85#[derive(Default)]
86pub(crate) struct TokenCache {
87    pub(crate) token: Option<String>,
88    pub(crate) expires_at: Option<i64>,
89}
90
91pub(crate) struct ClientInner {
92    pub(crate) config: Config,
93    pub(crate) http: reqwest::Client,
94    pub(crate) base_url: String,
95    pub(crate) can_refresh: bool,
96    pub(crate) token: Mutex<TokenCache>,
97    pub(crate) backoff: BackoffFn,
98}
99
100/// Per-call overrides layered on top of the shared [`ClientInner`].
101#[derive(Clone, Default)]
102pub(crate) struct Overrides {
103    pub(crate) max_retries: Option<u32>,
104    pub(crate) timeout: Option<Duration>,
105}
106
107/// The VideoSDK server SDK client.
108///
109/// Cloning is cheap: clones share one connection pool and one token cache. The
110/// client mints and attaches the API token for its own requests automatically,
111/// so you never hand-roll a JWT or set the `Authorization` header.
112///
113/// ```no_run
114/// # #[tokio::main]
115/// # async fn main() -> Result<(), videosdk::Error> {
116/// // Reads VIDEOSDK_API_KEY and VIDEOSDK_SECRET from the environment.
117/// let client = videosdk::Client::new()?;
118/// # Ok(())
119/// # }
120/// ```
121#[derive(Clone)]
122pub struct Client {
123    inner: Arc<ClientInner>,
124    overrides: Overrides,
125}
126
127impl std::fmt::Debug for Client {
128    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
129        f.debug_struct("Client")
130            .field("base_url", &self.inner.base_url)
131            .finish_non_exhaustive()
132    }
133}
134
135impl Client {
136    /// Constructs a client from the `VIDEOSDK_API_KEY`, `VIDEOSDK_SECRET` and
137    /// `VIDEOSDK_API_ENDPOINT` environment variables.
138    pub fn new() -> Result<Self> {
139        ClientBuilder::new().build()
140    }
141
142    /// Starts building a client.
143    pub fn builder() -> ClientBuilder {
144        ClientBuilder::new()
145    }
146
147    pub(crate) fn from_parts(inner: Arc<ClientInner>, overrides: Overrides) -> Self {
148        Self { inner, overrides }
149    }
150
151    /// The resolved API endpoint.
152    pub fn base_url(&self) -> &str {
153        &self.inner.base_url
154    }
155
156    /// Returns a clone of this client that retries a different number of times.
157    ///
158    /// The clone shares the same connection pool and token cache.
159    pub fn with_max_retries(&self, max_retries: u32) -> Self {
160        let mut client = self.clone();
161        client.overrides.max_retries = Some(max_retries);
162        client
163    }
164
165    /// Returns a clone of this client with a different per-request timeout.
166    /// [`Duration::ZERO`] disables the timeout.
167    pub fn with_request_timeout(&self, timeout: Duration) -> Self {
168        let mut client = self.clone();
169        client.overrides.timeout = Some(timeout);
170        client
171    }
172
173    fn max_retries(&self) -> u32 {
174        self.overrides
175            .max_retries
176            .unwrap_or(self.inner.config.max_retries)
177    }
178
179    fn timeout(&self) -> Duration {
180        self.overrides.timeout.unwrap_or(self.inner.config.timeout)
181    }
182
183    /* --------------------------------- tokens --------------------------------- */
184
185    fn credentials(&self, what: &str) -> Result<(&str, &str)> {
186        match (
187            self.inner.config.api_key.as_deref(),
188            self.inner.config.secret.as_deref(),
189        ) {
190            (Some(api_key), Some(secret)) => Ok((api_key, secret)),
191            _ => Err(Error::config(format!(
192                "{what} requires the client to be constructed with an API key and secret"
193            ))),
194        }
195    }
196
197    /// Mints a fresh management token from the client's credentials, applying any
198    /// configured [`TokenOptions`].
199    pub fn generate_token(&self) -> Result<String> {
200        let (api_key, secret) = self.credentials("generate_token")?;
201        let TokenOptions {
202            permissions,
203            roles,
204            version,
205            expires_in,
206        } = &self.inner.config.token_options;
207        generate_token(&GenerateTokenParams {
208            api_key: api_key.to_string(),
209            secret_key: secret.to_string(),
210            permissions: permissions.clone(),
211            roles: roles.clone(),
212            version: *version,
213            expires_in: *expires_in,
214            claims: Map::new(),
215        })
216    }
217
218    /// Starts building an access token.
219    ///
220    /// Defaults to a participant token (`roles: ["rtc"]`); call
221    /// [`AccessTokenBuilder::for_api`] for a management token.
222    pub fn access_token(&self) -> Result<AccessTokenBuilder> {
223        let (api_key, secret) = self.credentials("access_token")?;
224        let mut builder = AccessTokenBuilder::new(api_key, secret);
225        if let Some(ttl) = self.inner.config.token_options.expires_in {
226            builder = builder.expires_in(ttl);
227        }
228        Ok(builder)
229    }
230
231    /// Verifies a token against the client's secret and returns its claims.
232    pub fn verify_token(&self, token: &str) -> Result<TokenClaims> {
233        let secret = self.inner.config.secret.as_deref().ok_or_else(|| {
234            Error::config("verify_token requires the client to be constructed with a secret")
235        })?;
236        verify_token(token, secret)
237    }
238
239    /// Mints a short-lived API-scoped token (`roles: ["crawler"]`), used to
240    /// authorize WHIP/WHEP publish and play. A zero TTL defaults to one hour.
241    pub(crate) fn mint_api_token(&self, ttl: Duration) -> Result<String> {
242        let ttl = if ttl.is_zero() {
243            Duration::from_secs(3600)
244        } else {
245            ttl
246        };
247        self.access_token()?.for_api().expires_in(ttl).to_jwt()
248    }
249
250    fn resolve_token(&self) -> Result<String> {
251        let mut cache = self
252            .inner
253            .token
254            .lock()
255            .unwrap_or_else(|poisoned| poisoned.into_inner());
256
257        if let Some(token) = &cache.token {
258            let fresh = match cache.expires_at {
259                None => true,
260                Some(exp) => exp - TOKEN_REFRESH_BUFFER > unix_now(),
261            };
262            if fresh {
263                return Ok(token.clone());
264            }
265        }
266
267        if !self.inner.can_refresh {
268            // A static token that may have expired: let the API decide.
269            return cache.token.clone().ok_or_else(|| {
270                Error::config("no token available and the client cannot generate one")
271            });
272        }
273
274        let token = self.generate_token()?;
275        cache.expires_at = decode_token(&token).ok().and_then(|c| c.expires_at);
276        cache.token = Some(token.clone());
277        Ok(token)
278    }
279
280    /* -------------------------------- execution -------------------------------- */
281
282    /// Issues a request with retries, returning the raw response body. Empty for
283    /// a 204, or when [`Expect::None`] was requested.
284    pub(crate) async fn execute(
285        &self,
286        method: Method,
287        path: &str,
288        options: &CallOptions,
289    ) -> Result<Vec<u8>> {
290        let idempotent = is_idempotent(&method);
291        let max_retries = self.max_retries();
292
293        let mut attempt = 0u32;
294        loop {
295            match self.attempt(&method, path, options).await {
296                Ok(body) => return Ok(body),
297                Err(err) => {
298                    if !should_retry(&err, idempotent) || attempt >= max_retries {
299                        return Err(err);
300                    }
301                    // Honor the server's cooldown on a 429; otherwise back off.
302                    let delay = err
303                        .retry_after()
304                        .filter(|_| err.status() == Some(429))
305                        .unwrap_or_else(|| (self.inner.backoff)(attempt));
306                    tokio::time::sleep(delay).await;
307                    attempt += 1;
308                }
309            }
310        }
311    }
312
313    /// Assembles the request headers, later layers replacing earlier ones:
314    /// the SDK's defaults, then the client's, then this call's.
315    fn build_headers(&self, token: &str, options: &CallOptions) -> Result<HeaderMap> {
316        let mut headers = HeaderMap::new();
317
318        // A raw JWT — no "Bearer " prefix.
319        let mut authorization = HeaderValue::from_str(token)
320            .map_err(|_| Error::config("the token contains characters invalid in a header"))?;
321        authorization.set_sensitive(true);
322        headers.insert(AUTHORIZATION, authorization);
323
324        let accept = match options.expect {
325            // A text response is whatever the endpoint produces, e.g. CSV.
326            Expect::Text => "*/*",
327            Expect::Json | Expect::None => "application/json",
328        };
329        headers.insert(ACCEPT, HeaderValue::from_static(accept));
330        headers.insert(USER_AGENT, HeaderValue::from_static(USER_AGENT_VALUE));
331
332        for (name, value) in &self.inner.config.headers {
333            headers.insert(name.clone(), value.clone());
334        }
335        for (name, value) in &options.headers {
336            let name = HeaderName::try_from(name.as_str())
337                .map_err(|_| Error::validation(format!("invalid header name {name:?}")))?;
338            let value = HeaderValue::try_from(value.as_str())
339                .map_err(|_| Error::validation(format!("invalid value for header {name:?}")))?;
340            headers.insert(name, value);
341        }
342
343        Ok(headers)
344    }
345
346    async fn attempt(&self, method: &Method, path: &str, options: &CallOptions) -> Result<Vec<u8>> {
347        let token = self.resolve_token()?;
348        let url = format!("{}{}", self.inner.base_url, ensure_leading_slash(path));
349
350        let mut request = self.inner.http.request(method.clone(), &url);
351        if !options.query.is_empty() {
352            request = request.query(&options.query);
353        }
354
355        // `RequestBuilder::header` *appends*, so a caller-supplied `Authorization`
356        // would be sent alongside ours rather than replacing it. Build the map
357        // ourselves — `HeaderMap::insert` replaces — and hand it over in one go.
358        request = request.headers(self.build_headers(&token, options)?);
359
360        match &options.body {
361            // `json` sets Content-Type only when it is not already present, so a
362            // caller-supplied Content-Type still wins.
363            Some(Body::Json(value)) => request = request.json(value),
364            Some(Body::Raw(bytes)) => request = request.body(bytes.clone()),
365            None => {}
366        }
367
368        let timeout = self.timeout();
369        if !timeout.is_zero() {
370            request = request.timeout(timeout);
371        }
372
373        let response = request
374            .send()
375            .await
376            .map_err(|e| transport_error(e, method, path, timeout))?;
377
378        let status = response.status();
379        let request_id = read_request_id(response.headers());
380        let retry_after = (status == StatusCode::TOO_MANY_REQUESTS)
381            .then(|| parse_retry_after(response.headers()))
382            .flatten();
383
384        let body = response
385            .bytes()
386            .await
387            .map_err(|e| transport_error(e, method, path, timeout))?;
388
389        if !status.is_success() {
390            return Err(Error::api(normalize_api_error(
391                status.as_u16(),
392                parse_maybe_json(&body),
393                method.as_str(),
394                path,
395                request_id,
396                retry_after,
397            )));
398        }
399
400        if options.expect == Expect::None || status == StatusCode::NO_CONTENT {
401            return Ok(Vec::new());
402        }
403        Ok(body.to_vec())
404    }
405
406    /// Uploads raw bytes to an absolute URL — typically an S3 or GCS presigned
407    /// URL — with `PUT` and no `Authorization` header, since the presigned URL
408    /// carries its own signature. Retries idempotently.
409    pub(crate) async fn put_binary(
410        &self,
411        url: &str,
412        body: &[u8],
413        content_type: &str,
414    ) -> Result<()> {
415        let mut attempt = 0u32;
416        loop {
417            match self.attempt_put(url, body, content_type).await {
418                Ok(()) => return Ok(()),
419                Err(err) => {
420                    if !should_retry(&err, true) || attempt >= self.max_retries() {
421                        return Err(err);
422                    }
423                    tokio::time::sleep((self.inner.backoff)(attempt)).await;
424                    attempt += 1;
425                }
426            }
427        }
428    }
429
430    async fn attempt_put(&self, url: &str, body: &[u8], content_type: &str) -> Result<()> {
431        // Keep the presigned signature out of error messages and logs.
432        let path = safe_url_path(url);
433        let timeout = self.timeout();
434
435        let mut request = self
436            .inner
437            .http
438            .put(url)
439            .header(reqwest::header::CONTENT_TYPE, content_type)
440            .body(body.to_vec());
441        if !timeout.is_zero() {
442            request = request.timeout(timeout);
443        }
444
445        let response = request
446            .send()
447            .await
448            .map_err(|e| transport_error(e, &Method::PUT, path, timeout))?;
449
450        let status = response.status();
451        if status.is_success() {
452            return Ok(());
453        }
454
455        let body = response.bytes().await.unwrap_or_default();
456        let mut api_error = normalize_api_error(
457            status.as_u16(),
458            parse_maybe_json(&body),
459            Method::PUT.as_str(),
460            path,
461            None,
462            None,
463        );
464        api_error.message = format!("uploading the file to storage failed (HTTP {status})");
465        api_error.code = Some("upload_failed".to_string());
466        Err(Error::api(api_error))
467    }
468
469    fn decode<T: DeserializeOwned>(bytes: &[u8], method: &Method, path: &str) -> Result<T> {
470        let context = format!("{method} {path}");
471        if bytes.iter().all(u8::is_ascii_whitespace) {
472            // Mirrors the Go SDK's zero-value return for an empty success body.
473            // Only types with a `null` representation (Option, unit) can absorb it.
474            return serde_json::from_str("null").map_err(|e| Error::decode(context, e));
475        }
476        serde_json::from_slice(bytes).map_err(|e| Error::decode(context, e))
477    }
478}
479
480/* ------------------------- typed executors for resources ------------------------ */
481
482/// One helper per response shape the API uses. There is no single generic
483/// envelope: endpoints return a flat object, a `{"data": T}` envelope, a
484/// envelope under a name of their own, a bare confirmation string, raw text, or
485/// nothing at all.
486impl Client {
487    /// Issues a request and decodes a flat JSON response body into `T`.
488    pub(crate) async fn json<T: DeserializeOwned>(
489        &self,
490        method: Method,
491        path: &str,
492        options: CallOptions,
493    ) -> Result<T> {
494        let bytes = self.execute(method.clone(), path, &options).await?;
495        Self::decode(&bytes, &method, path)
496    }
497
498    /// Issues a request and returns the response as a [`Value`], falling back to
499    /// a JSON string for a non-JSON body.
500    ///
501    /// Egress `start` endpoints answer with a bare confirmation string on some
502    /// paths and an id-bearing object on others, so neither can be assumed.
503    pub(crate) async fn maybe_json(
504        &self,
505        method: Method,
506        path: &str,
507        options: CallOptions,
508    ) -> Result<Value> {
509        let bytes = self.execute(method, path, &options).await?;
510        Ok(parse_maybe_json(&bytes).unwrap_or(Value::Null))
511    }
512
513    /// Issues a request and decodes the `data` field of a `{"data": T}` envelope.
514    pub(crate) async fn data<T: DeserializeOwned>(
515        &self,
516        method: Method,
517        path: &str,
518        options: CallOptions,
519    ) -> Result<T> {
520        self.wrapped(method, path, "data", options).await
521    }
522
523    /// Issues a request and decodes a single-key envelope, e.g. `{"alertRule": T}`.
524    ///
525    /// A missing key decodes as `null`, matching the Go SDK's zero value.
526    pub(crate) async fn wrapped<T: DeserializeOwned>(
527        &self,
528        method: Method,
529        path: &str,
530        key: &str,
531        options: CallOptions,
532    ) -> Result<T> {
533        let bytes = self.execute(method.clone(), path, &options).await?;
534        let context = format!("{method} {path}");
535        if bytes.iter().all(u8::is_ascii_whitespace) {
536            return serde_json::from_str("null").map_err(|e| Error::decode(context, e));
537        }
538        let mut envelope: Map<String, Value> =
539            serde_json::from_slice(&bytes).map_err(|e| Error::decode(&context, e))?;
540        let inner = envelope.remove(key).unwrap_or(Value::Null);
541        serde_json::from_value(inner).map_err(|e| Error::decode(context, e))
542    }
543
544    /// Issues a request and returns a string confirmation. Mirrors the server's
545    /// message endpoints: a JSON-string body is unquoted, anything else is
546    /// returned as raw text.
547    pub(crate) async fn message(
548        &self,
549        method: Method,
550        path: &str,
551        options: CallOptions,
552    ) -> Result<String> {
553        let bytes = self.execute(method, path, &options).await?;
554        let text = String::from_utf8_lossy(&bytes);
555        let trimmed = text.trim();
556        if trimmed.is_empty() {
557            return Ok(String::new());
558        }
559        if trimmed.starts_with('"') {
560            if let Ok(unquoted) = serde_json::from_str::<String>(trimmed) {
561                return Ok(unquoted);
562            }
563        }
564        Ok(trimmed.to_string())
565    }
566
567    /// Issues a request and returns the raw response body as text, e.g. a CSV export.
568    pub(crate) async fn text(
569        &self,
570        method: Method,
571        path: &str,
572        options: CallOptions,
573    ) -> Result<String> {
574        let bytes = self
575            .execute(method, path, &options.expect(Expect::Text))
576            .await?;
577        Ok(String::from_utf8_lossy(&bytes).into_owned())
578    }
579
580    /// Issues a request and discards the response body.
581    pub(crate) async fn none(
582        &self,
583        method: Method,
584        path: &str,
585        options: CallOptions,
586    ) -> Result<()> {
587        self.execute(method, path, &options.expect(Expect::None))
588            .await?;
589        Ok(())
590    }
591}
592
593/* --------------------------------- escape hatch --------------------------------- */
594
595impl Client {
596    /// Calls any endpoint with an arbitrary HTTP method, applying auth and
597    /// retries, and returns the undecoded JSON response.
598    ///
599    /// For the common verbs, [`Client::api`] reads better.
600    pub async fn request(&self, method: Method, path: &str, request: RawRequest) -> Result<Value> {
601        let expect = request.expect;
602        // A raw body wins over a JSON one, as it is the more specific request.
603        let body = match (request.raw_body, request.body) {
604            (Some(bytes), _) => Some(Body::Raw(bytes)),
605            (None, Some(value)) => Some(Body::Json(value)),
606            (None, None) => None,
607        };
608        let options = CallOptions {
609            query: request.query,
610            body,
611            headers: request.headers,
612            expect,
613        };
614        let bytes = self.execute(method.clone(), path, &options).await?;
615        match expect {
616            Expect::None => Ok(Value::Null),
617            // A text response is not JSON, so hand it back as a JSON string
618            // rather than failing to parse it.
619            Expect::Text => Ok(Value::String(String::from_utf8_lossy(&bytes).into_owned())),
620            Expect::Json if bytes.iter().all(u8::is_ascii_whitespace) => Ok(Value::Null),
621            Expect::Json => Self::decode(&bytes, &method, path),
622        }
623    }
624
625    /// The raw HTTP escape hatch. Each method applies auth and retries, and
626    /// returns the undecoded JSON response.
627    pub fn api(&self) -> Api<'_> {
628        Api { client: self }
629    }
630}
631
632/// Configures a raw [`Api`] or [`Client::request`] call.
633#[derive(Debug, Default)]
634pub struct RawRequest {
635    /// Query-string parameters.
636    pub query: Vec<(String, String)>,
637    /// A JSON request body. Ignored when [`raw_body`](RawRequest::raw_body) is set.
638    pub body: Option<Value>,
639    /// A pre-encoded request body, sent verbatim.
640    ///
641    /// Set `Content-Type` yourself via [`header`](RawRequest::header); unlike a
642    /// JSON body, none is inferred.
643    pub raw_body: Option<Vec<u8>>,
644    /// Extra headers for this request.
645    pub headers: Vec<(String, String)>,
646    /// What to expect back. Defaults to [`Expect::Json`].
647    pub expect: Expect,
648}
649
650impl RawRequest {
651    /// A request with no query, body or headers.
652    pub fn new() -> Self {
653        Self::default()
654    }
655
656    /// Attaches a JSON body.
657    pub fn body(mut self, body: Value) -> Self {
658        self.body = Some(body);
659        self
660    }
661
662    /// Attaches a pre-encoded body, sent verbatim.
663    ///
664    /// Pair it with a `Content-Type`, e.g. for an SDP offer:
665    ///
666    /// ```no_run
667    /// # use videosdk::RawRequest;
668    /// let request = RawRequest::new()
669    ///     .raw_body(b"v=0\r\n".to_vec())
670    ///     .header("content-type", "application/sdp");
671    /// ```
672    pub fn raw_body(mut self, body: impl Into<Vec<u8>>) -> Self {
673        self.raw_body = Some(body.into());
674        self
675    }
676
677    /// Attaches a header, replacing any the SDK would otherwise set.
678    pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
679        self.headers.push((name.into(), value.into()));
680        self
681    }
682
683    /// Attaches a query parameter.
684    pub fn query(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
685        self.query.push((name.into(), value.into()));
686        self
687    }
688
689    /// Sets what to expect back.
690    pub fn expect(mut self, expect: Expect) -> Self {
691        self.expect = expect;
692        self
693    }
694}
695
696/// The raw HTTP escape hatch, reachable via [`Client::api`].
697#[derive(Debug, Clone, Copy)]
698pub struct Api<'a> {
699    client: &'a Client,
700}
701
702macro_rules! api_method {
703    ($name:ident, $method:expr, $doc:literal) => {
704        #[doc = $doc]
705        pub async fn $name(&self, path: &str, request: RawRequest) -> Result<Value> {
706            self.client.request($method, path, request).await
707        }
708    };
709}
710
711impl Api<'_> {
712    api_method!(get, Method::GET, "Issues a raw `GET`.");
713    api_method!(post, Method::POST, "Issues a raw `POST`.");
714    api_method!(put, Method::PUT, "Issues a raw `PUT`.");
715    api_method!(patch, Method::PATCH, "Issues a raw `PATCH`.");
716    api_method!(delete, Method::DELETE, "Issues a raw `DELETE`.");
717}
718
719/* ---------------------------------- helpers ---------------------------------- */
720
721fn unix_now() -> i64 {
722    SystemTime::now()
723        .duration_since(UNIX_EPOCH)
724        .map(|d| d.as_secs() as i64)
725        .unwrap_or(0)
726}
727
728fn ensure_leading_slash(path: &str) -> String {
729    if path.starts_with('/') {
730        path.to_string()
731    } else {
732        format!("/{path}")
733    }
734}
735
736/// Strips the query string, which carries a presigned URL's signature.
737fn safe_url_path(url: &str) -> &str {
738    url.split('?').next().unwrap_or(url)
739}
740
741fn transport_error(
742    source: reqwest::Error,
743    method: &Method,
744    path: &str,
745    timeout: Duration,
746) -> Error {
747    if source.is_timeout() {
748        Error::Timeout {
749            method: method.to_string(),
750            path: path.to_string(),
751            elapsed: timeout,
752        }
753    } else {
754        Error::Network {
755            method: method.to_string(),
756            path: path.to_string(),
757            source,
758        }
759    }
760}
761
762/// Parses a body as JSON, falling back to the raw text for non-JSON bodies so
763/// the error normalizer can still read a plain-text message.
764fn parse_maybe_json(body: &[u8]) -> Option<Value> {
765    if body.iter().all(u8::is_ascii_whitespace) {
766        return None;
767    }
768    match serde_json::from_slice(body) {
769        Ok(value) => Some(value),
770        Err(_) => Some(Value::String(String::from_utf8_lossy(body).into_owned())),
771    }
772}
773
774fn read_request_id(headers: &HeaderMap) -> Option<String> {
775    [
776        "x-request-id",
777        "request-id",
778        "x-amzn-requestid",
779        "x-amz-request-id",
780    ]
781    .iter()
782    .find_map(|name| headers.get(*name))
783    .and_then(|value| value.to_str().ok())
784    .filter(|value| !value.is_empty())
785    .map(str::to_string)
786}
787
788/// Parses a `Retry-After` header, which is either delta-seconds or an HTTP-date.
789fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
790    let value = headers.get("retry-after")?.to_str().ok()?.trim();
791    if value.is_empty() {
792        return None;
793    }
794    if let Ok(seconds) = value.parse::<i64>() {
795        return (seconds > 0).then(|| Duration::from_secs(seconds as u64));
796    }
797    let deadline = httpdate::parse_http_date(value).ok()?;
798    deadline.duration_since(SystemTime::now()).ok()
799}
800
801fn is_idempotent(method: &Method) -> bool {
802    matches!(
803        *method,
804        Method::GET | Method::PUT | Method::DELETE | Method::HEAD | Method::OPTIONS
805    )
806}
807
808/// Decides whether a failed attempt is worth retrying.
809///
810/// A 429 is always safe — the server rejected the request before processing it.
811/// Otherwise only idempotent methods retry: a timeout or 5xx on a `POST` may
812/// already have applied server-side, and retrying would duplicate the effect.
813fn should_retry(error: &Error, idempotent: bool) -> bool {
814    if error.status() == Some(429) {
815        return true;
816    }
817    if !idempotent {
818        return false;
819    }
820    match error {
821        Error::Api(api) => api.status >= 500,
822        Error::Network { .. } | Error::Timeout { .. } => true,
823        _ => false,
824    }
825}
826
827/// Exponential backoff capped at 8 seconds, plus jitter.
828pub(crate) fn default_backoff(attempt: u32) -> Duration {
829    let base = Duration::from_secs(1)
830        .saturating_mul(1u32.checked_shl(attempt).unwrap_or(u32::MAX))
831        .min(MAX_BACKOFF);
832    base + Duration::from_millis(jitter_ms())
833}
834
835/// Up to 249 ms of jitter, derived from the clock. Backoff jitter only needs to
836/// decorrelate concurrent retries, so this avoids pulling in an RNG dependency.
837fn jitter_ms() -> u64 {
838    let nanos = SystemTime::now()
839        .duration_since(UNIX_EPOCH)
840        .map(|d| d.subsec_nanos() as u64)
841        .unwrap_or(0);
842    let mut x = nanos
843        .wrapping_mul(6_364_136_223_846_793_005)
844        .wrapping_add(1_442_695_040_888_963_407);
845    x ^= x >> 33;
846    x = x.wrapping_mul(0xff51_afd7_ed55_8ccd);
847    x ^= x >> 33;
848    x % 250
849}
850
851#[cfg(test)]
852mod tests {
853    use super::*;
854    use crate::error::ApiError;
855    use crate::error::ErrorKind;
856
857    fn api_error(status: u16) -> Error {
858        Error::api(ApiError {
859            message: String::new(),
860            kind: ErrorKind::from_status(status),
861            code: None,
862            status,
863            request_id: None,
864            details: None,
865            method: "GET".into(),
866            path: "/x".into(),
867            retry_after: None,
868        })
869    }
870
871    #[test]
872    fn idempotent_methods() {
873        for method in [
874            Method::GET,
875            Method::PUT,
876            Method::DELETE,
877            Method::HEAD,
878            Method::OPTIONS,
879        ] {
880            assert!(is_idempotent(&method), "{method} should be idempotent");
881        }
882        assert!(!is_idempotent(&Method::POST));
883        assert!(!is_idempotent(&Method::PATCH));
884    }
885
886    #[test]
887    fn rate_limits_retry_for_any_method() {
888        assert!(should_retry(&api_error(429), true));
889        assert!(should_retry(&api_error(429), false));
890    }
891
892    #[test]
893    fn server_errors_retry_only_when_idempotent() {
894        assert!(should_retry(&api_error(500), true));
895        assert!(should_retry(&api_error(503), true));
896        assert!(!should_retry(&api_error(500), false));
897    }
898
899    #[test]
900    fn client_errors_never_retry() {
901        for status in [400, 401, 403, 404, 409] {
902            assert!(!should_retry(&api_error(status), true), "status {status}");
903        }
904    }
905
906    #[test]
907    fn timeouts_retry_only_when_idempotent() {
908        let err = Error::Timeout {
909            method: "GET".into(),
910            path: "/x".into(),
911            elapsed: Duration::from_secs(1),
912        };
913        assert!(should_retry(&err, true));
914        assert!(!should_retry(&err, false));
915    }
916
917    #[test]
918    fn config_and_decode_errors_never_retry() {
919        assert!(!should_retry(&Error::config("x"), true));
920        assert!(!should_retry(&Error::validation("x"), true));
921    }
922
923    #[test]
924    fn backoff_grows_exponentially_and_caps() {
925        let bounds = |attempt: u32| {
926            let d = default_backoff(attempt);
927            (d.as_millis() as u64) / 1000
928        };
929        assert_eq!(bounds(0), 1);
930        assert_eq!(bounds(1), 2);
931        assert_eq!(bounds(2), 4);
932        assert_eq!(bounds(3), 8);
933        // Capped, plus jitter under 250ms.
934        assert!(default_backoff(10) < Duration::from_millis(8_250));
935        assert!(default_backoff(64) < Duration::from_millis(8_250));
936    }
937
938    #[test]
939    fn jitter_stays_in_range() {
940        for _ in 0..100 {
941            assert!(jitter_ms() < 250);
942        }
943    }
944
945    #[test]
946    fn parses_retry_after_seconds() {
947        let mut headers = HeaderMap::new();
948        headers.insert("retry-after", "3".parse().unwrap());
949        assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(3)));
950    }
951
952    #[test]
953    fn ignores_non_positive_or_missing_retry_after() {
954        assert_eq!(parse_retry_after(&HeaderMap::new()), None);
955        let mut headers = HeaderMap::new();
956        headers.insert("retry-after", "0".parse().unwrap());
957        assert_eq!(parse_retry_after(&headers), None);
958        headers.insert("retry-after", "-5".parse().unwrap());
959        assert_eq!(parse_retry_after(&headers), None);
960        headers.insert("retry-after", "garbage".parse().unwrap());
961        assert_eq!(parse_retry_after(&headers), None);
962    }
963
964    #[test]
965    fn parses_retry_after_http_date() {
966        let future = SystemTime::now() + Duration::from_secs(120);
967        let mut headers = HeaderMap::new();
968        headers.insert(
969            "retry-after",
970            httpdate::fmt_http_date(future).parse().unwrap(),
971        );
972        let parsed = parse_retry_after(&headers).expect("should parse an HTTP-date");
973        assert!(parsed > Duration::from_secs(60) && parsed <= Duration::from_secs(120));
974
975        // A date in the past yields no delay.
976        let past = SystemTime::now() - Duration::from_secs(60);
977        headers.insert(
978            "retry-after",
979            httpdate::fmt_http_date(past).parse().unwrap(),
980        );
981        assert_eq!(parse_retry_after(&headers), None);
982    }
983
984    #[test]
985    fn reads_request_id_in_priority_order() {
986        let mut headers = HeaderMap::new();
987        assert_eq!(read_request_id(&headers), None);
988        headers.insert("x-amz-request-id", "amz".parse().unwrap());
989        assert_eq!(read_request_id(&headers).as_deref(), Some("amz"));
990        headers.insert("x-request-id", "primary".parse().unwrap());
991        assert_eq!(read_request_id(&headers).as_deref(), Some("primary"));
992    }
993
994    #[test]
995    fn parse_maybe_json_falls_back_to_text() {
996        assert_eq!(parse_maybe_json(b"   "), None);
997        assert_eq!(
998            parse_maybe_json(b"{\"a\":1}"),
999            Some(serde_json::json!({"a": 1}))
1000        );
1001        assert_eq!(
1002            parse_maybe_json(b"<html>oops</html>"),
1003            Some(Value::String("<html>oops</html>".into()))
1004        );
1005    }
1006
1007    #[test]
1008    fn safe_url_path_strips_the_signature() {
1009        assert_eq!(
1010            safe_url_path("https://s3/bucket/key?X-Amz-Signature=secret"),
1011            "https://s3/bucket/key"
1012        );
1013        assert_eq!(safe_url_path("https://s3/key"), "https://s3/key");
1014    }
1015
1016    #[test]
1017    fn ensures_a_leading_slash() {
1018        assert_eq!(ensure_leading_slash("v2/rooms"), "/v2/rooms");
1019        assert_eq!(ensure_leading_slash("/v2/rooms"), "/v2/rooms");
1020    }
1021}
videosdk/client.rs

videosdk/
client.rs
