cachekit/

client.rs

use std::borrow::Cow;
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 L1 never serves an entry past 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()
        )));
    }
    if !has_control_byte(key.as_bytes()) {
        return Ok(());
    }
    // Rejection only: find the first offending byte to name it.
    key.bytes()
        .find(|&b| is_control_byte(b))
        .map_or(Ok(()), |b| {
            Err(CachekitError::InvalidKey(format!(
                "key contains illegal control character 0x{b:02X}"
            )))
        })
}

/// A byte no key may hold: a control character below 0x20, or DEL.
const fn is_control_byte(b: u8) -> bool {
    (b < 0x20) | (b == 0x7F)
}

/// Whether any byte of `bytes` is a control byte.
///
/// Every validated key passes through here, so it reads 16 bytes at a time:
/// a block's compares are folded with no early exit, which lets the compiler
/// vectorise them where a byte loop spends a branch per byte. When the length
/// is not a multiple of 16, the last block overlaps the one before it instead
/// of leaving a byte-at-a-time tail.
fn has_control_byte(bytes: &[u8]) -> bool {
    fn any_control(block: &[u8]) -> bool {
        block
            .iter()
            .fold(false, |found, &b| found | is_control_byte(b))
    }
    if bytes.len() < 16 {
        return any_control(bytes);
    }
    let (blocks, _) = bytes.as_chunks::<16>();
    blocks.iter().any(|block| any_control(block)) || any_control(&bytes[bytes.len() - 16..])
}

