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

acta_maker_sdk/http/
mod.rs
