cachekit/

client.rs

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

use bytes::Bytes;

use serde::{de::DeserializeOwned, Serialize};

use crate::backend::Backend;
#[cfg(feature = "l1")]
use crate::backend::{ttl_wire_secs, Freshness};
use crate::error::CachekitError;
use crate::metrics::{CacheCounters, L1Stats, ReadOutcome};
use crate::serializer;

// ── SharedBackend type alias ──────────────────────────────────────────────────

/// Reference-counted pointer to a heap-allocated backend.
///
/// On native targets (without `unsync`) we require `Send + Sync` via `Arc`.
/// On `wasm32` or with the `unsync` feature, `Rc` is used instead — the runtime
/// is single-threaded so `Send` bounds are unnecessary.
#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
pub type SharedBackend = std::sync::Arc<dyn Backend>;

/// Reference-counted pointer to a heap-allocated backend (`?Send` variant).
#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
pub type SharedBackend = std::rc::Rc<dyn Backend>;

// ── SharedFlight type alias ──────────────────────────────────────────────────

/// Reference-counted pointer to the single-flight map, so client clones share
/// fill-dedup state (two clones racing a cold miss must collapse to one fill).
#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
pub(crate) type SharedFlight = std::sync::Arc<crate::flight::FlightMap>;

#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
pub(crate) type SharedFlight = std::rc::Rc<crate::flight::FlightMap>;

/// Separate same-key ordering from single-flight: a refresh holds the flight
/// lock while computing, then takes this lock only for its version-checked
/// commit. Direct reads/writes/deletes take the same mutation lock.
#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
type SharedMutations = std::sync::Arc<crate::flight::MutationMap>;

// ── SharedEncryption type alias ──────────────────────────────────────────────

/// Reference-counted pointer to the encryption layer.
///
/// On native targets (without `unsync`) `Arc` is used (requires `Sync`).
/// On `wasm32` or with `unsync`, `Rc` is used — avoids the `!Sync` problem
/// caused by `Cell<u64>` inside cachekit-core's nonce counter.
#[cfg(all(
    feature = "encryption",
    not(any(target_arch = "wasm32", feature = "unsync"))
))]
type SharedEncryption = std::sync::Arc<crate::encryption::EncryptionLayer>;

#[cfg(all(
    feature = "encryption",
    any(target_arch = "wasm32", feature = "unsync")
))]
type SharedEncryption = std::rc::Rc<crate::encryption::EncryptionLayer>;

// ── Key validation ────────────────────────────────────────────────────────────

const MAX_KEY_BYTES: usize = 1024;

/// Maximum TTL for L1 entries populated from L2 cache hits.
/// Uses a short ceiling to limit staleness when the original TTL is unknown.
/// A server `X-CacheKit-Fresh-For` bounds it further, and a stale read or a
/// zero bound is not backfilled at all — see [`l1_backfill_ttl`].
///
/// Reconciliation with stale-while-revalidate: a backfilled entry's SWR
/// freshness window derives from this capped TTL (window = ratio × entry
/// TTL), **not** from the write-path TTL — the cap is the staleness bound
/// for L2-derived data, deliberately kept. SWR removes the cap's expiry
/// cliff instead: past ~ratio × 30 s the entry is served stale while one
/// background refresh re-executes the origin. If no newer mutation replaced
/// the entry, that refresh writes both layers and renews L1 hard expiry with
/// the caller's full TTL. Without SWR the entry simply hard-expires at the cap
/// and the next read blocks on L2, as before.
#[cfg(feature = "l1")]
const L1_BACKFILL_TTL_SECS: u64 = 30;

/// L1 lifetime for a backfill from an L2 hit, or `None` when the read must not
/// be backfilled (`spec/saas-api.md` § Remaining Freshness). A stale label or a
/// zero `Fresh-For` forbids it; a present `Fresh-For` is a hard bound under the
/// 30 s cap, which SWR cannot outlive because moka drops the entry at its TTL.
/// With no header the cap applies unchanged.
#[cfg(feature = "l1")]
fn l1_backfill_ttl(default_ttl: Duration, freshness: Freshness) -> Option<Duration> {
    let cap = std::cmp::min(default_ttl, Duration::from_secs(L1_BACKFILL_TTL_SECS));
    let ttl = freshness
        .fresh_for
        .map_or(cap, |fresh_for| cap.min(fresh_for));
    (!freshness.is_stale && !ttl.is_zero()).then_some(ttl)
}

fn validate_key(key: &str) -> Result<(), CachekitError> {
    if key.is_empty() {
        return Err(CachekitError::InvalidKey(
            "key must not be empty".to_owned(),
        ));
    }
    if key.len() > MAX_KEY_BYTES {
        return Err(CachekitError::InvalidKey(format!(
            "key is {} bytes (limit: {MAX_KEY_BYTES})",
            key.len()
        )));
    }
    for b in key.bytes() {
        if b < 0x20 || b == 0x7F {
            return Err(CachekitError::InvalidKey(format!(
                "key contains illegal control character 0x{b:02X}"
            )));
        }
    }
    Ok(())
}

// ── Stale-while-revalidate ───────────────────────────────────────────────────
//
// SWR needs an L1 to age entries in and a spawnable (`Send`) runtime for the
// background refresh — native, non-`unsync` targets with the `l1` feature.
// Everywhere else the SWR read path degrades to the plain read path and
// `SwrRead::Stale` is never produced.

/// Outcome of an SWR-aware typed read — see [`CacheKit::interop_get_swr`].
#[derive(Debug, Clone, PartialEq)]
pub enum SwrRead<T> {
    /// Cache hit within the freshness window (or an L2 hit): use directly.
    Fresh(T),
    /// L1 hit past the freshness threshold but before hard expiry: use the
    /// value now, and schedule a background refresh (the `#[cachekit]` macro
    /// re-executes the function, deduplicated on the [`CacheKit::single_flight`]
    /// locks without waiting on them). The token
    /// makes refresh completion conditional: a newer set or delete wins.
    Stale(T, SwrToken),
    /// No usable entry: fall through to a normal blocking miss + fill.
    Miss,
}

/// Mutation version captured by an SWR stale read.
///
/// Pass this back only through the `#[cachekit]`-generated refresh path. It
/// prevents an older background computation from overwriting a newer write
/// or resurrecting a deleted entry on this client or any of its clones.
#[derive(Clone)]
pub struct SwrToken {
    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    state: std::sync::Arc<crate::flight::MutationState>,
    version: u64,
}

impl std::fmt::Debug for SwrToken {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SwrToken")
            .field("version", &self.version)
            .finish_non_exhaustive()
    }
}

impl PartialEq for SwrToken {
    fn eq(&self, other: &Self) -> bool {
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        {
            self.version == other.version && std::sync::Arc::ptr_eq(&self.state, &other.state)
        }
        #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
        {
            self.version == other.version
        }
    }
}

impl Eq for SwrToken {}

// ── CacheKit ─────────────────────────────────────────────────────────────────

/// Cache client with optional L1 in-process cache layer.
///
/// `Clone` is cheap and shares everything: backend, L1 cache, single-flight
/// state, and encryption layer. Clones exist so `'static` background work
/// (e.g. the SWR refresh spawned by `#[cachekit]`) can hold the client
/// without borrowing it.
///
/// # Encryption
///
/// A client built with encryption configured (by any of the builder's
/// encryption methods, the `secure` / `secure_from_env` preset, or
/// [`CacheKit::from_env`] with `CACHEKIT_MASTER_KEY` set) encrypts every
/// value it writes and decrypts every value it reads — [`Self::get`],
/// [`Self::interop_get`], [`Self::interop_get_swr`], [`Self::set`] and
/// [`Self::set_with_ttl`] — and L1 holds ciphertext. Such a client has no
/// plaintext value path, and the `SecureCache` handle from `secure_cache()`
/// reads and writes the same entries. [`Self::delete`] and [`Self::exists`]
/// carry no value and are the same either way.
#[derive(Clone)]
pub struct CacheKit {
    backend: SharedBackend,
    default_ttl: Duration,
    namespace: Option<String>,
    max_payload_bytes: usize,
    flight: SharedFlight,
    /// Read counters behind [`Self::stats`]; the backend holds a
    /// `MetricsProvider` over the same allocation.
    counters: Arc<CacheCounters>,

