use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use reqwest::Method;
use serde::de::DeserializeOwned;
use serde_json::Value;
use crate::error::{map_gateway_error, SolariError};
use crate::types::GatewayErrorBody;
pub(crate) fn encode_uri_component(s: &str) -> String {
let mut out = String::with_capacity(s.len());
for &b in s.as_bytes() {
let c = b as char;
if c.is_ascii_alphanumeric()
|| matches!(c, '-' | '_' | '.' | '!' | '~' | '*' | '\'' | '(' | ')')
{
out.push(c);
} else {
out.push('%');
out.push_str(&format!("{b:02X}"));
}
}
out
}
#[derive(Debug, Clone)]
pub struct HttpOptions {
pub api_key: String,
pub base_url: String,
pub max_retries: u32,
pub request_timeout: Duration,
pub retry_delay_ms: Option<u64>,
}
impl HttpOptions {
pub fn new(api_key: impl Into<String>, base_url: impl Into<String>) -> Self {
HttpOptions {
api_key: api_key.into(),
base_url: base_url.into(),
max_retries: 5,
request_timeout: Duration::from_millis(300_000),
retry_delay_ms: None,
}
}
}
pub struct HttpTransport {
api_key: String,
base_url: String,
client: reqwest::Client,
max_retries: u32,
retry_delay_ms: Option<u64>,
}
impl HttpTransport {
pub fn new(opts: HttpOptions) -> Result<Self, SolariError> {
if opts.api_key.is_empty() {
return Err(SolariError::Other("HttpTransport requires an apiKey".into()));
}
if opts.base_url.is_empty() {
return Err(SolariError::Other("HttpTransport requires a baseUrl".into()));
}
let base_url = opts.base_url.trim_end_matches('/').to_string();
let client = reqwest::Client::builder()
.timeout(opts.request_timeout)
.build()
.map_err(|e| SolariError::Other(format!("failed to build HTTP client: {e}")))?;
Ok(HttpTransport {
api_key: opts.api_key,
base_url,
client,
max_retries: opts.max_retries,
retry_delay_ms: opts.retry_delay_ms,
})
}
pub fn api_key(&self) -> &str {
&self.api_key
}
pub fn auth_headers(&self) -> Vec<(String, String)> {
vec![("Authorization".into(), format!("Bearer {}", self.api_key))]
}
pub fn ws_origin(&self) -> String {
self.base_url.replacen("http", "ws", 1)
}
pub async fn request<T: DeserializeOwned>(
&self,
method: &str,
path: &str,
body: Option<Value>,
idempotency_key: Option<String>,
) -> Result<T, SolariError> {
let idempotent =
method == "GET" || method == "DELETE" || idempotency_key.is_some();
let m = Method::from_bytes(method.as_bytes())
.map_err(|e| SolariError::Other(format!("bad method {method}: {e}")))?;
let url = format!("{}{}", self.base_url, path);
let body_str = body.as_ref().map(|b| serde_json::to_string(b).unwrap());
let mut attempt: u32 = 0;
loop {
let mut rb = self
.client
.request(m.clone(), &url)
.header("Authorization", format!("Bearer {}", self.api_key))
.header("Accept", "application/json");
if let Some(k) = &idempotency_key {
rb = rb.header("Idempotency-Key", k);
}
if let Some(b) = &body_str {
rb = rb.header("Content-Type", "application/json").body(b.clone());
}
let resp = match rb.send().await {
Ok(r) => r,
Err(err) => {
if idempotent && attempt < self.max_retries {
tokio::time::sleep(self.backoff(attempt)).await;
attempt += 1;
continue;
}
return Err(SolariError::connection(format!(
"{method} {path} failed: {err}"
)));
}
};
let status = resp.status().as_u16();
if !(200..300).contains(&status) {
let text = resp.text().await.unwrap_or_default();
let err_body: Option<GatewayErrorBody> =
if text.is_empty() { None } else { serde_json::from_str(&text).ok() };
let retryable = matches!(status, 502 | 503 | 504 | 507)
|| err_body.as_ref().and_then(|b| b.retryable).unwrap_or(false);
if idempotent && retryable && attempt < self.max_retries {
tokio::time::sleep(self.backoff(attempt)).await;
attempt += 1;
continue;
}
return Err(map_gateway_error(status, err_body));
}
let text = resp.text().await.unwrap_or_default();
if text.is_empty() {
return serde_json::from_str("null")
.map_err(|e| SolariError::Other(format!("empty body decode: {e}")));
}
return serde_json::from_str(&text)
.map_err(|e| SolariError::Other(format!("bad response body: {e}")));
}
}
fn backoff(&self, attempt: u32) -> Duration {
if let Some(ms) = self.retry_delay_ms {
return Duration::from_millis(ms);
}
let base = (150u64.saturating_mul(1u64 << attempt.min(20))).min(8000);
Duration::from_millis(base + jitter_ms())
}
}
pub fn new_idempotency_key() -> String {
static COUNTER: AtomicU64 = AtomicU64::new(0);
let n = COUNTER.fetch_add(1, Ordering::Relaxed);
let t = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos() as u64)
.unwrap_or(0);
let a = t ^ n.rotate_left(17) ^ 0x9E37_79B9_7F4A_7C15;
let b = t.rotate_left(31) ^ n.wrapping_mul(0x2545_F491_4F6C_DD1D);
format!(
"{:08x}-{:04x}-4{:03x}-{:04x}-{:012x}",
(a >> 32) as u32,
(a >> 16) as u16,
(a & 0x0fff) as u16,
((b >> 48) as u16 & 0x3fff) | 0x8000,
b & 0xffff_ffff_ffff
)
}
fn jitter_ms() -> u64 {
let t = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.subsec_nanos() as u64)
.unwrap_or(0);
(t ^ t.rotate_left(13)) % 250
}