pas-external 0.7.0

//! `sv` claim cache port — STANDARDS_AUTH_INVALIDATION §5.3.
//!
//! [`SvCachePort`] is the public extension axis: a narrow substrate
//! trait (`load`, `store`) that consumers implement for multi-pod
//! deployments (KVRocks, Redis). The trait is namespace- and
//! TTL-agnostic — the SDK uses [`ppoppo_token::sv_cache_key`] and
//! [`ppoppo_token::SV_CACHE_TTL`] (re-exported below as
//! [`SV_CACHE_TTL`]) as the single source of truth shared with PAS
//! server and PCS chat-auth, and applies the [`CheckResult`] decision
//! logic in [`super::sv::adapter`] before the substrate is ever called.
//!
//! Phase 6.1.D (Finding G): the previously-duplicated constants
//! `SV_CACHE_KEY_PREFIX` / `SV_CACHE_TTL` are gone — the engine now
//! owns the namespace + TTL contract, and a type-system-enforced
//! re-export makes drift impossible (a contract change in
//! `ppoppo-token` requires a recompile of every reader).
//!
//! [`MemorySvBackend`] is the default per-pod in-memory implementation
//! used when no backend is supplied via
//! [`super::PasAuth::session_validator_with_backend`].

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use tokio::sync::RwLock;

/// TTL for the `sv:{ppnum_id}` shared cache (60 s, spec-fixed).
/// Re-export of [`ppoppo_token::SV_CACHE_TTL`] — the engine owns the
/// invariant; the SDK reads it through this path so a future change to
/// the contract surfaces here as a recompile, not as silent drift.
pub(super) use ppoppo_token::SV_CACHE_TTL;

// Memory-bound for [`MemorySvBackend`]. With `SV_CACHE_TTL = 60 s`,
// an unbounded HashMap leaks one entry per unique `ppnum_id` ever resolved on
// the pod. 10_000 entries × ~80 bytes ≈ 800 KB worst case — comfortable for
// the SDK's "small to mid per-pod" sweet spot. Consumers needing higher caps
// should plug in their own backend via
// [`super::PasAuth::session_validator_with_backend`].
const MAX_ENTRIES: usize = 10_000;

/// Pluggable cache substrate.
///
/// `load` returns `None` on cache miss OR any transient backend error
/// (network blip, deserialization fault). Implementors **MUST** swallow
/// transient errors and return `None`; the policy treats `None` as
/// "verify with PAS." Surfacing an error here would defeat S-L6
/// fail-closed semantics (a cache outage would brick all sessions).
///
/// `store` is best-effort. The policy passes the SDK-owned TTL down so
/// backends with native expiry (Redis `SETEX`, KVRocks TTL) honor it.
/// In-memory backends honor it directly. A failed `store` only costs
/// one extra `/userinfo` round-trip on the next validate.
#[async_trait]
pub trait SvCachePort: Send + Sync + 'static {
    async fn load(&self, key: &str) -> Option<i64>;
    async fn store(&self, key: &str, value: i64, ttl: Duration);
}

/// Three-arm result of a cache lookup, computed by the SDK-internal
/// adapter. SDK-internal — consumers never observe this; the public
/// extension axis is the [`SvCachePort`] trait.
///
/// Kept distinct (rather than collapsed to a `bool`) so telemetry can
/// distinguish "saw a fresher sv" (break-glass converged across pods)
/// vs "had nothing cached" (cold cache) without changing dispatch.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum CheckResult {
    /// `token_sv >= cached_sv`. Admit.
    Fresh,
    /// `token_sv < cached_sv`. Token's sv is older than what we last
    /// observed — break-glass propagated. Refresh.
    Stale,
    /// Cache miss or transient backend error. Refresh.
    Unknown,
}

