asana_cli/api/
client.rs

1//! Core asynchronous HTTP client for interacting with Asana's REST API.
2
3use crate::api::{
4    auth::AuthToken,
5    error::{ApiError, RateLimitInfo},
6    pagination::ListResponse,
7};
8use async_stream::try_stream;
9use base64::{Engine as _, engine::general_purpose};
10use directories::ProjectDirs;
11use futures_core::Stream;
12use reqwest::{
13    Method, StatusCode,
14    header::{ACCEPT, AUTHORIZATION, HeaderMap, HeaderValue, RETRY_AFTER, USER_AGENT},
15};
16use serde::de::DeserializeOwned;
17use serde::{Deserialize, Serialize};
18use serde_json::Value;
19use sha2::{Digest, Sha256};
20use std::{
21    collections::HashMap,
22    path::PathBuf,
23    sync::{
24        Arc,
25        atomic::{AtomicBool, Ordering},
26    },
27    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
28};
29use tokio::{fs, sync::RwLock, time::sleep};
30use tracing::{debug, warn};
31
32const VERSION: &str = match option_env!("CARGO_PKG_VERSION") {
33    Some(version) => version,
34    None => "unknown",
35};
36
37/// In-memory cache entry.
38#[derive(Clone)]
39struct CacheEntry {
40    expires_at: Instant,
41    value: Arc<Vec<u8>>,
42}
43
44/// On-disk cache entry representation.
45#[derive(Debug, Serialize, Deserialize)]
46struct DiskCacheEntry {
47    expires_at: u64,
48    body: String,
49}
50
51/// Configurable options for the API client.
52#[derive(Debug, Clone)]
53pub struct ApiClientOptions {
54    /// Base URL for the Asana API.
55    pub base_url: String,
56    /// User agent string sent with every request.
57    pub user_agent: String,
58    /// Total request timeout applied to HTTP calls.
59    pub timeout: Duration,
60    /// Maximum number of retry attempts for transient failures.
61    pub max_retries: usize,
62    /// Initial backoff delay applied between retries.
63    pub retry_base_delay: Duration,
64    /// Time-to-live for cached responses.
65    pub cache_ttl: Duration,
66    /// Directory used to persist cached responses across runs.
67    pub cache_dir: PathBuf,
68    /// Whether the client should avoid network calls and use cached data only.
69    pub offline: bool,
70}
71
72impl ApiClientOptions {
73    /// Construct options with a specific base URL.
74    #[must_use]
75    pub fn new(base_url: impl Into<String>) -> Self {
76        Self {
77            base_url: base_url.into(),
78            ..Self::default()
79        }
80    }
81
82    /// Override the cache directory used for disk persistence.
83    #[must_use]
84    pub fn with_cache_dir(mut self, cache_dir: PathBuf) -> Self {
85        self.cache_dir = cache_dir;
86        self
87    }
88
89    /// Override the request timeout.
90    #[must_use]
91    pub const fn with_timeout(mut self, timeout: Duration) -> Self {
92        self.timeout = timeout;
93        self
94    }
95
96    /// Override retry attempts.
97    #[must_use]
98    pub const fn with_max_retries(mut self, retries: usize) -> Self {
99        self.max_retries = retries;
100        self
101    }
102
103    /// Override retry backoff base delay.
104    #[must_use]
105    pub const fn with_retry_base_delay(mut self, delay: Duration) -> Self {
106        self.retry_base_delay = delay;
107        self
108    }
109
110    /// Override cache TTL.
111    #[must_use]
112    pub const fn with_cache_ttl(mut self, ttl: Duration) -> Self {
113        self.cache_ttl = ttl;
114        self
115    }
116
117    /// Start the client in offline mode.
118    #[must_use]
119    pub const fn with_offline(mut self, offline: bool) -> Self {
120        self.offline = offline;
121        self
122    }
123}
124
125impl Default for ApiClientOptions {
126    fn default() -> Self {
127        let base_url = crate::config::DEFAULT_API_BASE_URL.to_string();
128        let user_agent = format!("asana-cli/{VERSION}");
129        let cache_dir = default_cache_dir();
130        Self {
131            base_url,
132            user_agent,
133            timeout: Duration::from_secs(30),
134            max_retries: 3,
135            retry_base_delay: Duration::from_millis(500),
136            // 5 minutes. `Duration::from_mins` is still unstable on
137            // our pinned toolchain (1.95.0); use seconds.
138            cache_ttl: Duration::from_secs(5 * 60),
139            cache_dir,
140            offline: false,
141        }
142    }
143}
144
145fn default_cache_dir() -> PathBuf {
146    ProjectDirs::from("com", "asana", "asana-cli").map_or_else(
147        || {
148            let mut path = std::env::temp_dir();
149            path.push("asana-cli-cache");
150            path
151        },
152        |dirs| dirs.data_local_dir().join("cache"),
153    )
154}
155
156/// Builder for [`ApiClient`].
157pub struct ApiClientBuilder {
158    token: AuthToken,
159    options: ApiClientOptions,
160}
161
162impl ApiClientBuilder {
163    /// Create a new builder.
164    #[must_use]
165    pub fn new(token: AuthToken) -> Self {
166        Self {
167            token,
168            options: ApiClientOptions::default(),
169        }
170    }
171
172    /// Set the base URL.
173    #[must_use]
174    pub fn base_url(mut self, base_url: impl Into<String>) -> Self {
175        self.options.base_url = base_url.into();
176        self
177    }
178
179    /// Override the user agent.
180    #[must_use]
181    pub fn user_agent(mut self, user_agent: impl Into<String>) -> Self {
182        self.options.user_agent = user_agent.into();
183        self
184    }
185
186    /// Override the cache directory.
187    #[must_use]
188    pub fn cache_dir(mut self, cache_dir: PathBuf) -> Self {
189        self.options.cache_dir = cache_dir;
190        self
191    }
192
193    /// Override timeout.
194    #[must_use]
195    pub const fn timeout(mut self, timeout: Duration) -> Self {
196        self.options.timeout = timeout;
197        self
198    }
199
200    /// Override retry attempts.
201    #[must_use]
202    pub const fn max_retries(mut self, retries: usize) -> Self {
203        self.options.max_retries = retries;
204        self
205    }
206
207    /// Override retry base delay.
208    #[must_use]
209    pub const fn retry_base_delay(mut self, delay: Duration) -> Self {
210        self.options.retry_base_delay = delay;
211        self
212    }
213
214    /// Override cache TTL.
215    #[must_use]
216    pub const fn cache_ttl(mut self, ttl: Duration) -> Self {
217        self.options.cache_ttl = ttl;
218        self
219    }
220
221    /// Configure offline mode.
222    #[must_use]
223    pub const fn offline(mut self, offline: bool) -> Self {
224        self.options.offline = offline;
225        self
226    }
227
228    /// Finalise the builder, creating an [`ApiClient`].
229    ///
230    /// # Errors
231    ///
232    /// Returns an error if the cache directory cannot be created or if the HTTP client fails to initialize.
233    pub fn build(self) -> Result<ApiClient, ApiError> {
234        ApiClient::with_options(self.token, self.options)
235    }
236}
237
238/// Asynchronous Asana API client handling retries, rate limiting, and caching.
239pub struct ApiClient {
240    http: reqwest::Client,
241    token: AuthToken,
242    options: ApiClientOptions,
243    memory_cache: Arc<RwLock<HashMap<String, CacheEntry>>>,
244    offline: AtomicBool,
245    rate_limit: Arc<RwLock<Option<RateLimitInfo>>>,
246}
247
248impl Clone for ApiClient {
249    fn clone(&self) -> Self {
250        Self {
251            http: self.http.clone(),
252            token: self.token.clone(),
253            options: self.options.clone(),
254            memory_cache: Arc::clone(&self.memory_cache),
255            offline: AtomicBool::new(self.offline.load(Ordering::Relaxed)),
256            rate_limit: Arc::clone(&self.rate_limit),
257        }
258    }
259}
260
261impl ApiClient {
262    /// Create a builder for configuring the client.
263    #[must_use]
264    pub fn builder(token: AuthToken) -> ApiClientBuilder {
265        ApiClientBuilder::new(token)
266    }
267
268    /// Construct a client with default options.
269    ///
270    /// # Errors
271    ///
272    /// Returns an error if the cache directory cannot be created or if the HTTP client fails to initialize.
273    pub fn new(token: AuthToken) -> Result<Self, ApiError> {
274        Self::with_options(token, ApiClientOptions::default())
275    }
276
277    /// Construct a client with specific options.
278    ///
279    /// # Errors
280    ///
281    /// Returns an error if the cache directory cannot be created or if the HTTP client fails to initialize.
282    pub fn with_options(token: AuthToken, options: ApiClientOptions) -> Result<Self, ApiError> {
283        std::fs::create_dir_all(&options.cache_dir)?;
284
285        let mut default_headers = HeaderMap::new();
286        default_headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
287        let user_agent_value = HeaderValue::from_str(&options.user_agent)
288            .unwrap_or_else(|_| HeaderValue::from_static("asana-cli"));
289        default_headers.insert(USER_AGENT, user_agent_value);
290
291        let http = reqwest::Client::builder()
292            .timeout(options.timeout)
293            .connect_timeout(Duration::from_secs(10))
294            .default_headers(default_headers)
295            .build()?;
296
297        let offline = options.offline;
298        Ok(Self {
299            http,
300            token,
301            options,
302            memory_cache: Arc::new(RwLock::new(HashMap::new())),
303            offline: AtomicBool::new(offline),
304            rate_limit: Arc::new(RwLock::new(None)),
305        })
306    }
307
308    /// Update offline mode at runtime.
309    pub fn set_offline(&self, offline: bool) {
310        self.offline.store(offline, Ordering::Relaxed);
311    }
312
313    /// Determine if offline mode is active.
314    #[must_use]
315    pub fn is_offline(&self) -> bool {
316        self.offline.load(Ordering::Relaxed)
317    }
318
319    /// Retrieve the most recent rate-limit information captured from the API.
320    #[must_use]
321    pub async fn rate_limit_info(&self) -> Option<RateLimitInfo> {
322        let guard = self.rate_limit.read().await;
323        guard.clone()
324    }
325
326    /// Return the base URL currently configured.
327    #[must_use]
328    pub fn base_url(&self) -> &str {
329        &self.options.base_url
330    }
331
332    /// Retrieve JSON from an endpoint and deserialize into `T`.
333    ///
334    /// # Errors
335    ///
336    /// Returns an error if the network request fails, the response is invalid, or deserialization fails.
337    pub async fn get_json<T>(&self, path: &str, query: &[(&str, &str)]) -> Result<T, ApiError>
338    where
339        T: DeserializeOwned,
340    {
341        let query_pairs = build_query_pairs(query);
342        self.get_json_with_pairs(path, query_pairs).await
343    }
344
345    /// Retrieve the current authenticated user (`/users/me`).
346    ///
347    /// # Errors
348    ///
349    /// Returns an error if the network request fails, authentication is invalid, or the response cannot be parsed.
350    pub async fn get_current_user(&self) -> Result<Value, ApiError> {
351        self.get_json("/users/me", &[]).await
352    }
353
354    /// POST helper for JSON endpoints returning a structured payload.
355    ///
356    /// # Errors
357    ///
358    /// Returns an error if serialization fails, the network request fails, or the response cannot be deserialized.
359    pub async fn post_json<T, R>(&self, path: &str, body: &T) -> Result<R, ApiError>
360    where
361        T: Serialize + ?Sized + Sync,
362        R: DeserializeOwned,
363    {
364        self.post_json_with_query(path, Vec::new(), body).await
365    }
366
367    /// POST helper accepting explicit query parameters.
368    ///
369    /// # Errors
370    ///
371    /// Returns an error if serialization fails, the network request fails, or the response cannot be deserialized.
372    pub async fn post_json_with_query<T, R>(
373        &self,
374        path: &str,
375        query_pairs: Vec<(String, String)>,
376        body: &T,
377    ) -> Result<R, ApiError>
378    where
379        T: Serialize + ?Sized + Sync,
380        R: DeserializeOwned,
381    {
382        let bytes = self
383            .execute_serialized(Method::POST, path, query_pairs, Some(body))
384            .await?;
385        Self::parse_response(path, &bytes)
386    }
387
388    /// POST helper for endpoints that do not return a payload.
389    ///
390    /// # Errors
391    ///
392    /// Returns an error if serialization fails or the network request fails.
393    pub async fn post_void<T>(&self, path: &str, body: &T) -> Result<(), ApiError>
394    where
395        T: Serialize + ?Sized + Sync,
396    {
397        let _ = self
398            .execute_serialized(Method::POST, path, Vec::new(), Some(body))
399            .await?;
400        Ok(())
401    }
402
403    /// PUT helper for JSON endpoints returning a structured payload.
404    ///
405    /// # Errors
406    ///
407    /// Returns an error if serialization fails, the network request fails, or the response cannot be deserialized.
408    pub async fn put_json<T, R>(&self, path: &str, body: &T) -> Result<R, ApiError>
409    where
410        T: Serialize + ?Sized + Sync,
411        R: DeserializeOwned,
412    {
413        self.put_json_with_query(path, Vec::new(), body).await
414    }
415
416    /// PUT helper accepting explicit query parameters.
417    ///
418    /// # Errors
419    ///
420    /// Returns an error if serialization fails, the network request fails, or the response cannot be deserialized.
421    pub async fn put_json_with_query<T, R>(
422        &self,
423        path: &str,
424        query_pairs: Vec<(String, String)>,
425        body: &T,
426    ) -> Result<R, ApiError>
427    where
428        T: Serialize + ?Sized + Sync,
429        R: DeserializeOwned,
430    {
431        let bytes = self
432            .execute_serialized(Method::PUT, path, query_pairs, Some(body))
433            .await?;
434        Self::parse_response(path, &bytes)
435    }
436
437    /// DELETE helper ignoring the response payload.
438    ///
439    /// # Errors
440    ///
441    /// Returns an error if the network request fails.
442    pub async fn delete(
443        &self,
444        path: &str,
445        query_pairs: Vec<(String, String)>,
446    ) -> Result<(), ApiError> {
447        let _ = self
448            .execute(Method::DELETE, path, query_pairs, None)
449            .await?;
450        Ok(())
451    }
452
453    /// POST multipart form data.
454    ///
455    /// # Errors
456    ///
457    /// Returns an error if the network request fails or the response cannot be deserialized.
458    pub async fn post_multipart<R>(
459        &self,
460        path: &str,
461        form: reqwest::multipart::Form,
462    ) -> Result<R, ApiError>
463    where
464        R: DeserializeOwned,
465    {
466        let url = format!("{}{path}", self.options.base_url);
467
468        let response = self
469            .http
470            .post(&url)
471            .header("Authorization", format!("Bearer {}", self.token.expose()))
472            .multipart(form)
473            .send()
474            .await?;
475
476        let status = response.status();
477        let bytes = response.bytes().await?;
478
479        if !status.is_success() {
480            let message = String::from_utf8_lossy(&bytes).to_string();
481            return Err(ApiError::Http {
482                status,
483                message,
484                details: None,
485            });
486        }
487
488        Self::parse_response(path, &bytes)
489    }
490
491    /// Download file from URL.
492    ///
493    /// # Errors
494    ///
495    /// Returns an error if the network request fails or the download fails.
496    pub async fn download_file(&self, url: &str) -> Result<Vec<u8>, ApiError> {
497        let response = self
498            .http
499            .get(url)
500            .header("Authorization", format!("Bearer {}", self.token.expose()))
501            .send()
502            .await?;
503
504        if !response.status().is_success() {
505            let status = response.status();
506            let message = response
507                .text()
508                .await
509                .unwrap_or_else(|_| "Download failed".to_string());
510            return Err(ApiError::Http {
511                status,
512                message,
513                details: None,
514            });
515        }
516
517        let bytes = response.bytes().await?;
518        Ok(bytes.to_vec())
519    }
520
521    /// Stream paginated endpoints as a series of pages (`Vec<T>`).
522    pub fn paginate<T>(
523        &self,
524        path: impl Into<String>,
525        query: Vec<(String, String)>,
526    ) -> impl Stream<Item = Result<Vec<T>, ApiError>> + '_
527    where
528        T: DeserializeOwned + Send + 'static,
529    {
530        self.paginate_with_limit(path, query, None)
531    }
532
533    /// Stream paginated endpoints with an optional global item limit.
534    pub fn paginate_with_limit<T>(
535        &self,
536        path: impl Into<String>,
537        query: Vec<(String, String)>,
538        max_items: Option<usize>,
539    ) -> impl Stream<Item = Result<Vec<T>, ApiError>> + '_
540    where
541        T: DeserializeOwned + Send + 'static,
542    {
543        let path = path.into();
544        let client = self.clone();
545
546        try_stream! {
547            let mut next_offset: Option<String> = None;
548            let mut emitted: usize = 0;
549            loop {
550                if let Some(max) = max_items
551                    && emitted >= max {
552                        break;
553                    }
554
555                let mut query_pairs = query.clone();
556                if let Some(offset) = next_offset.clone() {
557                    query_pairs.push(("offset".to_string(), offset));
558                }
559
560                let response: ListResponse<T> = match client
561                    .get_json_with_pairs(&path, query_pairs.clone())
562                    .await
563                {
564                    Ok(resp) => resp,
565                    Err(ApiError::Http { status: StatusCode::BAD_REQUEST, details, message })
566                        if is_offset_expired(details.as_ref(), &message) =>
567                    {
568                        break;
569                    }
570                    Err(err) => {
571                        Err(err)?;
572                        unreachable!();
573                    }
574                };
575
576                let mut items = response.data;
577                let next_offset_candidate = response
578                    .next_page
579                    .as_ref()
580                    .and_then(|meta| meta.offset.clone());
581
582                if let Some(max) = max_items
583                    && emitted + items.len() > max {
584                        items.truncate(max - emitted);
585                    }
586
587                emitted += items.len();
588                let continue_after_page = next_offset_candidate.is_some()
589                    && max_items.is_none_or(|max| emitted < max);
590
591                yield items;
592
593                if !continue_after_page {
594                    break;
595                }
596
597                next_offset = next_offset_candidate;
598            }
599        }
600    }
601
602    pub(crate) async fn get_json_with_pairs<T>(
603        &self,
604        path: &str,
605        query_pairs: Vec<(String, String)>,
606    ) -> Result<T, ApiError>
607    where
608        T: DeserializeOwned,
609    {
610        let bytes = self.execute(Method::GET, path, query_pairs, None).await?;
611        Self::parse_response(path, &bytes)
612    }
613
614    async fn execute_serialized<T>(
615        &self,
616        method: Method,
617        path: &str,
618        query_pairs: Vec<(String, String)>,
619        body: Option<&T>,
620    ) -> Result<Vec<u8>, ApiError>
621    where
622        T: Serialize + ?Sized + Sync,
623    {
624        let json_body = match body {
625            Some(payload) => Some(serde_json::to_value(payload)?),
626            None => None,
627        };
628        self.execute(method, path, query_pairs, json_body).await
629    }
630
631    fn parse_response<T>(path: &str, bytes: &[u8]) -> Result<T, ApiError>
632    where
633        T: DeserializeOwned,
634    {
635        if bytes.is_empty() {
636            return Err(ApiError::Other(format!("empty response body for {path}")));
637        }
638        let value: Value = serde_json::from_slice(bytes)?;
639        validate_response_schema(&value)?;
640        Ok(serde_json::from_value::<T>(value)?)
641    }
642
643    async fn execute(
644        &self,
645        method: Method,
646        path: &str,
647        query_pairs: Vec<(String, String)>,
648        body: Option<Value>,
649    ) -> Result<Vec<u8>, ApiError> {
650        let mut cache_key = None;
651        if method == Method::GET {
652            let key = Self::build_cache_key(&method, path, &query_pairs);
653            if let Some(bytes) = self.get_from_cache(&key).await? {
654                return Ok(bytes);
655            }
656            cache_key = Some(key);
657            if self.is_offline() {
658                return Err(ApiError::Offline {
659                    resource: path.to_string(),
660                });
661            }
662        }
663
664        let url = self.build_url(path);
665        let mut attempt = 0usize;
666        let max_retries = self.options.max_retries;
667        let body_clone = body.clone();
668
669        loop {
670            let mut request = self.http.request(method.clone(), &url);
671            request = request.header(AUTHORIZATION, format!("Bearer {}", self.token.expose()));
672            if !query_pairs.is_empty() {
673                request = request.query(&query_pairs);
674            }
675            if let Some(ref json) = body_clone {
676                request = request.json(json);
677            }
678
679            let response = request.send().await;
680            match response {
681                Err(err) => {
682                    if (err.is_timeout() || err.is_connect()) && attempt < max_retries {
683                        let delay = self.backoff_delay(attempt);
684                        debug!("retrying after network error: {err}; sleeping {delay:?}");
685                        sleep(delay).await;
686                        attempt += 1;
687                        continue;
688                    }
689                    return Err(ApiError::Network(err));
690                }
691                Ok(resp) => {
692                    if resp.status().is_success() {
693                        let headers = resp.headers().clone();
694                        let bytes = resp.bytes().await?.to_vec();
695                        if let Some(info) = Self::extract_rate_limit_headers(&headers) {
696                            let mut guard = self.rate_limit.write().await;
697                            *guard = Some(info);
698                        }
699                        if let Some(ref key) = cache_key {
700                            self.write_cache(key, &bytes).await?;
701                        }
702                        return Ok(bytes);
703                    }
704
705                    let status = resp.status();
706
707                    if status == StatusCode::TOO_MANY_REQUESTS {
708                        if let Some(info) = Self::extract_rate_limit_headers(resp.headers()) {
709                            let mut guard = self.rate_limit.write().await;
710                            *guard = Some(info.clone());
711                        }
712                        let retry_after = Self::parse_retry_after(resp.headers())
713                            .unwrap_or_else(|| self.backoff_delay(attempt));
714                        if attempt < max_retries {
715                            debug!(
716                                "rate limited, waiting {:?} before retry (attempt {})",
717                                retry_after,
718                                attempt + 1
719                            );
720                            sleep(retry_after).await;
721                            attempt += 1;
722                            continue;
723                        }
724                        let body = resp.text().await.unwrap_or_default();
725                        return Err(ApiError::RateLimited { retry_after, body });
726                    }
727
728                    if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN {
729                        let body = resp.text().await.unwrap_or_default();
730                        return Err(ApiError::Authentication(body));
731                    }
732
733                    if status.is_server_error() && attempt < max_retries {
734                        let delay = self.backoff_delay(attempt);
735                        warn!("server error {status}; retrying after {delay:?}");
736                        sleep(delay).await;
737                        attempt += 1;
738                        continue;
739                    }
740
741                    let text = resp.text().await.unwrap_or_default();
742                    let details = serde_json::from_str::<Value>(&text).ok();
743                    return Err(ApiError::http(
744                        status,
745                        if text.is_empty() {
746                            status
747                                .canonical_reason()
748                                .unwrap_or("unknown error")
749                                .to_string()
750                        } else {
751                            text
752                        },
753                        details,
754                    ));
755                }
756            }
757        }
758    }
759
760    fn build_url(&self, path: &str) -> String {
761        let trimmed_base = self.options.base_url.trim_end_matches('/');
762        let trimmed_path = path.trim_start_matches('/');
763        format!("{trimmed_base}/{trimmed_path}")
764    }
765
766    fn build_cache_key(method: &Method, path: &str, query_pairs: &[(String, String)]) -> String {
767        let mut hasher = Sha256::new();
768        hasher.update(method.as_str());
769        hasher.update("::");
770        hasher.update(path);
771        hasher.update("::");
772
773        let mut sorted = query_pairs.to_vec();
774        sorted.sort();
775        if let Ok(serialized) = serde_json::to_string(&sorted) {
776            hasher.update(serialized);
777        }
778
779        format!("{:x}", hasher.finalize())
780    }
781
782    async fn get_from_cache(&self, key: &str) -> Result<Option<Vec<u8>>, ApiError> {
783        let now = Instant::now();
784        if let Some(entry) = {
785            let guard = self.memory_cache.read().await;
786            guard.get(key).cloned()
787        } && entry.expires_at > now
788        {
789            debug!("cache hit (memory) for {key}");
790            return Ok(Some((*entry.value).clone()));
791        }
792
793        let path = self.cache_file_path(key);
794        match fs::read(&path).await {
795            Ok(bytes) => {
796                match serde_json::from_slice::<DiskCacheEntry>(&bytes) {
797                    Ok(entry) => {
798                        let expires_at = UNIX_EPOCH + Duration::from_secs(entry.expires_at);
799                        if SystemTime::now() <= expires_at {
800                            match general_purpose::STANDARD.decode(entry.body) {
801                                Ok(body) => {
802                                    self.store_in_memory(key.to_string(), body.clone()).await;
803                                    return Ok(Some(body));
804                                }
805                                Err(err) => {
806                                    warn!("failed to decode cache entry: {err}");
807                                    fs::remove_file(&path).await.ok();
808                                }
809                            }
810                        } else {
811                            fs::remove_file(&path).await.ok();
812                        }
813                    }
814                    Err(err) => {
815                        warn!("failed to parse cache entry: {err}");
816                        fs::remove_file(&path).await.ok();
817                    }
818                }
819                Ok(None)
820            }
821            Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(None),
822            Err(err) => Err(ApiError::Cache(err)),
823        }
824    }
825
826    async fn write_cache(&self, key: &str, body: &[u8]) -> Result<(), ApiError> {
827        self.store_in_memory(key.to_string(), body.to_vec()).await;
828
829        let expires_at = SystemTime::now()
830            .checked_add(self.options.cache_ttl)
831            .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
832            .map_or_else(
833                || {
834                    SystemTime::now()
835                        .duration_since(UNIX_EPOCH)
836                        .map_or(0, |duration| duration.as_secs())
837                },
838                |duration| duration.as_secs(),
839            );
840
841        let entry = DiskCacheEntry {
842            expires_at,
843            body: general_purpose::STANDARD.encode(body),
844        };
845
846        let path = self.cache_file_path(key);
847        if let Some(parent) = path.parent() {
848            fs::create_dir_all(parent).await.ok();
849        }
850        let serialized = serde_json::to_vec(&entry)?;
851        fs::write(path, serialized).await?;
852
853        debug!("cached response for key {key}");
854        Ok(())
855    }
856
857    async fn store_in_memory(&self, key: String, body: Vec<u8>) {
858        let entry = CacheEntry {
859            expires_at: Instant::now() + self.options.cache_ttl,
860            value: Arc::new(body),
861        };
862        let mut guard = self.memory_cache.write().await;
863        guard.insert(key, entry);
864    }
865
866    fn cache_file_path(&self, key: &str) -> PathBuf {
867        let mut filename = String::from(key);
868        filename.push_str(".json");
869        self.options.cache_dir.join(filename)
870    }
871
872    fn backoff_delay(&self, attempt: usize) -> Duration {
873        let multiplier = 1u32
874            .checked_shl(u32::try_from(attempt).unwrap_or(u32::MAX))
875            .unwrap_or(1);
876        self.options
877            .retry_base_delay
878            .checked_mul(multiplier)
879            .unwrap_or(self.options.retry_base_delay)
880    }
881
882    fn parse_retry_after(headers: &HeaderMap) -> Option<Duration> {
883        headers.get(RETRY_AFTER).and_then(|value| {
884            value.to_str().ok().and_then(|retry| {
885                if let Ok(seconds) = retry.parse::<f64>()
886                    && seconds.is_finite()
887                    && seconds >= 0.0
888                {
889                    return Some(Duration::from_secs_f64(seconds));
890                }
891                None
892            })
893        })
894    }
895
896    fn extract_rate_limit_headers(headers: &HeaderMap) -> Option<RateLimitInfo> {
897        let limit = headers
898            .get("X-RateLimit-Limit")
899            .and_then(|value| value.to_str().ok())
900            .and_then(|v| v.parse::<u32>().ok());
901        let remaining = headers
902            .get("X-RateLimit-Remaining")
903            .and_then(|value| value.to_str().ok())
904            .and_then(|v| v.parse::<u32>().ok());
905        let reset = headers
906            .get("X-RateLimit-Reset")
907            .and_then(|value| value.to_str().ok())
908            .and_then(|v| v.parse::<u64>().ok());
909        let retry_after = Self::parse_retry_after(headers);
910
911        if limit.is_none() && remaining.is_none() && reset.is_none() && retry_after.is_none() {
912            None
913        } else {
914            Some(RateLimitInfo {
915                limit,
916                remaining,
917                reset,
918                retry_after,
919            })
920        }
921    }
922}
923
924fn build_query_pairs(query: &[(&str, &str)]) -> Vec<(String, String)> {
925    query
926        .iter()
927        .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
928        .collect()
929}
930
931fn is_offset_expired(details: Option<&Value>, message: &str) -> bool {
932    let matches = |text: &str| {
933        let lowered = text.to_ascii_lowercase();
934        lowered.contains("offset") && (lowered.contains("expired") || lowered.contains("invalid"))
935    };
936
937    if let Some(payload) = details
938        && let Some(errors) = payload.get("errors").and_then(|v| v.as_array())
939        && errors.iter().any(|err| {
940            err.get("message")
941                .and_then(Value::as_str)
942                .is_some_and(matches)
943        })
944    {
945        return true;
946    }
947
948    matches(message)
949}
950
951fn validate_response_schema(value: &Value) -> Result<(), ApiError> {
952    if let Value::Object(obj) = value {
953        if obj.contains_key("data") || obj.contains_key("errors") {
954            return Ok(());
955        }
956        Err(ApiError::Other(
957            "response missing required `data` or `errors` field".to_string(),
958        ))
959    } else {
960        Ok(())
961    }
962}
963
964#[cfg(test)]
965mod tests {
966    use super::*;
967    use crate::api::auth::AuthToken;
968    use futures_util::StreamExt;
969    use mockito::{Matcher, Server};
970    use secrecy::SecretString;
971    use tempfile::TempDir;
972
973    #[tokio::test]
974    async fn get_current_user_fetches_data() {
975        let mut server = Server::new_async().await;
976        let _m = server
977            .mock("GET", "/users/me")
978            .match_header("authorization", "Bearer test-token")
979            .with_status(200)
980            .with_body(r#"{ "data": { "name": "Test User" } }"#)
981            .create_async()
982            .await;
983
984        let tmp = TempDir::new().unwrap();
985        let token = AuthToken::new(SecretString::new("test-token".into()));
986        let base_url = server.url();
987        let client = ApiClient::builder(token)
988            .base_url(base_url)
989            .cache_dir(tmp.path().join("cache"))
990            .build()
991            .unwrap();
992
993        let user: Value = client.get_current_user().await.unwrap();
994        assert_eq!(user["data"]["name"], "Test User");
995        drop(server);
996    }
997
998    #[tokio::test]
999    async fn rate_limit_retries_then_succeeds() {
1000        let mut server = Server::new_async().await;
1001        let _m1 = server
1002            .mock("GET", "/users/me")
1003            .with_status(429)
1004            .with_header("Retry-After", "0.1")
1005            .create_async()
1006            .await;
1007        let _m2 = server
1008            .mock("GET", "/users/me")
1009            .with_status(200)
1010            .with_body(r#"{ "data": { "name": "Retry User" } }"#)
1011            .create_async()
1012            .await;
1013
1014        let tmp = TempDir::new().unwrap();
1015        let token = AuthToken::new(SecretString::new("retry-token".into()));
1016        let base_url = server.url();
1017        let client = ApiClient::builder(token)
1018            .base_url(base_url)
1019            .cache_dir(tmp.path().join("cache"))
1020            .retry_base_delay(Duration::from_millis(50))
1021            .max_retries(2)
1022            .build()
1023            .unwrap();
1024
1025        let user: Value = client.get_current_user().await.unwrap();
1026        assert_eq!(user["data"]["name"], "Retry User");
1027        drop(server);
1028    }
1029
1030    #[tokio::test]
1031    async fn offline_uses_cache() {
1032        let mut server = Server::new_async().await;
1033        let _m = server
1034            .mock("GET", "/users/me")
1035            .with_status(200)
1036            .with_body(r#"{ "data": { "name": "Cached User" } }"#)
1037            .create_async()
1038            .await;
1039
1040        let tmp = TempDir::new().unwrap();
1041        let token = AuthToken::new(SecretString::new("cache-token".into()));
1042        let url = server.url();
1043        let client = ApiClient::builder(token)
1044            .base_url(url)
1045            .cache_dir(tmp.path().join("cache"))
1046            .cache_ttl(Duration::from_secs(60))
1047            .build()
1048            .unwrap();
1049
1050        let user: Value = client.get_current_user().await.unwrap();
1051        assert_eq!(user["data"]["name"], "Cached User");
1052
1053        client.set_offline(true);
1054
1055        // Drop all mocks to ensure no HTTP requests are made.
1056        server.reset();
1057        drop(server);
1058
1059        let cached: Value = client.get_current_user().await.unwrap();
1060        assert_eq!(cached["data"]["name"], "Cached User");
1061    }
1062
1063    #[tokio::test]
1064    async fn rate_limit_headers_captured_on_success() {
1065        let mut server = Server::new_async().await;
1066        let _m = server
1067            .mock("GET", "/users/me")
1068            .with_status(200)
1069            .with_header("X-RateLimit-Limit", "150")
1070            .with_header("X-RateLimit-Remaining", "149")
1071            .with_header("X-RateLimit-Reset", "1234567890")
1072            .with_body(r#"{ "data": { "name": "Metrics User" } }"#)
1073            .create_async()
1074            .await;
1075
1076        let tmp = TempDir::new().unwrap();
1077        let token = AuthToken::new(SecretString::new("metrics-token".into()));
1078        let base_url = server.url();
1079        let client = ApiClient::builder(token)
1080            .base_url(base_url)
1081            .cache_dir(tmp.path().join("cache"))
1082            .build()
1083            .unwrap();
1084
1085        client.get_current_user().await.unwrap();
1086        let info = client.rate_limit_info().await.expect("rate limit info");
1087        assert_eq!(info.limit, Some(150));
1088        assert_eq!(info.remaining, Some(149));
1089        assert_eq!(info.reset, Some(1_234_567_890));
1090        assert!(info.retry_after.is_none());
1091        drop(server);
1092    }
1093
1094    #[tokio::test]
1095    async fn rate_limit_headers_captured_on_429() {
1096        let mut server = Server::new_async().await;
1097        let _m1 = server
1098            .mock("GET", "/users/me")
1099            .with_status(429)
1100            .with_header("Retry-After", "0.1")
1101            .with_header("X-RateLimit-Limit", "150")
1102            .with_header("X-RateLimit-Remaining", "0")
1103            .with_body("rate limited")
1104            .create_async()
1105            .await;
1106        let _m2 = server
1107            .mock("GET", "/users/me")
1108            .with_status(200)
1109            .with_body(r#"{ "data": { "name": "Retry User" } }"#)
1110            .create_async()
1111            .await;
1112
1113        let tmp = TempDir::new().unwrap();
1114        let token = AuthToken::new(SecretString::new("metrics-rate-limit".into()));
1115        let base_url = server.url();
1116        let client = ApiClient::builder(token)
1117            .base_url(base_url)
1118            .cache_dir(tmp.path().join("cache"))
1119            .retry_base_delay(Duration::from_millis(50))
1120            .max_retries(2)
1121            .build()
1122            .unwrap();
1123
1124        client.get_current_user().await.unwrap();
1125        let info = client.rate_limit_info().await.expect("rate limit info");
1126        assert_eq!(info.limit, Some(150));
1127        assert_eq!(info.remaining, Some(0));
1128        assert!(info.retry_after.is_some());
1129        drop(server);
1130    }
1131
1132    #[tokio::test]
1133    async fn paginate_respects_manual_limit() {
1134        let mut server = Server::new_async().await;
1135        let _first = server
1136            .mock("GET", "/items")
1137            .with_status(200)
1138            .with_body(
1139                r#"{
1140                    "data": [ { "gid": "1" } ],
1141                    "next_page": { "offset": "abc", "path": "/items" }
1142                }"#,
1143            )
1144            .create_async()
1145            .await;
1146        let _second = server
1147            .mock("GET", "/items")
1148            .match_query(Matcher::UrlEncoded("offset".into(), "abc".into()))
1149            .with_status(200)
1150            .with_body(r#"{ "data": [ { "gid": "2" } ] }"#)
1151            .expect(0)
1152            .create_async()
1153            .await;
1154
1155        let tmp = TempDir::new().unwrap();
1156        let token = AuthToken::new(SecretString::new("paginate-limit".into()));
1157        let base_url = server.url();
1158        let client = ApiClient::builder(token)
1159            .base_url(base_url)
1160            .cache_dir(tmp.path().join("cache"))
1161            .build()
1162            .unwrap();
1163
1164        let stream = client.paginate_with_limit::<Value>("/items", vec![], Some(1));
1165        tokio::pin!(stream);
1166        let mut ids = Vec::new();
1167        while let Some(page) = stream.next().await {
1168            let page = page.expect("page result");
1169            ids.extend(
1170                page.iter()
1171                    .filter_map(|item| item.get("gid"))
1172                    .filter_map(Value::as_str)
1173                    .map(ToString::to_string),
1174            );
1175        }
1176        assert_eq!(ids, vec!["1".to_string()]);
1177        drop(server);
1178    }
1179
1180    #[tokio::test]
1181    async fn paginate_handles_offset_expired() {
1182        let mut server = Server::new_async().await;
1183        let _first = server
1184            .mock("GET", "/items")
1185            .with_status(200)
1186            .with_body(
1187                r#"{
1188                    "data": [ { "gid": "1" } ],
1189                    "next_page": { "offset": "abc", "path": "/items" }
1190                }"#,
1191            )
1192            .create_async()
1193            .await;
1194        let _second = server
1195            .mock("GET", "/items")
1196            .match_query(Matcher::UrlEncoded("offset".into(), "abc".into()))
1197            .with_status(400)
1198            .with_body(
1199                r#"{
1200                    "errors": [ { "message": "offset is invalid or expired" } ]
1201                }"#,
1202            )
1203            .create_async()
1204            .await;
1205
1206        let tmp = TempDir::new().unwrap();
1207        let token = AuthToken::new(SecretString::new("paginate-offset".into()));
1208        let base_url = server.url();
1209        let client = ApiClient::builder(token)
1210            .base_url(base_url)
1211            .cache_dir(tmp.path().join("cache"))
1212            .build()
1213            .unwrap();
1214
1215        let stream = client.paginate::<Value>("/items", vec![]);
1216        tokio::pin!(stream);
1217        let mut ids = Vec::new();
1218        while let Some(page) = stream.next().await {
1219            let page = page.expect("page result");
1220            ids.extend(
1221                page.iter()
1222                    .filter_map(|item| item.get("gid"))
1223                    .filter_map(Value::as_str)
1224                    .map(ToString::to_string),
1225            );
1226        }
1227        assert_eq!(ids, vec!["1".to_string()]);
1228        drop(server);
1229    }
1230
1231    #[tokio::test]
1232    async fn empty_response_returns_error() {
1233        let mut server = Server::new_async().await;
1234        let _m = server
1235            .mock("GET", "/users/me")
1236            .with_status(200)
1237            .with_body("")
1238            .create_async()
1239            .await;
1240
1241        let tmp = TempDir::new().unwrap();
1242        let token = AuthToken::new(SecretString::new("empty-response".into()));
1243        let url = server.url();
1244        let client = ApiClient::builder(token)
1245            .base_url(url)
1246            .cache_dir(tmp.path().join("cache"))
1247            .build()
1248            .unwrap();
1249
1250        let err = client.get_current_user().await.expect_err("should error");
1251        assert!(matches!(err, ApiError::Other(message) if message.contains("empty response")));
1252        drop(server);
1253    }
1254
1255    #[tokio::test]
1256    async fn response_missing_data_field_errors() {
1257        let mut server = Server::new_async().await;
1258        let _m = server
1259            .mock("GET", "/users/me")
1260            .with_status(200)
1261            .with_body(r#"{ "foo": "bar" }"#)
1262            .create_async()
1263            .await;
1264
1265        let tmp = TempDir::new().unwrap();
1266        let token = AuthToken::new(SecretString::new("missing-data".into()));
1267        let url = server.url();
1268        let client = ApiClient::builder(token)
1269            .base_url(url)
1270            .cache_dir(tmp.path().join("cache"))
1271            .build()
1272            .unwrap();
1273
1274        let err = client.get_current_user().await.expect_err("should error");
1275        assert!(matches!(err, ApiError::Other(message) if message.contains("data")));
1276        drop(server);
1277    }
1278}
asana_cli/api/client.rs

asana_cli/api/
client.rs
