simulator_client/
client.rs

1use std::time::Duration;
2
3use bon::Builder;
4use chrono::{DateTime, Utc};
5use simulator_api::{AvailableRange, BacktestResponse, usage::UsageReport};
6use tokio_tungstenite::{
7    connect_async,
8    tungstenite::{client::IntoClientRequest, http::HeaderValue},
9};
10
11use crate::{
12    BacktestClientError, BacktestClientResult, BacktestSession, CreateSession,
13    session::CreateRequestResult,
14    urls::{backtest_ws_url, http_base_from_ws_url},
15};
16
17const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
18
19/// Backtest WebSocket client configured with a base URL and API key.
20///
21/// Build with [`BacktestClient::builder`], which supports optional timeouts and
22/// raw message logging.
23#[derive(Debug, Clone, Builder)]
24#[builder(on(String, into))]
25pub struct BacktestClient {
26    /// WebSocket endpoint. Accepts a bare hostname (normalized to
27    /// `wss://{host}/backtest`) or an explicit `ws(s)://` URL, whose
28    /// `/backtest` path is appended only if missing.
29    #[builder(with = |url: impl Into<String>| backtest_ws_url(&url.into()))]
30    url: String,
31    /// API key injected as the `X-API-Key` header.
32    #[builder(default)]
33    api_key: String,
34
35    /// Timeout for the initial connect handshake.
36    #[builder(default = DEFAULT_CONNECT_TIMEOUT)]
37    connect_timeout: Duration,
38
39    /// Default timeout for request/response operations.
40    request_timeout: Option<Duration>,
41
42    /// Log raw inbound responses at debug level.
43    #[builder(default)]
44    log_raw: bool,
45}
46
47impl BacktestClient {
48    async fn connect(&self) -> BacktestClientResult<BacktestSession> {
49        let mut request = self.url.clone().into_client_request().map_err(|source| {
50            BacktestClientError::BuildRequest {
51                url: self.url.clone(),
52                source: Box::new(source),
53            }
54        })?;
55
56        request
57            .headers_mut()
58            .insert("X-API-Key", HeaderValue::from_str(&self.api_key)?);
59
60        let (stream, _) = tokio::time::timeout(self.connect_timeout, connect_async(request))
61            .await
62            .map_err(|_| BacktestClientError::Timeout {
63                action: "connecting",
64                duration: self.connect_timeout,
65            })?
66            .map_err(|source| BacktestClientError::Connect {
67                url: self.url.clone(),
68                source: Box::new(source),
69            })?;
70        Ok(BacktestSession::new(
71            stream,
72            self.request_timeout,
73            self.log_raw,
74        ))
75    }
76
77    /// Create a backtest session by connecting and sending a `CreateBacktestSession` request.
78    pub async fn create_session(
79        &self,
80        create: CreateSession,
81    ) -> BacktestClientResult<BacktestSession> {
82        let request = create.into_request()?;
83        let rpc_base_url = http_base_from_ws_url(&self.url);
84        let mut session = self.connect().await?;
85        match session
86            .create_with_request(request, rpc_base_url, None)
87            .await?
88        {
89            CreateRequestResult::Single { .. } => {}
90            CreateRequestResult::Parallel { session_ids, .. } => {
91                return Err(BacktestClientError::UnexpectedResponse {
92                    context: "creating single session",
93                    response: Box::new(BacktestResponse::SessionsCreated { session_ids }),
94                });
95            }
96        }
97        Ok(session)
98    }
99
100    /// Create one or many sessions and return the resulting session IDs.
101    ///
102    /// When `CreateSession.parallel` is true, this may return many IDs.
103    pub async fn create_sessions(
104        &self,
105        create: CreateSession,
106    ) -> BacktestClientResult<Vec<String>> {
107        self.create_sessions_with_progress(create, |_| {}).await
108    }
109
110    /// Create one or many sessions and stream each successfully created session ID.
111    ///
112    /// `on_session_created` is called for each streamed `SessionCreated` response.
113    pub async fn create_sessions_with_progress(
114        &self,
115        create: CreateSession,
116        mut on_session_created: impl FnMut(String) + Send,
117    ) -> BacktestClientResult<Vec<String>> {
118        let request = create.into_request()?;
119        let rpc_base_url = http_base_from_ws_url(&self.url);
120        let mut session = self.connect().await?;
121        match session
122            .create_with_request(request, rpc_base_url, Some(&mut on_session_created))
123            .await?
124        {
125            CreateRequestResult::Single { session_id, .. } => Ok(vec![session_id]),
126            CreateRequestResult::Parallel { session_ids, .. } => Ok(session_ids),
127        }
128    }
129
130    /// Like [`Self::create_sessions`], but additionally returns the opaque
131    /// per-session `task_id` reported by the server (if any).
132    pub async fn create_sessions_with_task_ids(
133        &self,
134        create: CreateSession,
135    ) -> BacktestClientResult<Vec<(String, Option<String>)>> {
136        let request = create.into_request()?;
137        let rpc_base_url = http_base_from_ws_url(&self.url);
138        let mut session = self.connect().await?;
139        match session
140            .create_with_request(request, rpc_base_url, None)
141            .await?
142        {
143            CreateRequestResult::Single {
144                session_id,
145                task_id,
146            } => Ok(vec![(session_id, task_id)]),
147            CreateRequestResult::Parallel {
148                session_ids,
149                task_ids,
150            } => Ok(session_ids.into_iter().zip(task_ids).collect()),
151        }
152    }
153
154    /// Fetch the available slot ranges from the server's `/available-ranges` endpoint.
155    pub async fn available_ranges(&self) -> BacktestClientResult<Vec<AvailableRange>> {
156        let base = http_base_from_ws_url(&self.url);
157        let url = format!("{base}/available-ranges");
158        // Forwarded (when set) so the endpoint adds this key's own ranges,
159        // keeping the client's split in sync with the server's. Omitted when
160        // unset so a keyless client gets global ranges, not an empty-key 401.
161        let mut request = reqwest::Client::new().get(&url);
162        if !self.api_key.is_empty() {
163            request = request.header("X-API-Key", &self.api_key);
164        }
165        let ranges = request
166            .send()
167            .await
168            .map_err(|e| BacktestClientError::Http {
169                url: url.clone(),
170                source: Box::new(e),
171            })?
172            .error_for_status()
173            .map_err(|e| BacktestClientError::Http {
174                url: url.clone(),
175                source: Box::new(e),
176            })?
177            .json::<Vec<AvailableRange>>()
178            .await
179            .map_err(|e| BacktestClientError::Http {
180                url: url.clone(),
181                source: Box::new(e),
182            })?;
183        Ok(ranges)
184    }
185
186    /// Fetch per-API-key usage metrics from the server's `/usage` endpoint.
187    ///
188    /// `since` and `until` narrow the reporting window; both default to the
189    /// server's configured window when omitted.
190    pub async fn usage(
191        &self,
192        since: Option<DateTime<Utc>>,
193        until: Option<DateTime<Utc>>,
194    ) -> BacktestClientResult<UsageReport> {
195        let base = http_base_from_ws_url(&self.url);
196        let url = format!("{base}/usage");
197        let mut req = reqwest::Client::new()
198            .get(&url)
199            .header("X-API-Key", &self.api_key);
200        if let Some(t) = since {
201            req = req.query(&[(
202                "since",
203                t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
204            )]);
205        }
206        if let Some(t) = until {
207            req = req.query(&[(
208                "until",
209                t.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
210            )]);
211        }
212        let resp = req.send().await.map_err(|e| BacktestClientError::Http {
213            url: url.clone(),
214            source: Box::new(e),
215        })?;
216        let status = resp.status();
217        if !status.is_success() {
218            let body = resp.text().await.unwrap_or_default();
219            return Err(BacktestClientError::HttpStatus { url, status, body });
220        }
221        resp.json::<UsageReport>()
222            .await
223            .map_err(|e| BacktestClientError::Http {
224                url: url.clone(),
225                source: Box::new(e),
226            })
227    }
228
229    /// Attach to an existing backtest session over the control websocket.
230    pub async fn attach_session(
231        &self,
232        session_id: impl Into<String>,
233        last_sequence: Option<u64>,
234    ) -> BacktestClientResult<BacktestSession> {
235        let rpc_base_url = http_base_from_ws_url(&self.url);
236        let mut session = self.connect().await?;
237        session
238            .attach(session_id.into(), last_sequence, rpc_base_url)
239            .await?;
240        Ok(session)
241    }
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247
248    #[test]
249    fn builder_normalizes_bare_hosts_and_keeps_full_urls() {
250        let from_host = BacktestClient::builder()
251            .url("simulator.termina.technology")
252            .build();
253        assert_eq!(from_host.url, "wss://simulator.termina.technology/backtest");
254
255        let from_full_url = BacktestClient::builder()
256            .url("ws://localhost:8900/backtest".to_string())
257            .build();
258        assert_eq!(from_full_url.url, "ws://localhost:8900/backtest");
259    }
260}
simulator_client/client.rs

simulator_client/
client.rs
