use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, AUTHORIZATION, USER_AGENT};
use reqwest::{Method, StatusCode};
use serde::de::DeserializeOwned;
use serde_json::{Map, Value};
use crate::builder::{ClientBuilder, Config, TokenOptions};
use crate::error::{normalize_api_error, Error, Result};
use crate::token::{
decode_token, generate_token, verify_token, AccessTokenBuilder, GenerateTokenParams,
TokenClaims,
};
pub const DEFAULT_BASE_URL: &str = "https://api.videosdk.live";
pub const VERSION: &str = env!("CARGO_PKG_VERSION");
const USER_AGENT_VALUE: &str = concat!("videosdk-rs/", env!("CARGO_PKG_VERSION"));
const TOKEN_REFRESH_BUFFER: i64 = 60;
const MAX_BACKOFF: Duration = Duration::from_secs(8);
pub(crate) type BackoffFn = Arc<dyn Fn(u32) -> Duration + Send + Sync>;
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum Expect {
#[default]
Json,
Text,
None,
}
pub(crate) enum Body {
Json(Value),
Raw(Vec<u8>),
}
#[derive(Default)]
pub(crate) struct CallOptions {
pub(crate) query: Vec<(String, String)>,
pub(crate) body: Option<Body>,
pub(crate) headers: Vec<(String, String)>,
pub(crate) expect: Expect,
}
impl CallOptions {
pub(crate) fn new() -> Self {
Self::default()
}
pub(crate) fn json(body: impl serde::Serialize) -> Result<Self> {
let body = serde_json::to_value(body).map_err(|e| Error::Encode { source: e })?;
Ok(Self {
body: Some(Body::Json(body)),
..Default::default()
})
}
pub(crate) fn query(mut self, query: Vec<(String, String)>) -> Self {
self.query = query;
self
}
pub(crate) fn expect(mut self, expect: Expect) -> Self {
self.expect = expect;
self
}
}
#[derive(Default)]
pub(crate) struct TokenCache {
pub(crate) token: Option<String>,
pub(crate) expires_at: Option<i64>,
}
pub(crate) struct ClientInner {
pub(crate) config: Config,
pub(crate) http: reqwest::Client,
pub(crate) base_url: String,
pub(crate) can_refresh: bool,
pub(crate) token: Mutex<TokenCache>,
pub(crate) backoff: BackoffFn,
}
#[derive(Clone, Default)]
pub(crate) struct Overrides {
pub(crate) max_retries: Option<u32>,
pub(crate) timeout: Option<Duration>,
}
#[derive(Clone)]
pub struct Client {
inner: Arc<ClientInner>,
overrides: Overrides,
}
impl std::fmt::Debug for Client {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Client")
.field("base_url", &self.inner.base_url)
.finish_non_exhaustive()
}
}
impl Client {
pub fn new() -> Result<Self> {
ClientBuilder::new().build()
}
pub fn builder() -> ClientBuilder {
ClientBuilder::new()
}
pub(crate) fn from_parts(inner: Arc<ClientInner>, overrides: Overrides) -> Self {
Self { inner, overrides }
}
pub fn base_url(&self) -> &str {
&self.inner.base_url
}
pub fn with_max_retries(&self, max_retries: u32) -> Self {
let mut client = self.clone();
client.overrides.max_retries = Some(max_retries);
client
}
pub fn with_request_timeout(&self, timeout: Duration) -> Self {
let mut client = self.clone();
client.overrides.timeout = Some(timeout);
client
}
fn max_retries(&self) -> u32 {
self.overrides
.max_retries
.unwrap_or(self.inner.config.max_retries)
}
fn timeout(&self) -> Duration {
self.overrides.timeout.unwrap_or(self.inner.config.timeout)
}
fn credentials(&self, what: &str) -> Result<(&str, &str)> {
match (
self.inner.config.api_key.as_deref(),
self.inner.config.secret.as_deref(),
) {
(Some(api_key), Some(secret)) => Ok((api_key, secret)),
_ => Err(Error::config(format!(
"{what} requires the client to be constructed with an API key and secret"
))),
}
}
pub fn generate_token(&self) -> Result<String> {
let (api_key, secret) = self.credentials("generate_token")?;
let TokenOptions {
permissions,
roles,
version,
expires_in,
} = &self.inner.config.token_options;
generate_token(&GenerateTokenParams {
api_key: api_key.to_string(),
secret_key: secret.to_string(),
permissions: permissions.clone(),
roles: roles.clone(),
version: *version,
expires_in: *expires_in,
claims: Map::new(),
})
}
pub fn access_token(&self) -> Result<AccessTokenBuilder> {
let (api_key, secret) = self.credentials("access_token")?;
let mut builder = AccessTokenBuilder::new(api_key, secret);
if let Some(ttl) = self.inner.config.token_options.expires_in {
builder = builder.expires_in(ttl);
}
Ok(builder)
}
pub fn verify_token(&self, token: &str) -> Result<TokenClaims> {
let secret = self.inner.config.secret.as_deref().ok_or_else(|| {
Error::config("verify_token requires the client to be constructed with a secret")
})?;
verify_token(token, secret)
}
pub(crate) fn mint_api_token(&self, ttl: Duration) -> Result<String> {
let ttl = if ttl.is_zero() {
Duration::from_secs(3600)
} else {
ttl
};
self.access_token()?.for_api().expires_in(ttl).to_jwt()
}
fn resolve_token(&self) -> Result<String> {
let mut cache = self
.inner
.token
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(token) = &cache.token {
let fresh = match cache.expires_at {
None => true,
Some(exp) => exp - TOKEN_REFRESH_BUFFER > unix_now(),
};
if fresh {
return Ok(token.clone());
}
}
if !self.inner.can_refresh {
return cache.token.clone().ok_or_else(|| {
Error::config("no token available and the client cannot generate one")
});
}
let token = self.generate_token()?;
cache.expires_at = decode_token(&token).ok().and_then(|c| c.expires_at);
cache.token = Some(token.clone());
Ok(token)
}
pub(crate) async fn execute(
&self,
method: Method,
path: &str,
options: &CallOptions,
) -> Result<Vec<u8>> {
let idempotent = is_idempotent(&method);
let max_retries = self.max_retries();
let mut attempt = 0u32;
loop {
match self.attempt(&method, path, options).await {
Ok(body) => return Ok(body),
Err(err) => {
if !should_retry(&err, idempotent) || attempt >= max_retries {
return Err(err);
}
let delay = err
.retry_after()
.filter(|_| err.status() == Some(429))
.unwrap_or_else(|| (self.inner.backoff)(attempt));
tokio::time::sleep(delay).await;
attempt += 1;
}
}
}
}
fn build_headers(&self, token: &str, options: &CallOptions) -> Result<HeaderMap> {
let mut headers = HeaderMap::new();
let mut authorization = HeaderValue::from_str(token)
.map_err(|_| Error::config("the token contains characters invalid in a header"))?;
authorization.set_sensitive(true);
headers.insert(AUTHORIZATION, authorization);
let accept = match options.expect {
Expect::Text => "*/*",
Expect::Json | Expect::None => "application/json",
};
headers.insert(ACCEPT, HeaderValue::from_static(accept));
headers.insert(USER_AGENT, HeaderValue::from_static(USER_AGENT_VALUE));
for (name, value) in &self.inner.config.headers {
headers.insert(name.clone(), value.clone());
}
for (name, value) in &options.headers {
let name = HeaderName::try_from(name.as_str())
.map_err(|_| Error::validation(format!("invalid header name {name:?}")))?;
let value = HeaderValue::try_from(value.as_str())
.map_err(|_| Error::validation(format!("invalid value for header {name:?}")))?;
headers.insert(name, value);
}
Ok(headers)
}
async fn attempt(&self, method: &Method, path: &str, options: &CallOptions) -> Result<Vec<u8>> {
let token = self.resolve_token()?;
let url = format!("{}{}", self.inner.base_url, ensure_leading_slash(path));
let mut request = self.inner.http.request(method.clone(), &url);
if !options.query.is_empty() {
request = request.query(&options.query);
}
request = request.headers(self.build_headers(&token, options)?);
match &options.body {
Some(Body::Json(value)) => request = request.json(value),
Some(Body::Raw(bytes)) => request = request.body(bytes.clone()),
None => {}
}
let timeout = self.timeout();
if !timeout.is_zero() {
request = request.timeout(timeout);
}
let response = request
.send()
.await
.map_err(|e| transport_error(e, method, path, timeout))?;
let status = response.status();
let request_id = read_request_id(response.headers());
let retry_after = (status == StatusCode::TOO_MANY_REQUESTS)
.then(|| parse_retry_after(response.headers()))
.flatten();
let body = response
.bytes()
.await
.map_err(|e| transport_error(e, method, path, timeout))?;
if !status.is_success() {
return Err(Error::api(normalize_api_error(
status.as_u16(),
parse_maybe_json(&body),
method.as_str(),
path,
request_id,
retry_after,
)));
}
if options.expect == Expect::None || status == StatusCode::NO_CONTENT {
return Ok(Vec::new());
}
Ok(body.to_vec())
}
pub(crate) async fn put_binary(
&self,
url: &str,
body: &[u8],
content_type: &str,
) -> Result<()> {
let mut attempt = 0u32;
loop {
match self.attempt_put(url, body, content_type).await {
Ok(()) => return Ok(()),
Err(err) => {
if !should_retry(&err, true) || attempt >= self.max_retries() {
return Err(err);
}
tokio::time::sleep((self.inner.backoff)(attempt)).await;
attempt += 1;
}
}
}
}
async fn attempt_put(&self, url: &str, body: &[u8], content_type: &str) -> Result<()> {
let path = safe_url_path(url);
let timeout = self.timeout();
let mut request = self
.inner
.http
.put(url)
.header(reqwest::header::CONTENT_TYPE, content_type)
.body(body.to_vec());
if !timeout.is_zero() {
request = request.timeout(timeout);
}
let response = request
.send()
.await
.map_err(|e| transport_error(e, &Method::PUT, path, timeout))?;
let status = response.status();
if status.is_success() {
return Ok(());
}
let body = response.bytes().await.unwrap_or_default();
let mut api_error = normalize_api_error(
status.as_u16(),
parse_maybe_json(&body),
Method::PUT.as_str(),
path,
None,
None,
);
api_error.message = format!("uploading the file to storage failed (HTTP {status})");
api_error.code = Some("upload_failed".to_string());
Err(Error::api(api_error))
}
fn decode<T: DeserializeOwned>(bytes: &[u8], method: &Method, path: &str) -> Result<T> {
let context = format!("{method} {path}");
if bytes.iter().all(u8::is_ascii_whitespace) {
return serde_json::from_str("null").map_err(|e| Error::decode(context, e));
}
serde_json::from_slice(bytes).map_err(|e| Error::decode(context, e))
}
}
impl Client {
pub(crate) async fn json<T: DeserializeOwned>(
&self,
method: Method,
path: &str,
options: CallOptions,
) -> Result<T> {
let bytes = self.execute(method.clone(), path, &options).await?;
Self::decode(&bytes, &method, path)
}
pub(crate) async fn maybe_json(
&self,
method: Method,
path: &str,
options: CallOptions,
) -> Result<Value> {
let bytes = self.execute(method, path, &options).await?;
Ok(parse_maybe_json(&bytes).unwrap_or(Value::Null))
}
pub(crate) async fn data<T: DeserializeOwned>(
&self,
method: Method,
path: &str,
options: CallOptions,
) -> Result<T> {
self.wrapped(method, path, "data", options).await
}
pub(crate) async fn wrapped<T: DeserializeOwned>(
&self,
method: Method,
path: &str,
key: &str,
options: CallOptions,
) -> Result<T> {
let bytes = self.execute(method.clone(), path, &options).await?;
let context = format!("{method} {path}");
if bytes.iter().all(u8::is_ascii_whitespace) {
return serde_json::from_str("null").map_err(|e| Error::decode(context, e));
}
let mut envelope: Map<String, Value> =
serde_json::from_slice(&bytes).map_err(|e| Error::decode(&context, e))?;
let inner = envelope.remove(key).unwrap_or(Value::Null);
serde_json::from_value(inner).map_err(|e| Error::decode(context, e))
}
pub(crate) async fn message(
&self,
method: Method,
path: &str,
options: CallOptions,
) -> Result<String> {
let bytes = self.execute(method, path, &options).await?;
let text = String::from_utf8_lossy(&bytes);
let trimmed = text.trim();
if trimmed.is_empty() {
return Ok(String::new());
}
if trimmed.starts_with('"') {
if let Ok(unquoted) = serde_json::from_str::<String>(trimmed) {
return Ok(unquoted);
}
}
Ok(trimmed.to_string())
}
pub(crate) async fn text(
&self,
method: Method,
path: &str,
options: CallOptions,
) -> Result<String> {
let bytes = self
.execute(method, path, &options.expect(Expect::Text))
.await?;
Ok(String::from_utf8_lossy(&bytes).into_owned())
}
pub(crate) async fn none(
&self,
method: Method,
path: &str,
options: CallOptions,
) -> Result<()> {
self.execute(method, path, &options.expect(Expect::None))
.await?;
Ok(())
}
}
impl Client {
pub async fn request(&self, method: Method, path: &str, request: RawRequest) -> Result<Value> {
let expect = request.expect;
let body = match (request.raw_body, request.body) {
(Some(bytes), _) => Some(Body::Raw(bytes)),
(None, Some(value)) => Some(Body::Json(value)),
(None, None) => None,
};
let options = CallOptions {
query: request.query,
body,
headers: request.headers,
expect,
};
let bytes = self.execute(method.clone(), path, &options).await?;
match expect {
Expect::None => Ok(Value::Null),
Expect::Text => Ok(Value::String(String::from_utf8_lossy(&bytes).into_owned())),
Expect::Json if bytes.iter().all(u8::is_ascii_whitespace) => Ok(Value::Null),
Expect::Json => Self::decode(&bytes, &method, path),
}
}
pub fn api(&self) -> Api<'_> {
Api { client: self }
}
}
#[derive(Debug, Default)]
pub struct RawRequest {
pub query: Vec<(String, String)>,
pub body: Option<Value>,
pub raw_body: Option<Vec<u8>>,
pub headers: Vec<(String, String)>,
pub expect: Expect,
}
impl RawRequest {
pub fn new() -> Self {
Self::default()
}
pub fn body(mut self, body: Value) -> Self {
self.body = Some(body);
self
}
pub fn raw_body(mut self, body: impl Into<Vec<u8>>) -> Self {
self.raw_body = Some(body.into());
self
}
pub fn header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
self.headers.push((name.into(), value.into()));
self
}
pub fn query(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
self.query.push((name.into(), value.into()));
self
}
pub fn expect(mut self, expect: Expect) -> Self {
self.expect = expect;
self
}
}
#[derive(Debug, Clone, Copy)]
pub struct Api<'a> {
client: &'a Client,
}
macro_rules! api_method {
($name:ident, $method:expr, $doc:literal) => {
#[doc = $doc]
pub async fn $name(&self, path: &str, request: RawRequest) -> Result<Value> {
self.client.request($method, path, request).await
}
};
}
impl Api<'_> {
api_method!(get, Method::GET, "Issues a raw `GET`.");
api_method!(post, Method::POST, "Issues a raw `POST`.");
api_method!(put, Method::PUT, "Issues a raw `PUT`.");
api_method!(patch, Method::PATCH, "Issues a raw `PATCH`.");
api_method!(delete, Method::DELETE, "Issues a raw `DELETE`.");
}
fn unix_now() -> i64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as i64)
.unwrap_or(0)
}
fn ensure_leading_slash(path: &str) -> String {
if path.starts_with('/') {
path.to_string()
} else {
format!("/{path}")
}
}
fn safe_url_path(url: &str) -> &str {
url.split('?').next().unwrap_or(url)
}
fn transport_error(
source: reqwest::Error,
method: &Method,
path: &str,
timeout: Duration,
) -> Error {
if source.is_timeout() {
Error::Timeout {
method: method.to_string(),
path: path.to_string(),
elapsed: timeout,
}
} else {
Error::Network {
method: method.to_string(),
path: path.to_string(),
source,
}
}
}
fn parse_maybe_json(body: &[u8]) -> Option<Value> {
if body.iter().all(u8::is_ascii_whitespace) {
return None;
}
match serde_json::from_slice(body) {
Ok(value) => Some(value),
Err(_) => Some(Value::String(String::from_utf8_lossy(body).into_owned())),
}
}
fn read_request_id(headers: &HeaderMap) -> Option<String> {
[
"x-request-id",
"request-id",
"x-amzn-requestid",
"x-amz-request-id",
]
.iter()
.find_map(|name| headers.get(*name))
.and_then(|value| value.to_str().ok())
.filter(|value| !value.is_empty())
.map(str::to_string)
}
fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
let value = headers.get("retry-after")?.to_str().ok()?.trim();
if value.is_empty() {
return None;
}
if let Ok(seconds) = value.parse::<i64>() {
return (seconds > 0).then(|| Duration::from_secs(seconds as u64));
}
let deadline = httpdate::parse_http_date(value).ok()?;
deadline.duration_since(SystemTime::now()).ok()
}
fn is_idempotent(method: &Method) -> bool {
matches!(
*method,
Method::GET | Method::PUT | Method::DELETE | Method::HEAD | Method::OPTIONS
)
}
fn should_retry(error: &Error, idempotent: bool) -> bool {
if error.status() == Some(429) {
return true;
}
if !idempotent {
return false;
}
match error {
Error::Api(api) => api.status >= 500,
Error::Network { .. } | Error::Timeout { .. } => true,
_ => false,
}
}
pub(crate) fn default_backoff(attempt: u32) -> Duration {
let base = Duration::from_secs(1)
.saturating_mul(1u32.checked_shl(attempt).unwrap_or(u32::MAX))
.min(MAX_BACKOFF);
base + Duration::from_millis(jitter_ms())
}
fn jitter_ms() -> u64 {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.subsec_nanos() as u64)
.unwrap_or(0);
let mut x = nanos
.wrapping_mul(6_364_136_223_846_793_005)
.wrapping_add(1_442_695_040_888_963_407);
x ^= x >> 33;
x = x.wrapping_mul(0xff51_afd7_ed55_8ccd);
x ^= x >> 33;
x % 250
}
#[cfg(test)]
mod tests {
use super::*;
use crate::error::ApiError;
use crate::error::ErrorKind;
fn api_error(status: u16) -> Error {
Error::api(ApiError {
message: String::new(),
kind: ErrorKind::from_status(status),
code: None,
status,
request_id: None,
details: None,
method: "GET".into(),
path: "/x".into(),
retry_after: None,
})
}
#[test]
fn idempotent_methods() {
for method in [
Method::GET,
Method::PUT,
Method::DELETE,
Method::HEAD,
Method::OPTIONS,
] {
assert!(is_idempotent(&method), "{method} should be idempotent");
}
assert!(!is_idempotent(&Method::POST));
assert!(!is_idempotent(&Method::PATCH));
}
#[test]
fn rate_limits_retry_for_any_method() {
assert!(should_retry(&api_error(429), true));
assert!(should_retry(&api_error(429), false));
}
#[test]
fn server_errors_retry_only_when_idempotent() {
assert!(should_retry(&api_error(500), true));
assert!(should_retry(&api_error(503), true));
assert!(!should_retry(&api_error(500), false));
}
#[test]
fn client_errors_never_retry() {
for status in [400, 401, 403, 404, 409] {
assert!(!should_retry(&api_error(status), true), "status {status}");
}
}
#[test]
fn timeouts_retry_only_when_idempotent() {
let err = Error::Timeout {
method: "GET".into(),
path: "/x".into(),
elapsed: Duration::from_secs(1),
};
assert!(should_retry(&err, true));
assert!(!should_retry(&err, false));
}
#[test]
fn config_and_decode_errors_never_retry() {
assert!(!should_retry(&Error::config("x"), true));
assert!(!should_retry(&Error::validation("x"), true));
}
#[test]
fn backoff_grows_exponentially_and_caps() {
let bounds = |attempt: u32| {
let d = default_backoff(attempt);
(d.as_millis() as u64) / 1000
};
assert_eq!(bounds(0), 1);
assert_eq!(bounds(1), 2);
assert_eq!(bounds(2), 4);
assert_eq!(bounds(3), 8);
assert!(default_backoff(10) < Duration::from_millis(8_250));
assert!(default_backoff(64) < Duration::from_millis(8_250));
}
#[test]
fn jitter_stays_in_range() {
for _ in 0..100 {
assert!(jitter_ms() < 250);
}
}
#[test]
fn parses_retry_after_seconds() {
let mut headers = HeaderMap::new();
headers.insert("retry-after", "3".parse().unwrap());
assert_eq!(parse_retry_after(&headers), Some(Duration::from_secs(3)));
}
#[test]
fn ignores_non_positive_or_missing_retry_after() {
assert_eq!(parse_retry_after(&HeaderMap::new()), None);
let mut headers = HeaderMap::new();
headers.insert("retry-after", "0".parse().unwrap());
assert_eq!(parse_retry_after(&headers), None);
headers.insert("retry-after", "-5".parse().unwrap());
assert_eq!(parse_retry_after(&headers), None);
headers.insert("retry-after", "garbage".parse().unwrap());
assert_eq!(parse_retry_after(&headers), None);
}
#[test]
fn parses_retry_after_http_date() {
let future = SystemTime::now() + Duration::from_secs(120);
let mut headers = HeaderMap::new();
headers.insert(
"retry-after",
httpdate::fmt_http_date(future).parse().unwrap(),
);
let parsed = parse_retry_after(&headers).expect("should parse an HTTP-date");
assert!(parsed > Duration::from_secs(60) && parsed <= Duration::from_secs(120));
let past = SystemTime::now() - Duration::from_secs(60);
headers.insert(
"retry-after",
httpdate::fmt_http_date(past).parse().unwrap(),
);
assert_eq!(parse_retry_after(&headers), None);
}
#[test]
fn reads_request_id_in_priority_order() {
let mut headers = HeaderMap::new();
assert_eq!(read_request_id(&headers), None);
headers.insert("x-amz-request-id", "amz".parse().unwrap());
assert_eq!(read_request_id(&headers).as_deref(), Some("amz"));
headers.insert("x-request-id", "primary".parse().unwrap());
assert_eq!(read_request_id(&headers).as_deref(), Some("primary"));
}
#[test]
fn parse_maybe_json_falls_back_to_text() {
assert_eq!(parse_maybe_json(b" "), None);
assert_eq!(
parse_maybe_json(b"{\"a\":1}"),
Some(serde_json::json!({"a": 1}))
);
assert_eq!(
parse_maybe_json(b"<html>oops</html>"),
Some(Value::String("<html>oops</html>".into()))
);
}
#[test]
fn safe_url_path_strips_the_signature() {
assert_eq!(
safe_url_path("https://s3/bucket/key?X-Amz-Signature=secret"),
"https://s3/bucket/key"
);
assert_eq!(safe_url_path("https://s3/key"), "https://s3/key");
}
#[test]
fn ensures_a_leading_slash() {
assert_eq!(ensure_leading_slash("v2/rooms"), "/v2/rooms");
assert_eq!(ensure_leading_slash("/v2/rooms"), "/v2/rooms");
}
}