rust_ynab/ynab/
client.rs

1use crate::ynab::errors::{Error, ErrorResponse};
2use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
3use reqwest::RequestBuilder;
4use secrecy::ExposeSecret;
5use std::num::NonZeroU32;
6use std::sync::Arc;
7use std::time::Duration;
8
9/// Controls automatic retry of transient failures. See [`Client::with_retry`].
10#[derive(Debug, Clone, Copy)]
11pub struct RetryPolicy {
12    pub max_retries: u32,
13    pub base_delay: Duration,
14    pub max_delay: Duration,
15}
16
17#[derive(Debug)]
18/// Client is the YNAB API client. Use Client::new() to create one.
19pub struct Client {
20    pub(crate) base_url: reqwest::Url,
21    pub(crate) http_client: reqwest::Client,
22    pub(crate) limiter: Option<Arc<DefaultDirectRateLimiter>>,
23    #[allow(dead_code)]
24    api_key: secrecy::SecretBox<String>, // in case we need to use this later on
25    pub(crate) timeout: Option<Duration>,
26    pub(crate) retry_policy: Option<RetryPolicy>,
27}
28
29impl Client {
30    /// Creates a new client with the given Personal Access Token.
31    ///
32    /// # Examples
33    ///
34    /// ```no_run
35    /// use rust_ynab::Client;
36    ///
37    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
38    /// let client = Client::new(&std::env::var("YNAB_TOKEN")?)?;
39    /// # Ok(()) }
40    /// ```
41    pub fn new(api_key: impl Into<String>) -> Result<Self, Error> {
42        let api_key = secrecy::SecretBox::new(Box::new(api_key.into()));
43        let http_client = Self::build_http_client(api_key.expose_secret())?;
44        Ok(Self {
45            base_url: reqwest::Url::parse("https://api.ynab.com/v1")
46                .expect("hardcoded base URL is always valid"),
47            http_client,
48            limiter: None,
49            api_key,
50            timeout: None,
51            retry_policy: None,
52        })
53    }
54
55    fn build_http_client(api_key: &str) -> Result<reqwest::Client, Error> {
56        let mut headers = reqwest::header::HeaderMap::new();
57        headers.insert(
58            reqwest::header::AUTHORIZATION,
59            format!("Bearer {}", api_key)
60                .parse()
61                .expect("api key must be valid ASCII"),
62        );
63        let builder = reqwest::Client::builder().default_headers(headers);
64        builder.build().map_err(Into::into)
65    }
66
67    /// Sets the request timeout. Returns `self` for chaining.
68    ///
69    /// # Examples
70    ///
71    /// ```no_run
72    /// use rust_ynab::Client;
73    /// use std::time::Duration;
74    ///
75    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
76    /// let client = Client::new(&std::env::var("YNAB_TOKEN")?)?
77    ///     .with_timeout(Duration::from_secs(30))?;
78    /// # Ok(()) }
79    /// ```
80    pub fn with_timeout(mut self, timeout: Duration) -> Result<Self, Error> {
81        self.timeout = Some(timeout);
82        Ok(self)
83    }
84
85    /// Overrides the base URL. Primarily useful for testing.
86    pub fn with_base_url(mut self, base_url: impl AsRef<str>) -> Result<Self, Error> {
87        self.base_url = reqwest::Url::parse(base_url.as_ref())?;
88        Ok(self)
89    }
90
91    /// Configures a token bucket rate limiter on the client. Returns `self` for chaining.
92    ///
93    /// `requests_per_hour` is the total allowed requests per hour.
94    /// `burst_volume` optionally allows a number of requests to be made immediately
95    /// before throttling begins. The effective sustained rate becomes
96    /// `requests_per_hour - burst_volume` to account for burst consumption.
97    /// If `None`, no burst is allowed and the full rate is sustained evenly.
98    ///
99    /// # Examples
100    ///
101    /// ```no_run
102    /// use rust_ynab::Client;
103    /// use std::time::Duration;
104    ///
105    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
106    /// let client = Client::new(&std::env::var("YNAB_TOKEN")?)?
107    ///     .with_rate_limiter(200, Some(10))?  // 10 burst, then 190/hr
108    ///     .with_timeout(Duration::from_secs(30))?;
109    /// # Ok(()) }
110    /// ```
111    pub fn with_rate_limiter(
112        mut self,
113        requests_per_hour: usize,
114        burst_volume: Option<usize>,
115    ) -> Result<Self, Error> {
116        let requests = NonZeroU32::new(requests_per_hour as u32)
117            .ok_or_else(|| Error::InvalidRateLimit("requests_per_hour must be non-zero".into()))?;
118
119        let quota = match burst_volume {
120            None => Quota::per_hour(requests),
121            Some(burst) => {
122                let effective = (requests_per_hour as u32)
123                    .checked_sub(burst as u32)
124                    .ok_or_else(|| {
125                        Error::InvalidRateLimit(
126                            "requests_per_hour must be greater than burst_volume".into(),
127                        )
128                    })?;
129                let effective_rate = NonZeroU32::new(effective).ok_or_else(|| {
130                    Error::InvalidRateLimit(
131                        "requests_per_hour - burst_volume must be non-zero".into(),
132                    )
133                })?;
134                let burst = NonZeroU32::new(burst as u32).ok_or_else(|| {
135                    Error::InvalidRateLimit("burst_volume must be non-zero".into())
136                })?;
137                Quota::per_hour(effective_rate).allow_burst(burst)
138            }
139        };
140
141        self.limiter = Some(Arc::new(RateLimiter::direct(quota)));
142        Ok(self)
143    }
144
145    /// Enables automatic retry of transient failures (429, 503, and — for `GET` requests only —
146    /// connection-level errors). Returns `self` for chaining.
147    ///
148    /// A `GET` is always safe to retry. A write (`POST`/`PUT`/`PATCH`/`DELETE`) is only retried
149    /// when a response was actually received with status 429 or 503 — that means YNAB rejected
150    /// the request before processing it. A write is never retried on a connection error or
151    /// timeout, since it's impossible to tell whether the server already applied it before the
152    /// connection dropped; retrying blind there risks creating a duplicate transaction.
153    ///
154    /// If the response carries a `Retry-After` header, that delay is used as-is. Otherwise the
155    /// wait is `base_delay * 2^attempt`, capped at `max_delay`.
156    ///
157    /// # Examples
158    ///
159    /// ```no_run
160    /// use rust_ynab::Client;
161    /// use std::time::Duration;
162    ///
163    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
164    /// let client = Client::new(&std::env::var("YNAB_TOKEN")?)?
165    ///     .with_retry(3, Duration::from_millis(200), Duration::from_secs(10));
166    /// # Ok(()) }
167    /// ```
168    pub fn with_retry(
169        mut self,
170        max_retries: u32,
171        base_delay: Duration,
172        max_delay: Duration,
173    ) -> Self {
174        self.retry_policy = Some(RetryPolicy {
175            max_retries,
176            base_delay,
177            max_delay,
178        });
179        self
180    }
181
182    /// Returns the delay before the next retry attempt, or `None` if no retry policy is
183    /// configured or the retry budget for this request is exhausted.
184    fn next_retry_delay(&self, attempt: u32) -> Option<Duration> {
185        let policy = self.retry_policy?;
186        if attempt >= policy.max_retries {
187            return None;
188        }
189        let exp = policy
190            .base_delay
191            .saturating_mul(2u32.saturating_pow(attempt));
192        Some(exp.min(policy.max_delay))
193    }
194
195    pub(crate) fn make_url(&self, endpoint: &str) -> reqwest::Url {
196        let mut url = self.base_url.clone();
197        url.path_segments_mut()
198            .expect("base URL must be a valid base")
199            .extend(endpoint.split('/').filter(|s| !s.is_empty()));
200        url
201    }
202
203    // shared: rate-limit wait + URL build + method selection
204    async fn prepare(&self, method: reqwest::Method, endpoint: &str) -> RequestBuilder {
205        if let Some(limiter) = &self.limiter {
206            let start = std::time::Instant::now();
207            limiter.until_ready().await;
208            tracing::debug!(
209                waited_ms = start.elapsed().as_millis() as u64,
210                "rate limiter delayed request"
211            );
212        }
213        let url = self.make_url(endpoint);
214        self.http_client.request(method, url)
215    }
216
217    // shared: timeout + send + status check + error decode + body decode + retry
218    async fn send_json<T: serde::de::DeserializeOwned>(
219        &self,
220        method: reqwest::Method,
221        mut builder: RequestBuilder,
222    ) -> Result<T, Error> {
223        if let Some(t) = self.timeout {
224            builder = builder.timeout(t);
225        }
226        let mut attempt: u32 = 0;
227        loop {
228            let start = std::time::Instant::now();
229            // try_clone should never fail since we aren't building a stream
230            let req = builder
231                .try_clone()
232                .expect("body must be clonable for retry");
233            let res = match req.send().await {
234                Ok(res) => res,
235                Err(e) => {
236                    // a write that never got a response might already have landed server-side —
237                    // only a GET is safe to blindly retry on a connection-level failure.
238                    if method == reqwest::Method::GET
239                        && let Some(delay) = self.next_retry_delay(attempt)
240                    {
241                        tracing::debug!(
242                            attempt,
243                            delay_ms = delay.as_millis() as u64,
244                            error = %e,
245                            "retrying after connection error"
246                        );
247                        tokio::time::sleep(delay).await;
248                        attempt += 1;
249                        continue;
250                    }
251                    return Err(e.into());
252                }
253            };
254            let status = res.status();
255            let elapsed_ms = start.elapsed().as_millis() as u64;
256
257            if !status.is_success() {
258                let retry_after = res
259                    .headers()
260                    .get(reqwest::header::RETRY_AFTER)
261                    .and_then(|v| v.to_str().ok())
262                    .and_then(|s| s.parse::<u64>().ok())
263                    .map(Duration::from_secs);
264                let err_body: ErrorResponse = res.json().await?;
265                tracing::warn!(status = status.as_u16(), elapsed_ms, error = %err_body.error, "YNAB API error");
266
267                // 429/503 mean the request was rejected, not processed — safe to retry
268                // regardless of method, unlike a connection-level failure.
269                let retryable_status = matches!(
270                    status,
271                    reqwest::StatusCode::TOO_MANY_REQUESTS
272                        | reqwest::StatusCode::SERVICE_UNAVAILABLE
273                );
274                if retryable_status && let Some(backoff) = self.next_retry_delay(attempt) {
275                    let delay = retry_after.unwrap_or(backoff);
276                    tracing::debug!(
277                        attempt,
278                        delay_ms = delay.as_millis() as u64,
279                        status = status.as_u16(),
280                        "retrying request"
281                    );
282                    tokio::time::sleep(delay).await;
283                    attempt += 1;
284                    continue;
285                }
286                return Err(Error::new_api_error(status, err_body.error));
287            }
288            tracing::debug!(status = status.as_u16(), elapsed_ms, "request succeeded");
289            return res.json().await.map_err(Into::into);
290        }
291    }
292
293    #[tracing::instrument(skip(self, params), fields(endpoint = %endpoint))]
294    pub(crate) async fn get<T: serde::de::DeserializeOwned, Q: serde::ser::Serialize + ?Sized>(
295        &self,
296        endpoint: &str,
297        params: Option<&Q>,
298    ) -> Result<T, Error> {
299        let mut builder = self.prepare(reqwest::Method::GET, endpoint).await;
300        if let Some(p) = params {
301            builder = builder.query(p);
302        }
303        self.send_json(reqwest::Method::GET, builder).await
304    }
305
306    #[tracing::instrument(skip(self, body), fields(endpoint = %endpoint))]
307    pub(crate) async fn post<T: serde::de::DeserializeOwned, B: serde::ser::Serialize>(
308        &self,
309        endpoint: &str,
310        body: B,
311    ) -> Result<T, Error> {
312        let builder = self
313            .prepare(reqwest::Method::POST, endpoint)
314            .await
315            .json(&body);
316        self.send_json(reqwest::Method::POST, builder).await
317    }
318
319    #[tracing::instrument(skip(self, body), fields(endpoint = %endpoint))]
320    pub(crate) async fn patch<T: serde::de::DeserializeOwned, B: serde::ser::Serialize>(
321        &self,
322        endpoint: &str,
323        body: B,
324    ) -> Result<T, Error> {
325        let builder = self
326            .prepare(reqwest::Method::PATCH, endpoint)
327            .await
328            .json(&body);
329        self.send_json(reqwest::Method::PATCH, builder).await
330    }
331
332    #[tracing::instrument(skip(self, body), fields(endpoint = %endpoint))]
333    pub(crate) async fn put<T: serde::de::DeserializeOwned, B: serde::ser::Serialize>(
334        &self,
335        endpoint: &str,
336        body: B,
337    ) -> Result<T, Error> {
338        let builder = self
339            .prepare(reqwest::Method::PUT, endpoint)
340            .await
341            .json(&body);
342        self.send_json(reqwest::Method::PUT, builder).await
343    }
344
345    #[tracing::instrument(skip(self), fields(endpoint = %endpoint))]
346    pub(crate) async fn delete<T: serde::de::DeserializeOwned>(
347        &self,
348        endpoint: &str,
349    ) -> Result<T, Error> {
350        let builder = self.prepare(reqwest::Method::DELETE, endpoint).await;
351        self.send_json(reqwest::Method::DELETE, builder).await
352    }
353}
354
355#[cfg(test)]
356mod tests {
357    use super::*;
358    use crate::ynab::common::NO_PARAMS;
359    use crate::ynab::testutil::{error_body, new_test_client};
360    use serde::Deserialize;
361    use wiremock::matchers::{method, path};
362    use wiremock::{Mock, ResponseTemplate};
363
364    #[derive(Debug, Deserialize)]
365    struct Pong {
366        ok: bool,
367    }
368
369    fn too_many_requests_body() -> serde_json::Value {
370        error_body("429", "too_many_requests", "Too many requests")
371    }
372
373    #[tokio::test]
374    async fn retries_on_429_then_succeeds() {
375        let (client, server) = new_test_client().await;
376        let client = client.with_retry(3, Duration::from_millis(5), Duration::from_millis(50));
377
378        Mock::given(method("GET"))
379            .and(path("/plans"))
380            .respond_with(ResponseTemplate::new(429).set_body_json(too_many_requests_body()))
381            .up_to_n_times(2)
382            .expect(2)
383            .mount(&server)
384            .await;
385        Mock::given(method("GET"))
386            .and(path("/plans"))
387            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({"ok": true})))
388            .expect(1)
389            .mount(&server)
390            .await;
391
392        let result: Pong = client.get("plans", NO_PARAMS).await.unwrap();
393        assert!(result.ok);
394    }
395
396    #[tokio::test]
397    async fn stops_retrying_once_budget_exhausted() {
398        let (client, server) = new_test_client().await;
399        let client = client.with_retry(2, Duration::from_millis(5), Duration::from_millis(50));
400
401        Mock::given(method("GET"))
402            .and(path("/plans"))
403            .respond_with(ResponseTemplate::new(429).set_body_json(too_many_requests_body()))
404            // initial attempt + 2 retries = 3 total requests
405            .expect(3)
406            .mount(&server)
407            .await;
408
409        let result: Result<Pong, Error> = client.get("plans", NO_PARAMS).await;
410        assert!(matches!(result, Err(Error::RateLimited(_))));
411    }
412
413    #[tokio::test]
414    async fn no_retry_without_policy_configured() {
415        let (client, server) = new_test_client().await;
416
417        Mock::given(method("GET"))
418            .and(path("/plans"))
419            .respond_with(ResponseTemplate::new(429).set_body_json(too_many_requests_body()))
420            .expect(1)
421            .mount(&server)
422            .await;
423
424        let result: Result<Pong, Error> = client.get("plans", NO_PARAMS).await;
425        assert!(matches!(result, Err(Error::RateLimited(_))));
426    }
427}
rust_ynab/ynab/client.rs

rust_ynab/ynab/
client.rs
