use crate::internal::crypto_key::is_key_expired;
use crate::internal::CryptoKey;
use crate::types::KeyMeta;
use std::sync::atomic::{AtomicU64, AtomicU8, Ordering};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

/// Jitter factor for cache TTL to prevent thundering herd.
/// Each entry gets up to 10% of the TTL added as random jitter,
/// so entries loaded at the same time don't all expire simultaneously.
const TTL_JITTER_FRACTION: f64 = 0.10;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CachePolicy {
    Simple,
    Lru,
    Lfu,
    Slru,
    TinyLfu,
}

impl CachePolicy {
    pub fn parse(s: &str, default: CachePolicy) -> CachePolicy {
        match s.to_ascii_lowercase().as_str() {
            "simple" => CachePolicy::Simple,
            "lru" => CachePolicy::Lru,
            "lfu" => CachePolicy::Lfu,
            "slru" => CachePolicy::Slru,
            "tinylfu" => CachePolicy::TinyLfu,
            _ => default,
        }
    }
}

const SEG_PROBATIONARY: u8 = 0;
const SEG_PROTECTED: u8 = 1;

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

struct CacheEntry {
    key: Arc<CryptoKey>,
    /// Epoch milliseconds when this entry was last loaded/validated from the metastore.
    /// Stored as AtomicU64 so stale-while-revalidate can CAS it without exclusive locks.
    loaded_at_ms: AtomicU64,
    /// Per-entry jitter (millis) added to TTL to prevent thundering herd expiry.
    ttl_jitter_ms: u64,
    last_access: AtomicU64,
    freq: AtomicU64,
    segment: AtomicU8,
}

// Use Arc<str> keys to avoid String clones on cache lookups.
type CacheKey = (Arc<str>, i64);

/// Result of a cache check (no loader call).
///
/// The four variants encode three orthogonal pieces of information:
/// freshness, who owns the reload claim, and whether we have any key
/// at all. `Hit` and `StaleOther` happen to be handled identically by
/// the current encrypt-side consumer (both return the cached key), but
/// the distinction matters for metrics — `StaleOther` records a
/// `cache_stale` event while `Hit` records `cache_hit`. Collapsing the
/// two would lose that observability. Likewise `StaleReload` and
/// `Miss` are both "go to the metastore" but the former carries a
/// fallback key for the loader-failure path. T-finding "CacheCheck
/// reinvents Result; Hit | StaleOther arms merged identically" in
/// `docs/review-2026-05-05-findings.md`.
#[derive(Debug)]
pub enum CacheCheck {
    /// Fresh hit — use this key directly.
    Hit(Arc<CryptoKey>),
    /// Stale entry and we claimed the reload — caller should reload and insert.
    /// Contains the stale key as fallback if reload fails.
    StaleReload(Arc<CryptoKey>),
    /// Stale entry but another thread is reloading — use this stale key.
    StaleOther(Arc<CryptoKey>),
    /// Cache miss — caller must load from metastore and insert.
    Miss,
}

pub trait KeyCacher: Send + Sync {
    fn get_or_load_latest(
        &self,
        id: &str,
        loader: &mut dyn FnMut() -> anyhow::Result<Arc<CryptoKey>>,
    ) -> anyhow::Result<Arc<CryptoKey>>;
    fn get_or_load(
        &self,
        meta: &KeyMeta,
        loader: &mut dyn FnMut() -> anyhow::Result<Arc<CryptoKey>>,
    ) -> anyhow::Result<Arc<CryptoKey>>;
    fn close(&self) -> anyhow::Result<()> {
        Ok(())
    }

    /// Check the cache without calling a loader.
    fn check_latest(&self, id: &str) -> CacheCheck {
        // Default for NeverCache: always miss
        let _ = id;
        CacheCheck::Miss
    }
    fn check_meta(&self, meta: &KeyMeta) -> CacheCheck {
        let _ = meta;
        CacheCheck::Miss
    }
    /// Insert a key into the cache after an async load.
    fn insert_latest_key(&self, _id: &str, _key: Arc<CryptoKey>) {}
    fn insert_meta_key(&self, _meta: &KeyMeta, _key: Arc<CryptoKey>) {}

    /// Approximate count of entries currently held. `0` for caches
    /// that don't hold state. Used by tests to assert eviction-policy
    /// bounds; the value is racy under concurrent insertions and is
    /// not intended for production decision-making.
    fn entry_count(&self) -> usize {
        0
    }
}

#[derive(Debug)]
pub struct NeverCache;