    /// The reliability stack's breaker, when configured — kept so
    /// [`Self::circuit_state`] can read it through the type-erased backend.
    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
    breaker: Option<Arc<crate::reliability::CircuitBreaker>>,

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    mutations: SharedMutations,

    #[cfg(feature = "l1")]
    pub(crate) l1: Option<crate::l1::L1Cache>,

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    pub(crate) swr_enabled: bool,

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    swr_threshold_ratio: f64,

    #[cfg(feature = "encryption")]
    encryption: Option<SharedEncryption>,
}

impl CacheKit {
    /// Create a new builder.
    pub fn builder() -> CacheKitBuilder {
        CacheKitBuilder::default()
    }

    /// Build from environment variables via [`crate::config::CachekitConfig::from_env`].
    ///
    /// Creates a [`crate::backend::cachekitio::CachekitIO`] backend from the
    /// config. Requires the `cachekitio` feature. With the `encryption`
    /// feature on and `CACHEKIT_MASTER_KEY` set, the client encrypts every
    /// value read and write (see [Encryption](CacheKit#encryption)); without
    /// the feature, a set `CACHEKIT_MASTER_KEY` is a [`CachekitError::Config`].
    #[cfg(all(feature = "cachekitio", not(target_arch = "wasm32")))]
    pub fn from_env() -> Result<CacheKitBuilder, CachekitError> {
        use crate::backend::cachekitio::CachekitIO;
        use crate::config::CachekitConfig;

        let config = CachekitConfig::from_env()?;

        // The same environment encrypts on a build with the feature, so
        // ignoring the key here would be a silent plaintext downgrade.
        #[cfg(not(feature = "encryption"))]
        if config.master_key.is_some() {
            return Err(encryption_feature_missing(
                "CacheKit::from_env() with CACHEKIT_MASTER_KEY set",
            ));
        }

        let api_key_z = config
            .api_key
            .ok_or_else(|| CachekitError::Config("CACHEKIT_API_KEY is required".to_owned()))?;

        let backend = CachekitIO::builder()
            .api_key(api_key_z.as_str())
            .api_url(config.api_url)
            .build()
            .map_err(|e| CachekitError::Config(e.to_string()))?;

        #[cfg(not(feature = "unsync"))]
        let shared: SharedBackend = std::sync::Arc::new(backend);
        #[cfg(feature = "unsync")]
        let shared: SharedBackend = std::rc::Rc::new(backend);

        let mut builder = CacheKitBuilder::default()
            .backend(shared)
            .default_ttl(config.default_ttl)
            .max_payload_bytes(config.max_payload_bytes)
            .l1_capacity(config.l1_capacity);

        if let Some(ns) = config.namespace.clone() {
            builder = builder.namespace(ns);
        }

        // Wire up encryption if master key is configured
        #[cfg(feature = "encryption")]
        if let Some(ref master_key) = config.master_key {
            let namespace = config.namespace.as_deref().unwrap_or("default");
            let previous: Vec<&[u8]> = config
                .previous_master_keys
                .iter()
                .map(|key| key.as_slice())
                .collect();
            builder = builder.encryption_from_hex_decoded(master_key, &previous, namespace)?;
        }

        Ok(builder)
    }

    // ── Namespacing ───────────────────────────────────────────────────────────

    fn namespaced_key(&self, key: &str) -> String {
        match &self.namespace {
            Some(ns) => format!("{ns}:{key}"),
            None => key.to_owned(),
        }
    }

    /// Validate key and return the namespaced version.
    fn resolve_key(&self, key: &str) -> Result<String, CachekitError> {
        validate_key(key)?;
        Ok(self.namespaced_key(key))
    }

    // ── L1 helpers ───────────────────────────────────────────────────────────

    /// Try L1 cache first. Returns Some(bytes) on hit.
    #[cfg(feature = "l1")]
    fn l1_get(&self, full_key: &str) -> Option<Bytes> {
        self.l1.as_ref().and_then(|l1| l1.get_shared(full_key))
    }

    /// Populate L1 from an L2 hit, for no longer than [`l1_backfill_ttl`] allows.
    #[cfg(feature = "l1")]
    fn l1_backfill(&self, full_key: &str, bytes: Bytes, freshness: Freshness) {
        if let (Some(l1), Some(ttl)) = (&self.l1, l1_backfill_ttl(self.default_ttl, freshness)) {
            l1.set_shared(full_key, bytes, ttl);
        }
    }

    /// Write-through to L1 after a write the backend accepted. The copy lives
    /// no longer than the TTL the write sent on the wire ([`ttl_wire_secs`]):
    /// `spec/saas-api.md` forbids serving it past `min(local_ttl, effective_ttl)`,
    /// and the wire truncates a fractional TTL from 1 s up (1.5 s goes as 1).
    #[cfg(feature = "l1")]
    fn l1_set(&self, full_key: &str, bytes: Bytes, ttl: Duration) {
        if let Some(ref l1) = self.l1 {
            let wire = Duration::from_secs(ttl_wire_secs(ttl));
            l1.set_shared(full_key, bytes, ttl.min(wire));
        }
    }

    /// Invalidate L1 entry.
    #[cfg(feature = "l1")]
    fn l1_delete(&self, full_key: &str) {
        if let Some(ref l1) = self.l1 {
            l1.delete(full_key);
        }
    }

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    async fn lock_l1_mutation(&self, full_key: &str) -> Option<crate::flight::MutationGuard> {
        if self.l1.is_some() {
            Some(self.mutations.lock(full_key).await)
        } else {
            None
        }
    }

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    async fn complete_swr_bytes(
        &self,
        key: &str,
        bytes: Vec<u8>,
        ttl: Duration,
        token: SwrToken,
    ) -> Result<bool, CachekitError> {
        Self::validate_ttl(ttl)?;
        self.check_payload_size(bytes.len())?;
        let full_key = self.resolve_key(key)?;
        let Some(mutation) = self.lock_l1_mutation(&full_key).await else {
            return Ok(false);
        };

        // Check before touching L2. The mutation state lives independently of
        // the moka entry, so hard expiry or capacity eviction does not look
        // like an explicit set/delete and waste a valid origin result.
        if !mutation.is_current(&token.state, token.version) {
            return Ok(false);
        }
        let l1_bytes = Bytes::copy_from_slice(&bytes);
        self.backend.set(&full_key, bytes, Some(ttl)).await?;
        self.l1_set(&full_key, l1_bytes, ttl);
        mutation.advance();
        crate::metrics::trace_write(&full_key, ttl);
        Ok(true)
    }

    #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
    async fn complete_swr_bytes(
        &self,
        _key: &str,
        _bytes: Vec<u8>,
        _ttl: Duration,
        _token: SwrToken,
    ) -> Result<bool, CachekitError> {
        // `SwrRead::Stale` is unreachable on this build, so there is no
        // versioned entry a caller could legitimately complete.
        Ok(false)
    }

    /// Rejects a zero TTL.
    fn validate_ttl(ttl: Duration) -> Result<(), CachekitError> {
        if ttl.is_zero() {
            return Err(CachekitError::Config(
                "TTL must be greater than zero".to_owned(),
            ));
        }
        Ok(())
    }

    // ── Public operations ─────────────────────────────────────────────────────

