ppoppo-sdk-core 0.6.2

//! JWT/token retention layer for Ppoppo client SDKs.
//!
//! [`TokenCache`] wraps a [`TokenSource`] and adds near-expiry refresh with
//! single-flight semantics: when multiple tasks call [`TokenCache::get`]
//! concurrently during an expiry window, only one performs the fetch; the
//! rest queue on the refresh lock and then reuse the result.
//!
//! ## Typical wiring
//!
//! ```ignore
//! use ppoppo_sdk_core::token_cache::{
//!     ClientCredentialsSource, TokenCache, TokenCacheConfig,
//! };
//!
//! let source = ClientCredentialsSource::new(token_url, client_id, client_secret);
//! let cache = Arc::new(TokenCache::new(Box::new(source), TokenCacheConfig::default()));
//! let token: String = cache.get().await?;
//! ```

use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use ppoppo_clock::ArcClock;
use ppoppo_clock::native::WallClock;
use tokio::sync::Mutex;

/// Errors returned by [`TokenCache`] and [`TokenSource`] implementations.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum TokenCacheError {
    #[error("token fetch failed: {0}")]
    Fetch(String),
    #[error("token response is malformed: {0}")]
    Malformed(String),
    /// A source-specific failure, preserved intact for the SDK layer that
    /// installed the source. A custom [`TokenSource`] wraps its own typed
    /// error here instead of flattening it to a string, so the SDK facade
    /// can `downcast` back to its native error type (e.g. distinguish
    /// "not authenticated — no durable credential" from a transport
    /// failure) after the error has passed through the cache.
    #[error("{0}")]
    Source(#[source] Box<dyn std::error::Error + Send + Sync>),
}

/// Object-safe async port for acquiring a fresh `(token, ttl)` pair.
///
/// Implement this for custom token acquisition strategies. The built-in
/// implementation is [`ClientCredentialsSource`].
#[async_trait]
pub trait TokenSource: Send + Sync {
    /// Fetch a fresh token. Returns the raw JWT string and its lifetime.
    /// The cache subtracts [`TokenCacheConfig::refresh_skew`] from the TTL
    /// before storing the expiry, so the token is treated as stale slightly
    /// before the server considers it expired.
    async fn fetch_token(&self) -> Result<(String, Duration), TokenCacheError>;
}

/// Configuration for [`TokenCache`].
#[derive(Debug, Clone)]
pub struct TokenCacheConfig {
    /// How far before a token's actual expiry to treat it as stale.
    /// Defaults to 60 s — guards against clock skew and network latency on
    /// the refresh call.
    pub refresh_skew: Duration,
}

impl Default for TokenCacheConfig {
    fn default() -> Self {
        // Single skew SSOT: the same 60 s constant drives the pure
        // `is_expired` helper and the cache's stored-expiry discount.
        #[allow(clippy::cast_sign_loss)] // compile-time positive constant
        Self { refresh_skew: Duration::from_secs(jwt_expiry::EXPIRY_SKEW_SECS as u64) }
    }
}

struct CacheInner {
    cached: Option<(String, i64)>,
}

/// Thread-safe JWT cache with near-expiry refresh and single-flight semantics.
///
/// Wrap in `Arc<TokenCache>` and share across tasks.
pub struct TokenCache {
    data: Mutex<CacheInner>,
    /// Serialises concurrent refreshes (single-flight pattern).
    /// Only the task that acquires this lock performs the HTTP round-trip;
    /// the rest wait, then double-check the cache before deciding to fetch.
    refresh: Mutex<()>,
    config: TokenCacheConfig,
    source: Box<dyn TokenSource>,
    clock: ArcClock,
}

impl TokenCache {
    /// Create a new cache backed by `source`.
    pub fn new(source: Box<dyn TokenSource>, config: TokenCacheConfig) -> Self {
        Self {
            data: Mutex::new(CacheInner { cached: None }),
            refresh: Mutex::new(()),
            config,
            source,
            clock: Arc::new(WallClock),
        }
    }

    #[must_use]
    pub fn with_clock(mut self, clock: ArcClock) -> Self {
        self.clock = clock;
        self
    }