// ── 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, such as
/// [`encryption_from_env`](CacheKitBuilder::encryption_from_env), or the
/// `secure` / `secure_from_env` preset) 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.
    ///
    /// `CACHEKIT_MASTER_KEY` does not make the client encrypt: a key in the
    /// environment is a key source, never a switch (protocol
    /// `spec/intent-presets.md` § Encryption Activation, rule 2). State the
    /// intent instead — encrypt with the environment's keys, or don't:
    ///
    /// ```no_run
    /// # fn demo() -> Result<(), cachekit::CachekitError> {
    /// let encrypted = cachekit::CacheKit::from_env()?
    ///     .encryption_from_env()?
    ///     .build()?;
    /// let plaintext = cachekit::CacheKit::from_env()?.no_encryption().build()?;
    /// # Ok(()) }
    /// ```
    ///
    /// Releases up to 0.11 encrypted whenever the variable was set. So that
    /// such a deployment cannot silently start writing plaintext, a builder
    /// from `from_env()` that saw `CACHEKIT_MASTER_KEY` and states neither
    /// intent fails at [`CacheKitBuilder::build`] with
    /// [`CachekitError::Config`]. This check is transitional and goes away
    /// in a later release; without the variable, `from_env()?.build()?`
    /// builds a plaintext client.
    ///
    /// A set key is validated here: one that is empty, not hex, shorter than
    /// 32 bytes or not UTF-8 is a [`CachekitError::Config`].
    ///
    /// The default TTL is 300 s, and no environment variable changes it. Chain
    /// [`CacheKitBuilder::default_ttl`] on the returned builder, or pass a TTL
    /// per call with [`CacheKit::set_with_ttl`].
    #[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()?;

        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 {
            builder = builder.namespace(ns);
        }
        builder.env_master_key = config.master_key.is_some();

        Ok(builder)
    }

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

    /// `key` behind the client's namespace prefix. Without a namespace the
    /// storage key is `key` itself, borrowed: an L1 hit then allocates
    /// nothing for its key.
    fn namespaced_key<'k>(&self, key: &'k str) -> Cow<'k, str> {
        match &self.namespace {
            Some(ns) => Cow::Owned(format!("{ns}:{key}")),
            None => Cow::Borrowed(key),
        }
    }

    /// Validate key and return the namespaced version.
    fn resolve_key<'k>(&self, key: &'k str) -> Result<Cow<'k, str>, 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 L1 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.
    ///
    /// Values are MessagePack. A `Vec<u8>` field written by [`Self::set`]
    /// as a plain `Vec<u8>` decodes back here unchanged, but it is stored as
    /// an array of integers; see [`Self::set`] for the compact form.
    ///
    /// # 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` — L1
    /// never serves an entry past its 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 and count the read:
    /// one outcome, by the layer that served it, or one miss.
    async fn get_bytes_swr(&self, key: &str) -> Result<SwrRead<Bytes>, CachekitError> {
        let full_key = self.resolve_key(key)?;
        Ok(match self.read_bytes_swr(&full_key).await? {
            SwrRead::Fresh((bytes, by)) => {
                self.counters.record(by, &full_key);
                SwrRead::Fresh(bytes)
            }
            SwrRead::Stale((bytes, by), token) => {
                self.counters.record(by, &full_key);
                SwrRead::Stale(bytes, token)
            }
            SwrRead::Miss => {
                self.counters.record(ReadOutcome::Miss, &full_key);
                SwrRead::Miss
            }
        })
    }

    /// Fetch raw payload bytes for `key` (L1, then L2 with L1 backfill) and
    /// count the read: one outcome, by the layer that served it, or one miss.
    async fn get_bytes(&self, key: &str) -> Result<Option<Bytes>, CachekitError> {
        let full_key = self.resolve_key(key)?;
        let read = self.read_bytes(&full_key).await?;
        let by = read.as_ref().map_or(ReadOutcome::Miss, |(_, by)| *by);
        self.counters.record(by, &full_key);
        Ok(read.map(|(bytes, _)| bytes))
    }

    /// The uncounted read behind [`Self::get_bytes_swr`]: an L1 hit is split
    /// into fresh vs stale against the configured freshness window; on L1
    /// miss this defers to [`Self::read_bytes`] (L2 + backfill), whose hit is
    /// always fresh. Each value comes with the layer that served it.
    async fn read_bytes_swr(
        &self,
        full_key: &str,
    ) -> Result<SwrRead<(Bytes, ReadOutcome)>, CachekitError> {
        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
        if self.swr_enabled {
            if let Some(ref l1) = self.l1 {
                match l1.get_with_swr_shared(full_key, self.swr_threshold_ratio) {
                    crate::l1::SharedSwrRead::Fresh(bytes) => {
                        self.check_payload_size(bytes.len())?;
                        return Ok(SwrRead::Fresh((bytes, ReadOutcome::L1Hit)));
                    }
                    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.check_payload_size(bytes.len())?;
                                return Ok(SwrRead::Fresh((bytes, ReadOutcome::L1Hit)));
                            }
                            crate::l1::SharedSwrRead::Stale(bytes) => {
                                self.check_payload_size(bytes.len())?;
                                let (state, version) = mutation.snapshot();
                                return Ok(SwrRead::Stale(
                                    (bytes, ReadOutcome::L1Stale),
                                    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.read_bytes(full_key).await? {
            Some(read) => SwrRead::Fresh(read),
            None => SwrRead::Miss,
        })
    }

    /// The uncounted read behind [`Self::get_bytes`]: L1, then L2 with L1
    /// backfill, with the layer that served the bytes.
    async fn read_bytes(
        &self,
        full_key: &str,
    ) -> Result<Option<(Bytes, ReadOutcome)>, CachekitError> {
        // L1 hit
        #[cfg(feature = "l1")]
        if let Some(bytes) = self.l1_get(full_key) {
            self.check_payload_size(bytes.len())?;
            return Ok(Some((bytes, ReadOutcome::L1Hit)));
        }

        // 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.check_payload_size(bytes.len())?;
            return Ok(Some((bytes, ReadOutcome::L1Hit)));
        }

        // L2 backend
        let Some((bytes, freshness)) = self.backend.get_with_freshness(full_key).await? else {
            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, ReadOutcome::L2Hit)))
    }

    /// 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`].
    ///
    /// # Binary payloads
    ///
    /// Values are serialized as MessagePack through serde. serde treats a
    /// plain `Vec<u8>` (or `&[u8]`) as a sequence, so it is stored as a
    /// MessagePack array of integers: up to 2 bytes per byte, and a per-element
    /// encode and decode. Mark the field with `serde_bytes` to store it as a
    /// single MessagePack `bin` value instead. `serde_bytes` is a crate you add
    /// to your own `Cargo.toml`:
    ///
    /// ```ignore
    /// #[derive(serde::Serialize, serde::Deserialize)]
    /// struct Response {
    ///     status: u16,
    ///     #[serde(with = "serde_bytes")]
    ///     body: Vec<u8>,
    /// }
    ///
    /// cache.set("resp", &Response { status: 200, body }).await?;
    /// ```
    ///
    /// For a bare byte value, store a `serde_bytes::ByteBuf`. Either form
    /// decodes into either type on [`Self::get`], so switching a field to
    /// `serde_bytes` still reads entries written before the switch.
    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 — counts one outcome: a hit by the layer
    /// that served it, or a miss when nothing was found. A `#[cachekit]` call
    /// also counts one outcome: a hit by the layer that served its value, or
    /// one miss when it runs the function. The reads inside that call (its
    /// first read, single-flight re-polls) count nothing themselves.
    /// `exists`, and reads that fail at the backend or on an oversized
    /// payload, are not counted. A plain read whose served entry then fails to
    /// decode or decrypt counts the layer that served it. 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: expired entries are swept out first,
    /// a pass over every entry that 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
    }

    /// [`Self::interop_get`] for a `#[cachekit]` call: the value with the
    /// layer that served it, counted by nobody. The call's own reads (first
    /// read, single-flight re-polls) count nothing; the call counts its one
    /// outcome with [`Self::__count`]. Macro plumbing for `#[cachekit]` — not
    /// public API.
    #[doc(hidden)]
    pub async fn __interop_read<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<Option<(T, ReadOutcome)>, CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.__interop_read(key).await;
        }
        self.reject_namespaced_interop()?;
        let full_key = self.resolve_key(key)?;
        match self.read_bytes(&full_key).await? {
            Some((bytes, by)) => Ok(Some((crate::interop::deserialize(&bytes)?, by))),
            None => Ok(None),
        }
    }

    /// [`Self::interop_get_swr`] for a `#[cachekit]` call, uncounted like
    /// [`Self::__interop_read`]. Macro plumbing for `#[cachekit]` — not
    /// public API.
    #[doc(hidden)]
    pub async fn __interop_read_swr<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<SwrRead<(T, ReadOutcome)>, CachekitError> {
        #[cfg(feature = "encryption")]
        if let Some(secure) = self.encrypting() {
            return secure.__interop_read_swr(key).await;
        }
        self.reject_namespaced_interop()?;
        let full_key = self.resolve_key(key)?;
        Ok(match self.read_bytes_swr(&full_key).await? {
            SwrRead::Fresh((bytes, by)) => {
                SwrRead::Fresh((crate::interop::deserialize(&bytes)?, by))
            }
            SwrRead::Stale((bytes, by), token) => {
                SwrRead::Stale((crate::interop::deserialize(&bytes)?, by), token)
            }
            SwrRead::Miss => SwrRead::Miss,
        })
    }

    /// Count a `#[cachekit]` call's one outcome: the layer that served its
    /// value, or [`ReadOutcome::Miss`] when it runs the function. Macro
    /// plumbing for `#[cachekit]` — not public API.
    #[doc(hidden)]
    pub fn __count(&self, key: &str, outcome: ReadOutcome) {
        self.counters.record(outcome, &self.namespaced_key(key));
    }

    // ── 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_env`] /
    /// [`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 \
                 (.encryption_from_env() to read CACHEKIT_MASTER_KEY)"
                    .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.