    /// Retrieve and deserialize a value stored under `key`.
    ///
    /// Returns `None` if the key does not exist.
    /// Checks L1 cache before hitting the backend.
    ///
    /// # Errors
    ///
    /// On a client with encryption configured, returns
    /// [`CachekitError::Encryption`] — not a miss — if the stored entry fails
    /// decryption. The key's L1 copy is dropped first and the backend entry
    /// is left in place, so the next read reaches the backend: a hit once the
    /// entry is replaced with ciphertext this client can decrypt, a miss once
    /// it expires or is deleted. A decrypted entry that does not decode as `T`
    /// is [`CachekitError::Serialization`] naming only `T`, never quoting the
    /// value.
    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.get(key).await;
        }
        match self.get_bytes(key).await? {
            Some(bytes) => Ok(Some(serializer::deserialize(&bytes)?)),
            None => Ok(None),
        }
    }

    /// Retrieve and deserialize an interop-mode value stored under `key`.
    ///
    /// Identical to [`Self::get`] except the payload is decoded with
    /// [`crate::interop::deserialize`]. Both consume exactly one MessagePack
    /// document and reject trailing bytes (interop/v1 spec MUST); this one
    /// also names a Python-SDK-internal CK frame with a specific diagnostic
    /// instead of a generic trailing-bytes error.
    ///
    /// Use with keys from [`crate::interop::interop_key`] on a client
    /// **without** a namespace prefix. There is no interop-specific write
    /// method: [`Self::set`] already writes plain MessagePack (no ByteStorage
    /// envelope), which is the interop value format. On a client with
    /// encryption configured the stored bytes are AES-GCM ciphertext of that
    /// MessagePack, decrypted here before decoding: in interop mode the
    /// AES-GCM plaintext is the plain MessagePack value bytes, so the AAD
    /// (v0x03, `format="msgpack"`, `compressed="False"`) verifies cross-SDK
    /// unchanged.
    ///
    /// Unlike this read, [`Self::set`] / [`Self::set_with_ttl`] do not check
    /// the client namespace: on a client with [`CacheKitBuilder::namespace`]
    /// set they return `Ok` and store the entry under
    /// `{namespace}:{interop_key}`, which no interop reader addresses. Interop
    /// writes need a client without a namespace.
    ///
    /// # Errors
    ///
    /// Returns [`CachekitError::Config`] if the client was built with
    /// [`CacheKitBuilder::namespace`]: the prefix
    /// would rewrite the storage key to `{prefix}:{interop_key}`, which no
    /// other SDK computes — every cross-SDK entry would silently miss. Interop
    /// keys carry their own namespace segment; failing loudly here beats a
    /// 100% miss rate that looks like a cold cache. With encryption
    /// configured, an entry that fails decryption is
    /// [`CachekitError::Encryption`], with the key's L1 copy dropped as for
    /// [`Self::get`].
    pub async fn interop_get<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<Option<T>, CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.interop_get(key).await;
        }
        self.reject_namespaced_interop()?;
        match self.get_bytes(key).await? {
            Some(bytes) => Ok(Some(crate::interop::deserialize(&bytes)?)),
            None => Ok(None),
        }
    }

    /// Interop keys must reach the backend verbatim; a client namespace prefix
    /// would silently produce storage keys no other SDK computes.
    fn reject_namespaced_interop(&self) -> Result<(), CachekitError> {
        match self.namespace {
            None => Ok(()),
            Some(_) => Err(CachekitError::Config(
                "interop reads require a client without a namespace prefix: .namespace() \
                 would store interop entries under {prefix}:{interop_key}, \
                 which other SDKs never compute (interop keys already carry a namespace \
                 segment) — use a dedicated non-namespaced client for interop entries"
                    .to_owned(),
            )),
        }
    }

    /// Retrieve and deserialize an interop-mode value with SWR classification.
    ///
    /// Identical to [`Self::interop_get`] except an L1 hit is classified
    /// against the client's stale-while-revalidate freshness window:
    ///
    /// - [`SwrRead::Fresh`] — L1 hit within `swr_threshold_ratio` of the
    ///   entry's TTL (±10% jitter), or any L2 hit. Use directly.
    /// - [`SwrRead::Stale`] — L1 hit past the threshold but **before hard
    ///   expiry**: the value is returned without touching the backend or
    ///   origin, and the caller should schedule exactly one background
    ///   refresh. The `#[cachekit]` macro dedups it on the cold-miss
    ///   [`Self::single_flight`] locks without waiting for another worker's
    ///   fill: a refresh that finds either lock held, or whose lock call
    ///   fails, stands down and the stale value keeps being served. The
    ///   accompanying [`SwrToken`] makes completion conditional,
    ///   so a newer set/delete always wins.
    /// - [`SwrRead::Miss`] — nothing usable anywhere: normal blocking miss.
    ///
    /// A hard-expired L1 entry is a [`SwrRead::Miss`], never `Stale` — moka
    /// drops entries at their TTL, so SWR cannot serve past hard expiry.
    ///
    /// With SWR disabled ([`CacheKitBuilder::swr_enabled`]`(false)`), without
    /// the `l1` feature, on wasm32, or under `unsync`, this behaves exactly
    /// like [`Self::interop_get`]: hits are `Fresh`, `Stale` is never
    /// produced.
    ///
    /// On a client with encryption configured, staleness is judged on the L1
    /// **ciphertext** entry (freshness metadata never exposes plaintext),
    /// then the value is decrypted as for [`Self::interop_get`].
    ///
    /// # Errors
    ///
    /// Same as [`Self::interop_get`] (including the namespaced-client
    /// rejection).
    pub async fn interop_get_swr<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<SwrRead<T>, CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.interop_get_swr(key).await;
        }
        self.reject_namespaced_interop()?;
        match self.get_bytes_swr(key).await? {
            SwrRead::Fresh(b) => Ok(SwrRead::Fresh(crate::interop::deserialize(&b)?)),
            SwrRead::Stale(b, token) => Ok(SwrRead::Stale(crate::interop::deserialize(&b)?, token)),
            SwrRead::Miss => Ok(SwrRead::Miss),
        }
    }

    /// Fetch raw payload bytes with SWR classification: an L1 hit is split
    /// into fresh vs stale against the configured freshness window; on L1
    /// miss this defers to [`Self::get_bytes`] (L2 + backfill), whose hit is
    /// always fresh.
    async fn get_bytes_swr(&self, key: &str) -> Result<SwrRead<Bytes>, CachekitError> {
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        if self.swr_enabled {
            if let Some(ref l1) = self.l1 {
                let full_key = self.resolve_key(key)?;
                match l1.get_with_swr_shared(&full_key, self.swr_threshold_ratio) {
                    crate::l1::SharedSwrRead::Fresh(bytes) => {
                        self.counters.record(ReadOutcome::L1Hit, &full_key);
                        self.check_payload_size(bytes.len())?;
                        return Ok(SwrRead::Fresh(bytes));
                    }
                    crate::l1::SharedSwrRead::Stale(_) => {
                        // Token capture and the stale snapshot must be atomic
                        // relative to explicit mutations. Re-check after
                        // taking the per-key guard; a write may have landed
                        // between the optimistic classification and here.
                        let mutation = self.mutations.lock(&full_key).await;
                        match l1.get_with_swr_shared(&full_key, self.swr_threshold_ratio) {
                            crate::l1::SharedSwrRead::Fresh(bytes) => {
                                self.counters.record(ReadOutcome::L1Hit, &full_key);
                                self.check_payload_size(bytes.len())?;
                                return Ok(SwrRead::Fresh(bytes));
                            }
                            crate::l1::SharedSwrRead::Stale(bytes) => {
                                self.counters.record(ReadOutcome::L1Stale, &full_key);
                                self.check_payload_size(bytes.len())?;
                                let (state, version) = mutation.snapshot();
                                return Ok(SwrRead::Stale(bytes, SwrToken { state, version }));
                            }
                            crate::l1::SharedSwrRead::Miss => {}
                        }
                    }
                    // Absent or hard-expired: fall through to the normal
                    // read path (the redundant L1 re-check there is a cheap
                    // in-process miss).
                    crate::l1::SharedSwrRead::Miss => {}
                }
            }
        }

        Ok(match self.get_bytes(key).await? {
            Some(bytes) => SwrRead::Fresh(bytes),
            None => SwrRead::Miss,
        })
    }

    /// Fetch raw payload bytes for `key` (L1, then L2 with L1 backfill).
    async fn get_bytes(&self, key: &str) -> Result<Option<Bytes>, CachekitError> {
        let full_key = self.resolve_key(key)?;

        // L1 hit
        #[cfg(feature = "l1")]
        if let Some(bytes) = self.l1_get(&full_key) {
            self.counters.record(ReadOutcome::L1Hit, &full_key);
            self.check_payload_size(bytes.len())?;
            return Ok(Some(bytes));
        }

        // Serialize an L2 read/backfill with same-key writes. Re-check L1
        // after taking the lock because another operation may have filled it
        // between the optimistic read above and lock acquisition.
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        let _mutation = self.lock_l1_mutation(&full_key).await;

        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        if let Some(bytes) = self.l1_get(&full_key) {
            self.counters.record(ReadOutcome::L1Hit, &full_key);
            self.check_payload_size(bytes.len())?;
            return Ok(Some(bytes));
        }

        // L2 backend
        let (bytes, freshness) = match self.backend.get_with_freshness(&full_key).await? {
            Some(hit) => {
                self.counters.record(ReadOutcome::L2Hit, &full_key);
                hit
            }
            None => {
                self.counters.record(ReadOutcome::Miss, &full_key);
                return Ok(None);
            }
        };

        self.check_payload_size(bytes.len())?;
        // Takes over the Vec's allocation: no copy, and L1 shares it below.
        let bytes = Bytes::from(bytes);

        // Populate L1 on L2 hit, bounded by the cap and the server's freshness
        #[cfg(feature = "l1")]
        self.l1_backfill(&full_key, bytes.clone(), freshness);
        #[cfg(not(feature = "l1"))]
        let _ = freshness;

        Ok(Some(bytes))
    }

    /// Serialize and store `value` under `key` using the client's default TTL.
    ///
    /// Encrypts first on a client with encryption configured — see
    /// [`Self::set_with_ttl`].
    pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
        self.set_with_ttl(key, value, self.default_ttl).await
    }

    /// Serialize and store `value` under `key` with an explicit `ttl`.
    ///
    /// On a client with encryption configured the value is encrypted first:
    /// the backend and L1 receive only the AES-256-GCM ciphertext, and the
    /// payload limit applies to the ciphertext.
    ///
    /// A positive sub-second `ttl` is sent to the backend as 1 second.
    ///
    /// Returns [`CachekitError::Config`] if `ttl` is zero.
    ///
    /// The client namespace is not checked here: an interop key written on a
    /// namespaced client lands under `{namespace}:{interop_key}`, which no
    /// interop reader addresses — see [`Self::interop_get`].
    pub async fn set_with_ttl<T: Serialize>(
        &self,
        key: &str,
        value: &T,
        ttl: Duration,
    ) -> Result<(), CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.set_with_ttl(key, value, ttl).await;
        }
        Self::validate_ttl(ttl)?;

        let bytes = serializer::serialize(value)?;
        self.check_payload_size(bytes.len())?;

        let full_key = self.resolve_key(key)?;

        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        let mutation = self.lock_l1_mutation(&full_key).await;

        // Invalidate older refresh tokens before the first backend await, so
        // cancellation cannot leave an applied/attempted write vulnerable to
        // a stale background result.
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        if let Some(ref mutation) = mutation {
            mutation.advance();
        }

        // L1 needs its own copy: the backend consumes the Vec.
        #[cfg(feature = "l1")]
        {
            let l1_bytes = Bytes::copy_from_slice(&bytes);
            self.backend.set(&full_key, bytes, Some(ttl)).await?;
            self.l1_set(&full_key, l1_bytes, ttl);
        }
        #[cfg(not(feature = "l1"))]
        {
            self.backend.set(&full_key, bytes, Some(ttl)).await?;
        }

        crate::metrics::trace_write(&full_key, ttl);
        Ok(())
    }

    /// Commit an SWR refresh only if its stale-read token is still current,
    /// encrypted when the client has encryption configured. Macro plumbing;
    /// ordinary writes use [`Self::set_with_ttl`].
    #[doc(hidden)]
    pub async fn __complete_swr_refresh<T: Serialize>(
        &self,
        key: &str,
        value: &T,
        ttl: Duration,
        token: SwrToken,
    ) -> Result<bool, CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.__complete_swr_refresh(key, value, ttl, token).await;
        }
        let bytes = serializer::serialize(value)?;
        self.complete_swr_bytes(key, bytes, ttl, token).await
    }

    /// Delete `key` and return `true` if it existed.
    ///
    /// `CachekitIO` and `WorkersCachekitIO` return `true` on every successful
    /// delete, whether or not the key existed: the server does not report
    /// existence on `DELETE`.
    ///
    /// Invalidates the L1 entry regardless of the backend result.
    pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
        let full_key = self.resolve_key(key)?;

        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        let mutation = self.lock_l1_mutation(&full_key).await;

        // Invalidate before L1 changes or backend I/O so task cancellation
        // cannot let an older refresh resurrect this key.
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        if let Some(ref mutation) = mutation {
            mutation.advance();
        }

        // Invalidate L1 first so callers never read a stale value even if the
        // backend delete fails partway through.
        #[cfg(feature = "l1")]
        self.l1_delete(&full_key);

        let existed = self.backend.delete(&full_key).await?;
        crate::metrics::trace_delete(&full_key, existed);
        Ok(existed)
    }

    /// Return `true` if `key` exists without fetching the value.
    ///
    /// A live L1 entry answers without a backend call. That check does not
    /// count as a use of the entry for L1 eviction; only reads do.
    pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
        let full_key = self.resolve_key(key)?;

        // Check L1 first — avoids a network round-trip for warm entries.
        #[cfg(feature = "l1")]
        if self.l1.as_ref().is_some_and(|l1| l1.contains(&full_key)) {
            return Ok(true);
        }

        Ok(self.backend.exists(&full_key).await?)
    }

    // ── Observability ─────────────────────────────────────────────────────────

    /// Live hit/miss counters for this client and all of its clones.
    ///
    /// Every value read — `get`, `interop_get`, their SWR variants, and the
    /// [`SecureCache`] equivalents — is counted by where it was served from.
    /// `exists` and reads that fail with a backend error are not counted. The
    /// same counters feed the cachekit.io backends' `X-CacheKit-*` telemetry
    /// headers, so for a backend used by this client alone, what you see here
    /// is what the SaaS sees (a backend shared by several clients reports the
    /// first client built over it — see [`Backend::attach_metrics`]).
    ///
    /// ```no_run
    /// # fn example(cache: &cachekit::CacheKit) {
    /// let stats = cache.stats();
    /// let reads = stats.l1_hits + stats.l2_hits + stats.misses;
    /// eprintln!("L1 hit rate {:.1}% over {reads} reads", stats.l1_hit_rate() * 100.0);
    /// # }
    /// ```
    pub fn stats(&self) -> L1Stats {
        self.counters.snapshot()
    }

    /// Number of entries currently held in L1, or `None` when L1 is disabled
    /// (or the `l1` feature is off).
    ///
    /// Exact at the time of the call: moka's pending expirations and
    /// evictions are applied first, which makes this a housekeeping call, not
    /// a free counter read — poll it, don't put it on a hot path.
    pub fn l1_entry_count(&self) -> Option<u64> {
        #[cfg(feature = "l1")]
        let count = self.l1.as_ref().map(crate::l1::L1Cache::entry_count);
        #[cfg(not(feature = "l1"))]
        let count = None;
        count
    }

    /// Current circuit-breaker state, or `None` when this client has no
    /// breaker (reliability disabled, or configured with
    /// `circuit_breaker: None`).
    ///
    /// Reads the live state: an open circuit whose `open_timeout` has elapsed
    /// reports [`CircuitState::HalfOpen`](crate::CircuitState::HalfOpen). With
    /// the `tracing` feature every transition is also emitted on the
    /// `cachekit::reliability` target — `warn` when the circuit opens, `info`
    /// for half-open and closed.
    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
    pub fn circuit_state(&self) -> Option<crate::reliability::CircuitState> {
        self.breaker.as_ref().map(|breaker| breaker.state())
    }

    // ── Single-flight ─────────────────────────────────────────────────────────

    /// Begin a cold-miss single-flight for `key` (see [`crate::flight`]).
    ///
    /// Call after a cache miss, before computing the value. Concurrent
    /// in-process fills of the same key are collapsed to one; with the
    /// `reliability` feature and a lock-capable backend (CachekitIO, Redis),
    /// fills are also suppressed across processes via a distributed fill
    /// lock. The `#[cachekit]` macro does this automatically.
    ///
    /// The key is namespaced like every cache operation but not validated —
    /// this call is infallible; an invalid key simply fails later at the
    /// actual cache operation.
    pub async fn single_flight(&self, key: &str) -> crate::flight::SingleFlight {
        let full_key = self.namespaced_key(key);
        crate::flight::SingleFlight::acquire(&self.flight, &self.backend, &full_key).await
    }

    /// Begin the single-flight for an SWR background refresh, or `None` when
    /// another worker already holds this key's in-process flight or
    /// distributed fill lock, or the lock call fails. Unlike
    /// [`Self::single_flight`] it never waits for another worker's fill.
    /// Macro plumbing for `#[cachekit]` — not public API.
    #[doc(hidden)]
    pub async fn __refresh_flight(&self, key: &str) -> Option<crate::flight::SingleFlight> {
        let full_key = self.namespaced_key(key);
        crate::flight::SingleFlight::try_lead(&self.flight, &self.backend, &full_key).await
    }

    // ── Secure cache ─────────────────────────────────────────────────────────

    /// Return a [`SecureCache`] handle that encrypts all values before storage.
    ///
    /// L1 stores **ciphertext** (not plaintext) to preserve the zero-knowledge
    /// property across all cache layers.
    ///
    /// The handle encrypts exactly what this client's own value methods
    /// already encrypt (see [Encryption](CacheKit#encryption)), and the two
    /// read each other's entries. What it adds is the guarantee: code that
    /// must never run unencrypted asks for the handle, which fails on a
    /// client without encryption, and it carries the
    /// [`previous_key_hits`](SecureCache::previous_key_hits) rotation signal.
    ///
    /// # Errors
    /// Returns `CachekitError::Config` if no encryption layer is configured.
    /// Configure encryption via [`CacheKitBuilder::encryption`] /
    /// [`CacheKitBuilder::encryption_from_bytes`], or start from the
    /// `CacheKit::secure(url, master_key_hex)` /
    /// `CacheKit::secure_from_env(url)` preset (`redis` feature).
    #[cfg(feature = "encryption")]
    pub fn secure_cache(&self) -> Result<SecureCache<'_>, CachekitError> {
        self.encrypting().ok_or_else(|| {
            CachekitError::Config(
                "encryption not configured: use the CacheKit::secure(url, master_key_hex) \
                 preset, CacheKit::secure_from_env(url) to read the hex key from \
                 CACHEKIT_MASTER_KEY, or call .encryption() on the builder"
                    .to_owned(),
            )
        })
    }

    /// The encrypting handle when encryption is configured. Every value
    /// method routes through it first, so a configured key leaves no
    /// plaintext path (protocol `spec/intent-presets.md` § Encryption
    /// Activation, rule 1).
    #[cfg(feature = "encryption")]
    fn encrypting(&self) -> Option<SecureCache<'_>> {
        self.encryption.as_deref().map(|encryption| SecureCache {
            client: self,
            encryption,
        })
    }

    // ── Private helpers ───────────────────────────────────────────────────────

    fn check_payload_size(&self, size: usize) -> Result<(), CachekitError> {
        if size > self.max_payload_bytes {
            return Err(CachekitError::PayloadTooLarge {
                size,
                limit: self.max_payload_bytes,
            });
        }
        Ok(())
    }
}