    /// Seed the cache with an already-held token (e.g. loaded from disk),
    /// builder-style, before the cache is shared.
    ///
    /// The stored expiry is derived from the token's JWT `exp` claim (via
    /// [`parse_expiry`]) minus [`TokenCacheConfig::refresh_skew`] — the same
    /// discount the fetch path applies. Tokens without a parseable `exp`
    /// are ignored, and an already-stale seed is filtered by `get()`'s
    /// freshness check — either way the first [`get`](Self::get) falls
    /// through to the source, which is the safe direction.
    ///
    /// This exists so a seeded cache never confuses "still-valid stored
    /// token" with "freshly minted token": seeding happens exactly once at
    /// construction, so [`invalidate`](Self::invalidate) always forces a
    /// genuine re-mint (a source that re-served stored state would hand the
    /// same dead token back after a 401).
    #[must_use]
    pub fn with_initial_token(mut self, token: impl Into<String>) -> Self {
        let token = token.into();
        if let Some(exp) = jwt_expiry::parse_expiry(&token) {
            let skew_ms = i64::try_from(self.config.refresh_skew.as_millis()).unwrap_or(i64::MAX);
            let exp_ms = exp.as_millisecond().saturating_sub(skew_ms);
            self.data.get_mut().cached = Some((token, exp_ms));
        }
        self
    }

    /// Return the current token, refreshing via the source if near-expired.
    pub async fn get(&self) -> Result<String, TokenCacheError> {
        // Fast path: cache is fresh.
        {
            let inner = self.data.lock().await;
            if let Some((token, exp)) = &inner.cached
                && self.clock.now_unix_millis() < *exp {
                    return Ok(token.clone());
                }
        }

        // Slow path: acquire the refresh lock (serialises concurrent refreshes).
        let _refresh_guard = self.refresh.lock().await;

        // Double-check: a previous holder may have already refreshed.
        {
            let inner = self.data.lock().await;
            if let Some((token, exp)) = &inner.cached
                && self.clock.now_unix_millis() < *exp {
                    return Ok(token.clone());
                }
        }

        // We are the refresh leader. Fetch a fresh token.
        let (token, ttl) = self.source.fetch_token().await?;
        let skewed_ttl = ttl.saturating_sub(self.config.refresh_skew);
        let exp = self.clock.now_unix_millis() + skewed_ttl.as_millis() as i64;
        self.data.lock().await.cached = Some((token.clone(), exp));

        Ok(token)
    }

    /// Discard the cached token so the next [`get`](Self::get) re-mints.
    ///
    /// Call this when the server rejects the current token (a 401 /
    /// `Unauthenticated`): the token is *dead* — revoked, session-epoch-bumped,
    /// or rotated out from under a still-valid clock TTL — so clock-based expiry
    /// is the wrong staleness signal. Forcing a fetch lets a caller recover on
    /// its next request instead of waiting out the remaining TTL.
    pub async fn invalidate(&self) {
        self.data.lock().await.cached = None;
    }
}

#[cfg(feature = "client-credentials")]
mod credentials;
mod jwt_expiry;

