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        let client = reqwest::Client::builder().timeout(TIMEOUT).build()?;
94        Ok(Self::with_client(client, base_url))
95    }
96
97    #[must_use]
98    pub fn with_client(client: reqwest::Client, base_url: impl Into<String>) -> Self {
99        let mut base_url = base_url.into();
100        while base_url.ends_with('/') {
101            let _ = base_url.pop();
102        }
103        Self {
104            client,
105            base_url,
106            cache: Arc::default(),
107        }
108    }
109
110    #[must_use]
111    pub fn base_url(&self) -> &str {
112        &self.base_url
113    }
114
115    pub async fn vaults(&self) -> Result<Vec<Vault>, HttpError> {
116        let response: VaultsResponse = self
117            .collect(
118                "/api/v1/vaults",
119                |all: &VaultsResponse| all.next_cursor.as_deref(),
120                |all, page| {
121                    all.vaults.extend(page.vaults);
122                    all.next_cursor = page.next_cursor;
123                },
124            )
125            .await?;
126        Ok(response.vaults)
127    }
128
129    /// Vaults whose primary delegate is `delegate` with an unexpired grant, the
130    /// same rule `/vault` authentication applies.
131    pub async fn delegated_vaults(
132        &self,
133        delegate: crate::Address32,
134    ) -> Result<Vec<Vault>, HttpError> {
135        let response: VaultsResponse = self
136            .collect(
137                &format!("/api/v1/vaults?delegate={delegate}"),
138                |all: &VaultsResponse| all.next_cursor.as_deref(),
139                |all, page| {
140                    all.vaults.extend(page.vaults);
141                    all.next_cursor = page.next_cursor;
142                },
143            )
144            .await?;
145        Ok(response.vaults)
146    }
147
148    pub async fn vault(&self, pda: crate::Address32) -> Result<Vault, HttpError> {
149        self.get(&format!("/api/v1/vaults/{pda}")).await
150    }
151
152    /// Every unsettled position of the vault, following the server cursor.
153    pub async fn vault_positions(
154        &self,
155        pda: crate::Address32,
156    ) -> Result<Vec<VaultPosition>, HttpError> {
157        let response: VaultPositionsResponse = self
158            .collect(
159                &format!("/api/v1/vaults/{pda}/positions?settled=false"),
160                |all: &VaultPositionsResponse| all.next_cursor.as_deref(),
161                |all, page| {
162                    all.positions.extend(page.positions);
163                    all.next_cursor = page.next_cursor;
164                },
165            )
166            .await?;
167        Ok(response.positions)
168    }
169
170    /// Every cycle of the vault, newest first.
171    pub async fn vault_cycles(&self, pda: crate::Address32) -> Result<Vec<VaultCycle>, HttpError> {
172        let response: VaultCyclesResponse = self
173            .collect(
174                &format!("/api/v1/vaults/{pda}/cycles"),
175                |all: &VaultCyclesResponse| all.next_cursor.as_deref(),
176                |all, page| {
177                    all.cycles.extend(page.cycles);
178                    all.next_cursor = page.next_cursor;
179                },
180            )
181            .await?;
182        Ok(response.cycles)
183    }
184
185    /// Every pending deposit of the vault, read without Solana RPC.
186    pub async fn vault_deposit_requests(
187        &self,
188        pda: crate::Address32,
189    ) -> Result<Vec<VaultDepositRequest>, HttpError> {
190        self.vault_requests(&format!("/api/v1/vaults/{pda}/deposit-requests"))
191            .await
192    }
193
194    /// Every pending withdrawal of the vault, read without Solana RPC.
195    pub async fn vault_withdraw_requests(
196        &self,
197        pda: crate::Address32,
198    ) -> Result<Vec<VaultWithdrawRequest>, HttpError> {
199        self.vault_requests(&format!("/api/v1/vaults/{pda}/withdraw-requests"))
200            .await
201    }
202
203    async fn vault_requests<T: DeserializeOwned>(&self, path: &str) -> Result<Vec<T>, HttpError> {
204        let response: VaultRequestsResponse<T> = crate::paging::fetch_all(
205            |cursor| async move { self.get(&Self::page_path(path, 100, &cursor)?).await },
206            |all: &VaultRequestsResponse<T>| all.next_cursor.as_deref(),
207            |all, page| {
208                all.requests.extend(page.requests);
209                all.next_cursor = page.next_cursor;
210            },
211        )
212        .await?;
213        Ok(response.requests)
214    }
215
216    /// Open and funded positions with indicative spot, moneyness and intrinsic value.
217    pub async fn vault_exposure(&self, pda: crate::Address32) -> Result<VaultExposure, HttpError> {
218        self.collect(
219            &format!("/api/v1/vaults/{pda}/exposure"),
220            |all: &VaultExposure| all.next_cursor.as_deref(),
221            |all, page| {
222                all.positions.extend(page.positions);
223                all.next_cursor = page.next_cursor;
224            },
225        )
226        .await
227    }
228
229    /// One exposure page. The totals describe the full cached snapshot.
230    pub async fn vault_exposure_page(
231        &self,
232        pda: crate::Address32,
233        limit: u16,
234        cursor: Option<&str>,
235    ) -> Result<VaultExposure, HttpError> {
236        let cursor = cursor.map_or(crate::ws::types::Cursor::Start {}, |id| {
237            crate::ws::types::Cursor::After { id: id.to_owned() }
238        });
239        let path = Self::page_path(&format!("/api/v1/vaults/{pda}/exposure"), limit, &cursor)?;
240        self.get(&path).await
241    }
242
243    pub async fn markets(&self, filter: &MarketFilter) -> Result<Vec<MarketInfo>, HttpError> {
244        let response: MarketsResponse = self
245            .collect(
246                &format!("/api/v1/markets{}", filter.query()),
247                |all: &MarketsResponse| all.next_cursor.as_deref(),
248                |all, page| {
249                    all.markets.extend(page.markets);
250                    all.next_cursor = page.next_cursor;
251                },
252            )
253            .await?;
254        Ok(response.markets)
255    }
256
257    pub async fn tokens(&self) -> Result<BTreeMap<crate::Address32, TokenInfo>, HttpError> {
258        let response: TokensResponse = self
259            .collect(
260                "/api/v1/tokens",
261                |all: &TokensResponse| all.next_cursor.as_deref(),
262                |all, page| {
263                    all.tokens.extend(page.tokens);
264                    all.next_cursor = page.next_cursor;
265                },
266            )
267            .await?;
268        Ok(response
269            .tokens
270            .into_iter()
271            .map(|token| (token.mint, token))
272            .collect())
273    }
274
275    fn page_path(
276        path: &str,
277        limit: u16,
278        cursor: &crate::ws::types::Cursor,
279    ) -> Result<String, HttpError> {
280        let mut url = reqwest::Url::parse(&format!("http://localhost{path}"))
281            .map_err(|_| HttpError::InvalidCursor)?;
282        url.query_pairs_mut()
283            .append_pair("limit", &limit.to_string());
284        if let crate::ws::types::Cursor::After { id } = cursor {
285            url.query_pairs_mut().append_pair("cursor", id);
286        }
287        Ok(format!(
288            "{}?{}",
289            url.path(),
290            url.query().unwrap_or_default()
291        ))
292    }
293
294    async fn collect<T: DeserializeOwned>(
295        &self,
296        path: &str,
297        next: impl Fn(&T) -> Option<&str>,
298        append: impl Fn(&mut T, T),
299    ) -> Result<T, HttpError> {
300        crate::paging::fetch_all(
301            |cursor| async move { self.get(&Self::page_path(path, 200, &cursor)?).await },
302            next,
303            append,
304        )
305        .await
306    }
307
308    pub async fn venue(&self) -> Result<Venue, HttpError> {
309        self.get("/api/v1/venue").await
310    }
311
312    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, HttpError> {
313        let mut request = self.client.get(format!("{}{path}", self.base_url));
314        if let Some(etag) = self.cached(path, |cached| cached.etag.clone()) {
315            request = request.header(IF_NONE_MATCH, etag);
316        }
317        let response = request.send().await?;
318        let status = response.status();
319        if status == StatusCode::NOT_MODIFIED
320            && let Some(body) = self.cached(path, |cached| Arc::clone(&cached.body))
321        {
322            return Ok(serde_json::from_slice(&body)?);
323        }
324        if !status.is_success() {
325            let body = response.text().await?;
326            return Err(HttpError::Status {
327                status: status.as_u16(),
328                body,
329            });
330        }
331        let etag = response
332            .headers()
333            .get(ETAG)
334            .and_then(|value| value.to_str().ok())
335            .map(str::to_owned);
336        let body = response.bytes().await?;
337        let value = serde_json::from_slice(&body)?;
338        if let Some(etag) = etag {
339            let _ = self.cache().insert(
340                path.to_owned(),
341                Cached {
342                    etag,
343                    body: Arc::from(body.as_ref()),
344                },
345            );
346        }
347        Ok(value)
348    }
349
350    fn cached<R>(&self, path: &str, read: impl FnOnce(&Cached) -> R) -> Option<R> {
351        self.cache().get(path).map(read)
352    }
353
354    fn cache(&self) -> MutexGuard<'_, HashMap<String, Cached>> {
355        match self.cache.lock() {
356            Ok(cache) => cache,
357            Err(poisoned) => poisoned.into_inner(),
358        }
359    }
360}
361
362#[cfg(test)]
363mod tests;
acta_maker_sdk/http/mod.rs

acta_maker_sdk/http/
mod.rs