impl KeyCacher for NeverCache {
    fn get_or_load_latest(
        &self,
        _id: &str,
        loader: &mut dyn FnMut() -> anyhow::Result<Arc<CryptoKey>>,
    ) -> anyhow::Result<Arc<CryptoKey>> {
        loader()
    }
    fn get_or_load(
        &self,
        _meta: &KeyMeta,
        loader: &mut dyn FnMut() -> anyhow::Result<Arc<CryptoKey>>,
    ) -> anyhow::Result<Arc<CryptoKey>> {
        loader()
    }
}

pub struct SimpleKeyCache {
    by_meta: scc::HashMap<CacheKey, CacheEntry>,
    latest: scc::HashMap<Arc<str>, i64>,
    ttl_ms: u64,
    max: usize,
    policy: CachePolicy,
    expire_after_s: i64,
    access_ctr: AtomicU64,
    decay_ctr: AtomicU64,
    /// Strictly-increasing per-insert seed mixer used by
    /// `random_jitter_ms`. Decoupled from `access_ctr` so two inserts
    /// without intervening reads still see distinct seeds.
    jitter_ctr: AtomicU64,
}

impl std::fmt::Debug for SimpleKeyCache {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SimpleKeyCache")
            .field("ttl_ms", &self.ttl_ms)
            .field("max", &self.max)
            .field("policy", &self.policy)
            .finish()
    }
}

impl SimpleKeyCache {
    pub fn new_with_ttl(ttl_s: i64) -> Self {
        Self::new_with_policy(ttl_s, 0, CachePolicy::Simple, 0)
    }

    pub fn new_with_policy(
        ttl_s: i64,
        max: usize,
        policy: CachePolicy,
        expire_after_s: i64,
    ) -> Self {
        // The `Simple` policy is intentionally unbounded — it never evicts.
        // Pairing it with `max > 0` is a misconfiguration: the bound will
        // be silently ignored. Surface a warning so operators noticing
        // unbounded memory growth can correct the policy. T-finding
        // "Simple policy bypasses eviction entirely" in
        // docs/review-2026-05-05-findings.md.
        if matches!(policy, CachePolicy::Simple) && max > 0 {
            log::warn!(
                "SimpleKeyCache: policy=Simple is unbounded; max={max} will be ignored. \
                 Use Lru, Lfu, Slru, or TinyLfu to enforce a bound."
            );
        }
        Self {
            by_meta: scc::HashMap::new(),
            latest: scc::HashMap::new(),
            ttl_ms: if ttl_s <= 0 { 0 } else { ttl_s as u64 * 1000 },
            max,
            policy,
            expire_after_s,
            access_ctr: AtomicU64::new(0),
            decay_ctr: AtomicU64::new(0),
            jitter_ctr: AtomicU64::new(0),
        }
    }
    pub fn new() -> Self {
        Self::new_with_ttl(60 * 60)
    }

    fn next_access(&self) -> u64 {
        self.access_ctr.fetch_add(1, Ordering::Relaxed) + 1
    }

    fn is_expired(&self, loaded_at_ms: u64, jitter_ms: u64) -> bool {
        // ttl_ms == 0 means "no TTL" — entries do not expire on a clock,
        // they are only invalidated by metastore-level events. Returning
        // `true` here would force a reload on every access (cache thrash).
        // The previous behavior was the source of T-finding "is_expired
        // returns true when ttl_ms == 0" in
        // docs/review-2026-05-05-findings.md.
        if self.ttl_ms == 0 {
            return false;
        }
        now_ms().saturating_sub(loaded_at_ms) >= self.ttl_ms + jitter_ms
    }

    /// Generate a random jitter in milliseconds, up to TTL_JITTER_FRACTION of the TTL.
    fn random_jitter_ms(&self, seed: u64) -> u64 {
        if self.ttl_ms == 0 {
            return 0;
        }
        let max_jitter_ms = (self.ttl_ms as f64 * TTL_JITTER_FRACTION) as u64;
        if max_jitter_ms == 0 {
            return 0;
        }
        // Mix the per-entry `seed` (a hash of `(id, created)`) with a
        // strictly-increasing `jitter_ctr` so two inserts at the same
        // wall-clock value still get distinct jitter. The previous
        // implementation read `access_ctr` non-atomically, which gave
        // identical jitter to entries inserted between accesses and
        // defeated the thundering-herd protection. T-finding
        // "random_jitter_ms is sequential LCG" in
        // `docs/review-2026-05-05-findings.md`.
        let bump = self.jitter_ctr.fetch_add(1, Ordering::Relaxed);
        // SplitMix64-style mixer — full avalanche from a small input
        // delta, no allocation, no syscall.
        let mut z = seed.wrapping_add(bump.wrapping_mul(0x9E3779B97F4A7C15));
        z = (z ^ (z >> 30)).wrapping_mul(0xBF58476D1CE4E5B9);
        z = (z ^ (z >> 27)).wrapping_mul(0x94D049BB133111EB);
        z ^= z >> 31;
        z % max_jitter_ms
    }

