diff --git a/src/config.rs b/src/config.rs index 6ffdbe2..64ac134 100644 --- a/src/config.rs +++ b/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>>>, +} + +impl Default for TokenRequestLimiters { + fn default() -> Self { + Self { + inner: Arc::new(Mutex::new(HashMap::new())), + } + } +} + +impl TokenRequestLimiters { + pub fn semaphore(&self, token: &str) -> Arc { + 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, 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 { fn optional(name: &str) -> Option { 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); + } +} diff --git a/src/intervals.rs b/src/intervals.rs index 782cf17..28c36c0 100644 --- a/src/intervals.rs +++ b/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, + authentication: Option, } 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) -> 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, + ) -> 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 { - Ok(request.basic_auth("API_KEY", Some(self.api_key()?))) + fn authenticated( + &self, + request: reqwest::RequestBuilder, + ) -> Result { + 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 { + 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() diff --git a/src/main.rs b/src/main.rs index 8c4290c..758274d 100644 --- a/src/main.rs +++ b/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 }