// ── SecureCache ──────────────────────────────────────────────────────────────

/// Encrypted cache handle returned by [`CacheKit::secure_cache()`].
///
/// All values are serialized, then encrypted with AES-256-GCM before storage.
/// L1 stores ciphertext to maintain zero-knowledge guarantees. The client's
/// own value methods route through these same methods once encryption is
/// configured, so the handle and the client read each other's entries.
#[cfg(feature = "encryption")]
pub struct SecureCache<'a> {
    client: &'a CacheKit,
    encryption: &'a crate::encryption::EncryptionLayer,
}

#[cfg(feature = "encryption")]
impl std::fmt::Debug for SecureCache<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SecureCache")
            .field("tenant_id", &self.encryption.tenant_id())
            .finish()
    }
}

#[cfg(feature = "encryption")]
impl SecureCache<'_> {
    /// Rotation drain signal; see
    /// [`EncryptionLayer::previous_key_hits`](crate::EncryptionLayer::previous_key_hits).
    pub fn previous_key_hits(&self) -> Vec<u64> {
        self.encryption.previous_key_hits()
    }

    /// Whether AES is hardware-accelerated on this host (informational); see
    /// [`EncryptionLayer::hardware_acceleration_enabled`](crate::EncryptionLayer::hardware_acceleration_enabled).
    pub fn hardware_acceleration_enabled(&self) -> bool {
        self.encryption.hardware_acceleration_enabled()
    }

    /// Encrypt and store `value` under `key` using the client's default TTL.
    pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
        self.set_with_ttl(key, value, self.client.default_ttl).await
    }

    /// Encrypt and store `value` under `key` with an explicit `ttl`.
    ///
    /// The client namespace is not checked here: an interop key written on a
    /// namespaced client lands under `{namespace}:{interop_key}`, which no
    /// interop reader addresses — see [`CacheKit::interop_get`].
    pub async fn set_with_ttl<T: Serialize>(
        &self,
        key: &str,
        value: &T,
        ttl: Duration,
    ) -> Result<(), CachekitError> {
        CacheKit::validate_ttl(ttl)?;
        let (full_key, ciphertext) = self.seal(key, value)?;
        // Size-check what is actually persisted (nonce + ciphertext + tag).
        // The get paths check the stored ciphertext length, so checking the
        // plaintext here would let a value within 28 bytes of the limit write
        // successfully and then fail EVERY subsequent read with PayloadTooLarge.
        self.client.check_payload_size(ciphertext.len())?;

        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        let mutation = self.client.lock_l1_mutation(&full_key).await;

        // Match the plain write path: invalidate older refresh tokens before
        // the first backend await, including on cancellation or failure.
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        if let Some(ref mutation) = mutation {
            mutation.advance();
        }

        // L1 needs its own copy: the backend consumes the Vec.
        #[cfg(feature = "l1")]
        {
            let l1_bytes = Bytes::copy_from_slice(&ciphertext);
            self.client
                .backend
                .set(&full_key, ciphertext, Some(ttl))
                .await?;
            self.client.l1_set(&full_key, l1_bytes, ttl);
        }
        #[cfg(not(feature = "l1"))]
        {
            self.client
                .backend
                .set(&full_key, ciphertext, Some(ttl))
                .await?;
        }

        crate::metrics::trace_write(&full_key, ttl);
        Ok(())
    }

    /// Commit an encrypted SWR refresh only if its stale-read token is still
    /// current. Macro plumbing; ordinary writes use [`Self::set_with_ttl`].
    #[doc(hidden)]
    pub async fn __complete_swr_refresh<T: Serialize>(
        &self,
        key: &str,
        value: &T,
        ttl: Duration,
        token: SwrToken,
    ) -> Result<bool, CachekitError> {
        let (_, ciphertext) = self.seal(key, value)?;
        self.client
            .complete_swr_bytes(key, ciphertext, ttl, token)
            .await
    }

    /// Serialize and encrypt `value` for `key`, returning the key this client
    /// passes to its backend (`.namespace()` prefix included) with the
    /// ciphertext. Every write encrypts here, so the AAD binds that key and an
    /// entry copied to another namespace fails to decrypt.
    fn seal<T: Serialize>(&self, key: &str, value: &T) -> Result<(String, Vec<u8>), CachekitError> {
        let full_key = self.client.resolve_key(key)?;
        let plaintext = serializer::serialize(value)?;
        let ciphertext = self.encryption.encrypt(&plaintext, &full_key)?;
        Ok((full_key, ciphertext))
    }

    /// Retrieve, decrypt, and deserialize a value stored under `key`.
    ///
    /// Checks L1 (which holds ciphertext) before the backend.
    ///
    /// # Errors
    ///
    /// Returns [`CachekitError::Encryption`] if the stored entry fails
    /// decryption. The key's L1 copy is dropped first and the backend entry
    /// is left in place, so the next read reaches the backend: a hit once the
    /// entry is replaced with ciphertext this client can decrypt, a miss once
    /// it expires or is deleted. A decrypted entry that does not decode as `T`
    /// is [`CachekitError::Serialization`] naming only `T`: the decoder's own
    /// message would quote the plaintext.
    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
        match self.get_plaintext(key).await? {
            Some(plaintext) => Ok(Some(redact_decode(serializer::deserialize(&plaintext))?)),
            None => Ok(None),
        }
    }

    /// Retrieve, decrypt, and deserialize an interop-mode value stored under `key`.
    ///
    /// Identical to [`Self::get`] except the decrypted plaintext is decoded
    /// with [`crate::interop::deserialize`] — exactly one MessagePack document,
    /// trailing bytes rejected (interop/v1 spec MUST), as in [`Self::get`]. In interop mode the
    /// AES-GCM plaintext is the plain MessagePack value bytes, so the AAD
    /// (v0x03, `format="msgpack"`, `compressed="False"`) verifies cross-SDK
    /// unchanged.
    ///
    /// # Errors
    ///
    /// Returns [`CachekitError::Config`] on a namespace-prefixed client — see
    /// [`CacheKit::interop_get`] — and [`CachekitError::Encryption`] if the
    /// stored entry fails decryption, with the key's L1 copy dropped as for
    /// [`Self::get`].
    pub async fn interop_get<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<Option<T>, CachekitError> {
        self.client.reject_namespaced_interop()?;
        match self.get_plaintext(key).await? {
            Some(plaintext) => Ok(Some(redact_decode(crate::interop::deserialize(
                &plaintext,
            ))?)),
            None => Ok(None),
        }
    }

    /// Retrieve, decrypt, and deserialize an interop-mode value with SWR
    /// classification. The secure twin of [`CacheKit::interop_get_swr`]:
    /// staleness is judged on the L1 **ciphertext** entry (zero-knowledge is
    /// preserved — freshness metadata never exposes plaintext), then the
    /// value is decrypted and decoded per [`Self::interop_get`].
    ///
    /// # Errors
    ///
    /// Same as [`Self::interop_get`] — the secure path fails closed on every
    /// backend and decryption error.
    pub async fn interop_get_swr<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<SwrRead<T>, CachekitError> {
        self.client.reject_namespaced_interop()?;
        match self.client.get_bytes_swr(key).await? {
            SwrRead::Fresh(ct) => Ok(SwrRead::Fresh(redact_decode(crate::interop::deserialize(
                &self.decrypt_or_evict(&ct, key)?,
            ))?)),
            SwrRead::Stale(ct, token) => Ok(SwrRead::Stale(
                redact_decode(crate::interop::deserialize(
                    &self.decrypt_or_evict(&ct, key)?,
                ))?,
                token,
            )),
            SwrRead::Miss => Ok(SwrRead::Miss),
        }
    }

    /// Fetch ciphertext (L1, then L2 with L1 backfill) and decrypt it.
    ///
    /// Ciphertext retrieval delegates to [`CacheKit::get_bytes`], which returns
    /// the stored bytes untransformed — for a secure cache exactly the AES-GCM
    /// ciphertext, so decrypt receives the same bytes the backend holds.
    async fn get_plaintext(&self, key: &str) -> Result<Option<Vec<u8>>, CachekitError> {
        match self.client.get_bytes(key).await? {
            Some(ciphertext) => Ok(Some(self.decrypt_or_evict(&ciphertext, key)?)),
            None => Ok(None),
        }
    }

    /// Decrypt ciphertext read for `key` against the key this client passes to
    /// its backend (`.namespace()` prefix included); on failure, drop the key's L1 copy before propagating the
    /// error.
    ///
    /// An L2 hit is backfilled into L1 before it is decrypted, so an entry
    /// that fails authentication would otherwise keep failing from L1 after
    /// the backend entry is fixed or deleted. The backend entry itself is
    /// left untouched as evidence.
    fn decrypt_or_evict(&self, ciphertext: &[u8], key: &str) -> Result<Vec<u8>, CachekitError> {
        let full_key = self.client.resolve_key(key)?;
        self.encryption
            .decrypt(ciphertext, &full_key)
            .inspect_err(|_| {
                #[cfg(feature = "l1")]
                self.client.l1_delete(&full_key);
            })
    }

    /// Delete an encrypted key. Behaves identically to [`CacheKit::delete`].
    pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
        self.client.delete(key).await
    }

    /// Check if an encrypted key exists. Behaves identically to [`CacheKit::exists`].
    pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
        self.client.exists(key).await
    }
}