#[cfg(feature = "client-credentials")]
pub use credentials::ClientCredentialsSource;
pub use jwt_expiry::{EXPIRY_SKEW_SECS, is_expired, parse_expiry};

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;
    use std::sync::{
        Arc,
        atomic::{AtomicUsize, Ordering},
    };

    struct CountingSource {
        count: Arc<AtomicUsize>,
        ttl: Duration,
        token: &'static str,
    }

    #[async_trait]
    impl TokenSource for CountingSource {
        async fn fetch_token(&self) -> Result<(String, Duration), TokenCacheError> {
            self.count.fetch_add(1, Ordering::SeqCst);
            // Simulate a brief network hop so concurrent callers actually
            // overlap on the refresh lock.
            tokio::time::sleep(Duration::from_millis(10)).await;
            Ok((self.token.to_string(), self.ttl))
        }
    }

    #[tokio::test]
    async fn token_cache_returns_cached_until_near_expiry() {
        let count = Arc::new(AtomicUsize::new(0));
        let source = CountingSource {
            count: Arc::clone(&count),
            ttl: Duration::from_secs(3600),
            token: "tok-abc",
        };
        let cache = TokenCache::new(Box::new(source), TokenCacheConfig::default());

        let t1 = cache.get().await.unwrap();
        let t2 = cache.get().await.unwrap();

        assert_eq!(t1, "tok-abc");
        assert_eq!(t2, "tok-abc");
        assert_eq!(count.load(Ordering::SeqCst), 1, "source called more than once for valid cache");
    }

    #[tokio::test]
    async fn invalidate_forces_refetch_despite_valid_ttl() {
        let count = Arc::new(AtomicUsize::new(0));
        let source = CountingSource {
            count: Arc::clone(&count),
            ttl: Duration::from_secs(3600), // long TTL: clock says "still fresh"
            token: "tok-1",
        };
        let cache = TokenCache::new(Box::new(source), TokenCacheConfig::default());

        let _ = cache.get().await.unwrap(); // fetch #1, cached
        let _ = cache.get().await.unwrap(); // served from cache
        assert_eq!(count.load(Ordering::SeqCst), 1, "second get should hit cache");

        // The server rejected this token — it is dead despite a valid clock TTL.
        cache.invalidate().await;

        let _ = cache.get().await.unwrap(); // must re-fetch, not serve the dead token
        assert_eq!(
            count.load(Ordering::SeqCst),
            2,
            "invalidate() must force the next get() to re-mint",
        );
    }

    fn fake_jwt(exp: jiff::Timestamp) -> String {
        jwt_expiry::tests::make_fake_token(exp)
    }

    #[tokio::test]
    async fn with_initial_token_serves_seed_without_fetch() {
        let count = Arc::new(AtomicUsize::new(0));
        let source = CountingSource {
            count: Arc::clone(&count),
            ttl: Duration::from_secs(3600),
            token: "tok-fresh",
        };
        let seed = fake_jwt(jiff::Timestamp::now() + jiff::SignedDuration::from_hours(1));
        let cache = TokenCache::new(Box::new(source), TokenCacheConfig::default())
            .with_initial_token(seed.clone());

        assert_eq!(cache.get().await.unwrap(), seed);
        assert_eq!(count.load(Ordering::SeqCst), 0, "a valid seed must not trigger a fetch");

        // A 401-invalidate must force a genuine re-mint, never re-serve the seed.
        cache.invalidate().await;
        assert_eq!(cache.get().await.unwrap(), "tok-fresh");
        assert_eq!(count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn with_initial_token_ignores_unparseable_and_stale_seeds() {
        for seed in [
            "not-a-jwt".to_string(),
            fake_jwt(jiff::Timestamp::now() - jiff::SignedDuration::from_hours(1)),
        ] {
            let count = Arc::new(AtomicUsize::new(0));
            let source = CountingSource {
                count: Arc::clone(&count),
                ttl: Duration::from_secs(3600),
                token: "tok-fresh",
            };
            let cache = TokenCache::new(Box::new(source), TokenCacheConfig::default())
                .with_initial_token(seed);

            assert_eq!(cache.get().await.unwrap(), "tok-fresh");
            assert_eq!(count.load(Ordering::SeqCst), 1, "unusable seed must fall through to fetch");
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn token_cache_single_flights_concurrent_refresh() {
        let count = Arc::new(AtomicUsize::new(0));
        let source = CountingSource {
            count: Arc::clone(&count),
            ttl: Duration::from_secs(3600),
            token: "tok-xyz",
        };
        let cache = Arc::new(TokenCache::new(Box::new(source), TokenCacheConfig::default()));

        let mut handles = Vec::new();
        for _ in 0..8 {
            let cache = Arc::clone(&cache);
            handles.push(tokio::spawn(async move { cache.get().await }));
        }
        for h in handles {
            assert_eq!(h.await.unwrap().unwrap(), "tok-xyz");
        }

        // All 8 concurrent callers must share one fetch (single-flight).
        assert_eq!(count.load(Ordering::SeqCst), 1, "source called more than once (single-flight broken)");
    }
}