highlevel_api/http/
mod.rs

1use std::sync::atomic::{AtomicU32, Ordering};
2use std::sync::Arc;
3use std::time::{Duration, SystemTime, UNIX_EPOCH};
4
5use reqwest::{Client, RequestBuilder, Response};
6use serde::{de::DeserializeOwned, Serialize};
7use tokio::sync::RwLock;
8use tokio::time::Instant;
9
10use crate::{
11    config::HighLevelConfig,
12    error::{Error, Result},
13};
14
15const DEFAULT_RATE: u32 = 95;
16const DEFAULT_WINDOW_SECS: u64 = 10;
17const MAX_RETRIES: u32 = 3;
18const CIRCUIT_BREAK_THRESHOLD: u32 = 3;
19const CIRCUIT_BREAK_RESET_SECS: u64 = 30;
20
21struct TokenBucket {
22    capacity: u32,
23    tokens: f64,
24    refill_per_sec: f64,
25    last_refill: Instant,
26}
27
28impl TokenBucket {
29    fn new(capacity: u32, window_secs: u64) -> Self {
30        Self {
31            capacity,
32            tokens: capacity as f64,
33            refill_per_sec: capacity as f64 / window_secs as f64,
34            last_refill: Instant::now(),
35        }
36    }
37
38    fn try_acquire(&mut self) -> bool {
39        self.refill();
40        if self.tokens >= 1.0 {
41            self.tokens -= 1.0;
42            true
43        } else {
44            false
45        }
46    }
47
48    fn refill(&mut self) {
49        let elapsed = self.last_refill.elapsed().as_secs_f64();
50        self.tokens = (self.tokens + elapsed * self.refill_per_sec).min(self.capacity as f64);
51        self.last_refill = Instant::now();
52    }
53}
54
55struct CircuitBreaker {
56    consecutive_429s: AtomicU32,
57    open_until: RwLock<Option<Instant>>,
58}
59
60impl CircuitBreaker {
61    fn new() -> Self {
62        Self {
63            consecutive_429s: AtomicU32::new(0),
64            open_until: RwLock::new(None),
65        }
66    }
67
68    async fn is_open(&self) -> bool {
69        if let Some(until) = *self.open_until.read().await {
70            if Instant::now() < until {
71                return true;
72            }
73            self.open_until.write().await.take();
74            self.consecutive_429s.store(0, Ordering::SeqCst);
75        }
76        false
77    }
78
79    async fn record_429(&self) {
80        let count = self.consecutive_429s.fetch_add(1, Ordering::SeqCst) + 1;
81        if count >= CIRCUIT_BREAK_THRESHOLD {
82            *self.open_until.write().await =
83                Some(Instant::now() + Duration::from_secs(CIRCUIT_BREAK_RESET_SECS));
84        }
85    }
86
87    async fn record_success(&self) {
88        self.consecutive_429s.store(0, Ordering::SeqCst);
89        self.open_until.write().await.take();
90    }
91}
92
93pub struct HttpClient {
94    inner: Client,
95    pub base_url: String,
96    pub api_version: String,
97    token: RwLock<Option<String>>,
98    bucket: RwLock<TokenBucket>,
99    breaker: CircuitBreaker,
100}
101
102impl HttpClient {
103    pub fn new(config: &HighLevelConfig) -> Self {
104        Self {
105            inner: Client::new(),
106            base_url: config.base_url.clone(),
107            api_version: config.api_version.clone(),
108            token: RwLock::new(None),
109            bucket: RwLock::new(TokenBucket::new(DEFAULT_RATE, DEFAULT_WINDOW_SECS)),
110            breaker: CircuitBreaker::new(),
111        }
112    }
113
114    pub async fn set_token(&self, token: String) {
115        *self.token.write().await = Some(token);
116    }
117
118    pub async fn get_token(&self) -> Option<String> {
119        self.token.read().await.clone()
120    }
121
122    fn url(&self, path: &str) -> String {
123        format!("{}{}", self.base_url, path)
124    }
125
126    async fn acquire_token(&self) -> Result<()> {
127        if self.breaker.is_open().await {
128            return Err(Error::RateLimited {
129                retry_after: Some(CIRCUIT_BREAK_RESET_SECS),
130                circuit_breaker: true,
131            });
132        }
133        loop {
134            if self.bucket.write().await.try_acquire() {
135                return Ok(());
136            }
137            tokio::time::sleep(Duration::from_millis(50)).await;
138        }
139    }
140
141    /// Execute a request with rate-limit awareness, retry-on-429, and circuit
142    /// breaking. `build` is a closure (called once per attempt) that receives a
143    /// `&Client` and returns a `RequestBuilder`. Auth is applied automatically.
144    async fn execute_with_retry<F>(&self, build: F) -> Result<Response>
145    where
146        F: Fn(&Client) -> RequestBuilder,
147    {
148        self.acquire_token().await?;
149
150        let token = self.token.read().await.clone();
151        let mut builder = build(&self.inner);
152        if let Some(ref t) = token {
153            builder = builder.bearer_auth(t);
154        }
155        let req = builder.build()?;
156        let mut response = self.inner.execute(req).await?;
157
158        let mut retries = 0u32;
159        while response.status() == 429 && retries < MAX_RETRIES {
160            self.breaker.record_429().await;
161
162            let delay = exponential_backoff(retries) + jitter();
163            tokio::time::sleep(delay).await;
164
165            let token = self.token.read().await.clone();
166            let mut builder = build(&self.inner);
167            if let Some(ref t) = token {
168                builder = builder.bearer_auth(t);
169            }
170            let req = builder.build()?;
171            response = self.inner.execute(req).await?;
172            retries += 1;
173        }
174
175        if response.status() == 429 {
176            let retry_after = response
177                .headers()
178                .get("Retry-After")
179                .and_then(|v| v.to_str().ok())
180                .and_then(|v| v.parse::<u64>().ok());
181            return Err(Error::RateLimited {
182                retry_after,
183                circuit_breaker: self.breaker.is_open().await,
184            });
185        }
186
187        self.breaker.record_success().await;
188        Ok(response)
189    }
190
191    async fn handle_response<T: DeserializeOwned>(&self, resp: Response) -> Result<T> {
192        let status = resp.status();
193        if status.is_success() {
194            Ok(resp.json::<T>().await?)
195        } else {
196            let code = status.as_u16();
197            let body = resp.text().await.unwrap_or_default();
198            let message = serde_json::from_str::<serde_json::Value>(&body)
199                .ok()
200                .and_then(|v| {
201                    v.get("message")
202                        .and_then(|m| m.as_str())
203                        .map(|s| s.to_string())
204                })
205                .unwrap_or(body);
206            Err(Error::Api {
207                status: code,
208                message,
209            })
210        }
211    }
212
213    async fn handle_no_content(&self, resp: Response) -> Result<()> {
214        let status = resp.status();
215        if status.is_success() {
216            Ok(())
217        } else {
218            let code = status.as_u16();
219            let message = resp.text().await.unwrap_or_else(|_| status.to_string());
220            Err(Error::Api {
221                status: code,
222                message,
223            })
224        }
225    }
226
227    // ── Typed HTTP methods ────────────────────────────────────────────────────
228
229    pub async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
230        let u = Arc::new(self.url(path));
231        let v = self.api_version.clone();
232        let resp = self
233            .execute_with_retry(|client| {
234                client
235                    .get(u.as_str())
236                    .header("Version", &v)
237                    .header("Content-Type", "application/json")
238            })
239            .await?;
240        self.handle_response(resp).await
241    }
242
243    pub async fn get_with_query<T: DeserializeOwned, Q: Serialize + ?Sized>(
244        &self,
245        path: &str,
246        query: &Q,
247    ) -> Result<T> {
248        let u = Arc::new(self.url(path));
249        let v = self.api_version.clone();
250        // Serialize query params so the closure can be Fn + Send
251        let qv = serde_json::to_value(query).map_err(Error::Json)?;
252        let resp = self
253            .execute_with_retry(move |client| {
254                client
255                    .get(u.as_str())
256                    .header("Version", &v)
257                    .header("Content-Type", "application/json")
258                    .query(&qv)
259            })
260            .await?;
261        self.handle_response(resp).await
262    }
263
264    pub async fn post<T: DeserializeOwned, B: Serialize>(&self, path: &str, body: &B) -> Result<T> {
265        let u = Arc::new(self.url(path));
266        let v = self.api_version.clone();
267        let body_json = serde_json::to_value(body).map_err(Error::Json)?;
268        let resp = self
269            .execute_with_retry(move |client| {
270                client
271                    .post(u.as_str())
272                    .header("Version", &v)
273                    .json(&body_json)
274            })
275            .await?;
276        self.handle_response(resp).await
277    }
278
279    pub async fn put<T: DeserializeOwned, B: Serialize>(&self, path: &str, body: &B) -> Result<T> {
280        let u = Arc::new(self.url(path));
281        let v = self.api_version.clone();
282        let body_json = serde_json::to_value(body).map_err(Error::Json)?;
283        let resp = self
284            .execute_with_retry(move |client| {
285                client
286                    .put(u.as_str())
287                    .header("Version", &v)
288                    .json(&body_json)
289            })
290            .await?;
291        self.handle_response(resp).await
292    }
293
294    pub async fn patch<T: DeserializeOwned, B: Serialize>(
295        &self,
296        path: &str,
297        body: &B,
298    ) -> Result<T> {
299        let u = Arc::new(self.url(path));
300        let v = self.api_version.clone();
301        let body_json = serde_json::to_value(body).map_err(Error::Json)?;
302        let resp = self
303            .execute_with_retry(move |client| {
304                client
305                    .patch(u.as_str())
306                    .header("Version", &v)
307                    .json(&body_json)
308            })
309            .await?;
310        self.handle_response(resp).await
311    }
312
313    pub async fn delete<T: DeserializeOwned>(&self, path: &str) -> Result<T> {
314        let u = Arc::new(self.url(path));
315        let v = self.api_version.clone();
316        let resp = self
317            .execute_with_retry(|client| client.delete(u.as_str()).header("Version", &v))
318            .await?;
319        self.handle_response(resp).await
320    }
321
322    pub async fn delete_with_body<T: DeserializeOwned, B: Serialize>(
323        &self,
324        path: &str,
325        body: &B,
326    ) -> Result<T> {
327        let u = Arc::new(self.url(path));
328        let v = self.api_version.clone();
329        let body_json = serde_json::to_value(body).map_err(Error::Json)?;
330        let resp = self
331            .execute_with_retry(move |client| {
332                client
333                    .delete(u.as_str())
334                    .header("Version", &v)
335                    .json(&body_json)
336            })
337            .await?;
338        self.handle_response(resp).await
339    }
340
341    pub async fn delete_no_content(&self, path: &str) -> Result<()> {
342        let u = Arc::new(self.url(path));
343        let v = self.api_version.clone();
344        let resp = self
345            .execute_with_retry(|client| client.delete(u.as_str()).header("Version", &v))
346            .await?;
347        self.handle_no_content(resp).await
348    }
349
350    // ── Raw escape-hatch methods ──────────────────────────────────────────────
351
352    pub async fn get_raw(&self, path: &str) -> Result<Response> {
353        let u = Arc::new(self.url(path));
354        let v = self.api_version.clone();
355        self.execute_with_retry(|client| {
356            client
357                .get(u.as_str())
358                .header("Version", &v)
359                .header("Content-Type", "application/json")
360        })
361        .await
362    }
363
364    pub async fn post_raw<B: Serialize>(&self, path: &str, body: &B) -> Result<Response> {
365        let u = Arc::new(self.url(path));
366        let v = self.api_version.clone();
367        let body_json = serde_json::to_value(body).map_err(Error::Json)?;
368        self.execute_with_retry(move |client| {
369            client
370                .post(u.as_str())
371                .header("Version", &v)
372                .json(&body_json)
373        })
374        .await
375    }
376}
377
378fn exponential_backoff(retry: u32) -> Duration {
379    let base_ms = 200u64 * 2u64.pow(retry);
380    Duration::from_millis(base_ms.min(10_000))
381}
382
383fn jitter() -> Duration {
384    let t = SystemTime::now()
385        .duration_since(UNIX_EPOCH)
386        .map(|d| d.as_nanos() as u64)
387        .unwrap_or(0);
388    let hash = t
389        .wrapping_mul(6364136223846793005)
390        .wrapping_add(1442695040888963407);
391    Duration::from_millis(hash % 200)
392}
highlevel_api/http/mod.rs

highlevel_api/http/
mod.rs