/// Decoding decrypted plaintext keeps the `Serialization` class, which
/// `#[cachekit]` reads as a miss, but replaces the decoder's message: serde
/// quotes the offending value (`invalid type: string "…", expected struct
/// …`), and that value is the plaintext encryption exists to protect
/// (CWE-532).
#[cfg(feature = "encryption")]
fn redact_decode<T>(decoded: Result<T, CachekitError>) -> Result<T, CachekitError> {
    decoded.map_err(|e| match e {
        CachekitError::Serialization(_) => CachekitError::Serialization(format!(
            "decrypted entry does not decode as {}",
            std::any::type_name::<T>()
        )),
        other => other,
    })
}

// ── CacheKitBuilder ───────────────────────────────────────────────────────────

/// Fluent builder for [`CacheKit`].
#[derive(Default)]
#[must_use]
pub struct CacheKitBuilder {
    backend: Option<SharedBackend>,
    default_ttl: Option<Duration>,
    namespace: Option<String>,
    max_payload_bytes: Option<usize>,

    #[cfg(feature = "l1")]
    l1_capacity: Option<usize>,

    #[cfg(feature = "l1")]
    no_l1: bool,

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    swr_enabled: Option<bool>,

    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    swr_threshold_ratio: Option<f64>,

    // pub(crate): the `secure` preset's key-path tests read the layer.
    #[cfg(feature = "encryption")]
    pub(crate) encryption: Option<SharedEncryption>,

    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
    reliability: Option<crate::reliability::ReliabilityConfig>,

