limit Intervals requests to five per credential

src/config.rs

@@ -1,5 +1,47 @@
 use anyhow::{Context, Result};
-use std::env;
+use sha2::{Digest, Sha256};
+use std::{
+    collections::HashMap,
+    env,
+    sync::{Arc, Mutex},
+};
+use tokio::sync::Semaphore;
+
+const MAX_ACTIVE_REQUESTS_PER_TOKEN: usize = 5;
+
+#[derive(Clone, Debug)]
+pub struct TokenRequestLimiters {
+    inner: Arc<Mutex<HashMap<[u8; 32], Arc<Semaphore>>>>,
+}
+
+impl Default for TokenRequestLimiters {
+    fn default() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(HashMap::new())),
+        }
+    }
+}
+
+impl TokenRequestLimiters {
+    pub fn semaphore(&self, token: &str) -> Arc<Semaphore> {
+        let digest = Sha256::digest(token.as_bytes());
+        let key: [u8; 32] = digest.into();
+
+        let mut limiters = self
+            .inner
+            .lock()
+            .expect("token request limiter mutex poisoned");
+
+        limiters
+            .entry(key)
+            .or_insert_with(|| {
+                Arc::new(Semaphore::new(
+                    MAX_ACTIVE_REQUESTS_PER_TOKEN,
+                ))
+            })
+            .clone()
+    }
+}
 
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -16,6 +58,10 @@ pub struct Config {
     pub intervals_webhook_secret: Option<String>,
 
     pub cookie_secure: bool,
+
+    /// Shared per-credential concurrency limit for Intervals.icu requests.
+    /// Each API/OAuth credential is limited to five active requests.
+    pub intervals_request_limiters: TokenRequestLimiters,
 }
 
 impl Config {
@@ -62,6 +108,8 @@ impl Config {
             intervals_webhook_secret,
 
             cookie_secure,
+
+            intervals_request_limiters: TokenRequestLimiters::default(),
         })
     }
 }
@@ -79,3 +127,58 @@ fn required(name: &str) -> Result<String> {
 fn optional(name: &str) -> Option<String> {
     env::var(name).ok().filter(|value| !value.trim().is_empty())
 }
+
+#[cfg(test)]
+mod tests {
+    use super::TokenRequestLimiters;
+    use std::sync::Arc;
+    use tokio::time::{Duration, timeout};
+
+    #[tokio::test]
+    async fn allows_at_most_five_active_requests_per_token() {
+        let limiters = TokenRequestLimiters::default();
+        let semaphore = limiters.semaphore("same-token");
+
+        let mut permits = Vec::new();
+        for _ in 0..5 {
+            permits.push(
+                semaphore
+                    .clone()
+                    .acquire_owned()
+                    .await
+                    .expect("semaphore should be open"),
+            );
+        }
+
+        let sixth = timeout(
+            Duration::from_millis(20),
+            semaphore.clone().acquire_owned(),
+        )
+        .await;
+
+        assert!(sixth.is_err(), "sixth request must wait");
+
+        drop(permits.pop());
+
+        let sixth = timeout(
+            Duration::from_millis(100),
+            semaphore.acquire_owned(),
+        )
+        .await
+        .expect("a freed permit should become available")
+        .expect("semaphore should be open");
+
+        drop(sixth);
+    }
+
+    #[test]
+    fn different_tokens_get_independent_limiters() {
+        let limiters = TokenRequestLimiters::default();
+        let first = limiters.semaphore("token-a");
+        let second = limiters.semaphore("token-b");
+
+        assert!(!Arc::ptr_eq(&first, &second));
+        assert_eq!(first.available_permits(), 5);
+        assert_eq!(second.available_permits(), 5);
+    }
+}

src/intervals.rs

@@ -5,32 +5,52 @@ use serde_json::Value;
 
 use crate::config::Config;
 
+#[derive(Clone)]
+enum Authentication {
+    ApiKey(String),
+    Bearer(String),
+}
+
 #[derive(Clone)]
 pub struct IntervalsClient {
     client: Client,
     config: Config,
-    api_key: Option<String>,
+    authentication: Option<Authentication>,
 }
 
 impl IntervalsClient {
     pub fn new(config: Config) -> Self {
-        let api_key = config.intervals_api_key.clone();
+        let authentication = config
+            .intervals_api_key
+            .clone()
+            .map(Authentication::ApiKey);
 
         Self {
             client: Client::new(),
             config,
-            api_key,
+            authentication,
         }
     }
 
     /// Create a client using an explicitly supplied personal API key.
-    ///
-    /// Primarily used by /dev/sync/{api_key}/{athlete_id}.
     pub fn with_api_key(config: Config, api_key: impl Into<String>) -> Self {
         Self {
             client: Client::new(),
             config,
-            api_key: Some(api_key.into()),
+            authentication: Some(Authentication::ApiKey(api_key.into())),
+        }
+    }
+
+    /// Create a client using an OAuth access token.
+    /// Requests are authenticated with HTTP Bearer auth.
+    pub fn with_access_token(
+        config: Config,
+        access_token: impl Into<String>,
+    ) -> Self {
+        Self {
+            client: Client::new(),
+            config,
+            authentication: Some(Authentication::Bearer(access_token.into())),
         }
     }
 
@@ -38,10 +58,14 @@ impl IntervalsClient {
         self.config.intervals_base_url.trim_end_matches('/')
     }
 
-    fn api_key(&self) -> Result<&str> {
-        self.api_key
-            .as_deref()
-            .context("Intervals.icu API key is not configured")
+    fn authentication_token(&self) -> Result<&str> {
+        match self.authentication.as_ref() {
+            Some(Authentication::ApiKey(token))
+            | Some(Authentication::Bearer(token)) => Ok(token),
+            None => Err(anyhow!(
+                "Intervals.icu authentication is not configured"
+            )),
+        }
     }
 
     fn api_url(&self, path: &str) -> String {
@@ -52,11 +76,38 @@ impl IntervalsClient {
         )
     }
 
-    fn authenticated(&self, request: reqwest::RequestBuilder) -> Result<reqwest::RequestBuilder> {
-        Ok(request.basic_auth("API_KEY", Some(self.api_key()?)))
+    fn authenticated(
+        &self,
+        request: reqwest::RequestBuilder,
+    ) -> Result<reqwest::RequestBuilder> {
+        match self.authentication.as_ref() {
+            Some(Authentication::ApiKey(token)) => {
+                Ok(request.basic_auth("API_KEY", Some(token)))
+            }
+            Some(Authentication::Bearer(token)) => {
+                Ok(request.bearer_auth(token))
+            }
+            None => Err(anyhow!(
+                "Intervals.icu authentication is not configured"
+            )),
+        }
     }
 
     async fn get_json(&self, url: String) -> Result<Value> {
+        let token = self.authentication_token()?;
+        let semaphore = self
+            .config
+            .intervals_request_limiters
+            .semaphore(token);
+
+        // The permit covers the complete active request, including reading
+        // the response body. This enforces a real concurrency limit rather
+        // than merely limiting request creation.
+        let _permit = semaphore
+            .acquire()
+            .await
+            .context("Intervals.icu request limiter closed")?;
+
         let response = self
             .authenticated(self.client.get(&url))?
             .send()

src/main.rs

@@ -1029,7 +1029,7 @@ pub async fn process_activity(
         ));
     }
 
-    let client = IntervalsClient::with_api_key(state.config.clone(), access_token);
+    let client = IntervalsClient::with_access_token(state.config.clone(), access_token);
 
     process_activity_with_client(state, intervals_athlete_id, activity_id, &client).await
 }