    fn is_invalid(&self, key: &CryptoKey) -> bool {
        if key.revoked() {
            return true;
        }
        if self.expire_after_s <= 0 {
            return false;
        }
        let now_s = crate::session::now_s();
        is_key_expired(key.created(), self.expire_after_s, now_s)
    }

    fn get_latest_if_fresh(&self, id: &str) -> Option<(Arc<CryptoKey>, bool, bool)> {
        // read() takes a shared lock on the bucket, not exclusive
        let (interned_id, created) = self.latest.read_sync(id, |k, &v| (k.clone(), v))?;

        let result = self
            .by_meta
            .read_sync(&(interned_id, created), |_, entry| {
                // Acquire to synchronize with `try_claim_reload_*`'s
                // Release-side CAS — without an Acquire load here a thread
                // can observe the bumped `loaded_at_ms` from another thread's
                // claim without seeing any subsequent metadata writes that
                // claim might have published. T-finding "Mixed Relaxed reads
                // + AcqRel CAS" in `docs/review-2026-05-05-findings.md`.
                let loaded = entry.loaded_at_ms.load(Ordering::Acquire);
                let expired = self.is_expired(loaded, entry.ttl_jitter_ms);
                let invalid = self.is_invalid(&entry.key);
                entry
                    .last_access
                    .store(self.next_access(), Ordering::Relaxed);
                entry.freq.fetch_add(1, Ordering::Relaxed);
                let seg = entry.segment.load(Ordering::Relaxed);
                let promote = self.policy == CachePolicy::Slru && seg == SEG_PROBATIONARY;
                if promote {
                    entry.segment.store(SEG_PROTECTED, Ordering::Relaxed);
                }
                (entry.key.clone(), expired, invalid, promote)
            })?;
        let (key_ref, expired, invalid, promote) = result;
        if promote {
            self.slru_rebalance();
        }
        Some((key_ref, expired, invalid))
    }

    fn get_meta_if_fresh(&self, meta: &KeyMeta) -> Option<(Arc<CryptoKey>, bool)> {
        let cache_key = (Arc::<str>::from(meta.id.as_str()), meta.created);
        let result = self.by_meta.read_sync(&cache_key, |_, entry| {
            // Acquire — see the matching note in `get_latest_if_fresh`.
            let loaded = entry.loaded_at_ms.load(Ordering::Acquire);
            let mut expired = self.is_expired(loaded, entry.ttl_jitter_ms);
            if entry.key.revoked() {
                expired = false;
            }
            entry
                .last_access
                .store(self.next_access(), Ordering::Relaxed);
            entry.freq.fetch_add(1, Ordering::Relaxed);
            let seg = entry.segment.load(Ordering::Relaxed);
            let promote = self.policy == CachePolicy::Slru && seg == SEG_PROBATIONARY;
            if promote {
                entry.segment.store(SEG_PROTECTED, Ordering::Relaxed);
            }
            (entry.key.clone(), expired, promote)
        })?;
        let (key_ref, expired, promote) = result;
        if promote {
            self.slru_rebalance();
        }
        Some((key_ref, expired))
    }

    fn insert_latest(&self, id: &str, key: Arc<CryptoKey>) {
        let meta = KeyMeta {
            id: id.to_string(),
            created: key.created(),
        };
        self.insert_meta(&meta, key);
    }