    /// Retry deadlines for the backend `CacheKit::io` built; cleared by
    /// [`Self::backend`] so a replaced backend never inherits them.
    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
    pub(crate) retry_deadlines: Option<crate::reliability::RetryDeadlines>,
}

#[cfg(not(feature = "encryption"))]
fn encryption_feature_missing(what: &str) -> CachekitError {
    CachekitError::Config(format!(
        "{what} needs the `encryption` cargo feature, which this build of \
         cachekit-rs was compiled without; enable it to encrypt"
    ))
}

impl CacheKitBuilder {
    /// Set the storage backend.
    pub fn backend(mut self, backend: SharedBackend) -> Self {
        self.backend = Some(backend);
        #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
        {
            self.retry_deadlines = None;
        }
        self
    }

    /// Override the default TTL (used when no per-call TTL is specified).
    pub fn default_ttl(mut self, ttl: Duration) -> Self {
        self.default_ttl = Some(ttl);
        self
    }

    /// Set a namespace prefix. All keys will be stored as `{namespace}:{key}`.
    pub fn namespace(mut self, ns: impl Into<String>) -> Self {
        self.namespace = Some(ns.into());
        self
    }

    /// Set the maximum accepted payload size in bytes.
    pub fn max_payload_bytes(mut self, limit: usize) -> Self {
        self.max_payload_bytes = Some(limit);
        self
    }

    /// Set the L1 cache capacity (max entries).
    #[cfg(feature = "l1")]
    pub fn l1_capacity(mut self, capacity: usize) -> Self {
        self.l1_capacity = Some(capacity);
        self
    }

    /// Disable the L1 cache entirely.
    #[cfg(feature = "l1")]
    pub fn no_l1(mut self) -> Self {
        self.no_l1 = true;
        self
    }

    /// Enable or disable L1 stale-while-revalidate (default: **enabled**,
    /// matching the Python and TypeScript SDKs).
    ///
    /// With SWR on, an L1 hit older than `swr_threshold_ratio` of its TTL is
    /// still served immediately, and the `#[cachekit]` macro schedules at
    /// most one background refresh (deduplicated on the
    /// [`CacheKit::single_flight`] locks, in-process and — on lock-capable
    /// backends — across processes). A refresh that finds either lock held,
    /// or whose lock call fails, stands down instead of waiting, and the
    /// stale value keeps being served. A hard-expired entry is never served:
    /// it falls through to a normal blocking miss.
    ///
    /// Native targets only: this knob does not exist on wasm32, under the
    /// `unsync` feature, or without `l1` — calling it there is a compile
    /// error rather than a silent no-op.
    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    pub fn swr_enabled(mut self, enabled: bool) -> Self {
        self.swr_enabled = Some(enabled);
        self
    }

