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
13use types::{MarketsResponse, TokensResponse, VaultPositionsResponse, VaultsResponse};
14pub use types::{
15    ProjectionCursor, TokenProgramKind, TrackRecord, Vault, VaultPerformance, VaultPolicy,
16    VaultPosition, VaultPositionRole, VaultPositionStatus, VaultPositionType, Venue,
17};
18
19use crate::ws::types::{MarketInfo, TokenInfo};
20
21use crate::wire::{WireError, decode_base58_32};
22
23const TIMEOUT: Duration = Duration::from_secs(10);
24const POSITION_PAGE: u16 = 200;
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 address: {0}")]
34    Address(#[from] WireError),
35    #[error("invalid response body: {0}")]
36    Decode(#[from] serde_json::Error),
37}
38
39#[derive(Debug, Clone, Default, PartialEq, Eq)]
40pub struct MarketFilter {
41    pub underlying: Option<String>,
42    pub quote: Option<String>,
43    pub expiry: Option<u64>,
44    pub is_put: Option<bool>,
45    pub active: Option<bool>,
46}
47
48impl MarketFilter {
49    fn query(&self) -> String {
50        let mut params = Vec::new();
51        if let Some(underlying) = &self.underlying {
52            params.push(format!("underlying={underlying}"));
53        }
54        if let Some(quote) = &self.quote {
55            params.push(format!("quote={quote}"));
56        }
57        if let Some(expiry) = self.expiry {
58            params.push(format!("expiry={expiry}"));
59        }
60        if let Some(is_put) = self.is_put {
61            params.push(format!("is_put={is_put}"));
62        }
63        if let Some(active) = self.active {
64            params.push(format!("active={active}"));
65        }
66        if params.is_empty() {
67            String::new()
68        } else {
69            format!("?{}", params.join("&"))
70        }
71    }
72}
73
74#[derive(Debug, Clone)]
75pub struct HttpClient {
76    client: reqwest::Client,
77    base_url: String,
78    cache: Arc<Mutex<HashMap<String, Cached>>>,
79}
80
81#[derive(Debug)]
82struct Cached {
83    etag: String,
84    body: Arc<[u8]>,
85}
86
87impl HttpClient {
88    /// `base_url` is the API origin, e.g. `https://beta-api.acta.markets`.
89    pub fn new(base_url: impl Into<String>) -> Result<Self, HttpError> {
90        let client = reqwest::Client::builder().timeout(TIMEOUT).build()?;
91        Ok(Self::with_client(client, base_url))
92    }
93
94    #[must_use]
95    pub fn with_client(client: reqwest::Client, base_url: impl Into<String>) -> Self {
96        let mut base_url = base_url.into();
97        while base_url.ends_with('/') {
98            let _ = base_url.pop();
99        }
100        Self {
101            client,
102            base_url,
103            cache: Arc::default(),
104        }
105    }
106
107    #[must_use]
108    pub fn base_url(&self) -> &str {
109        &self.base_url
110    }
111
112    pub async fn vaults(&self) -> Result<Vec<Vault>, HttpError> {
113        let response: VaultsResponse = self.get("/api/v1/vaults").await?;
114        Ok(response.vaults)
115    }
116
117    /// Vaults whose primary delegate is `delegate` with an unexpired grant, the
118    /// same rule `/vault` authentication applies.
119    pub async fn delegated_vaults(&self, delegate: &str) -> Result<Vec<Vault>, HttpError> {
120        let _ = decode_base58_32(delegate)?;
121        let response: VaultsResponse = self
122            .get(&format!("/api/v1/vaults?delegate={delegate}"))
123            .await?;
124        Ok(response.vaults)
125    }
126
127    pub async fn vault(&self, pda: &str) -> Result<Vault, HttpError> {
128        let _ = decode_base58_32(pda)?;
129        self.get(&format!("/api/v1/vaults/{pda}")).await
130    }
131
132    /// Every unsettled position of the vault, following the server cursor.
133    pub async fn vault_positions(&self, pda: &str) -> Result<Vec<VaultPosition>, HttpError> {
134        let _ = decode_base58_32(pda)?;
135        let base = format!("/api/v1/vaults/{pda}/positions?settled=false&limit={POSITION_PAGE}");
136        let mut positions = Vec::new();
137        let mut path = base.clone();
138        loop {
139            let page: VaultPositionsResponse = self.get(&path).await?;
140            positions.extend(page.positions);
141            match page.next_cursor {
142                Some(cursor) => path = format!("{base}&cursor={cursor}"),
143                None => return Ok(positions),
144            }
145        }
146    }
147
148    pub async fn markets(&self, filter: &MarketFilter) -> Result<Vec<MarketInfo>, HttpError> {
149        let response: MarketsResponse = self
150            .get(&format!("/api/v1/markets{}", filter.query()))
151            .await?;
152        Ok(response.markets)
153    }
154
155    pub async fn tokens(&self) -> Result<BTreeMap<String, TokenInfo>, HttpError> {
156        let response: TokensResponse = self.get("/api/v1/tokens").await?;
157        Ok(response
158            .tokens
159            .into_iter()
160            .map(|token| (token.mint.clone(), token))
161            .collect())
162    }
163
164    pub async fn venue(&self) -> Result<Venue, HttpError> {
165        self.get("/api/v1/venue").await
166    }
167
168    async fn get<T: DeserializeOwned>(&self, path: &str) -> Result<T, HttpError> {
169        let mut request = self.client.get(format!("{}{path}", self.base_url));
170        if let Some(etag) = self.cached(path, |cached| cached.etag.clone()) {
171            request = request.header(IF_NONE_MATCH, etag);
172        }
173        let response = request.send().await?;
174        let status = response.status();
175        if status == StatusCode::NOT_MODIFIED
176            && let Some(body) = self.cached(path, |cached| Arc::clone(&cached.body))
177        {
178            return Ok(serde_json::from_slice(&body)?);
179        }
180        if !status.is_success() {
181            let body = response.text().await?;
182            return Err(HttpError::Status {
183                status: status.as_u16(),
184                body,
185            });
186        }
187        let etag = response
188            .headers()
189            .get(ETAG)
190            .and_then(|value| value.to_str().ok())
191            .map(str::to_owned);
192        let body = response.bytes().await?;
193        let value = serde_json::from_slice(&body)?;
194        if let Some(etag) = etag {
195            let _ = self.cache().insert(
196                path.to_owned(),
197                Cached {
198                    etag,
199                    body: Arc::from(body.as_ref()),
200                },
201            );
202        }
203        Ok(value)
204    }
205
206    fn cached<R>(&self, path: &str, read: impl FnOnce(&Cached) -> R) -> Option<R> {
207        self.cache().get(path).map(read)
208    }
209
210    fn cache(&self) -> MutexGuard<'_, HashMap<String, Cached>> {
211        match self.cache.lock() {
212            Ok(cache) => cache,
213            Err(poisoned) => poisoned.into_inner(),
214        }
215    }
216}
217
218#[cfg(test)]
219mod tests;
acta_maker_sdk/http/mod.rs

acta_maker_sdk/http/
mod.rs