    fn insert_meta(&self, meta: &KeyMeta, key: Arc<CryptoKey>) {
        let interned_id: Arc<str> = Arc::from(meta.id.as_str());
        let cache_key = (interned_id.clone(), meta.created);
        // Per-entry seed: hash of `(id, created)` so each cache key
        // contributes distinct entropy to the jitter calculation. Two
        // entries inserted at the same wall-clock value but with
        // different ids will get different jitter values.
        use std::hash::{BuildHasher, Hasher};
        let mut hasher = std::collections::hash_map::RandomState::new().build_hasher();
        hasher.write(meta.id.as_bytes());
        hasher.write_i64(meta.created);
        let entry_seed = hasher.finish();
        let entry = CacheEntry {
            key,
            loaded_at_ms: AtomicU64::new(now_ms()),
            ttl_jitter_ms: self.random_jitter_ms(entry_seed),
            last_access: AtomicU64::new(self.next_access()),
            freq: AtomicU64::new(1),
            segment: AtomicU8::new(SEG_PROBATIONARY),
        };
        // upsert replaces existing entries — scc::HashMap::insert does NOT,
        // which would leave stale entries with old loaded_at forever.
        self.by_meta.upsert_sync(cache_key, entry);
        // Update latest pointer
        let should_update = self
            .latest
            .read_sync(&interned_id, |_, &existing| existing < meta.created)
            .unwrap_or(true);
        if should_update {
            let _ = self.latest.upsert_sync(interned_id, meta.created);
        }
        self.evict_if_needed();
    }

    fn evict_if_needed(&self) {
        if self.policy == CachePolicy::Simple || self.max == 0 {
            return;
        }
        while self.by_meta.len() > self.max {
            self.evict_one();
        }
    }

    fn evict_one(&self) {
        match self.policy {
            CachePolicy::Lru => {
                let mut victim: Option<(CacheKey, u64)> = None;
                self.by_meta.iter_sync(|k, v| {
                    let access = v.last_access.load(Ordering::Relaxed);
                    let dominated = match &victim {
                        None => true,
                        Some((_, min_access)) => access < *min_access,
                    };
                    if dominated {
                        victim = Some((k.clone(), access));
                    }
                    true
                });
                if let Some((k, _)) = victim {
                    self.by_meta.remove_sync(&k);
                }
            }
            CachePolicy::Lfu => {
                let mut victim: Option<(CacheKey, (u64, u64))> = None;
                self.by_meta.iter_sync(|k, v| {
                    let score = (
                        v.freq.load(Ordering::Relaxed),
                        v.last_access.load(Ordering::Relaxed),
                    );
                    let dominated = match &victim {
                        None => true,
                        Some((_, min_score)) => score < *min_score,
                    };
                    if dominated {
                        victim = Some((k.clone(), score));
                    }
                    true
                });
                if let Some((k, _)) = victim {
                    self.by_meta.remove_sync(&k);
                }
            }
            CachePolicy::TinyLfu => {
                self.decay_if_needed();
                let mut victim: Option<(CacheKey, (u64, u64))> = None;
                self.by_meta.iter_sync(|k, v| {
                    let score = (
                        v.freq.load(Ordering::Relaxed),
                        v.last_access.load(Ordering::Relaxed),
                    );
                    let dominated = match &victim {
                        None => true,
                        Some((_, min_score)) => score < *min_score,
                    };
                    if dominated {
                        victim = Some((k.clone(), score));
                    }
                    true
                });
                if let Some((k, _)) = victim {
                    self.by_meta.remove_sync(&k);
                }
            }
            CachePolicy::Slru => {
                self.slru_rebalance();
                let mut victim: Option<(CacheKey, u64)> = None;
                self.by_meta.iter_sync(|k, v| {
                    if v.segment.load(Ordering::Relaxed) == SEG_PROBATIONARY {
                        let access = v.last_access.load(Ordering::Relaxed);
                        let dominated = match &victim {
                            None => true,
                            Some((_, min_access)) => access < *min_access,
                        };
                        if dominated {
                            victim = Some((k.clone(), access));
                        }
                    }
                    true
                });
                if let Some((k, _)) = victim {
                    self.by_meta.remove_sync(&k);
                    return;
                }
                let mut victim: Option<(CacheKey, u64)> = None;
                self.by_meta.iter_sync(|k, v| {
                    if v.segment.load(Ordering::Relaxed) == SEG_PROTECTED {
                        let access = v.last_access.load(Ordering::Relaxed);
                        let dominated = match &victim {
                            None => true,
                            Some((_, min_access)) => access < *min_access,
                        };
                        if dominated {
                            victim = Some((k.clone(), access));
                        }
                    }
                    true
                });
                if let Some((k, _)) = victim {
                    self.by_meta.remove_sync(&k);
                }
            }
            CachePolicy::Simple => {}
        }
    }

