Skip to main content

cachekit/
client.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use serde::{de::DeserializeOwned, Serialize};
5
6use crate::backend::Backend;
7#[cfg(feature = "l1")]
8use crate::backend::Freshness;
9use crate::error::CachekitError;
10use crate::metrics::{CacheCounters, L1Stats, ReadOutcome};
11use crate::serializer;
12
13// ── SharedBackend type alias ──────────────────────────────────────────────────
14
15/// Reference-counted pointer to a heap-allocated backend.
16///
17/// On native targets (without `unsync`) we require `Send + Sync` via `Arc`.
18/// On `wasm32` or with the `unsync` feature, `Rc` is used instead — the runtime
19/// is single-threaded so `Send` bounds are unnecessary.
20#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
21pub type SharedBackend = std::sync::Arc<dyn Backend>;
22
23/// Reference-counted pointer to a heap-allocated backend (`?Send` variant).
24#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
25pub type SharedBackend = std::rc::Rc<dyn Backend>;
26
27// ── SharedFlight type alias ──────────────────────────────────────────────────
28
29/// Reference-counted pointer to the single-flight map, so client clones share
30/// fill-dedup state (two clones racing a cold miss must collapse to one fill).
31#[cfg(not(any(target_arch = "wasm32", feature = "unsync")))]
32type SharedFlight = std::sync::Arc<crate::flight::FlightMap>;
33
34#[cfg(any(target_arch = "wasm32", feature = "unsync"))]
35type SharedFlight = std::rc::Rc<crate::flight::FlightMap>;
36
37/// Separate same-key ordering from single-flight: a refresh holds the flight
38/// lock while computing, then takes this lock only for its version-checked
39/// commit. Direct reads/writes/deletes take the same mutation lock.
40#[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
41type SharedMutations = std::sync::Arc<crate::flight::MutationMap>;
42
43// ── SharedEncryption type alias ──────────────────────────────────────────────
44
45/// Reference-counted pointer to the encryption layer.
46///
47/// On native targets (without `unsync`) `Arc` is used (requires `Sync`).
48/// On `wasm32` or with `unsync`, `Rc` is used — avoids the `!Sync` problem
49/// caused by `Cell<u64>` inside cachekit-core's nonce counter.
50#[cfg(all(
51    feature = "encryption",
52    not(any(target_arch = "wasm32", feature = "unsync"))
53))]
54type SharedEncryption = std::sync::Arc<crate::encryption::EncryptionLayer>;
55
56#[cfg(all(
57    feature = "encryption",
58    any(target_arch = "wasm32", feature = "unsync")
59))]
60type SharedEncryption = std::rc::Rc<crate::encryption::EncryptionLayer>;
61
62// ── Key validation ────────────────────────────────────────────────────────────
63
64const MAX_KEY_BYTES: usize = 1024;
65
66/// Maximum TTL for L1 entries populated from L2 cache hits.
67/// Uses a short ceiling to limit staleness when the original TTL is unknown.
68/// A server `X-CacheKit-Fresh-For` bounds it further, and a stale read or a
69/// zero bound is not backfilled at all — see [`l1_backfill_ttl`].
70///
71/// Reconciliation with stale-while-revalidate: a backfilled entry's SWR
72/// freshness window derives from this capped TTL (window = ratio × entry
73/// TTL), **not** from the write-path TTL — the cap is the staleness bound
74/// for L2-derived data, deliberately kept. SWR removes the cap's expiry
75/// cliff instead: past ~ratio × 30 s the entry is served stale while one
76/// background refresh re-executes the origin. If no newer mutation replaced
77/// the entry, that refresh writes both layers and renews L1 hard expiry with
78/// the caller's full TTL. Without SWR the entry simply hard-expires at the cap
79/// and the next read blocks on L2, as before.
80const L1_BACKFILL_TTL_SECS: u64 = 30;
81
82/// L1 lifetime for a backfill from an L2 hit, or `None` when the read must not
83/// be backfilled (`spec/saas-api.md` § Remaining Freshness). A stale label or a
84/// zero `Fresh-For` forbids it; a present `Fresh-For` is a hard bound under the
85/// 30 s cap, which SWR cannot outlive because moka drops the entry at its TTL.
86/// With no header the cap applies unchanged.
87#[cfg(feature = "l1")]
88fn l1_backfill_ttl(default_ttl: Duration, freshness: Freshness) -> Option<Duration> {
89    let cap = std::cmp::min(default_ttl, Duration::from_secs(L1_BACKFILL_TTL_SECS));
90    let ttl = freshness
91        .fresh_for
92        .map_or(cap, |fresh_for| cap.min(fresh_for));
93    (!freshness.is_stale && !ttl.is_zero()).then_some(ttl)
94}
95
96fn validate_key(key: &str) -> Result<(), CachekitError> {
97    if key.is_empty() {
98        return Err(CachekitError::InvalidKey(
99            "key must not be empty".to_owned(),
100        ));
101    }
102    if key.len() > MAX_KEY_BYTES {
103        return Err(CachekitError::InvalidKey(format!(
104            "key is {} bytes (limit: {MAX_KEY_BYTES})",
105            key.len()
106        )));
107    }
108    for b in key.bytes() {
109        if b < 0x20 || b == 0x7F {
110            return Err(CachekitError::InvalidKey(format!(
111                "key contains illegal control character 0x{b:02X}"
112            )));
113        }
114    }
115    Ok(())
116}
117
118// ── Stale-while-revalidate ───────────────────────────────────────────────────
119//
120// SWR needs an L1 to age entries in and a spawnable (`Send`) runtime for the
121// background refresh — native, non-`unsync` targets with the `l1` feature.
122// Everywhere else the SWR read path degrades to the plain read path and
123// `SwrRead::Stale` is never produced.
124
125/// Outcome of an SWR-aware typed read — see [`CacheKit::interop_get_swr`].
126#[derive(Debug, Clone, PartialEq)]
127pub enum SwrRead<T> {
128    /// Cache hit within the freshness window (or an L2 hit): use directly.
129    Fresh(T),
130    /// L1 hit past the freshness threshold but before hard expiry: use the
131    /// value now, and schedule a background refresh (the `#[cachekit]` macro
132    /// does this via [`CacheKit::single_flight`] + re-execution). The token
133    /// makes refresh completion conditional: a newer set or delete wins.
134    Stale(T, SwrToken),
135    /// No usable entry: fall through to a normal blocking miss + fill.
136    Miss,
137}
138
139/// Mutation version captured by an SWR stale read.
140///
141/// Pass this back only through the `#[cachekit]`-generated refresh path. It
142/// prevents an older background computation from overwriting a newer write
143/// or resurrecting a deleted entry on this client or any of its clones.
144#[derive(Clone)]
145pub struct SwrToken {
146    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
147    state: std::sync::Arc<crate::flight::MutationState>,
148    version: u64,
149}
150
151impl std::fmt::Debug for SwrToken {
152    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153        f.debug_struct("SwrToken")
154            .field("version", &self.version)
155            .finish_non_exhaustive()
156    }
157}
158
159impl PartialEq for SwrToken {
160    fn eq(&self, other: &Self) -> bool {
161        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
162        {
163            self.version == other.version && std::sync::Arc::ptr_eq(&self.state, &other.state)
164        }
165        #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
166        {
167            self.version == other.version
168        }
169    }
170}
171
172impl Eq for SwrToken {}
173
174// ── CacheKit ─────────────────────────────────────────────────────────────────
175
176/// Cache client with optional L1 in-process cache layer.
177///
178/// `Clone` is cheap and shares everything: backend, L1 cache, single-flight
179/// state, and encryption layer. Clones exist so `'static` background work
180/// (e.g. the SWR refresh spawned by `#[cachekit]`) can hold the client
181/// without borrowing it.
182#[derive(Clone)]
183pub struct CacheKit {
184    backend: SharedBackend,
185    default_ttl: Duration,
186    namespace: Option<String>,
187    max_payload_bytes: usize,
188    flight: SharedFlight,
189    /// Read counters behind [`Self::stats`]; the backend holds a
190    /// `MetricsProvider` over the same allocation.
191    counters: Arc<CacheCounters>,
192
193    /// The reliability stack's breaker, when configured — kept so
194    /// [`Self::circuit_state`] can read it through the type-erased backend.
195    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
196    breaker: Option<Arc<crate::reliability::CircuitBreaker>>,
197
198    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
199    mutations: SharedMutations,
200
201    #[cfg(feature = "l1")]
202    pub(crate) l1: Option<crate::l1::L1Cache>,
203
204    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
205    pub(crate) swr_enabled: bool,
206
207    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
208    swr_threshold_ratio: f64,
209
210    #[cfg(feature = "encryption")]
211    encryption: Option<SharedEncryption>,
212}
213
214impl CacheKit {
215    /// Create a new builder.
216    pub fn builder() -> CacheKitBuilder {
217        CacheKitBuilder::default()
218    }
219
220    /// Build from environment variables via [`crate::config::CachekitConfig::from_env`].
221    ///
222    /// Creates a [`crate::backend::cachekitio::CachekitIO`] backend from the
223    /// config. Requires the `cachekitio` feature.
224    #[cfg(all(feature = "cachekitio", not(target_arch = "wasm32")))]
225    pub fn from_env() -> Result<CacheKitBuilder, CachekitError> {
226        use crate::backend::cachekitio::CachekitIO;
227        use crate::config::CachekitConfig;
228
229        let config = CachekitConfig::from_env()?;
230
231        let api_key_z = config
232            .api_key
233            .ok_or_else(|| CachekitError::Config("CACHEKIT_API_KEY is required".to_owned()))?;
234
235        let backend = CachekitIO::builder()
236            .api_key(api_key_z.as_str())
237            .api_url(config.api_url)
238            .build()
239            .map_err(|e| CachekitError::Config(e.to_string()))?;
240
241        #[cfg(not(feature = "unsync"))]
242        let shared: SharedBackend = std::sync::Arc::new(backend);
243        #[cfg(feature = "unsync")]
244        let shared: SharedBackend = std::rc::Rc::new(backend);
245
246        let mut builder = CacheKitBuilder::default()
247            .backend(shared)
248            .default_ttl(config.default_ttl)
249            .max_payload_bytes(config.max_payload_bytes)
250            .l1_capacity(config.l1_capacity);
251
252        if let Some(ns) = config.namespace.clone() {
253            builder = builder.namespace(ns);
254        }
255
256        // Wire up encryption if master key is configured
257        #[cfg(feature = "encryption")]
258        if let Some(ref master_key) = config.master_key {
259            let namespace = config.namespace.as_deref().unwrap_or("default");
260            let previous: Vec<&[u8]> = config
261                .previous_master_keys
262                .iter()
263                .map(|key| key.as_slice())
264                .collect();
265            builder = builder.encryption_from_hex_decoded(master_key, &previous, namespace)?;
266        }
267
268        Ok(builder)
269    }
270
271    // ── Namespacing ───────────────────────────────────────────────────────────
272
273    fn namespaced_key(&self, key: &str) -> String {
274        match &self.namespace {
275            Some(ns) => format!("{ns}:{key}"),
276            None => key.to_owned(),
277        }
278    }
279
280    /// Validate key and return the namespaced version.
281    fn resolve_key(&self, key: &str) -> Result<String, CachekitError> {
282        validate_key(key)?;
283        Ok(self.namespaced_key(key))
284    }
285
286    // ── L1 helpers ───────────────────────────────────────────────────────────
287
288    /// Try L1 cache first. Returns Some(bytes) on hit.
289    #[cfg(feature = "l1")]
290    fn l1_get(&self, full_key: &str) -> Option<Vec<u8>> {
291        self.l1.as_ref().and_then(|l1| l1.get(full_key))
292    }
293
294    /// Populate L1 from an L2 hit, for no longer than [`l1_backfill_ttl`] allows.
295    #[cfg(feature = "l1")]
296    fn l1_backfill(&self, full_key: &str, bytes: &[u8], freshness: Freshness) {
297        if let (Some(l1), Some(ttl)) = (&self.l1, l1_backfill_ttl(self.default_ttl, freshness)) {
298            l1.set(full_key, bytes, ttl);
299        }
300    }
301
302    /// Write-through to L1.
303    #[cfg(feature = "l1")]
304    fn l1_set(&self, full_key: &str, bytes: &[u8], ttl: Duration) {
305        if let Some(ref l1) = self.l1 {
306            l1.set(full_key, bytes, ttl);
307        }
308    }
309
310    /// Invalidate L1 entry.
311    #[cfg(feature = "l1")]
312    fn l1_delete(&self, full_key: &str) {
313        if let Some(ref l1) = self.l1 {
314            l1.delete(full_key);
315        }
316    }
317
318    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
319    async fn lock_l1_mutation(&self, full_key: &str) -> Option<crate::flight::MutationGuard> {
320        if self.l1.is_some() {
321            Some(self.mutations.lock(full_key).await)
322        } else {
323            None
324        }
325    }
326
327    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
328    async fn complete_swr_bytes(
329        &self,
330        key: &str,
331        bytes: Vec<u8>,
332        ttl: Duration,
333        token: SwrToken,
334    ) -> Result<bool, CachekitError> {
335        Self::validate_ttl(ttl)?;
336        self.check_payload_size(bytes.len())?;
337        let full_key = self.resolve_key(key)?;
338        let Some(mutation) = self.lock_l1_mutation(&full_key).await else {
339            return Ok(false);
340        };
341
342        // Check before touching L2. The mutation state lives independently of
343        // the moka entry, so hard expiry or capacity eviction does not look
344        // like an explicit set/delete and waste a valid origin result.
345        if !mutation.is_current(&token.state, token.version) {
346            return Ok(false);
347        }
348        let l1_bytes = bytes.clone();
349        self.backend.set(&full_key, bytes, Some(ttl)).await?;
350        self.l1_set(&full_key, &l1_bytes, ttl);
351        mutation.advance();
352        crate::metrics::trace_write(&full_key, ttl);
353        Ok(true)
354    }
355
356    #[cfg(not(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32"))))]
357    async fn complete_swr_bytes(
358        &self,
359        _key: &str,
360        _bytes: Vec<u8>,
361        _ttl: Duration,
362        _token: SwrToken,
363    ) -> Result<bool, CachekitError> {
364        // `SwrRead::Stale` is unreachable on this build, so there is no
365        // versioned entry a caller could legitimately complete.
366        Ok(false)
367    }
368
369    /// Validate TTL is at least 1 second.
370    fn validate_ttl(ttl: Duration) -> Result<(), CachekitError> {
371        if ttl < Duration::from_secs(1) {
372            return Err(CachekitError::Config(format!(
373                "TTL must be at least 1 second; got {ttl:?}"
374            )));
375        }
376        Ok(())
377    }
378
379    // ── Public operations ─────────────────────────────────────────────────────
380
381    /// Retrieve and deserialize a value stored under `key`.
382    ///
383    /// Returns `None` if the key does not exist.
384    /// Checks L1 cache before hitting the backend.
385    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
386        match self.get_bytes(key).await? {
387            Some(bytes) => Ok(Some(serializer::deserialize(&bytes)?)),
388            None => Ok(None),
389        }
390    }
391
392    /// Retrieve and deserialize an interop-mode value stored under `key`.
393    ///
394    /// Identical to [`Self::get`] except the payload is decoded with
395    /// [`crate::interop::deserialize`], which consumes exactly one MessagePack
396    /// document and rejects trailing bytes (interop/v1 spec MUST). A
397    /// Python-SDK-internal CK frame is rejected with a specific diagnostic
398    /// instead of silently decoding as the integer 67.
399    ///
400    /// Use with keys from [`crate::interop::interop_key`] on a client
401    /// **without** a namespace prefix. There is no interop-specific write
402    /// method: [`Self::set`] already writes plain MessagePack (no ByteStorage
403    /// envelope), which is the interop value format.
404    ///
405    /// # Errors
406    ///
407    /// Returns [`CachekitError::Config`] if the client was built with
408    /// [`CacheKitBuilder::namespace`]: the prefix
409    /// would rewrite the storage key to `{prefix}:{interop_key}`, which no
410    /// other SDK computes — every cross-SDK entry would silently miss. Interop
411    /// keys carry their own namespace segment; failing loudly here beats a
412    /// 100% miss rate that looks like a cold cache.
413    pub async fn interop_get<T: DeserializeOwned>(
414        &self,
415        key: &str,
416    ) -> Result<Option<T>, CachekitError> {
417        self.reject_namespaced_interop()?;
418        match self.get_bytes(key).await? {
419            Some(bytes) => Ok(Some(crate::interop::deserialize(&bytes)?)),
420            None => Ok(None),
421        }
422    }
423
424    /// Interop keys must reach the backend verbatim; a client namespace prefix
425    /// would silently produce storage keys no other SDK computes.
426    fn reject_namespaced_interop(&self) -> Result<(), CachekitError> {
427        match self.namespace {
428            None => Ok(()),
429            Some(_) => Err(CachekitError::Config(
430                "interop reads require a client without a namespace prefix: .namespace() \
431                 would store interop entries under {prefix}:{interop_key}, \
432                 which other SDKs never compute (interop keys already carry a namespace \
433                 segment) — use a dedicated non-namespaced client for interop entries"
434                    .to_owned(),
435            )),
436        }
437    }
438
439    /// Retrieve and deserialize an interop-mode value with SWR classification.
440    ///
441    /// Identical to [`Self::interop_get`] except an L1 hit is classified
442    /// against the client's stale-while-revalidate freshness window:
443    ///
444    /// - [`SwrRead::Fresh`] — L1 hit within `swr_threshold_ratio` of the
445    ///   entry's TTL (±10% jitter), or any L2 hit. Use directly.
446    /// - [`SwrRead::Stale`] — L1 hit past the threshold but **before hard
447    ///   expiry**: the value is returned without touching the backend or
448    ///   origin, and the caller should schedule exactly one background
449    ///   refresh (dedup via [`Self::single_flight`] — this is what the
450    ///   `#[cachekit]` macro generates). The accompanying [`SwrToken`] makes
451    ///   completion conditional, so a newer set/delete always wins.
452    /// - [`SwrRead::Miss`] — nothing usable anywhere: normal blocking miss.
453    ///
454    /// A hard-expired L1 entry is a [`SwrRead::Miss`], never `Stale` — moka
455    /// drops entries at their TTL, so SWR cannot serve past hard expiry.
456    ///
457    /// With SWR disabled ([`CacheKitBuilder::swr_enabled`]`(false)`), without
458    /// the `l1` feature, on wasm32, or under `unsync`, this behaves exactly
459    /// like [`Self::interop_get`]: hits are `Fresh`, `Stale` is never
460    /// produced.
461    ///
462    /// # Errors
463    ///
464    /// Same as [`Self::interop_get`] (including the namespaced-client
465    /// rejection).
466    pub async fn interop_get_swr<T: DeserializeOwned>(
467        &self,
468        key: &str,
469    ) -> Result<SwrRead<T>, CachekitError> {
470        self.reject_namespaced_interop()?;
471        match self.get_bytes_swr(key).await? {
472            SwrRead::Fresh(b) => Ok(SwrRead::Fresh(crate::interop::deserialize(&b)?)),
473            SwrRead::Stale(b, token) => Ok(SwrRead::Stale(crate::interop::deserialize(&b)?, token)),
474            SwrRead::Miss => Ok(SwrRead::Miss),
475        }
476    }
477
478    /// Fetch raw payload bytes with SWR classification: an L1 hit is split
479    /// into fresh vs stale against the configured freshness window; on L1
480    /// miss this defers to [`Self::get_bytes`] (L2 + backfill), whose hit is
481    /// always fresh.
482    async fn get_bytes_swr(&self, key: &str) -> Result<SwrRead<Vec<u8>>, CachekitError> {
483        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
484        if self.swr_enabled {
485            if let Some(ref l1) = self.l1 {
486                let full_key = self.resolve_key(key)?;
487                match l1.get_with_swr(&full_key, self.swr_threshold_ratio) {
488                    crate::l1::L1SwrRead::Fresh(bytes) => {
489                        self.counters.record(ReadOutcome::L1Hit, &full_key);
490                        self.check_payload_size(bytes.len())?;
491                        return Ok(SwrRead::Fresh(bytes));
492                    }
493                    crate::l1::L1SwrRead::Stale(_) => {
494                        // Token capture and the stale snapshot must be atomic
495                        // relative to explicit mutations. Re-check after
496                        // taking the per-key guard; a write may have landed
497                        // between the optimistic classification and here.
498                        let mutation = self.mutations.lock(&full_key).await;
499                        match l1.get_with_swr(&full_key, self.swr_threshold_ratio) {
500                            crate::l1::L1SwrRead::Fresh(bytes) => {
501                                self.counters.record(ReadOutcome::L1Hit, &full_key);
502                                self.check_payload_size(bytes.len())?;
503                                return Ok(SwrRead::Fresh(bytes));
504                            }
505                            crate::l1::L1SwrRead::Stale(bytes) => {
506                                self.counters.record(ReadOutcome::L1Stale, &full_key);
507                                self.check_payload_size(bytes.len())?;
508                                let (state, version) = mutation.snapshot();
509                                return Ok(SwrRead::Stale(bytes, SwrToken { state, version }));
510                            }
511                            crate::l1::L1SwrRead::Miss => {}
512                        }
513                    }
514                    // Absent or hard-expired: fall through to the normal
515                    // read path (the redundant L1 re-check there is a cheap
516                    // in-process miss).
517                    crate::l1::L1SwrRead::Miss => {}
518                }
519            }
520        }
521
522        Ok(match self.get_bytes(key).await? {
523            Some(bytes) => SwrRead::Fresh(bytes),
524            None => SwrRead::Miss,
525        })
526    }
527
528    /// Fetch raw payload bytes for `key` (L1, then L2 with L1 backfill).
529    async fn get_bytes(&self, key: &str) -> Result<Option<Vec<u8>>, CachekitError> {
530        let full_key = self.resolve_key(key)?;
531
532        // L1 hit
533        #[cfg(feature = "l1")]
534        if let Some(bytes) = self.l1_get(&full_key) {
535            self.counters.record(ReadOutcome::L1Hit, &full_key);
536            self.check_payload_size(bytes.len())?;
537            return Ok(Some(bytes));
538        }
539
540        // Serialize an L2 read/backfill with same-key writes. Re-check L1
541        // after taking the lock because another operation may have filled it
542        // between the optimistic read above and lock acquisition.
543        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
544        let _mutation = self.lock_l1_mutation(&full_key).await;
545
546        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
547        if let Some(bytes) = self.l1_get(&full_key) {
548            self.counters.record(ReadOutcome::L1Hit, &full_key);
549            self.check_payload_size(bytes.len())?;
550            return Ok(Some(bytes));
551        }
552
553        // L2 backend
554        let (bytes, freshness) = match self.backend.get_with_freshness(&full_key).await? {
555            Some(hit) => {
556                self.counters.record(ReadOutcome::L2Hit, &full_key);
557                hit
558            }
559            None => {
560                self.counters.record(ReadOutcome::Miss, &full_key);
561                return Ok(None);
562            }
563        };
564
565        self.check_payload_size(bytes.len())?;
566
567        // Populate L1 on L2 hit, bounded by the cap and the server's freshness
568        #[cfg(feature = "l1")]
569        self.l1_backfill(&full_key, &bytes, freshness);
570        #[cfg(not(feature = "l1"))]
571        let _ = freshness;
572
573        Ok(Some(bytes))
574    }
575
576    /// Serialize and store `value` under `key` using the client's default TTL.
577    pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
578        self.set_with_ttl(key, value, self.default_ttl).await
579    }
580
581    /// Serialize and store `value` under `key` with an explicit `ttl`.
582    ///
583    /// Returns [`CachekitError::Config`] if `ttl` is less than 1 second.
584    pub async fn set_with_ttl<T: Serialize>(
585        &self,
586        key: &str,
587        value: &T,
588        ttl: Duration,
589    ) -> Result<(), CachekitError> {
590        Self::validate_ttl(ttl)?;
591
592        let bytes = serializer::serialize(value)?;
593        self.check_payload_size(bytes.len())?;
594
595        let full_key = self.resolve_key(key)?;
596
597        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
598        let mutation = self.lock_l1_mutation(&full_key).await;
599
600        // Invalidate older refresh tokens before the first backend await, so
601        // cancellation cannot leave an applied/attempted write vulnerable to
602        // a stale background result.
603        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
604        if let Some(ref mutation) = mutation {
605            mutation.advance();
606        }
607
608        // Only clone bytes when L1 needs a copy after the backend consumes them.
609        #[cfg(feature = "l1")]
610        {
611            let l1_bytes = bytes.clone();
612            self.backend.set(&full_key, bytes, Some(ttl)).await?;
613            self.l1_set(&full_key, &l1_bytes, ttl);
614        }
615        #[cfg(not(feature = "l1"))]
616        {
617            self.backend.set(&full_key, bytes, Some(ttl)).await?;
618        }
619
620        crate::metrics::trace_write(&full_key, ttl);
621        Ok(())
622    }
623
624    /// Commit an unencrypted SWR refresh only if its stale-read token is
625    /// still current. Macro plumbing; ordinary writes use [`Self::set_with_ttl`].
626    #[doc(hidden)]
627    pub async fn __complete_swr_refresh<T: Serialize>(
628        &self,
629        key: &str,
630        value: &T,
631        ttl: Duration,
632        token: SwrToken,
633    ) -> Result<bool, CachekitError> {
634        let bytes = serializer::serialize(value)?;
635        self.complete_swr_bytes(key, bytes, ttl, token).await
636    }
637
638    /// Delete `key` and return `true` if it existed.
639    ///
640    /// Invalidates the L1 entry regardless of the backend result.
641    pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
642        let full_key = self.resolve_key(key)?;
643
644        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
645        let mutation = self.lock_l1_mutation(&full_key).await;
646
647        // Invalidate before L1 changes or backend I/O so task cancellation
648        // cannot let an older refresh resurrect this key.
649        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
650        if let Some(ref mutation) = mutation {
651            mutation.advance();
652        }
653
654        // Invalidate L1 first so callers never read a stale value even if the
655        // backend delete fails partway through.
656        #[cfg(feature = "l1")]
657        self.l1_delete(&full_key);
658
659        let existed = self.backend.delete(&full_key).await?;
660        crate::metrics::trace_delete(&full_key, existed);
661        Ok(existed)
662    }
663
664    /// Return `true` if `key` exists without fetching the value.
665    pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
666        let full_key = self.resolve_key(key)?;
667
668        // Check L1 first — avoids a network round-trip for warm entries.
669        #[cfg(feature = "l1")]
670        if self.l1_get(&full_key).is_some() {
671            return Ok(true);
672        }
673
674        Ok(self.backend.exists(&full_key).await?)
675    }
676
677    // ── Observability ─────────────────────────────────────────────────────────
678
679    /// Live hit/miss counters for this client and all of its clones.
680    ///
681    /// Every value read — `get`, `interop_get`, their SWR variants, and the
682    /// [`SecureCache`] equivalents — is counted by where it was served from.
683    /// `exists` and reads that fail with a backend error are not counted. The
684    /// same counters feed the cachekit.io backends' `X-CacheKit-*` telemetry
685    /// headers, so for a backend used by this client alone, what you see here
686    /// is what the SaaS sees (a backend shared by several clients reports the
687    /// first client built over it — see [`Backend::attach_metrics`]).
688    ///
689    /// ```no_run
690    /// # fn example(cache: &cachekit::CacheKit) {
691    /// let stats = cache.stats();
692    /// let reads = stats.l1_hits + stats.l2_hits + stats.misses;
693    /// eprintln!("L1 hit rate {:.1}% over {reads} reads", stats.l1_hit_rate() * 100.0);
694    /// # }
695    /// ```
696    pub fn stats(&self) -> L1Stats {
697        self.counters.snapshot()
698    }
699
700    /// Number of entries currently held in L1, or `None` when L1 is disabled
701    /// (or the `l1` feature is off).
702    ///
703    /// Exact at the time of the call: moka's pending expirations and
704    /// evictions are applied first, which makes this a housekeeping call, not
705    /// a free counter read — poll it, don't put it on a hot path.
706    pub fn l1_entry_count(&self) -> Option<u64> {
707        #[cfg(feature = "l1")]
708        let count = self.l1.as_ref().map(crate::l1::L1Cache::entry_count);
709        #[cfg(not(feature = "l1"))]
710        let count = None;
711        count
712    }
713
714    /// Current circuit-breaker state, or `None` when this client has no
715    /// breaker (reliability disabled, or configured with
716    /// `circuit_breaker: None`).
717    ///
718    /// Reads the live state: an open circuit whose `open_timeout` has elapsed
719    /// reports [`CircuitState::HalfOpen`](crate::CircuitState::HalfOpen). With
720    /// the `tracing` feature every transition is also emitted on the
721    /// `cachekit::reliability` target — `warn` when the circuit opens, `info`
722    /// for half-open and closed.
723    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
724    pub fn circuit_state(&self) -> Option<crate::reliability::CircuitState> {
725        self.breaker.as_ref().map(|breaker| breaker.state())
726    }
727
728    // ── Single-flight ─────────────────────────────────────────────────────────
729
730    /// Begin a cold-miss single-flight for `key` (see [`crate::flight`]).
731    ///
732    /// Call after a cache miss, before computing the value. Concurrent
733    /// in-process fills of the same key are collapsed to one; with the
734    /// `reliability` feature and a lock-capable backend (CachekitIO, Redis),
735    /// fills are also suppressed across processes via a distributed fill
736    /// lock. The `#[cachekit]` macro does this automatically.
737    ///
738    /// The key is namespaced like every cache operation but not validated —
739    /// this call is infallible; an invalid key simply fails later at the
740    /// actual cache operation.
741    pub async fn single_flight(&self, key: &str) -> crate::flight::SingleFlight {
742        let full_key = self.namespaced_key(key);
743        crate::flight::SingleFlight::acquire(&self.flight, &self.backend, &full_key).await
744    }
745
746    // ── Secure cache ─────────────────────────────────────────────────────────
747
748    /// Return a [`SecureCache`] handle that encrypts all values before storage.
749    ///
750    /// L1 stores **ciphertext** (not plaintext) to preserve the zero-knowledge
751    /// property across all cache layers.
752    ///
753    /// # Errors
754    /// Returns `CachekitError::Config` if no encryption layer is configured.
755    /// Configure encryption via [`CacheKitBuilder::encryption`] /
756    /// [`CacheKitBuilder::encryption_from_bytes`], or start from the
757    /// `CacheKit::secure(url, master_key_hex)` /
758    /// `CacheKit::secure_from_env(url)` preset (`redis` feature).
759    #[cfg(feature = "encryption")]
760    pub fn secure_cache(&self) -> Result<SecureCache<'_>, CachekitError> {
761        let enc = self.encryption.as_ref().ok_or_else(|| {
762            CachekitError::Config(
763                "encryption not configured: use the CacheKit::secure(url, master_key_hex) \
764                 preset, CacheKit::secure_from_env(url) to read the hex key from \
765                 CACHEKIT_MASTER_KEY, or call .encryption() on the builder"
766                    .to_owned(),
767            )
768        })?;
769        Ok(SecureCache {
770            client: self,
771            encryption: enc,
772        })
773    }
774
775    // ── Private helpers ───────────────────────────────────────────────────────
776
777    fn check_payload_size(&self, size: usize) -> Result<(), CachekitError> {
778        if size > self.max_payload_bytes {
779            return Err(CachekitError::PayloadTooLarge {
780                size,
781                limit: self.max_payload_bytes,
782            });
783        }
784        Ok(())
785    }
786}
787
788// ── SecureCache ──────────────────────────────────────────────────────────────
789
790/// Encrypted cache handle returned by [`CacheKit::secure_cache()`].
791///
792/// All values are serialized, then encrypted with AES-256-GCM before storage.
793/// L1 stores ciphertext to maintain zero-knowledge guarantees.
794#[cfg(feature = "encryption")]
795pub struct SecureCache<'a> {
796    client: &'a CacheKit,
797    encryption: &'a crate::encryption::EncryptionLayer,
798}
799
800#[cfg(feature = "encryption")]
801impl std::fmt::Debug for SecureCache<'_> {
802    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
803        f.debug_struct("SecureCache")
804            .field("tenant_id", &self.encryption.tenant_id())
805            .finish()
806    }
807}
808
809#[cfg(feature = "encryption")]
810impl SecureCache<'_> {
811    /// Rotation drain signal; see
812    /// [`EncryptionLayer::previous_key_hits`](crate::EncryptionLayer::previous_key_hits).
813    pub fn previous_key_hits(&self) -> Vec<u64> {
814        self.encryption.previous_key_hits()
815    }
816
817    /// Whether AES is hardware-accelerated on this host (informational); see
818    /// [`EncryptionLayer::hardware_acceleration_enabled`](crate::EncryptionLayer::hardware_acceleration_enabled).
819    pub fn hardware_acceleration_enabled(&self) -> bool {
820        self.encryption.hardware_acceleration_enabled()
821    }
822
823    /// Encrypt and store `value` under `key` using the client's default TTL.
824    pub async fn set<T: Serialize>(&self, key: &str, value: &T) -> Result<(), CachekitError> {
825        self.set_with_ttl(key, value, self.client.default_ttl).await
826    }
827
828    /// Encrypt and store `value` under `key` with an explicit `ttl`.
829    pub async fn set_with_ttl<T: Serialize>(
830        &self,
831        key: &str,
832        value: &T,
833        ttl: Duration,
834    ) -> Result<(), CachekitError> {
835        CacheKit::validate_ttl(ttl)?;
836
837        // Serialize then encrypt
838        let plaintext = serializer::serialize(value)?;
839        let ciphertext = self.encryption.encrypt(&plaintext, key)?;
840        // Size-check what is actually persisted (nonce + ciphertext + tag).
841        // The get paths check the stored ciphertext length, so checking the
842        // plaintext here would let a value within 28 bytes of the limit write
843        // successfully and then fail EVERY subsequent read with PayloadTooLarge.
844        self.client.check_payload_size(ciphertext.len())?;
845
846        let full_key = self.client.resolve_key(key)?;
847
848        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
849        let mutation = self.client.lock_l1_mutation(&full_key).await;
850
851        // Match the plain write path: invalidate older refresh tokens before
852        // the first backend await, including on cancellation or failure.
853        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
854        if let Some(ref mutation) = mutation {
855            mutation.advance();
856        }
857
858        // Only clone when L1 needs a copy after the backend consumes the data.
859        #[cfg(feature = "l1")]
860        {
861            let l1_bytes = ciphertext.clone();
862            self.client
863                .backend
864                .set(&full_key, ciphertext, Some(ttl))
865                .await?;
866            self.client.l1_set(&full_key, &l1_bytes, ttl);
867        }
868        #[cfg(not(feature = "l1"))]
869        {
870            self.client
871                .backend
872                .set(&full_key, ciphertext, Some(ttl))
873                .await?;
874        }
875
876        crate::metrics::trace_write(&full_key, ttl);
877        Ok(())
878    }
879
880    /// Commit an encrypted SWR refresh only if its stale-read token is still
881    /// current. Macro plumbing; ordinary writes use [`Self::set_with_ttl`].
882    #[doc(hidden)]
883    pub async fn __complete_swr_refresh<T: Serialize>(
884        &self,
885        key: &str,
886        value: &T,
887        ttl: Duration,
888        token: SwrToken,
889    ) -> Result<bool, CachekitError> {
890        let plaintext = serializer::serialize(value)?;
891        let ciphertext = self.encryption.encrypt(&plaintext, key)?;
892        self.client
893            .complete_swr_bytes(key, ciphertext, ttl, token)
894            .await
895    }
896
897    /// Retrieve, decrypt, and deserialize a value stored under `key`.
898    ///
899    /// Checks L1 (which holds ciphertext) before the backend.
900    ///
901    /// # Errors
902    ///
903    /// Returns [`CachekitError::Encryption`] if the stored entry fails
904    /// decryption. The key's L1 copy is dropped first and the backend entry
905    /// is left in place, so the next read reaches the backend: a hit once the
906    /// entry is replaced with ciphertext this client can decrypt, a miss once
907    /// it expires or is deleted.
908    pub async fn get<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>, CachekitError> {
909        match self.get_plaintext(key).await? {
910            Some(plaintext) => Ok(Some(serializer::deserialize(&plaintext)?)),
911            None => Ok(None),
912        }
913    }
914
915    /// Retrieve, decrypt, and deserialize an interop-mode value stored under `key`.
916    ///
917    /// Identical to [`Self::get`] except the decrypted plaintext is decoded
918    /// with [`crate::interop::deserialize`] — exactly one MessagePack document,
919    /// trailing bytes rejected (interop/v1 spec MUST). In interop mode the
920    /// AES-GCM plaintext is the plain MessagePack value bytes, so the AAD
921    /// (v0x03, `format="msgpack"`, `compressed="False"`) verifies cross-SDK
922    /// unchanged.
923    ///
924    /// # Errors
925    ///
926    /// Returns [`CachekitError::Config`] on a namespace-prefixed client — see
927    /// [`CacheKit::interop_get`] — and [`CachekitError::Encryption`] if the
928    /// stored entry fails decryption, with the key's L1 copy dropped as for
929    /// [`Self::get`].
930    pub async fn interop_get<T: DeserializeOwned>(
931        &self,
932        key: &str,
933    ) -> Result<Option<T>, CachekitError> {
934        self.client.reject_namespaced_interop()?;
935        match self.get_plaintext(key).await? {
936            Some(plaintext) => Ok(Some(crate::interop::deserialize(&plaintext)?)),
937            None => Ok(None),
938        }
939    }
940
941    /// Retrieve, decrypt, and deserialize an interop-mode value with SWR
942    /// classification. The secure twin of [`CacheKit::interop_get_swr`]:
943    /// staleness is judged on the L1 **ciphertext** entry (zero-knowledge is
944    /// preserved — freshness metadata never exposes plaintext), then the
945    /// value is decrypted and decoded per [`Self::interop_get`].
946    ///
947    /// # Errors
948    ///
949    /// Same as [`Self::interop_get`] — the secure path fails closed on every
950    /// backend and decryption error.
951    pub async fn interop_get_swr<T: DeserializeOwned>(
952        &self,
953        key: &str,
954    ) -> Result<SwrRead<T>, CachekitError> {
955        self.client.reject_namespaced_interop()?;
956        match self.client.get_bytes_swr(key).await? {
957            SwrRead::Fresh(ct) => Ok(SwrRead::Fresh(crate::interop::deserialize(
958                &self.decrypt_or_evict(&ct, key)?,
959            )?)),
960            SwrRead::Stale(ct, token) => Ok(SwrRead::Stale(
961                crate::interop::deserialize(&self.decrypt_or_evict(&ct, key)?)?,
962                token,
963            )),
964            SwrRead::Miss => Ok(SwrRead::Miss),
965        }
966    }
967
968    /// Fetch ciphertext (L1, then L2 with L1 backfill) and decrypt it.
969    ///
970    /// Ciphertext retrieval delegates to [`CacheKit::get_bytes`], which returns
971    /// the stored bytes untransformed — for a secure cache exactly the AES-GCM
972    /// ciphertext, so decrypt receives the same bytes the backend holds.
973    async fn get_plaintext(&self, key: &str) -> Result<Option<Vec<u8>>, CachekitError> {
974        match self.client.get_bytes(key).await? {
975            Some(ciphertext) => Ok(Some(self.decrypt_or_evict(&ciphertext, key)?)),
976            None => Ok(None),
977        }
978    }
979
980    /// Decrypt ciphertext read for `key`; on failure, drop the key's L1 copy
981    /// before propagating the error.
982    ///
983    /// An L2 hit is backfilled into L1 before it is decrypted, so an entry
984    /// that fails authentication would otherwise keep failing from L1 after
985    /// the backend entry is fixed or deleted. The backend entry itself is
986    /// left untouched as evidence.
987    fn decrypt_or_evict(&self, ciphertext: &[u8], key: &str) -> Result<Vec<u8>, CachekitError> {
988        self.encryption.decrypt(ciphertext, key).inspect_err(|_| {
989            // The read that produced `ciphertext` already resolved this key,
990            // so `resolve_key` cannot fail here.
991            #[cfg(feature = "l1")]
992            if let Ok(full_key) = self.client.resolve_key(key) {
993                self.client.l1_delete(&full_key);
994            }
995        })
996    }
997
998    /// Delete an encrypted key. Behaves identically to [`CacheKit::delete`].
999    pub async fn delete(&self, key: &str) -> Result<bool, CachekitError> {
1000        self.client.delete(key).await
1001    }
1002
1003    /// Check if an encrypted key exists. Behaves identically to [`CacheKit::exists`].
1004    pub async fn exists(&self, key: &str) -> Result<bool, CachekitError> {
1005        self.client.exists(key).await
1006    }
1007}
1008
1009// ── CacheKitBuilder ───────────────────────────────────────────────────────────
1010
1011/// Fluent builder for [`CacheKit`].
1012#[derive(Default)]
1013#[must_use]
1014pub struct CacheKitBuilder {
1015    backend: Option<SharedBackend>,
1016    default_ttl: Option<Duration>,
1017    namespace: Option<String>,
1018    max_payload_bytes: Option<usize>,
1019
1020    #[cfg(feature = "l1")]
1021    l1_capacity: Option<usize>,
1022
1023    #[cfg(feature = "l1")]
1024    no_l1: bool,
1025
1026    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1027    swr_enabled: Option<bool>,
1028
1029    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1030    swr_threshold_ratio: Option<f64>,
1031
1032    // pub(crate): the `secure` preset's key-path tests read the layer.
1033    #[cfg(feature = "encryption")]
1034    pub(crate) encryption: Option<SharedEncryption>,
1035
1036    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1037    reliability: Option<crate::reliability::ReliabilityConfig>,
1038}
1039
1040impl CacheKitBuilder {
1041    /// Set the storage backend.
1042    pub fn backend(mut self, backend: SharedBackend) -> Self {
1043        self.backend = Some(backend);
1044        self
1045    }
1046
1047    /// Override the default TTL (used when no per-call TTL is specified).
1048    pub fn default_ttl(mut self, ttl: Duration) -> Self {
1049        self.default_ttl = Some(ttl);
1050        self
1051    }
1052
1053    /// Set a namespace prefix. All keys will be stored as `{namespace}:{key}`.
1054    pub fn namespace(mut self, ns: impl Into<String>) -> Self {
1055        self.namespace = Some(ns.into());
1056        self
1057    }
1058
1059    /// Set the maximum accepted payload size in bytes.
1060    pub fn max_payload_bytes(mut self, limit: usize) -> Self {
1061        self.max_payload_bytes = Some(limit);
1062        self
1063    }
1064
1065    /// Set the L1 cache capacity (max entries).
1066    #[cfg(feature = "l1")]
1067    pub fn l1_capacity(mut self, capacity: usize) -> Self {
1068        self.l1_capacity = Some(capacity);
1069        self
1070    }
1071
1072    /// Disable the L1 cache entirely.
1073    #[cfg(feature = "l1")]
1074    pub fn no_l1(mut self) -> Self {
1075        self.no_l1 = true;
1076        self
1077    }
1078
1079    /// Enable or disable L1 stale-while-revalidate (default: **enabled**,
1080    /// matching the Python and TypeScript SDKs).
1081    ///
1082    /// With SWR on, an L1 hit older than `swr_threshold_ratio` of its TTL is
1083    /// still served immediately, and the `#[cachekit]` macro schedules
1084    /// exactly one background refresh (deduplicated through
1085    /// [`CacheKit::single_flight`], in-process and — on lock-capable
1086    /// backends — across processes). A hard-expired entry is never served:
1087    /// it falls through to a normal blocking miss.
1088    ///
1089    /// Native targets only: this knob does not exist on wasm32, under the
1090    /// `unsync` feature, or without `l1` — calling it there is a compile
1091    /// error rather than a silent no-op.
1092    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1093    pub fn swr_enabled(mut self, enabled: bool) -> Self {
1094        self.swr_enabled = Some(enabled);
1095        self
1096    }
1097
1098    /// Set the SWR freshness threshold as a fraction of each L1 entry's TTL
1099    /// (default: **0.5**, matching the Python and TypeScript SDKs).
1100    ///
1101    /// An entry is *fresh* until it has lived `ratio × TTL` (±10% jitter,
1102    /// drawn once when the entry is inserted, to de-synchronise refreshes
1103    /// across processes), then *stale* — served immediately with a background
1104    /// refresh — until hard expiry. Mirrors cachekit-py's
1105    /// `swr_threshold_ratio` semantics (elapsed-lifetime fraction). Must be in
1106    /// `(0.0, 1.0]`; validated at [`Self::build`].
1107    #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1108    pub fn swr_threshold_ratio(mut self, ratio: f64) -> Self {
1109        self.swr_threshold_ratio = Some(ratio);
1110        self
1111    }
1112
1113    // Stubs for when the l1 feature is disabled — still compile cleanly.
1114    #[cfg(not(feature = "l1"))]
1115    pub fn l1_capacity(self, _capacity: usize) -> Self {
1116        self
1117    }
1118
1119    #[cfg(not(feature = "l1"))]
1120    pub fn no_l1(self) -> Self {
1121        self
1122    }
1123
1124    /// Wrap the backend in the reliability stack (retry with exponential
1125    /// backoff + jitter, circuit breaker, backpressure) — see
1126    /// [`crate::reliability`].
1127    ///
1128    /// Enabled by default with production settings by the `production`,
1129    /// `secure`, and `io` intent presets; off for `minimal` and for
1130    /// manually-built clients. To opt a preset out, pass
1131    /// [`ReliabilityConfig::disabled()`](crate::reliability::ReliabilityConfig::disabled)
1132    /// — a disabled config applies no wrapping at all.
1133    #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1134    pub fn reliability(mut self, config: crate::reliability::ReliabilityConfig) -> Self {
1135        self.reliability = Some(config);
1136        self
1137    }
1138
1139    /// Configure encryption from raw master key bytes and tenant ID.
1140    ///
1141    /// The master key must be exactly 32 raw bytes; anything else is a
1142    /// [`CachekitError::Config`]. Hex-encoded keys go to [`Self::encryption`]
1143    /// — their ASCII bytes are 64 bytes long and rejected here.
1144    /// Keys are derived per-tenant via HKDF-SHA256.
1145    #[cfg(feature = "encryption")]
1146    pub fn encryption_from_bytes(
1147        mut self,
1148        master_key: &[u8],
1149        tenant_id: &str,
1150    ) -> Result<Self, CachekitError> {
1151        let layer = crate::encryption::EncryptionLayer::new(master_key, tenant_id)?;
1152        self.encryption = Some(SharedEncryption::new(layer));
1153        Ok(self)
1154    }
1155
1156    /// Configure encryption with decrypt-only previous master keys for
1157    /// key rotation.
1158    ///
1159    /// Writes encrypt under `master_key`; reads attempt it first, then each
1160    /// key in `previous_keys` sequentially (attempt order = slice order).
1161    /// At most 3 previous keys; supplying more is a config error, never
1162    /// truncated. See [`crate::encryption::EncryptionLayer::with_previous_keys`].
1163    ///
1164    /// Every key, current and previous, must be exactly 32 raw bytes.
1165    #[cfg(feature = "encryption")]
1166    pub fn encryption_from_bytes_with_previous(
1167        mut self,
1168        master_key: &[u8],
1169        previous_keys: &[&[u8]],
1170        tenant_id: &str,
1171    ) -> Result<Self, CachekitError> {
1172        let layer = crate::encryption::EncryptionLayer::with_previous_keys(
1173            master_key,
1174            previous_keys,
1175            tenant_id,
1176        )?;
1177        self.encryption = Some(SharedEncryption::new(layer));
1178        Ok(self)
1179    }
1180
1181    /// Configure encryption from a hex-encoded master key string.
1182    ///
1183    /// Hex-decodes with the same decoder as [`CacheKit::from_env`] and the
1184    /// `secure` preset. The key must decode to at least 32 bytes; use exactly
1185    /// 32 (64 hex chars), the only length every SDK accepts.
1186    #[cfg(feature = "encryption")]
1187    pub fn encryption(self, hex_key: &str, tenant_id: &str) -> Result<Self, CachekitError> {
1188        let bytes = crate::config::decode_master_key_hex(hex_key, "master key")?;
1189        self.encryption_from_hex_decoded(&bytes, &[], tenant_id)
1190    }
1191
1192    /// Shared tail of the hex paths: keys already validated by
1193    /// `decode_master_key_hex`, so the `>= 32` floor applies, not the
1194    /// raw-bytes exactly-32 rule.
1195    #[cfg(feature = "encryption")]
1196    pub(crate) fn encryption_from_hex_decoded(
1197        mut self,
1198        master_key: &[u8],
1199        previous_keys: &[&[u8]],
1200        tenant_id: &str,
1201    ) -> Result<Self, CachekitError> {
1202        let layer = crate::encryption::EncryptionLayer::from_hex_decoded_keys(
1203            master_key,
1204            previous_keys,
1205            tenant_id,
1206        )?;
1207        self.encryption = Some(SharedEncryption::new(layer));
1208        Ok(self)
1209    }
1210
1211    // Stub for when encryption feature is disabled.
1212    #[cfg(not(feature = "encryption"))]
1213    pub fn encryption_from_bytes(
1214        self,
1215        _master_key: &[u8],
1216        _tenant_id: &str,
1217    ) -> Result<Self, CachekitError> {
1218        Ok(self)
1219    }
1220
1221    #[cfg(not(feature = "encryption"))]
1222    pub fn encryption_from_bytes_with_previous(
1223        self,
1224        _master_key: &[u8],
1225        _previous_keys: &[&[u8]],
1226        _tenant_id: &str,
1227    ) -> Result<Self, CachekitError> {
1228        Ok(self)
1229    }
1230
1231    #[cfg(not(feature = "encryption"))]
1232    pub fn encryption(self, _hex_key: &str, _tenant_id: &str) -> Result<Self, CachekitError> {
1233        Ok(self)
1234    }
1235
1236    /// Finalise and build the [`CacheKit`] client.
1237    ///
1238    /// Returns an error if no backend was provided.
1239    pub fn build(self) -> Result<CacheKit, CachekitError> {
1240        let backend = self.backend.ok_or_else(|| {
1241            CachekitError::Config("a backend must be provided via .backend()".to_owned())
1242        })?;
1243
1244        // Validate namespace if provided
1245        if let Some(ref ns) = self.namespace {
1246            if ns.is_empty() {
1247                return Err(CachekitError::Config("namespace cannot be empty".into()));
1248            }
1249            if ns.len() > 255 {
1250                return Err(CachekitError::Config("namespace exceeds 255 bytes".into()));
1251            }
1252            if !ns.bytes().all(|b| (0x20..=0x7E).contains(&b)) {
1253                return Err(CachekitError::Config(
1254                    "namespace must be ASCII printable".into(),
1255                ));
1256            }
1257        }
1258
1259        #[cfg(feature = "l1")]
1260        let l1 = if self.no_l1 {
1261            None
1262        } else {
1263            let capacity = self.l1_capacity.unwrap_or(1000);
1264            Some(crate::l1::L1Cache::new(capacity))
1265        };
1266
1267        // SWR defaults mirror the sibling SDKs: enabled, threshold ratio 0.5,
1268        // ratio validated in (0.0, 1.0] exactly like py's L1CacheConfig.
1269        #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1270        let swr_threshold_ratio = {
1271            let ratio = self.swr_threshold_ratio.unwrap_or(0.5);
1272            if !(ratio > 0.0 && ratio <= 1.0) {
1273                return Err(CachekitError::Config(format!(
1274                    "swr_threshold_ratio must be in (0.0, 1.0]; got {ratio}"
1275                )));
1276            }
1277            ratio
1278        };
1279
1280        // Hand the backend a live view of this client's read counters so the
1281        // SaaS telemetry headers report real numbers with no user plumbing.
1282        // Attached to the raw backend, before the reliability decorator, so
1283        // no forwarding is needed; a provider set on the backend's own
1284        // builder wins (see `Backend::attach_metrics`).
1285        #[cfg(feature = "l1")]
1286        let l1_enabled = l1.is_some();
1287        #[cfg(not(feature = "l1"))]
1288        let l1_enabled = false;
1289        let counters = Arc::new(CacheCounters::new(l1_enabled));
1290        backend.attach_metrics(counters.provider());
1291
1292        // Apply the reliability stack last so it decorates the final backend.
1293        // A disabled config is the documented opt-out: skip the (no-op)
1294        // decorator entirely. The layer check lives on ReliabilityConfig
1295        // itself so a future layer can't be missed here (this gate
1296        // shipped that exact bug once already).
1297        #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1298        let (backend, breaker) = match self.reliability {
1299            Some(config) if !config.is_disabled() => {
1300                crate::reliability::wrap_reliable(backend, config)
1301            }
1302            _ => (backend, None),
1303        };
1304
1305        Ok(CacheKit {
1306            backend,
1307            default_ttl: self.default_ttl.unwrap_or(Duration::from_secs(300)),
1308            namespace: self.namespace,
1309            max_payload_bytes: self.max_payload_bytes.unwrap_or(5 * 1024 * 1024),
1310            flight: SharedFlight::default(),
1311            counters,
1312
1313            #[cfg(all(feature = "reliability", not(target_arch = "wasm32")))]
1314            breaker,
1315
1316            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1317            mutations: SharedMutations::default(),
1318
1319            #[cfg(feature = "l1")]
1320            l1,
1321
1322            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1323            swr_enabled: self.swr_enabled.unwrap_or(true),
1324
1325            #[cfg(all(feature = "l1", not(feature = "unsync"), not(target_arch = "wasm32")))]
1326            swr_threshold_ratio,
1327
1328            #[cfg(feature = "encryption")]
1329            encryption: self.encryption,
1330        })
1331    }
1332}
1333
1334#[cfg(all(test, feature = "l1"))]
1335mod backfill_ttl_tests {
1336    use std::time::Duration;
1337
1338    use super::l1_backfill_ttl;
1339    use crate::backend::Freshness;
1340
1341    const DEFAULT_TTL: Duration = Duration::from_secs(300);
1342
1343    fn fresh_for(secs: u64) -> Freshness {
1344        Freshness {
1345            is_stale: false,
1346            fresh_for: Some(Duration::from_secs(secs)),
1347        }
1348    }
1349
1350    /// (e) No header: today's `min(default_ttl, 30 s)`, unchanged.
1351    #[test]
1352    fn absent_header_keeps_the_30s_cap() {
1353        let absent = Freshness::default();
1354        assert_eq!(
1355            l1_backfill_ttl(DEFAULT_TTL, absent),
1356            Some(Duration::from_secs(30))
1357        );
1358        assert_eq!(
1359            l1_backfill_ttl(Duration::from_secs(10), absent),
1360            Some(Duration::from_secs(10))
1361        );
1362    }
1363
1364    #[test]
1365    fn fresh_for_bounds_but_never_lifts_the_cap() {
1366        assert_eq!(
1367            l1_backfill_ttl(DEFAULT_TTL, fresh_for(5)),
1368            Some(Duration::from_secs(5))
1369        );
1370        assert_eq!(
1371            l1_backfill_ttl(DEFAULT_TTL, fresh_for(3600)),
1372            Some(Duration::from_secs(30))
1373        );
1374    }
1375
1376    #[test]
1377    fn stale_or_zero_forbids_the_backfill() {
1378        assert_eq!(l1_backfill_ttl(DEFAULT_TTL, fresh_for(0)), None);
1379        let stale = |fresh_for| Freshness {
1380            is_stale: true,
1381            fresh_for,
1382        };
1383        assert_eq!(l1_backfill_ttl(DEFAULT_TTL, stale(None)), None);
1384        assert_eq!(
1385            l1_backfill_ttl(DEFAULT_TTL, stale(Some(Duration::from_secs(60)))),
1386            None
1387        );
1388    }
1389}