///
/// The handle borrows its [`CacheKit`] (`SecureCache<'a>` holds `&'a CacheKit`),
/// so it cannot be stored in the same struct as the client it came from.
/// Store the `CacheKit` and call [`CacheKit::secure_cache()`] where you need
/// the handle: it only checks that encryption is configured and copies two
/// references, so calling it per operation is cheap.
#[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`.
    ///
    /// 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 [`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.into_owned(), 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),
        }
    }

    /// The secure twin of [`CacheKit::__interop_read`], which hands an
    /// encrypting client's reads here.
    async fn __interop_read<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<Option<(T, ReadOutcome)>, CachekitError> {
        self.client.reject_namespaced_interop()?;
        let full_key = self.client.resolve_key(key)?;
        match self.client.read_bytes(&full_key).await? {
            Some((ct, by)) => Ok(Some((
                redact_decode(crate::interop::deserialize(
                    &self.decrypt_or_evict(&ct, key)?,
                ))?,
                by,
            ))),
            None => Ok(None),
        }
    }

    /// The secure twin of [`CacheKit::__interop_read_swr`], which hands an
    /// encrypting client's reads here.
    async fn __interop_read_swr<T: DeserializeOwned>(
        &self,
        key: &str,
    ) -> Result<SwrRead<(T, ReadOutcome)>, CachekitError> {
        self.client.reject_namespaced_interop()?;
        let full_key = self.client.resolve_key(key)?;
        let decode = |ct: &[u8]| -> Result<T, CachekitError> {
            redact_decode(crate::interop::deserialize(
                &self.decrypt_or_evict(ct, key)?,
            ))
        };
        Ok(match self.client.read_bytes_swr(&full_key).await? {
            SwrRead::Fresh((ct, by)) => SwrRead::Fresh((decode(&ct)?, by)),
            SwrRead::Stale((ct, by), token) => SwrRead::Stale((decode(&ct)?, by), token),
            SwrRead::Miss => 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>,

    /// [`Self::no_encryption`] was called.
    no_encryption: bool,

    /// `CacheKit::from_env` saw `CACHEKIT_MASTER_KEY`: [`Self::build`]
    /// refuses until an encryption call or [`Self::no_encryption`] states
    /// the intent (the transitional gate documented on `from_env`).
    env_master_key: bool,

    #[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)
    }

    /// Configure encryption from the environment: the hex key in
    /// `CACHEKIT_MASTER_KEY`, plus any decrypt-only rotation keys in
    /// `CACHEKIT_PREVIOUS_MASTER_KEYS`, for tenant `"default"`.
    ///
    /// The explicit call that turns a `CACHEKIT_MASTER_KEY` into encryption:
    /// no constructor, [`CacheKit::from_env`] included, encrypts because the
    /// variable is set. Keys are decoded and validated as for
    /// [`Self::encryption_with_previous`] and [`CacheKit::from_env`].
    ///
    /// ```no_run
    /// # fn demo() -> Result<(), cachekit::CachekitError> {
    /// let cache = cachekit::CacheKit::from_env()?
    ///     .encryption_from_env()?
    ///     .build()?;
    /// # Ok(()) }
    /// ```
    ///
    /// Encrypts every value read and write, as for
    /// [`Self::encryption_from_bytes`].
    ///
    /// # Errors
    ///
    /// [`CachekitError::Config`] when `CACHEKIT_MASTER_KEY` is unset, empty,
    /// not UTF-8, not hex or shorter than 32 bytes, or when
    /// `CACHEKIT_PREVIOUS_MASTER_KEYS` is malformed: asked to encrypt, the
    /// builder never falls back to plaintext.
    #[cfg(feature = "encryption")]
    pub fn encryption_from_env(self) -> Result<Self, CachekitError> {
        let (master_key, previous) = crate::config::master_keys_from_env(
            ".encryption_with_previous(hex_key, previous_hex_keys, \"default\")",
        )?;
        let previous: Vec<&[u8]> = previous.iter().map(|key| key.as_slice()).collect();
        self.encryption_from_hex_decoded(&master_key, &previous, "default")
    }

    /// 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()"))
    }

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

    /// State that this client does not encrypt, even with
    /// `CACHEKIT_MASTER_KEY` set: the explicit opt-out of protocol
    /// `spec/intent-presets.md` § Encryption Activation, rule 3.
    ///
    /// No client encrypts without an encryption call, so this changes
    /// nothing on its own. It answers the transitional check on a builder
    /// from [`CacheKit::from_env`] that saw `CACHEKIT_MASTER_KEY`, and it
    /// records the intent at the call site. Combined with any encryption
    /// call, or the `secure` preset, [`Self::build`] returns
    /// [`CachekitError::Config`]: an opt-out never quietly undoes one.
    pub fn no_encryption(mut self) -> Self {
        self.no_encryption = true;
        self
    }

    /// Finalise and build the [`CacheKit`] client.
    ///
    /// Returns an error if no backend was provided, if
    /// [`Self::no_encryption`] was combined with an encryption call, or if
    /// the builder came from [`CacheKit::from_env`] with
    /// `CACHEKIT_MASTER_KEY` set and states no encryption intent.
    pub fn build(self) -> Result<CacheKit, CachekitError> {
        #[cfg(feature = "encryption")]
        let encrypting = self.encryption.is_some();
        #[cfg(not(feature = "encryption"))]
        let encrypting = false;
        if self.no_encryption && encrypting {
            return Err(CachekitError::Config(
                ".no_encryption() conflicts with an encryption call (or the secure preset) \
                 on the same builder: keep one"
                    .to_owned(),
            ));
        }
        if self.env_master_key && !encrypting && !self.no_encryption {
            #[cfg(feature = "encryption")]
            let msg = "CACHEKIT_MASTER_KEY is set, and CacheKit::from_env() no longer encrypts \
                       because of it: chain .encryption_from_env() to encrypt with it, or \
                       .no_encryption() to build a plaintext client";
            #[cfg(not(feature = "encryption"))]
            let msg = "CACHEKIT_MASTER_KEY is set, but this build of cachekit-rs has no \
                       `encryption` cargo feature and cannot encrypt with it: chain \
                       .no_encryption() to build a plaintext client, or enable the feature \
                       to encrypt";
            return Err(CachekitError::Config(msg.to_owned()));
        }

        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(test)]
mod validate_key_tests {
    use super::{validate_key, MAX_KEY_BYTES};
    use crate::error::CachekitError;

    /// The byte loop `validate_key` ran before it checked 16 bytes at a time:
    /// the behaviour it must keep, error text included.
    fn byte_loop(key: &str) -> Result<(), String> {
        if key.is_empty() {
            return Err("key must not be empty".to_owned());
        }
        if key.len() > MAX_KEY_BYTES {
            return Err(format!(
                "key is {} bytes (limit: {MAX_KEY_BYTES})",
                key.len()
            ));
        }
        for b in key.bytes() {
            if b < 0x20 || b == 0x7F {
                return Err(format!("key contains illegal control character 0x{b:02X}"));
            }
        }
        Ok(())
    }

    fn check(key: &str) {
        let got = validate_key(key).map_err(|e| match e {
            CachekitError::InvalidKey(msg) => msg,
            other => format!("not InvalidKey: {other:?}"),
        });
        assert_eq!(got, byte_loop(key), "key {key:?}");
    }

    /// Every ASCII character, plus a 2-, 3- and 4-byte character (whose bytes
    /// are all 0x80 or more), at every position of every key length from 0
    /// to 48: every block boundary and overlap of the 16-byte check.
    #[test]
    fn same_verdict_and_message_as_the_byte_loop() {
        let fillers: Vec<char> = (0u8..0x80)
            .map(char::from)
            .chain(['é', '€', '😀'])
            .collect();
        for len in 0..=48 {
            for pos in 0..len {
                for &c in &fillers {
                    let mut key: String = "k".repeat(pos);
                    key.push(c);
                    key.push_str(&"k".repeat(len - pos - 1));
                    check(&key);
                }
            }
            check(&"k".repeat(len));
        }
    }

    /// With two offending bytes the message names the first, wherever the
    /// two fall relative to the blocks.
    #[test]
    fn names_the_first_offending_byte() {
        for len in 2..=48 {
            for first in 0..len {
                for second in first + 1..len {
                    let mut key = vec![b'k'; len];
                    key[first] = 0x7F;
                    key[second] = 0x01;
                    check(&String::from_utf8_lossy(&key));
                }
            }
        }
    }

    /// Near the length limit, where the last block overlaps the most.
    #[test]
    fn long_keys_around_the_limit() {
        for len in MAX_KEY_BYTES - 40..=MAX_KEY_BYTES + 1 {
            for pos in (0..len).filter(|p| *p < 40 || *p % 16 < 2 || *p % 16 > 13 || len - p < 40) {
                for c in ['\u{0}', '\u{1F}', ' ', '~', '\u{7F}', 'é'] {
                    let mut key: String = "k".repeat(pos);
                    key.push(c);
                    key.push_str(&"k".repeat(len - pos - 1));
                    check(&key);
                }
            }
        }
    }
}

#[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)
        ));
    }
}