    fn slru_rebalance(&self) {
        if self.max == 0 {
            return;
        }
        let protected_cap = std::cmp::max(1, self.max / 2);
        let mut protected_count = 0_usize;
        let mut victim: Option<(CacheKey, u64)> = None;
        self.by_meta.iter_sync(|k, v| {
            if v.segment.load(Ordering::Relaxed) == SEG_PROTECTED {
                protected_count += 1;
                let access = v.last_access.load(Ordering::Relaxed);
                let dominated = match &victim {
                    None => true,
                    Some((_, min_access)) => access < *min_access,
                };
                if dominated {
                    victim = Some((k.clone(), access));
                }
            }
            true
        });
        if protected_count > protected_cap {
            if let Some((k, _)) = victim {
                self.by_meta.read_sync(&k, |_, v| {
                    v.segment.store(SEG_PROBATIONARY, Ordering::Relaxed);
                });
            }
        }
    }

    fn decay_if_needed(&self) {
        if self.max == 0 {
            return;
        }
        let threshold = std::cmp::max(1, self.max as u64 * 10);
        let n = self.decay_ctr.fetch_add(1, Ordering::Relaxed) + 1;
        #[allow(clippy::manual_is_multiple_of)]
        if n % threshold == 0 {
            self.by_meta.iter_sync(|_, v| {
                let old = v.freq.load(Ordering::Relaxed);
                v.freq.store(std::cmp::max(1, old / 2), Ordering::Relaxed);
                true
            });
        }
    }

    /// Attempt to claim the reload for the latest key of `id` by CAS-ing loaded_at_ms.
    /// Returns true if this thread claimed the reload (and should do the metastore query).
    /// Other threads see the updated loaded_at and return the stale key without reloading.
    fn try_claim_reload_latest(&self, id: &str) -> bool {
        if self.ttl_ms == 0 {
            // No TTL means no reload — see `is_expired`.
            return false;
        }
        let Some((interned_id, created)) = self.latest.read_sync(id, |k, &v| (k.clone(), v)) else {
            return false;
        };
        let fresh = now_ms();
        let mut claimed = false;
        self.by_meta.read_sync(&(interned_id, created), |_, entry| {
            // Acquire load + AcqRel-success / Acquire-failure CAS so the
            // post-CAS reader's freshness signal is fully synchronized
            // with this thread's claim. T-finding "Mixed Relaxed reads +
            // AcqRel CAS" in `docs/review-2026-05-05-findings.md`.
            let old = entry.loaded_at_ms.load(Ordering::Acquire);
            if fresh.saturating_sub(old) >= self.ttl_ms + entry.ttl_jitter_ms {
                claimed = entry
                    .loaded_at_ms
                    .compare_exchange(old, fresh, Ordering::AcqRel, Ordering::Acquire)
                    .is_ok();
            }
        });
        claimed
    }

    /// Attempt to claim the reload for a specific key meta by CAS-ing loaded_at_ms.
    fn try_claim_reload_meta(&self, meta: &KeyMeta) -> bool {
        if self.ttl_ms == 0 {
            return false;
        }
        let cache_key = (Arc::<str>::from(meta.id.as_str()), meta.created);
        let fresh = now_ms();
        let mut claimed = false;
        self.by_meta.read_sync(&cache_key, |_, entry| {
            // Acquire load + AcqRel-success / Acquire-failure CAS so the
            // post-CAS reader's freshness signal is fully synchronized
            // with this thread's claim. T-finding "Mixed Relaxed reads +
            // AcqRel CAS" in `docs/review-2026-05-05-findings.md`.
            let old = entry.loaded_at_ms.load(Ordering::Acquire);
            if fresh.saturating_sub(old) >= self.ttl_ms + entry.ttl_jitter_ms {
                claimed = entry
                    .loaded_at_ms
                    .compare_exchange(old, fresh, Ordering::AcqRel, Ordering::Acquire)
                    .is_ok();
            }
        });
        claimed
    }
}

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

