truefix_ig_client/
client.rs

1use reqwest::{Method, RequestBuilder, StatusCode};
2use serde::{Serialize, de::DeserializeOwned};
3use std::sync::RwLock;
4use std::time::{Duration, Instant};
5use url::Url;
6
7use crate::{
8    config::{AuthenticationVersion, ClientConfig, Credentials},
9    error::{IgError, IgResult},
10    types::{
11        AccountsResponse, CreatePositionRequest, DealReferenceResponse, HistoricalPricesQuery,
12        HistoricalPricesResponse, LoginResponse, MarketDetails, MarketsResponse, OAuthToken,
13        PositionsResponse, V3LoginResponse,
14    },
15};
16
17#[derive(Debug, Clone)]
18struct SessionTokens {
19    cst: String,
20    x_security_token: String,
21}
22
23#[derive(Debug, Clone)]
24struct OAuthSession {
25    access_token: String,
26    refresh_token: String,
27    account_id: String,
28    expires_at: Instant,
29}
30
31#[derive(Debug, Clone)]
32enum Session {
33    V2(SessionTokens),
34    V3(OAuthSession),
35}
36
37const TOKEN_REFRESH_THRESHOLD: Duration = Duration::from_secs(10);
38
39/// IG REST trading client.
40///
41/// Session credentials are kept behind a lock so that one client can be shared among
42/// concurrent read-only tasks after [`Self::login`] succeeds. Calls which change server state are
43/// deliberately made once and are never automatically replayed.
44#[derive(Debug)]
45pub struct IgClient {
46    http: reqwest::Client,
47    config: ClientConfig,
48    base_url: Url,
49    session: RwLock<Option<Session>>,
50}
51
52impl IgClient {
53    /// Constructs a client without contacting IG.
54    pub fn new(config: ClientConfig) -> IgResult<Self> {
55        let base_url = Url::parse(config.environment.rest_base()).map_err(|error| {
56            IgError::InvalidConfiguration(format!("invalid REST base URL: {error}"))
57        })?;
58        let mut http = reqwest::Client::builder().timeout(config.timeout);
59        if let Some(proxy) = &config.proxy {
60            http = http.proxy(reqwest::Proxy::all(proxy).map_err(|error| {
61                IgError::InvalidConfiguration(format!("invalid proxy URL: {error}"))
62            })?);
63        }
64        Ok(Self {
65            http: http.build()?,
66            config,
67            base_url,
68            session: RwLock::new(None),
69        })
70    }
71
72    /// Authenticates using the protocol selected in [`ClientConfig::authentication`].
73    pub async fn login(&self) -> IgResult<LoginResponse> {
74        match &self.config.authentication {
75            AuthenticationVersion::V2 => self.login_v2().await,
76            AuthenticationVersion::V3 { account_id } => self.login_v3(account_id).await,
77        }
78    }
79
80    /// Authenticates with v2 and stores the CST/X-SECURITY-TOKEN pair.
81    pub async fn login_v2(&self) -> IgResult<LoginResponse> {
82        #[derive(Serialize)]
83        struct LoginRequest<'a> {
84            identifier: &'a str,
85            password: &'a str,
86        }
87
88        let credentials = self.credentials()?;
89        let response = self
90            .request(Method::POST, "session", 2, false)?
91            .json(&LoginRequest {
92                identifier: credentials.identifier(),
93                password: credentials.password(),
94            })
95            .send()
96            .await?;
97        let (tokens, login) = self.decode_login(response).await?;
98        *self.session.write().map_err(|_| IgError::MissingSession)? = Some(Session::V2(tokens));
99        Ok(login)
100    }
101
102    /// Authenticates with v3 OAuth and stores refreshable bearer credentials.
103    pub async fn login_v3(&self, account_id: &str) -> IgResult<LoginResponse> {
104        #[derive(Serialize)]
105        #[serde(rename_all = "camelCase")]
106        struct LoginRequest<'a> {
107            identifier: &'a str,
108            password: &'a str,
109            account_id: &'a str,
110        }
111
112        if account_id.is_empty() {
113            return Err(IgError::InvalidConfiguration(
114                "v3 authentication requires a non-empty account ID".to_owned(),
115            ));
116        }
117        let credentials = self.credentials()?;
118        let response = self
119            .request(Method::POST, "session", 3, false)?
120            .header("IG-ACCOUNT-ID", account_id)
121            .json(&LoginRequest {
122                identifier: credentials.identifier(),
123                password: credentials.password(),
124                account_id,
125            })
126            .send()
127            .await?;
128        let response = Self::ensure_success(response).await?;
129        let login: V3LoginResponse = response.json().await?;
130        let oauth = login.oauth_token.ok_or_else(|| {
131            IgError::InvalidConfiguration(
132                "IG v3 login response did not contain an OAuth token".to_owned(),
133            )
134        })?;
135        let selected_account = login
136            .account_id
137            .clone()
138            .unwrap_or_else(|| account_id.to_owned());
139        let result = LoginResponse {
140            current_account_id: login.current_account_id.or(login.account_id),
141            lightstreamer_endpoint: login.lightstreamer_endpoint,
142            client_id: login.client_id,
143            currency_iso_code: login.currency_iso_code,
144            dealing_enabled: login.dealing_enabled,
145        };
146        *self.session.write().map_err(|_| IgError::MissingSession)? =
147            Some(Session::V3(oauth_session(oauth, selected_account)));
148        Ok(result)
149    }
150
151    /// Deletes the current session and clears local tokens only after IG accepts the request.
152    pub async fn logout(&self) -> IgResult<()> {
153        self.refresh_oauth_if_needed().await?;
154        let response = self
155            .request(Method::DELETE, "session", 1, true)?
156            .send()
157            .await?;
158        Self::ensure_success(response).await?;
159        *self.session.write().map_err(|_| IgError::MissingSession)? = None;
160        Ok(())
161    }
162
163    /// Returns the accounts available to the authenticated client.
164    pub async fn accounts(&self) -> IgResult<AccountsResponse> {
165        self.get("accounts", 1).await
166    }
167
168    /// Returns all open positions for the active account.
169    pub async fn positions(&self) -> IgResult<PositionsResponse> {
170        self.get("positions", 2).await
171    }
172
173    /// Returns market metadata and a latest price snapshot for one IG epic.
174    pub async fn market(&self, epic: &str) -> IgResult<MarketDetails> {
175        self.get(&format!("markets/{}", encode_path_segment(epic)), 3)
176            .await
177    }
178
179    /// Searches the market catalogue using IG's `/markets` endpoint.
180    pub async fn search_markets(&self, query: &str) -> IgResult<MarketsResponse> {
181        self.refresh_oauth_if_needed().await?;
182        let mut url = self.url("markets")?;
183        url.query_pairs_mut().append_pair("searchTerm", query);
184        self.send_json(self.request_url(Method::GET, url, 1, true)?)
185            .await
186    }
187
188    /// Returns historical prices for an epic at the requested IG resolution.
189    pub async fn historical_prices(
190        &self,
191        epic: &str,
192        query: HistoricalPricesQuery<'_>,
193    ) -> IgResult<HistoricalPricesResponse> {
194        self.refresh_oauth_if_needed().await?;
195        let mut url = self.url(&format!("prices/{}", encode_path_segment(epic)))?;
196        let mut pairs = url.query_pairs_mut();
197        pairs.append_pair("resolution", query.resolution);
198        if let Some(value) = query.from {
199            pairs.append_pair("from", value);
200        }
201        if let Some(value) = query.to {
202            pairs.append_pair("to", value);
203        }
204        if let Some(value) = query.max {
205            pairs.append_pair("max", &value.to_string());
206        }
207        drop(pairs);
208        self.send_json(self.request_url(Method::GET, url, 3, true)?)
209            .await
210    }
211
212    /// Creates a position. This operation is never automatically retried.
213    pub async fn create_position(
214        &self,
215        request: &CreatePositionRequest,
216    ) -> IgResult<DealReferenceResponse> {
217        self.refresh_oauth_if_needed().await?;
218        let response = self
219            .request(Method::POST, "positions/otc", 2, true)?
220            .json(request)
221            .send()
222            .await?;
223        Self::decode_json(response).await
224    }
225
226    fn credentials(&self) -> IgResult<&Credentials> {
227        self.config
228            .credentials
229            .as_ref()
230            .ok_or(IgError::MissingCredentials)
231    }
232
233    fn url(&self, path: &str) -> IgResult<Url> {
234        let mut url = self.base_url.clone();
235        url.set_path(&format!(
236            "{}/{}",
237            self.base_url.path().trim_end_matches('/'),
238            path.trim_start_matches('/')
239        ));
240        url.set_query(None);
241        Ok(url)
242    }
243
244    fn request(
245        &self,
246        method: Method,
247        path: &str,
248        version: u8,
249        include_session: bool,
250    ) -> IgResult<RequestBuilder> {
251        self.request_url(method, self.url(path)?, version, include_session)
252    }
253
254    fn request_url(
255        &self,
256        method: Method,
257        url: Url,
258        version: u8,
259        include_session: bool,
260    ) -> IgResult<RequestBuilder> {
261        let credentials = self.credentials()?;
262        let mut request = self
263            .http
264            .request(method, url)
265            .header("X-IG-API-KEY", credentials.api_key())
266            .header("Accept", "application/json")
267            .header("Content-Type", "application/json")
268            .header("Version", version.to_string());
269        if include_session {
270            let session = self.session.read().map_err(|_| IgError::MissingSession)?;
271            let session = session.as_ref().ok_or(IgError::MissingSession)?;
272            request = match session {
273                Session::V2(tokens) => request
274                    .header("CST", &tokens.cst)
275                    .header("X-SECURITY-TOKEN", &tokens.x_security_token),
276                Session::V3(oauth) => request
277                    .header("Authorization", format!("Bearer {}", oauth.access_token))
278                    .header("IG-ACCOUNT-ID", &oauth.account_id),
279            };
280        }
281        Ok(request)
282    }
283
284    async fn get<T: DeserializeOwned>(&self, path: &str, version: u8) -> IgResult<T> {
285        self.refresh_oauth_if_needed().await?;
286        self.send_json(self.request(Method::GET, path, version, true)?)
287            .await
288    }
289
290    async fn send_json<T: DeserializeOwned>(&self, request: RequestBuilder) -> IgResult<T> {
291        Self::decode_json(request.send().await?).await
292    }
293
294    async fn decode_login(
295        &self,
296        response: reqwest::Response,
297    ) -> IgResult<(SessionTokens, LoginResponse)> {
298        let response = Self::ensure_success(response).await?;
299        let cst = response
300            .headers()
301            .get("CST")
302            .and_then(|value| value.to_str().ok())
303            .ok_or(IgError::MissingToken("CST"))?
304            .to_owned();
305        let x_security_token = response
306            .headers()
307            .get("X-SECURITY-TOKEN")
308            .and_then(|value| value.to_str().ok())
309            .ok_or(IgError::MissingToken("X-SECURITY-TOKEN"))?
310            .to_owned();
311        Ok((
312            SessionTokens {
313                cst,
314                x_security_token,
315            },
316            response.json().await?,
317        ))
318    }
319
320    async fn refresh_oauth_if_needed(&self) -> IgResult<()> {
321        let refresh_token = {
322            let session = self.session.read().map_err(|_| IgError::MissingSession)?;
323            match session.as_ref() {
324                Some(Session::V3(oauth))
325                    if oauth.expires_at.saturating_duration_since(Instant::now())
326                        <= TOKEN_REFRESH_THRESHOLD =>
327                {
328                    Some(oauth.refresh_token.clone())
329                }
330                _ => None,
331            }
332        };
333        let Some(refresh_token) = refresh_token else {
334            return Ok(());
335        };
336
337        #[derive(Serialize)]
338        #[serde(rename_all = "camelCase")]
339        struct RefreshTokenRequest<'a> {
340            refresh_token: &'a str,
341        }
342
343        let response = self
344            .request(Method::POST, "session/refresh-token", 1, false)?
345            .json(&RefreshTokenRequest {
346                refresh_token: &refresh_token,
347            })
348            .send()
349            .await?;
350        let oauth: OAuthToken = Self::decode_json(response).await?;
351        let mut session = self.session.write().map_err(|_| IgError::MissingSession)?;
352        let Some(Session::V3(current)) = session.as_mut() else {
353            return Ok(());
354        };
355        current.access_token = oauth.access_token;
356        current.refresh_token = oauth.refresh_token;
357        current.expires_at = expiry_from(oauth.expires_in);
358        Ok(())
359    }
360
361    async fn decode_json<T: DeserializeOwned>(response: reqwest::Response) -> IgResult<T> {
362        Self::ensure_success(response)
363            .await?
364            .json()
365            .await
366            .map_err(IgError::from)
367    }
368
369    async fn ensure_success(response: reqwest::Response) -> IgResult<reqwest::Response> {
370        if response.status().is_success() {
371            return Ok(response);
372        }
373        let status = response.status();
374        let body = response.text().await.unwrap_or_default();
375        Err(exchange_error(status, &body))
376    }
377}
378
379fn exchange_error(status: StatusCode, body: &str) -> IgError {
380    #[derive(serde::Deserialize)]
381    #[serde(rename_all = "camelCase")]
382    struct ErrorBody {
383        error_code: Option<String>,
384        message: Option<String>,
385    }
386    let parsed = serde_json::from_str::<ErrorBody>(body).ok();
387    let code = parsed
388        .as_ref()
389        .and_then(|value| value.error_code.clone())
390        .unwrap_or_else(|| "unknown".to_owned());
391    let message = parsed
392        .and_then(|value| value.message)
393        .unwrap_or_else(|| body.to_owned());
394    IgError::Exchange {
395        status: status.as_u16(),
396        code,
397        message,
398    }
399}
400
401fn oauth_session(token: OAuthToken, account_id: String) -> OAuthSession {
402    OAuthSession {
403        access_token: token.access_token,
404        refresh_token: token.refresh_token,
405        account_id,
406        expires_at: expiry_from(token.expires_in),
407    }
408}
409
410fn expiry_from(expires_in: u64) -> Instant {
411    Instant::now() + Duration::from_secs(expires_in.max(1))
412}
413
414fn encode_path_segment(value: &str) -> String {
415    let mut encoded = String::with_capacity(value.len());
416    for byte in value.bytes() {
417        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
418            encoded.push(char::from(byte));
419        } else {
420            encoded.push_str(&format!("%{byte:02X}"));
421        }
422    }
423    encoded
424}
425
426#[cfg(test)]
427mod tests {
428    use super::*;
429    use crate::config::Environment;
430
431    #[test]
432    fn custom_base_preserves_gateway_path() {
433        let config = ClientConfig {
434            environment: Environment::Custom {
435                rest_base: "http://127.0.0.1:8080/gateway/deal".to_owned(),
436            },
437            ..ClientConfig::default()
438        };
439        let client = IgClient::new(config).unwrap();
440        assert_eq!(
441            client.url("session").unwrap().as_str(),
442            "http://127.0.0.1:8080/gateway/deal/session"
443        );
444    }
445
446    #[test]
447    fn epics_are_encoded_as_one_path_segment() {
448        assert_eq!(encode_path_segment("CS.D/EUR USD"), "CS.D%2FEUR%20USD");
449    }
450
451    #[test]
452    fn encoded_epic_is_not_encoded_twice_in_a_url() {
453        let config = ClientConfig {
454            environment: Environment::Custom {
455                rest_base: "http://127.0.0.1:8080/gateway/deal".to_owned(),
456            },
457            ..ClientConfig::default()
458        };
459        let client = IgClient::new(config).unwrap();
460        assert_eq!(
461            client.url("markets/CS.D%2FEUR").unwrap().path(),
462            "/gateway/deal/markets/CS.D%2FEUR"
463        );
464    }
465}
truefix_ig_client/client.rs

truefix_ig_client/
client.rs