/// In-memory [`SvCachePort`]. Default choice for SDK consumers.
///
/// `tokio::sync::RwLock<HashMap<String, (sv, Instant, Duration)>>` with
/// lazy eviction on read (entries past their TTL are treated as miss)
/// plus a 10 000-entry cap with opportunistic pruning on `store` (see
/// source for the exact policy). Cheap-`Clone` (the inner `Arc` is
/// shared across handles), so a single backend can be observed from
/// multiple call sites — useful for tests that drive the backend
/// directly while the policy holds it. Production consumers with many
/// pods may want to plug in a shared backend (Redis, KVRocks) so a
/// break-glass on one pod converges on all pods within network RTT;
/// the in-memory default is per-pod with 60 s TTL convergence.
///
/// **Sustained workload above `MAX_ENTRIES = 10_000`**: the eviction
/// pass scans the map (`min_by_key` over all entries) on each
/// overflowing insert. At many-thousands of unique `ppnum_id` per
/// 60 s window, this becomes O(n) per write — switch to a shared
/// backend (`PasAuth::session_validator_with_backend`) before that point.
#[derive(Clone)]
pub struct MemorySvBackend {
    inner: Arc<RwLock<HashMap<String, (i64, Instant, Duration)>>>,
}

impl MemorySvBackend {
    #[must_use]
    pub fn new() -> Self {
        Self { inner: Arc::new(RwLock::new(HashMap::new())) }
    }
}

impl Default for MemorySvBackend {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl SvCachePort for MemorySvBackend {
    async fn load(&self, key: &str) -> Option<i64> {
        let guard = self.inner.read().await;
        let (sv, written_at, ttl) = guard.get(key)?;
        if written_at.elapsed() >= *ttl {
            return None;
        }
        Some(*sv)
    }

    async fn store(&self, key: &str, value: i64, ttl: Duration) {
        let mut guard = self.inner.write().await;
        if guard.len() >= MAX_ENTRIES && !guard.contains_key(key) {
            // First, free expired slots cheaply.
            guard.retain(|_, (_, written_at, slot_ttl)| written_at.elapsed() < *slot_ttl);
            // Still full? Evict the single oldest by write time (FIFO).
            // Under TTL=60s this is effectively LRU — entries don't live
            // long enough for hot-vs-cold patterns to develop.
            if guard.len() >= MAX_ENTRIES {
                let oldest_key = guard
                    .iter()
                    .min_by_key(|(_, (_, written, _))| *written)
                    .map(|(k, _)| k.clone());
                if let Some(k) = oldest_key {
                    guard.remove(&k);
                }
            }
        }
        guard.insert(key.to_string(), (value, Instant::now(), ttl));
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;

    // ---- MemorySvBackend ---------------------------------------------------

    #[tokio::test]
    async fn memory_backend_respects_ttl_on_load() {
        let backend = MemorySvBackend::new();
        backend.store("sv:abc", 42, SV_CACHE_TTL).await;
        assert_eq!(backend.load("sv:abc").await, Some(42));
        assert_eq!(backend.load("sv:missing").await, None);
    }

    #[tokio::test]
    async fn memory_backend_overwrite() {
        let backend = MemorySvBackend::new();
        backend.store("sv:xyz", 1, SV_CACHE_TTL).await;
        backend.store("sv:xyz", 2, SV_CACHE_TTL).await;
        assert_eq!(backend.load("sv:xyz").await, Some(2));
    }

    #[tokio::test]
    async fn memory_backend_bounded_by_max_entries() {
        // Insert MAX_ENTRIES + N unique keys within TTL; cap must hold.
        let backend = MemorySvBackend::new();
        for i in 0..(MAX_ENTRIES + 100) {
            backend.store(&format!("sv:{i}"), i as i64, SV_CACHE_TTL).await;
        }
        let len = backend.inner.read().await.len();
        assert!(len <= MAX_ENTRIES, "cache exceeded cap: {len} > {MAX_ENTRIES}");
        let last_key = format!("sv:{}", MAX_ENTRIES + 99);
        assert_eq!(
            backend.load(&last_key).await,
            Some((MAX_ENTRIES + 99) as i64),
            "newest entry must survive eviction",
        );
    }

    #[tokio::test]
    async fn memory_backend_honors_provided_ttl() {
        // Pins the lying-trait regression: a backend MUST honor the
        // TTL the SDK passes, not a hardcoded 60s. Write with a 0-tick
        // TTL — load must miss because elapsed >= ttl on the next read.
        let backend = MemorySvBackend::new();
        backend.store("sv:short", 7, Duration::from_nanos(1)).await;
        // Spin until elapsed exceeds the nanosecond TTL — Instant has
        // nanosecond resolution, so a yield is enough.
        tokio::task::yield_now().await;
        assert_eq!(
            backend.load("sv:short").await,
            None,
            "backend must honor caller-provided TTL, not hardcoded SV_CACHE_TTL",
        );
    }
}