impl KeyCacher for SimpleKeyCache {
    fn get_or_load_latest(
        &self,
        id: &str,
        loader: &mut dyn FnMut() -> anyhow::Result<Arc<CryptoKey>>,
    ) -> anyhow::Result<Arc<CryptoKey>> {
        if let Some((v, expired, invalid)) = self.get_latest_if_fresh(id) {
            if !expired && !invalid {
                if crate::metrics::is_enabled() {
                    crate::metrics::record_cache_hit("latest");
                }
                return Ok(v);
            }
            // Stale-while-revalidate: stale but not invalid (not revoked, not policy-expired).
            // Try to claim the reload via CAS. Only one thread does the metastore query;
            // all others return the stale key immediately.
            if expired && !invalid {
                if self.try_claim_reload_latest(id) {
                    // We claimed the reload. Do the metastore query.
                    // Other threads now see fresh loaded_at and won't reload.
                    if crate::metrics::is_enabled() {
                        crate::metrics::record_cache_stale("latest");
                    }
                    match loader() {
                        Ok(new_key) => {
                            self.insert_latest(id, new_key.clone());
                            // Return the loaded key: handles key rotation and
                            // revocation discovery (loader returns a new valid key
                            // when the cached one was revoked in the metastore).
                            return Ok(new_key);
                        }
                        Err(_) => {
                            // Metastore error: loaded_at was already bumped by the CAS,
                            // so we won't retry until next TTL expiry. This is acceptable
                            // since the metastore is unreachable anyway.
                        }
                    }
                } else if crate::metrics::is_enabled() {
                    crate::metrics::record_cache_stale("latest");
                }
                return Ok(v);
            }
            // Key is invalid (revoked or policy-expired): must do a full reload
        }
        // Cold miss or invalid key — must load from metastore
        if crate::metrics::is_enabled() {
            crate::metrics::record_cache_miss("latest");
        }
        let v = loader()?;
        self.insert_latest(id, v.clone());
        Ok(v)
    }
    fn get_or_load(
        &self,
        meta: &KeyMeta,
        loader: &mut dyn FnMut() -> anyhow::Result<Arc<CryptoKey>>,
    ) -> anyhow::Result<Arc<CryptoKey>> {
        if let Some((v, expired)) = self.get_meta_if_fresh(meta) {
            if !expired {
                if crate::metrics::is_enabled() {
                    crate::metrics::record_cache_hit("meta");
                }
                return Ok(v);
            }
            // Stale-while-revalidate for meta lookups (decrypt path).
            // Revoked keys are never marked expired by get_meta_if_fresh, so we
            // only reach here for non-revoked stale entries loaded by exact
            // (id, created). The key material is identical whether we reload or
            // not — just bump loaded_at via CAS so the entry stays fresh.
            // No loader call needed: zero metastore queries on the decrypt path.
            let _ = self.try_claim_reload_meta(meta);
            if crate::metrics::is_enabled() {
                crate::metrics::record_cache_stale("meta");
            }
            return Ok(v);
        }
        // Cold miss — must load from metastore
        if crate::metrics::is_enabled() {
            crate::metrics::record_cache_miss("meta");
        }
        let v = loader()?;
        self.insert_meta(meta, v.clone());
        Ok(v)
    }

    fn check_latest(&self, id: &str) -> CacheCheck {
        if let Some((v, expired, invalid)) = self.get_latest_if_fresh(id) {
            if !expired && !invalid {
                if crate::metrics::is_enabled() {
                    crate::metrics::record_cache_hit("latest");
                }
                return CacheCheck::Hit(v);
            }
            if expired && !invalid {
                if self.try_claim_reload_latest(id) {
                    if crate::metrics::is_enabled() {
                        crate::metrics::record_cache_stale("latest");
                    }
                    return CacheCheck::StaleReload(v);
                }
                if crate::metrics::is_enabled() {
                    crate::metrics::record_cache_stale("latest");
                }
                return CacheCheck::StaleOther(v);
            }
            // invalid — fall through to Miss
        }
        if crate::metrics::is_enabled() {
            crate::metrics::record_cache_miss("latest");
        }
        CacheCheck::Miss
    }

    fn check_meta(&self, meta: &KeyMeta) -> CacheCheck {
        if let Some((v, expired)) = self.get_meta_if_fresh(meta) {
            if !expired {
                if crate::metrics::is_enabled() {
                    crate::metrics::record_cache_hit("meta");
                }
                return CacheCheck::Hit(v);
            }
            // Stale-while-revalidate for meta: just bump loaded_at, no reload needed
            let _ = self.try_claim_reload_meta(meta);
            if crate::metrics::is_enabled() {
                crate::metrics::record_cache_stale("meta");
            }
            return CacheCheck::Hit(v); // Always return stale value for meta (decrypt path)
        }
        if crate::metrics::is_enabled() {
            crate::metrics::record_cache_miss("meta");
        }
        CacheCheck::Miss
    }

    fn insert_latest_key(&self, id: &str, key: Arc<CryptoKey>) {
        self.insert_latest(id, key);
    }

    fn insert_meta_key(&self, meta: &KeyMeta, key: Arc<CryptoKey>) {
        self.insert_meta(meta, key);
    }

    fn entry_count(&self) -> usize {
        self.by_meta.len()
    }
}