    /// Set the SWR freshness threshold as a fraction of each L1 entry's TTL
    /// (default: **0.5**, matching the Python and TypeScript SDKs).
    ///
    /// An entry is *fresh* until it has lived `ratio × TTL` (±10% jitter,
    /// drawn once when the entry is inserted, to de-synchronise refreshes
    /// across processes), then *stale* — served immediately with a background
    /// refresh — until hard expiry. Mirrors cachekit-py's
    /// `swr_threshold_ratio` semantics (elapsed-lifetime fraction). Must be in
    /// `(0.0, 1.0]`; validated at [`Self::build`].
    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
    pub fn swr_threshold_ratio(mut self, ratio: f64) -> Self {
        self.swr_threshold_ratio = Some(ratio);
        self
    }

    // Stubs for when the l1 feature is disabled — still compile cleanly.
    /// No-op: the `l1` feature is disabled, so there is no L1 to size.
    #[cfg(not(feature = "l1"))]
    pub fn l1_capacity(self, _capacity: usize) -> Self {
        self
    }

    /// No-op: the `l1` feature is disabled, so L1 is already off.
    #[cfg(not(feature = "l1"))]
    pub fn no_l1(self) -> Self {
        self
    }

    /// Wrap the backend in the reliability stack (retry with exponential
    /// backoff + jitter, circuit breaker, backpressure) — see
    /// [`crate::reliability`].
    ///
    /// Enabled by default with production settings by the `production`,
    /// `secure`, and `io` intent presets; off for `minimal` and for
    /// manually-built clients. To opt a preset out, pass
    /// [`ReliabilityConfig::disabled()`](crate::reliability::ReliabilityConfig::disabled)
    /// — a disabled config applies no wrapping at all.
    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
    pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self {
        self.reliability = Some(config);
        self
    }

    /// Configure encryption from raw master key bytes and tenant ID.
    ///
    /// The master key must be exactly 32 raw bytes; anything else is a
    /// [`CachekitError::Config`]. Hex-encoded keys go to [`Self::encryption`]
    /// (or [`Self::encryption_with_previous`] to rotate) — their ASCII bytes
    /// are 64 bytes long and rejected here.
    /// Keys are derived per-tenant via HKDF-SHA256.
    ///
    /// The built client encrypts every value read and write, not only those
    /// through the [`secure_cache()`](CacheKit::secure_cache) handle; see
    /// [Encryption](CacheKit#encryption). Calling any encryption method
    /// again replaces the earlier configuration.
    ///
    /// Without the `encryption` cargo feature this method, and every other
    /// builder encryption method, returns [`CachekitError::Config`]: a build
    /// that cannot encrypt never accepts a key.
    #[cfg(feature = "encryption")]
    pub fn encryption_from_bytes(
        mut self,
        master_key: &[u8],
        tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        let layer = crate::encryption::EncryptionLayer::new(master_key, tenant_id)?;
        self.encryption = Some(SharedEncryption::new(layer));
        Ok(self)
    }

    /// Configure encryption with decrypt-only previous master keys for
    /// key rotation.
    ///
    /// Writes encrypt under `master_key`; reads attempt it first, then each
    /// key in `previous_keys` sequentially (attempt order = slice order).
    /// At most 3 previous keys; supplying more is a config error, never
    /// truncated. See [`crate::encryption::EncryptionLayer::with_previous_keys`].
    ///
    /// Every key, current and previous, must be exactly 32 raw bytes. For
    /// hex keys, use the hex counterpart, [`Self::encryption_with_previous`].
    ///
    /// Encrypts every value read and write, as for
    /// [`Self::encryption_from_bytes`].
    #[cfg(feature = "encryption")]
    pub fn encryption_from_bytes_with_previous(
        mut self,
        master_key: &[u8],
        previous_keys: &[&[u8]],
        tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        let layer = crate::encryption::EncryptionLayer::with_previous_keys(
            master_key,
            previous_keys,
            tenant_id,
        )?;
        self.encryption = Some(SharedEncryption::new(layer));
        Ok(self)
    }

    /// Configure encryption from a hex-encoded master key string.
    ///
    /// Hex-decodes with the same decoder as [`CacheKit::from_env`] and the
    /// `secure` preset. The key must decode to at least 32 bytes; use exactly
    /// 32 (64 hex chars), the only length every SDK accepts.
    ///
    /// Encrypts every value read and write, as for
    /// [`Self::encryption_from_bytes`].
    #[cfg(feature = "encryption")]
    pub fn encryption(self, hex_key: &str, tenant_id: &str) -> Result<Self, CachekitError> {
        let bytes = crate::config::decode_master_key_hex(hex_key, "master key")?;
        self.encryption_from_hex_decoded(&bytes, &[], tenant_id)
    }

    /// Configure encryption from hex-encoded keys, with decrypt-only previous
    /// master keys for key rotation.
    ///
    /// The hex counterpart of [`Self::encryption_from_bytes_with_previous`].
    /// Every key, current and previous, is decoded as for [`Self::encryption`]
    /// and must decode to at least 32 bytes, so a key longer than 32 bytes
    /// that [`Self::encryption`] accepted can be listed as a previous key
    /// under the tenant it was written for (the env variables always derive
    /// for tenant `"default"`). Use exactly 32 bytes (64 hex chars) for new
    /// keys, the only length every SDK accepts.
    ///
    /// Writes encrypt under `hex_key`; reads attempt it first, then each key
    /// in `previous_hex_keys` sequentially (attempt order = slice order).
    /// At most 3 previous keys, and the current key must not be among them;
    /// either violation is a [`CachekitError::Config`], never truncated.
    ///
    /// ```no_run
    /// # fn demo(backend: cachekit::client::SharedBackend, k2_hex: &str, k1_hex: &str)
    /// #     -> Result<(), cachekit::CachekitError> {
    /// let cache = cachekit::CacheKit::builder()
    ///     .backend(backend)
    ///     .encryption_with_previous(k2_hex, &[k1_hex], "tenant-a")?
    ///     .build()?;
    /// # Ok(()) }
    /// ```
    ///
    /// Encrypts every value read and write, as for
    /// [`Self::encryption_from_bytes`].
    #[cfg(feature = "encryption")]
    pub fn encryption_with_previous(
        self,
        hex_key: &str,
        previous_hex_keys: &[&str],
        tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        let current = crate::config::decode_master_key_hex(hex_key, "master key")?;
        let previous = previous_hex_keys
            .iter()
            .enumerate()
            .map(|(i, key)| {
                crate::config::decode_master_key_hex(key, &format!("previous master key {i}"))
            })
            .collect::<Result<Vec<_>, _>>()?;
        let previous: Vec<&[u8]> = previous.iter().map(|key| key.as_slice()).collect();
        self.encryption_from_hex_decoded(&current, &previous, tenant_id)
    }

    /// Shared tail of the hex paths: keys already validated by
    /// `decode_master_key_hex`, so the `>= 32` floor applies, not the
    /// raw-bytes exactly-32 rule.
    #[cfg(feature = "encryption")]
    pub(crate) fn encryption_from_hex_decoded(
        mut self,
        master_key: &[u8],
        previous_keys: &[&[u8]],
        tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        let layer = crate::encryption::EncryptionLayer::from_hex_decoded_keys(
            master_key,
            previous_keys,
            tenant_id,
        )?;
        self.encryption = Some(SharedEncryption::new(layer));
        Ok(self)
    }

    // Without the `encryption` feature the builder methods still exist, so
    // feature-agnostic code compiles, but each one is an error: accepting a
    // key and building a plaintext client is the silent downgrade protocol
    // `spec/intent-presets.md` § Encryption Activation rule 1 forbids.

    /// Always [`CachekitError::Config`]: this build has no `encryption`
    /// feature.
    #[cfg(not(feature = "encryption"))]
    pub fn encryption_from_bytes(
        self,
        _master_key: &[u8],
        _tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        Err(encryption_feature_missing(".encryption_from_bytes()"))
    }

    /// Always [`CachekitError::Config`]: this build has no `encryption`
    /// feature.
    #[cfg(not(feature = "encryption"))]
    pub fn encryption_from_bytes_with_previous(
        self,
        _master_key: &[u8],
        _previous_keys: &[&[u8]],
        _tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        Err(encryption_feature_missing(
            ".encryption_from_bytes_with_previous()",
        ))
    }

