acta_maker_sdk/http/
mod.rs

1//! Read-only client for the public Acta HTTP API.
2
3mod types;
4
5use std::collections::{BTreeMap, HashMap};
6use std::sync::{Arc, Mutex, MutexGuard};
7use std::time::Duration;
8
9use reqwest::StatusCode;
10use reqwest::header::{ETAG, IF_NONE_MATCH};
11use serde::de::DeserializeOwned;
12
13pub use types::{
14    CountLimit, PremiumFlow, ProjectionCursor, TokenProgramKind, Vault, VaultCapacity, VaultCycle,
15    VaultDepositRequest, VaultExposure, VaultExposurePosition, VaultExposureTotal, VaultFees,
16    VaultLeg, VaultMetadata, VaultPerformance, VaultPolicy, VaultPosition, VaultPositionProduct,
17    VaultPositionRole, VaultPositionStatus, VaultPositionType, VaultPremium, VaultReturns,
18    VaultStaging, VaultWithdrawRequest, Venue, WindowReturn,
19};
20use types::{MarketsResponse, TokensResponse, VaultPositionsResponse, VaultsResponse};
21use types::{VaultCyclesResponse, VaultRequestsResponse};
22
23use crate::ws::types::{MarketInfo, TokenInfo};
24
25const TIMEOUT: Duration = Duration::from_secs(10);
26
27#[derive(Debug, thiserror::Error)]
28#[non_exhaustive]
29pub enum HttpError {
30    #[error("request failed: {0}")]
31    Request(#[from] reqwest::Error),
32    #[error("HTTP {status}: {body}")]
33    Status { status: u16, body: String },
34    #[error("invalid response body: {0}")]
35    Decode(#[from] serde_json::Error),
36    #[error("page cursor did not advance")]
37    InvalidCursor,
38    #[error(transparent)]
39    InvalidPage(#[from] crate::paging::InvalidPage),
40}
41
42#[derive(Debug, Clone, Default, PartialEq, Eq)]
43pub struct MarketFilter {
44    pub underlying: Option<crate::Address32>,
45    pub quote: Option<crate::Address32>,
46    pub expiry: Option<u64>,
47    pub is_put: Option<bool>,
48    pub active: Option<bool>,
49}
50
51impl MarketFilter {
52    fn query(&self) -> String {
53        let mut params = Vec::new();
54        if let Some(underlying) = &self.underlying {
55            params.push(format!("underlying={underlying}"));
56        }
57        if let Some(quote) = &self.quote {
58            params.push(format!("quote={quote}"));
59        }
60        if let Some(expiry) = self.expiry {
61            params.push(format!("expiry={expiry}"));
62        }
63        if let Some(is_put) = self.is_put {
64            params.push(format!("is_put={is_put}"));
65        }
66        if let Some(active) = self.active {
67            params.push(format!("active={active}"));
68        }
69        if params.is_empty() {
70            String::new()
71        } else {
72            format!("?{}", params.join("&"))
73        }
74    }
75}
76
77#[derive(Debug, Clone)]
78pub struct HttpClient {
79    client: reqwest::Client,
80    base_url: String,
81    cache: Arc<Mutex<HashMap<String, Cached>>>,
82}
83
84#[derive(Debug)]
85struct Cached {
86    etag: String,
87    body: Arc<[u8]>,
88}
89
90impl HttpClient {
91    /// `base_url` is the API origin, e.g. `https://beta-api.acta.markets`.
92    pub fn new(base_url: impl Into<String>) -> Result<Self, HttpError> {
93        static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
94        let client = match CLIENT.get() {
95            Some(client) => client.clone(),
96            None => {
97                let client = reqwest::Client::builder().timeout(TIMEOUT).build()?;
98                CLIENT.get_or_init(|| client).clone()
99            }
100        };
101        Ok(Self::with_client(client, base_url))
102    }
103
104    #[must_use]
105    pub fn with_client(client: reqwest::Client, base_url: impl Into<String>) -> Self {
106        let mut base_url = base_url.into();
107        while base_url.ends_with('/') {
108            let _ = base_url.pop();
109        }
110        Self {
111            client,
112            base_url,
113            cache: Arc::default(),
114        }
115    }
116
117    #[must_use]
118    pub fn base_url(&self) -> &str {
119        &self.base_url
120    }
121
122    pub async fn vaults(&self) -> Result<Vec<Vault>, HttpError> {
123        let response: VaultsResponse = self
124            .collect(
125                "/api/v1/vaults",
126                |all: &VaultsResponse| all.next_cursor.as_deref(),
127                |all, page| {
128                    all.vaults.extend(page.vaults);
129                    all.next_cursor = page.next_cursor;
130                },
131            )
132            .await?;
133        Ok(response.vaults)
134    }
135
136    /// Vaults whose primary delegate is `delegate` with an unexpired grant, the
137    /// same rule `/vault` authentication applies.
138    pub async fn delegated_vaults(
139        &self,
140        delegate: crate::Address32,
141    ) -> Result<Vec<Vault>, HttpError> {
142        let response: VaultsResponse = self
143            .collect(
144                &format!("/api/v1/vaults?delegate={delegate}"),
145                |all: &VaultsResponse| all.next_cursor.as_deref(),
146                |all, page| {
147                    all.vaults.extend(page.vaults);
148                    all.next_cursor = page.next_cursor;
149                },
150            )
151            .await?;
152        Ok(response.vaults)
153    }
154
155    pub async fn vault(&self, pda: crate::Address32) -> Result<Vault, HttpError> {
156        self.get(&format!("/api/v1/vaults/{pda}")).await
157    }
158
159    /// Every unsettled position of the vault, following the server cursor.
160    pub async fn vault_positions(
161        &self,
162        pda: crate::Address32,
163    ) -> Result<Vec<VaultPosition>, HttpError> {
164        let response: VaultPositionsResponse = self
165            .collect(
166                &format!("/api/v1/vaults/{pda}/positions?settled=false"),
167                |all: &VaultPositionsResponse| all.next_cursor.as_deref(),
168                |all, page| {
169                    all.positions.extend(page.positions);
170                    all.next_cursor = page.next_cursor;
171                },
172            )
173            .await?;
174        Ok(response.positions)
175    }
176
177    /// Every cycle of the vault, newest first.
178    pub async fn vault_cycles(&self, pda: crate::Address32) -> Result<Vec<VaultCycle>, HttpError> {
179        let response: VaultCyclesResponse = self
180            .collect(
181                &format!("/api/v1/vaults/{pda}/cycles"),
182                |all: &VaultCyclesResponse| all.next_cursor.as_deref(),
183                |all, page| {
184                    all.cycles.extend(page.cycles);
185                    all.next_cursor = page.next_cursor;
186                },
187            )
188            .await?;
189        Ok(response.cycles)
190    }
191
192    /// Every pending deposit of the vault, read without Solana RPC.
193    pub async fn vault_deposit_requests(
194        &self,
195        pda: crate::Address32,
196    ) -> Result<Vec<VaultDepositRequest>, HttpError> {
197        self.vault_requests(&format!("/api/v1/vaults/{pda}/deposit-requests"))
198            .await
199    }
200
201    /// Every pending withdrawal of the vault, read without Solana RPC.
202    pub async fn vault_withdraw_requests(
203        &self,
204        pda: crate::Address32,
205    ) -> Result<Vec<VaultWithdrawRequest>, HttpError> {
206        self.vault_requests(&format!("/api/v1/vaults/{pda}/withdraw-requests"))
207            .await
208    }
209
210    async fn vault_requests<T: DeserializeOwned>(&self, path: &str) -> Result<Vec<T>, HttpError> {
211        let response: VaultRequestsResponse<T> = crate::paging::fetch_all(
212            |cursor| async move { self.get(&Self::page_path(path, 100, &cursor)?).await },
213            |all: &VaultRequestsResponse<T>| all.next_cursor.as_deref(),
214            |all, page| {
215                all.requests.extend(page.requests);
216                all.next_cursor = page.next_cursor;
217            },
218        )
219        .await?;
220        Ok(response.requests)
221    }
222
223    /// Open and funded positions with indicative spot, moneyness and intrinsic value.
224    pub async fn vault_exposure(&self, pda: crate::Address32) -> Result<VaultExposure, HttpError> {
225        self.collect(
226            &format!("/api/v1/vaults/{pda}/exposure"),
227            |all: &VaultExposure| all.next_cursor.as_deref(),
228            |all, page| {
229                all.positions.extend(page.positions);
230                all.next_cursor = page.next_cursor;
231            },
232        )
233        .await
234    }
235
236    /// One exposure page. The totals describe the full cached snapshot.
237    pub async fn vault_exposure_page(
238        &self,
239        pda: crate::Address32,
240        limit: u16,
241        cursor: Option<&str>,
242    ) -> Result<VaultExposure, HttpError> {
243        let cursor = cursor.map_or(crate::ws::types::Cursor::Start {}, |id| {
244            crate::ws::types::Cursor::After { id: id.to_owned() }
245        });
246        let path = Self::page_path(&format!("/api/v1/vaults/{pda}/exposure"), limit, &cursor)?;
247        self.get(&path).await
248    }
249
250    pub async fn markets(&self, filter: &MarketFilter) -> Result<Vec<MarketInfo>, HttpError> {
251        let response: MarketsResponse = self
252            .collect(
253                &format!("/api/v1/markets{}", filter.query()),
254                |all: &MarketsResponse| all.next_cursor.as_deref(),
255                |all, page| {
256                    all.markets.extend(page.markets);
257                    all.next_cursor = page.next_cursor;
258                },
259            )
260            .await?;
261        Ok(response.markets)
262    }
263
264    pub async fn tokens(&self) -> Result<BTreeMap<crate::Address32, TokenInfo>, HttpError> {
265        let response: TokensResponse = self
266            .collect(
267                "/api/v1/tokens",
268                |all: &TokensResponse| all.next_cursor.as_deref(),
269                |all, page| {
270                    all.tokens.extend(page.tokens);
271                    all.next_cursor = page.next_cursor;
272                },
273            )
274            .await?;
275        Ok(response
276            .tokens
277            .into_iter()
278            .map(|token| (token.mint, token))
279            .collect())
280    }
281
282    fn page_path(
283        path: &str,
284        limit: u16,
285        cursor: &crate::ws::types::Cursor,
286    ) -> Result<String, HttpError> {
287        let mut url = reqwest::Url::parse(&format!("http://localhost{path}"))
288            .map_err(|_| HttpError::InvalidCursor)?;
289        url.query_pairs_mut()
290            .append_pair("limit", &limit.to_string());
291        if let crate::ws::types::Cursor::After { id } = cursor {
292            url.query_pairs_mut().append_pair("cursor", id);
293        }
294        Ok(match url.query() {
295            Some(query) => format!("{}?{query}", url.path()),
296            None => url.path().to_owned(),
297        })
298    }
299
300    async fn collect<T: DeserializeOwned>(
301        &self,
302        path: &str,
303        next: impl Fn(&T) -> Option<&str>,
304        append: impl Fn(&mut T, T),
305    ) -> Result<T, HttpError> {
306        crate::paging::fetch_all(
307            |cursor| async move { self.get(&Self::page_path(path, 200, &cursor)?).await },
308            next,
309            append,
310        )
311        .await
312    }
313
314    pub async fn venue(&self) -> Result<Venue, HttpError> {
315        self.get("/api/v1/venue").await
316    }
317
318    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, HttpError> {
319        let mut request = self.client.get(format!("{}{path}", self.base_url));
320        if let Some(etag) = self.cached(path, |cached| cached.etag.clone()) {
321            request = request.header(IF_NONE_MATCH, etag);
322        }
323        let response = crate::io_observer::send(request, "acta", path).await?;
324        let status = response.status();
325        if status == StatusCode::NOT_MODIFIED
326            && let Some(body) = self.cached(path, |cached| Arc::clone(&cached.body))
327        {
328            return Ok(serde_json::from_slice(&body)?);
329        }
330        if !status.is_success() {
331            let body =
332                String::from_utf8_lossy(&crate::io_observer::bytes(response, "acta", path).await?)
333                    .into_owned();
334            return Err(HttpError::Status {
335                status: status.as_u16(),
336                body,
337            });
338        }
339        let etag = response
340            .headers()
341            .get(ETAG)
342            .and_then(|value| value.to_str().ok())
343            .map(str::to_owned);
344        let body = crate::io_observer::bytes(response, "acta", path).await?;
345        let value = serde_json::from_slice(&body)?;
346        if let Some(etag) = etag {
347            let _ = self.cache().insert(
348                path.to_owned(),
349                Cached {
350                    etag,
351                    body: Arc::from(body.as_slice()),
352                },
353            );
354        }
355        Ok(value)
356    }
357
358    fn cached<R>(&self, path: &str, read: impl FnOnce(&Cached) -> R) -> Option<R> {
359        self.cache().get(path).map(read)
360    }
361
362    fn cache(&self) -> MutexGuard<'_, HashMap<String, Cached>> {
363        match self.cache.lock() {
364            Ok(cache) => cache,
365            Err(poisoned) => poisoned.into_inner(),
366        }
367    }
368}
369
370#[cfg(test)]
371mod tests;
acta_maker_sdk/http/mod.rs

acta_maker_sdk/http/
mod.rs