    /// Always [`CachekitError::Config`]: this build has no `encryption`
    /// feature.
    #[cfg(not(feature = "encryption"))]
    pub fn encryption(self, _hex_key: &str, _tenant_id: &str) -> Result<Self, CachekitError> {
        Err(encryption_feature_missing(".encryption()"))
    }

    /// Always [`CachekitError::Config`]: this build has no `encryption`
    /// feature.
    #[cfg(not(feature = "encryption"))]
    pub fn encryption_with_previous(
        self,
        _hex_key: &str,
        _previous_hex_keys: &[&str],
        _tenant_id: &str,
    ) -> Result<Self, CachekitError> {
        Err(encryption_feature_missing(".encryption_with_previous()"))
    }

    /// Finalise and build the [`CacheKit`] client.
    ///
    /// Returns an error if no backend was provided.
    pub fn build(self) -> Result<CacheKit, CachekitError> {
        let backend = self.backend.ok_or_else(|| {
            CachekitError::Config("a backend must be provided via .backend()".to_owned())
        })?;

        // Validate namespace if provided
        if let Some(ref ns) = self.namespace {
            if ns.is_empty() {
                return Err(CachekitError::Config("namespace cannot be empty".into()));
            }
            if ns.len() > 255 {
                return Err(CachekitError::Config("namespace exceeds 255 bytes".into()));
            }
            if !ns.bytes().all(|b| (0x20..=0x7E).contains(&b)) {
                return Err(CachekitError::Config(
                    "namespace must be ASCII printable".into(),
                ));
            }
        }

        #[cfg(feature = "l1")]
        let l1 = if self.no_l1 {
            None
        } else {
            let capacity = self.l1_capacity.unwrap_or(1000);
            Some(crate::l1::L1Cache::new(capacity))
        };

        // SWR defaults mirror the sibling SDKs: enabled, threshold ratio 0.5,
        // ratio validated in (0.0, 1.0] exactly like py's L1CacheConfig.
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        let swr_threshold_ratio = {
            let ratio = self.swr_threshold_ratio.unwrap_or(0.5);
            if !(ratio > 0.0 && ratio <= 1.0) {
                return Err(CachekitError::Config(format!(
                    "swr_threshold_ratio must be in (0.0, 1.0]; got {ratio}"
                )));
            }
            ratio
        };

        // Hand the backend a live view of this client's read counters so the
        // SaaS telemetry headers report real numbers with no user plumbing.
        // Attached to the raw backend, before the reliability decorator, so
        // no forwarding is needed; a provider set on the backend's own
        // builder wins (see `Backend::attach_metrics`).
        #[cfg(feature = "l1")]
        let l1_enabled = l1.is_some();
        #[cfg(not(feature = "l1"))]
        let l1_enabled = false;
        let counters = Arc::new(CacheCounters::new(l1_enabled));
        backend.attach_metrics(counters.provider());

        // Apply the reliability stack last so it decorates the final backend.
        // A disabled config is the documented opt-out: skip the (no-op)
        // decorator entirely. The layer check lives on ReliabilityConfig
        // itself so a future layer can't be missed here (this gate
        // shipped that exact bug once already).
        #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
        let (backend, breaker) = match self.reliability {
            Some(config) if !config.is_disabled() => {
                crate::reliability::wrap_reliable(backend, config, self.retry_deadlines)
            }
            _ => (backend, None),
        };

        Ok(CacheKit {
            backend,
            default_ttl: self.default_ttl.unwrap_or(Duration::from_secs(300)),
            namespace: self.namespace,
            max_payload_bytes: self.max_payload_bytes.unwrap_or(5 * 1024 * 1024),
            flight: SharedFlight::default(),
            counters,

            #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
            breaker,

            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
            mutations: SharedMutations::default(),

            #[cfg(feature = "l1")]
            l1,

            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
            swr_enabled: self.swr_enabled.unwrap_or(true),

            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
            swr_threshold_ratio,

            #[cfg(feature = "encryption")]
            encryption: self.encryption,
        })
    }
}

#[cfg(all(test, feature = "l1"))]
mod backfill_ttl_tests {
    use std::time::Duration;

    use super::l1_backfill_ttl;
    use crate::backend::Freshness;

    const DEFAULT_TTL: Duration = Duration::from_secs(300);

    fn fresh_for(secs: u64) -> Freshness {
        Freshness {
            is_stale: false,
            fresh_for: Some(Duration::from_secs(secs)),
        }
    }

    /// (e) No header: today's `min(default_ttl, 30 s)`, unchanged.
    #[test]
    fn absent_header_keeps_the_30s_cap() {
        let absent = Freshness::default();
        assert_eq!(
            l1_backfill_ttl(DEFAULT_TTL, absent),
            Some(Duration::from_secs(30))
        );
        assert_eq!(
            l1_backfill_ttl(Duration::from_secs(10), absent),
            Some(Duration::from_secs(10))
        );
    }

    #[test]
    fn fresh_for_bounds_but_never_lifts_the_cap() {
        assert_eq!(
            l1_backfill_ttl(DEFAULT_TTL, fresh_for(5)),
            Some(Duration::from_secs(5))
        );
        assert_eq!(
            l1_backfill_ttl(DEFAULT_TTL, fresh_for(3600)),
            Some(Duration::from_secs(30))
        );
    }

    #[test]
    fn stale_or_zero_forbids_the_backfill() {
        assert_eq!(l1_backfill_ttl(DEFAULT_TTL, fresh_for(0)), None);
        let stale = |fresh_for| Freshness {
            is_stale: true,
            fresh_for,
        };
        assert_eq!(l1_backfill_ttl(DEFAULT_TTL, stale(None)), None);
        assert_eq!(
            l1_backfill_ttl(DEFAULT_TTL, stale(Some(Duration::from_secs(60)))),
            None
        );
    }
}

// SWR staleness exists only on native builds with L1 (see tests/swr_tests.rs).
#[cfg(all(
    test,
    feature = "encryption",
    feature = "l1",
    not(feature = "unsync"),
    not(target_arch = "wasm32")
))]
#[allow(clippy::unwrap_used)] // test-only: a failed setup step should panic loudly
mod secure_l1_tests {
    use std::time::Duration;

    use super::{CacheKit, SharedBackend, SwrRead};
    use crate::backend::MemoryBackend;
    use crate::error::CachekitError;

    /// The Stale arm reads straight from L1, and every public write on an
    /// encrypted client stores decryptable ciphertext there, so the
    /// undecryptable bytes are planted in L1 directly.
    #[tokio::test]
    async fn secure_interop_get_swr_evicts_stale_l1_after_decrypt_failure() {
        let backend = MemoryBackend::default();
        let shared: SharedBackend = std::sync::Arc::new(backend.clone());
        let client = CacheKit::builder()
            .backend(shared)
            .l1_capacity(100)
            .swr_threshold_ratio(0.001)
            .encryption_from_bytes(&[7u8; 32], "test-tenant")
            .unwrap()
            .build()
            .unwrap();
        client.l1_set(
            "poisoned-stale",
            bytes::Bytes::from_static(b"not ciphertext"),
            Duration::from_secs(60),
        );
        // Real time, not tokio's paused clock: L1 ages entries by std
        // Instant. threshold = 0.001 × 60 s = 60 ms (±10%); hard expiry 60 s.
        tokio::time::sleep(Duration::from_millis(200)).await;

        let secure = client.secure_cache().unwrap();
        let err = secure
            .interop_get_swr::<String>("poisoned-stale")
            .await
            .unwrap_err();
        assert!(matches!(err, CachekitError::Encryption(_)), "got: {err:?}");
        // Served from L1, so the entry had not hard-expired into the L2 path.
        let stats = client.stats();
        assert_eq!((stats.l1_hits, stats.l2_hits), (1, 0), "{stats:?}");

        // Nothing in the backend: a surviving L1 copy would keep failing; an
        // evicted one yields a miss.
        assert!(backend.store.lock().await.is_empty());
        assert!(matches!(
            secure.interop_get_swr::<String>("poisoned-stale").await,
            Ok(SwrRead::Miss)
        ));
    }
}