alkblobs/src/store.rs

//! The `Store` facade type (store-api.md §What this is): the public
//! surface consumers program against — constructor modes, the
//! size-threshold dispatch seam, and the read-path ops (`get`/`stat`/
//! `read_range`/`has`/`list` with the fall-through general rule,
//! implemented below). The remaining ops land against this type in
//! `store-put-paths` (put, the kv-only rejection behavior) and
//! `gc-protocol` (sweep, direct delete, the pin/liveness machinery).
//! Direct `delete` is *not* a read path — it lands with the GC
//! protocol's arbitration. This module owns the engine representation
//! and the constructor-mode machinery; the pure routing function lives
//! in [`crate::dispatch`].
//!
//! **Named design point — engine representation: concrete `Store`,
//! crate-internal enum dispatch.** A `Store<Kv: Backend, Large: Backend>`
//! generic was the alternative: rejected. The engines are statically
//! known and in-crate (no `dyn` anywhere — wave-1 design point 3), and
//! they are `pub(crate)` behind the crate-internal [`Backend`][crate::backend]
//! trait — a generic facade parameterized over `Backend` implementers
//! would leak engine types through the public surface, force consumers
//! to name engines they must never see (the facade is **engine-blind**,
//! ADR-007/008/010), or push `dyn` onto the surface. The concrete type
//! holds two crate-internal engine-holding enums ([`KvEngine`],
//! [`LargeEngine`]) dispatching over the closed five-engine set; the
//! conformance factory's binding note anticipated exactly this collapse
//! (tests/conformance/factory.rs — "most plausibly the associated type
//! collapsing to the concrete facade type across all factories"). The
//! public surface is engine-blind: consumers name modes and engines
//! only through named constructors, feature-gated with their engine
//! (`mem_only` needs feature `mem`; the sqlite/postgres/pg-lo
//! constructors arrive with waves 3–5 and slot into the same enums and
//! constructor pattern without reshaping — adding an enum variant and
//! a named constructor is slotting in).
//!
//! **Constructor modes** (ADR-011 §4; ADR-012 §6.3):
//!
//! - **dual-tier** (the default): kv + large, size-threshold dispatch
//!   via [`crate::dispatch::route`]. The wave-2 constructible shape is
//!   `kv = mem` + `large = local` — the dispatch-coverage test shape
//!   ADR-012 §6.3 names (both tiers load-bearing). Over-threshold
//!   content routes to the large tier; at-or-below to kv.
//! - **kv-only**: no large tier; over-threshold puts reject **at put
//!   time** (ADR-012 §6.2 — known-length immediately; unknown-length at
//!   mid-stream threshold overflow). The rejection behavior itself
//!   lands with the put seam; the threshold this mode carries is the
//!   rejection point's value.
//! - **mem-only**: the mem engine alone, **no dispatch threshold in
//!   effect** (everything is local and small) — a testing/embedder-
//!   ephemeral posture, never production. Structurally distinct from
//!   kv-only precisely by that threshold absence.
//!
//! Modes are validated structurally — unrepresentable states are
//! unrepresentable by construction: dual-tier's signature constructs
//! the large engine itself, mem-only's constructor exists only with
//! feature `mem` and constructs the mem engine — and every instance
//! carries its mode for introspection ([`Store::mode`]).
//!
//! **Read-path design points (recorded for the review gate):**
//!
//! - **The fall-through general rule (ADR-012 §6.1; POC #1 finding
//!   6).** Every key-addressed op consults [`KvEngine`] first, then
//!   [`LargeEngine`] — there is no length routing on the read side and
//!   [`Store::tier_for`] is never consulted by it (the dispatch seam's
//!   write-half routing is the put path's; `route` feeds nothing on
//!   reads). A miss on both tiers resolves per op's shape: `Ok(None)`
//!   on get/stat, `false` on has, `Err(Missing)` on read_range.
//! - **The public stream type (named design point): `Blob`.** `get`
//!   returns the pinned `(len, stream)` shape as
//!   `Option<(u64, Blob)>` (store-api.md §Read path) — [`Blob`] is a
//!   bounded, re-entrant windowed reader over the crate-internal
//!   [`AnyCursor`], which dispatches the engine cursor bodies by tier
//!   (mem's materializing body for kv, `local`'s pread body for
//!   large). Windowed reads are the uniform stream shape across
//!   tiers; no chunk-iterator machinery was adopted — the cursor
//!   bodies *are* the streaming abstraction, and per-chunk `Stream`-
//!   wrapping would re-shape them without changing the underlying
//!   reads. Engine cursors never leak onto the public surface:
//!   [`AnyCursor`] is `pub(crate)`, `Blob` is opaque but drivable
//!   (constructed only by the store core, ADR-010 §1).
//! - **Verification-on-get (named design point, pinned against
//!   ADR-006's text).** The whole-blob hash-check fires where the
//!   whole content is already in hand — the kv tier's materializing
//!   reads. A kv-tier `get` materializes the stored value and
//!   re-derives the canonical digest over it, comparing against the
//!   key (ADR-006 decision 1: "put and get paths hash-check");
//!   mismatch is `Err(Error::Verification)` and **never falls
//!   through further** — a verified mismatch is a pool-integrity
//!   fact, and consulting another tier past it would be a guess.
//!   The large tier's streaming `get` does *not* re-derive at open:
//!   opening is metadata + handle only, and the tier's integrity
//!   story is the put path's whole-blob verification pass (the
//!   complete staged artifact was checked before commit) plus the
//!   per-range slice digests `read_range` carries out-of-band
//!   (ADR-006 decisions 1–2; the §Context line "git objects
//!   self-verify under git's own model when read whole"). `read_range`
//!   carries no canonical check on either tier — the slice digest is
//!   the convention's whole answer (ADR-006 decision 2). stat/has/
//!   list carry no verification (metadata and enumeration only).
//! - **`list` returns `Vec<Key>` (named design point: Vec vs
//!   stream).** Whole-pool enumeration, complete by contract
//!   (ADR-003 — load-bearing for GC), returned **materialized and
//!   sorted in the keys' total order** (deterministic output). The
//!   sweep consumes the same crate-internal enumeration directly from
//!   the engines (its live-set diffs sort independently), so the
//!   facade shape serves consumers, not internal needs. A lazy stream
//!   was rejected: every shipped engine enumerates fully and eagerly
//!   (the completeness contract *is* the GC-safety property), so a
//!   pseudo-lazy wrapper adds machinery without changing the
//!   underlying walk. Listed bytes failing [`Key::from_bytes`]
//!   reject with `Error::KeyInvalid` — a malformed pool list is
//!   fatal, never a silent skip (ADR-003's malformed-list ruling
//!   applied at the facade boundary).
//! - **`KeyInvalid` at the facade boundary (ADR-002/003).** Read ops
//!   take opaque key *bytes* and convert to typed `Key` at the
//!   boundary via [`Key::from_bytes`]: malformed bytes reject with
//!   `Error::KeyInvalid` on every key-addressed op, never a guess,
//!   and the engines receive only the physical byte encoding — they
//!   never see typed keys.
//!
//! **The fleet declaration defers to the engine waves.** ADR-012 §3
//! rules fleet mode a constructor declaration (`fleet: true`), never
//! inferred — but it is a constructor *error* for every non-fleet-valid
//! configuration, and no fleet-valid configuration is constructible yet
//! (mem is never fleet-valid; `local` needs the shared-media
//! declaration of ADR-008 §3; the fleet-valid engines — postgres, pg-lo
//! — are waves 4–5). Adding the declaration then is additive; carrying
//! a flag that could only ever be rejected today would be dead
//! machinery.
//!
//! **The dispatch seam the ops tasks program through:** tier selection
//! for a given length ([`Store::tier_for`] composing
//! [`crate::dispatch::route`] per mode), the fall-through consult
//! order — **kv first, then large** — that every key-addressed op
//! consumes (ADR-012 §6.1's general rule; [`Store::kv`] then
//! [`Store::large`]), and the GC state home ([`crate::gc::GcHome`])
//! the gc-protocol task fills.
//!
//! The `#![allow(dead_code)]` below marks the work still ahead, not the
//! read path (which is driven below): write-side `tier_for`/`threshold`
//! and the GC home wait on `store-put-paths`/`gc-protocol`, and the kv
//! engine set is empty in default builds (mem is feature-gated; sqlite
//! is wave 3). The allow shrinks as those land.

#![allow(dead_code)]

use crate::backend::{Backend, ReadCursor};
use crate::dispatch::{route, Tier};
use crate::facade::{Blob, EntryMeta};
use crate::gc::GcHome;
use crate::keys::Key;
use crate::local::{LocalBackend, LocalCursor};

#[cfg(feature = "mem")]
use crate::mem::{MemBackend, MemCursor};

use crate::Error;

#[cfg(feature = "mem")]
use std::path::PathBuf;

use self::StoreMode::{DualTier, KvOnly, MemOnly};

/// The store instance — the public facade above the tier engines
/// (ADR-011 §1's vocabulary: *one running instantiation of the store
/// facade — one engine set per the constructor mode, plus dispatch and
/// GC state*; **the GC arbitration domain** for single-instance
/// topologies).
///
/// Engine-blind by construction: the engines are held in crate-internal
/// enums dispatched statically (see the module's engine-representation
/// design point); consumers name modes only. The GC state home
/// ([`crate::gc::GcHome`]) rides as a field — the store instance owns
/// all GC state (ADR-010 §2), which the gc-protocol task fills.
pub struct Store {
    /// The kv-tier engine — the first consult in every fall-through.
    kv: KvEngine,
    /// The large-tier engine — `None` in single-tier modes.
    large: Option<LargeEngine>,
    /// The dispatch threshold in effect where the mode has one; mem-only
    /// carries the shipped default but never consults it — "no dispatch
    /// threshold in effect" is the mode's ruling (ADR-011 §4), not a
    /// value absence.
    threshold: u64,
    /// The constructor mode; the dispatch/put/read/gc paths branch on it.
    mode: StoreMode,
    /// The GC state home (ADR-010 §2: the store core owns GC state)
    /// — the gc-protocol task's machinery lands here.
    gc: GcHome,
}

/// The constructor modes (ADR-011 §4's pinned set; ADR-012 §6.3's
/// dual-tier ruling). Three variants, ADR-pinned — a new mode is a new
/// ADR.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StoreMode {
    /// kv + large, size-threshold dispatch between them (the default).
    DualTier,
    /// kv alone; over-threshold puts reject at put time (ADR-012 §6.2).
    KvOnly,
    /// The mem engine alone, no dispatch threshold in effect — never
    /// production (ADR-011 §4).
    MemOnly,
}

/// The kv-tier engine set — the closed per-tier engine enum of the
/// engine-representation design point. Feature-gated variants slot in
/// as their engine waves land: sqlite (wave 3), postgres (wave 4).
pub(crate) enum KvEngine {
    /// The contract-reference ephemeral engine (ADR-011 §2).
    #[cfg(feature = "mem")]
    Mem(MemBackend),
}

/// The large-tier engine set. `local` is ungated (dependency-free,
/// always present, ADR-013 §4); pg-lo slots in at wave 5.
pub(crate) enum LargeEngine {
    /// The sharded-filesystem engine (ADR-003/008).
    Local(LocalBackend),
}

/// The read path's cursor-dispatch enum (ADR-010 §1's "store core
/// programs the abstraction"): tier dispatch over the engine cursor
/// bodies — the kv tier's materializing body and the large tier's
/// streaming body — behind one crate-internal value that [`Blob`] holds
/// and drives. Crate-internal: engine cursors never appear on the
/// public surface; the facade's stream shape is this enum's uniform
/// [`ReadCursor`] view (`Blob`).
pub(crate) enum AnyCursor {
    /// The kv body family: the whole value materialized in memory
    /// (bounded by tier policy).
    #[cfg(feature = "mem")]
    Kv(MemCursor),
    /// The large body family: a pread cursor over the committed file.
    Large(LocalCursor),
}

impl ReadCursor for AnyCursor {
    fn len(&self) -> u64 {
        match self {
            #[cfg(feature = "mem")]
            Self::Kv(cursor) => cursor.len(),
            Self::Large(cursor) => cursor.len(),
        }
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        match self {
            #[cfg(feature = "mem")]
            Self::Kv(cursor) => cursor.read_at(offset, len).await,
            Self::Large(cursor) => cursor.read_at(offset, len).await,
        }
    }
}

impl KvEngine {
    #[allow(unused_variables)] // kv engine set empty until wave 3 (sqlite): no-mem builds never reference `key`
    async fn get(&self, key: &[u8]) -> Result<Option<AnyCursor>, Error> {
        match *self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(ref engine) => Ok(engine.get(key).await?.map(AnyCursor::Kv)),
        }
    }

    #[allow(unused_variables)] // same empty-set scope as `get`
    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        match *self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(ref engine) => engine.has(key).await,
        }
    }

    #[allow(unused_variables)] // same empty-set scope as `get`
    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        match *self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(ref engine) => engine.size(key).await,
        }
    }

    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        match *self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(ref engine) => engine.list().await,
        }
    }
}

impl LargeEngine {
    async fn get(&self, key: &[u8]) -> Result<Option<AnyCursor>, Error> {
        match self {
            LargeEngine::Local(engine) => Ok(engine.get(key).await?.map(AnyCursor::Large)),
        }
    }

    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        match self {
            LargeEngine::Local(engine) => engine.has(key).await,
        }
    }

    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        match self {
            LargeEngine::Local(engine) => engine.size(key).await,
        }
    }

    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        match self {
            LargeEngine::Local(engine) => engine.list().await,
        }
    }
}

/// The facade's key-bytes decode: typed `Key` conversion at the store
/// boundary (ADR-002/003) — the one place on the read path where key
/// bytes are interpreted; anything malformed is `Error::KeyInvalid`,
/// never a guess, and never reaches an engine typed.
fn boundary_key(key: &[u8]) -> Result<Key, Error> {
    Key::from_bytes(key)
}

/// Constitute the facade's typed keys from an engine's opaque list
/// rows: a failed decode is fatal (`Error::KeyInvalid`) — a malformed
/// pool list is as deadly as an incomplete one and is never silently
/// skipped (ADR-003).
fn typed_keys(raws: Vec<Vec<u8>>) -> Result<Vec<Key>, Error> {
    raws.into_iter().map(|raw| boundary_key(&raw)).collect()
}

impl Store {
    /// **mem-only mode** (ADR-011 §4): the mem engine alone, no
    /// dispatch threshold in effect — a testing/embedder-ephemeral
    /// posture, never production. Every length routes to the (sole)
    /// mem engine; there is no large tier and nothing to reject.
    #[cfg(feature = "mem")]
    pub fn mem_only() -> Store {
        Store {
            kv: KvEngine::Mem(MemBackend::default()),
            large: None,
            threshold: crate::dispatch::DEFAULT_DISPATCH_THRESHOLD,
            mode: MemOnly,
            gc: GcHome,
        }
    }

    /// **kv-only mode** (ADR-011 §4; ADR-012 §6.2) with the mem kv
    /// engine: no large tier; `threshold` is the put-time rejection
    /// point — known-length puts over it reject immediately,
    /// unknown-length puts at mid-stream threshold overflow (the put
    /// seam's behavior; zero residue on either rejection). Routing in
    /// this mode is kv for every length.
    #[cfg(feature = "mem")]
    pub fn kv_only_mem(threshold: u64) -> Store {
        Store {
            kv: KvEngine::Mem(MemBackend::default()),
            large: None,
            threshold,
            mode: KvOnly,
            gc: GcHome,
        }
    }

    /// **dual-tier mode** (the default, ADR-011 §4) in the wave-2
    /// constructible shape — `kv = mem` + `large = local`, the
    /// dispatch-coverage test shape ADR-012 §6.3 names: both tiers
    /// load-bearing, routed by size-threshold dispatch at `threshold`
    /// (constructor-tunable; [`crate::dispatch::DEFAULT_DISPATCH_THRESHOLD`]
    /// is the shipped default).
    ///
    /// Constructing the large engine creates the root directory —
    /// fallible I/O, hence a fallible constructor. The mem kv engine
    /// cannot fail.
    ///
    /// Not fleet-anything: mem is never fleet-valid (ADR-011 §4).
    #[cfg(feature = "mem")]
    pub async fn dual_tier_mem_local(
        root: impl Into<PathBuf>,
        threshold: u64,
    ) -> Result<Store, Error> {
        Ok(Store {
            kv: KvEngine::Mem(MemBackend::default()),
            large: Some(LargeEngine::Local(LocalBackend::new(root).await?)),
            threshold,
            mode: DualTier,
            gc: GcHome,
        })
    }

    /// The constructor mode (ADR-011 §4's vocabulary) — introspection
    /// for embedders and tests; the dispatch/put/read paths branch on
    /// it crate-internally.
    pub fn mode(&self) -> StoreMode {
        self.mode
    }

    /// The active dispatch threshold — `Some` in dual-tier (the routing
    /// boundary) and kv-only (the put-time rejection point); `None` in
    /// mem-only mode ("no dispatch threshold in effect", ADR-011 §4).
    pub(crate) fn threshold(&self) -> Option<u64> {
        match self.mode {
            DualTier | KvOnly => Some(self.threshold),
            MemOnly => None,
        }
    }

    /// **Write-side tier selection for a given content length** — the
    /// dispatch seam's routing half (the put seam programs through
    /// this; read paths do not consult it). Dual-tier:
    /// [`crate::dispatch::route`]'s pure function at the instance's
    /// threshold (exactly-at-threshold stays kv — the pinned boundary
    /// reading). kv-only: kv for every length — the over-threshold
    /// *rejection* is the put seam's behavior keyed off
    /// [`Store::threshold`] (ADR-012 §6.2), not a routing answer.
    /// Mem-only: kv — there is no other tier and no threshold in
    /// effect.
    pub(crate) fn tier_for(&self, len: u64) -> Tier {
        match self.mode {
            DualTier => route(len, self.threshold),
            KvOnly | MemOnly => Tier::Kv,
        }
    }

    /// **The fall-through consult order — kv first** (ADR-012 §6.1's
    /// general rule): every key-addressed op consults the kv tier
    /// first; large comes after ([`Store::large`]). A miss on both
    /// tiers resolves per op's shape (`None` on get/stat,
    /// `Err(Missing)` on read_range/delete).
    pub(crate) fn kv(&self) -> &KvEngine {
        &self.kv
    }

    /// **…then large** — the second consult; `None` in single-tier
    /// modes (mem-only, kv-only), where large-tier content cannot
    /// exist.
    pub(crate) fn large(&self) -> Option<&LargeEngine> {
        self.large.as_ref()
    }

    /// The GC state home (ADR-010 §2: the store core owns all GC
    /// state) — the gc-protocol task's pin registry, liveness sources,
    /// protect callback, and in-process delete window land here and
    /// drive the sweep and direct delete from this facade.
    pub(crate) fn gc(&self) -> &GcHome {
        &self.gc
    }
}

/// The read path (store-api.md §Read path): `get`, `stat`,
/// `read_range`, `has`, `list` — every key-addressed op falling through
/// kv first, then large (ADR-012 §6.1's general rule), typed key
/// conversion at the facade boundary, and the module-doc design points
/// (the `Blob` stream shape, verification-on-get, the `Vec<Key>` list
/// shape) carried in their implementations.
impl Store {
    /// `get(key) → Option<(len, stream)>`. Falls through kv → large;
    /// a both-tier miss is `Ok(None)` — never `Error::Missing`
    /// (ADR-012 §6.5 ruling 5). On a kv hit the materialized read is
    /// whole-blob-verified against the key at get time (mismatch:
    /// `Err(Error::Verification)`, never a further fall-through) and
    /// the returned length is the verified materialized byte length;
    /// the large tier's hit opens streaming (metadata + handle, no
    /// canonical re-derivation — see the module doc), its length the
    /// cursor's metadata-only probe. Both hits return the same
    /// `(len, stream)` shape: the stream is [`Blob`] over the tier's
    /// cursor body.
    pub async fn get(&self, key: &[u8]) -> Result<Option<(u64, Blob)>, Error> {
        let key = boundary_key(key)?;
        let bytes = key.as_bytes();

        let cursor = if let Some(cursor) = self.kv().get(&bytes).await? {
            cursor
        } else if let Some(cursor) = self.large_get(&bytes).await? {
            cursor
        } else {
            return Ok(None);
        };

        match cursor {
            #[cfg(feature = "mem")]
            AnyCursor::Kv(kv) => {
                let value = kv.read_at(0, kv.len()).await?;
                let derived = key.algorithm().derive(&value, value.len() as u64);
                if &derived != key.digest() {
                    return Err(Error::Verification);
                }
                Ok(Some((value.len() as u64, Blob::new(AnyCursor::Kv(kv)))))
            }
            AnyCursor::Large(large) => Ok(Some((large.len(), Blob::new(AnyCursor::Large(large))))),
        }
    }

    /// The large tier's consult — `None` where the mode has no large
    /// tier (single-tier modes cannot hold large-tier content).
    async fn large_get(&self, bytes: &[u8]) -> Result<Option<AnyCursor>, Error> {
        match self.large() {
            Some(large) => large.get(bytes).await,
            None => Ok(None),
        }
    }

    /// `stat(key) → Option<EntryMeta>` — the cheap probe riding the
    /// trait's `size` method (ADR-008): length plus the key's hash
    /// algorithm, never a content load, falling through kv → large.
    /// A both-tier miss is `Ok(None)`.
    pub async fn stat(&self, key: &[u8]) -> Result<Option<EntryMeta>, Error> {
        let key = boundary_key(key)?;
        let bytes = key.as_bytes();

        let len = match self.kv().size(&bytes).await? {
            Some(len) => Some(len),
            None => match self.large() {
                Some(large) => large.size(&bytes).await?,
                None => None,
            },
        };
        Ok(len.map(|len| EntryMeta {
            len,
            algorithm: key.algorithm(),
        }))
    }

    /// `read_range(key, offset, len) → (slice, slice_digest)` — the
    /// slice plus the out-of-band SHA-256 slice digest (ADR-006, fixed
    /// by convention), composed through the cursor's `read_range` seam.
    /// Falls through kv → large exactly as [`Store::get`] does; no
    /// canonical verification rides a range (the slice digest is the
    /// whole answer, ADR-006 decision 2). A both-tier miss is
    /// `Err(Error::Missing)` — the pinned shape difference vs `get`'s
    /// `None` (ADR-012 §6.5 ruling 5).
    pub async fn read_range(
        &self,
        key: &[u8],
        offset: u64,
        len: u64,
    ) -> Result<(Vec<u8>, [u8; 32]), Error> {
        let key = boundary_key(key)?;
        let bytes = key.as_bytes();

        if let Some(cursor) = self.kv().get(&bytes).await? {
            return cursor.read_range(offset, len).await;
        }
        match self.large_get(&bytes).await? {
            Some(cursor) => cursor.read_range(offset, len).await,
            None => Err(Error::Missing),
        }
    }

    /// `has(key) → bool` — the pure existence probe (ADR-008's `has`
    /// discipline: an existence probe per tier, `local`'s body being
    /// the literal size-probe), falling through kv → large. Absent on
    /// both tiers: `false`. Virgin-store reads are no-ops.
    pub async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        let key = boundary_key(key)?;
        let bytes = key.as_bytes();

        if self.kv().has(&bytes).await? {
            return Ok(true);
        }
        match self.large() {
            Some(large) => large.has(&bytes).await,
            None => Ok(false),
        }
    }

    /// `list() → Vec<Key>` — whole-pool enumeration of every tier,
    /// complete by contract (ADR-003), returned materialized and
    /// sorted in the keys' total order (the recorded Vec-vs-stream
    /// design point; see the module doc). Malformed pool bytes are
    /// fatal: `Err(Error::KeyInvalid)`, never a silent skip.
    pub async fn list(&self) -> Result<Vec<Key>, Error> {
        let mut keys = typed_keys(self.kv().list().await?)?;
        if let Some(large) = self.large() {
            keys.extend(typed_keys(large.list().await?)?);
        }
        keys.sort();
        Ok(keys)
    }
}

#[cfg(all(test, feature = "mem"))]
mod tests {
    use super::*;
    use crate::backend::{slice_digest, Backend};
    use crate::dispatch::DEFAULT_DISPATCH_THRESHOLD;
    use crate::hashing::Algorithm;
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};

    /// A fresh temp root per test (mirrors `local.rs`'s helper); Drop
    /// removes the whole tree idempotently.
    struct TempRoot(PathBuf);

    fn temp_root(tag: &'static str) -> TempRoot {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let dir = std::env::temp_dir().join(format!(
            "alkblobs-store-{tag}-{}-{:x}-{seq}",
            std::process::id(),
            nanos
        ));
        fs::create_dir_all(&dir).expect("temp root created");
        TempRoot(dir)
    }

    impl TempRoot {
        fn path(&self) -> &Path {
            &self.0
        }
    }

    impl Drop for TempRoot {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.0);
        }
    }

    /// mem-only constructs and validates: the ephemeral mode's shape —
    /// no large tier, no threshold in effect, kv routing for every
    /// length (ADR-011 §4).
    #[cfg(feature = "mem")]
    #[test]
    fn mem_only_constructs_and_validates() {
        let store = Store::mem_only();
        assert_eq!(store.mode(), MemOnly);
        assert_eq!(store.threshold(), None, "no dispatch threshold in effect");
        assert!(store.large().is_none(), "mem-only has no large tier");

        for len in [0u64, 1, DEFAULT_DISPATCH_THRESHOLD + 1, u64::MAX] {
            assert_eq!(
                store.tier_for(len),
                Tier::Kv,
                "mem-only routes every length to its sole tier"
            );
        }
    }

    /// kv-only constructs and validates: no large tier; the threshold
    /// is the *rejection point's* carrier (ADR-012 §6.2), not a routing
    /// answer — every length routes kv there, and the put seam rejects
    /// known-length over-threshold puts immediately (its behavior).
    #[cfg(feature = "mem")]
    #[test]
    fn kv_only_mem_constructs_and_validates() {
        let store = Store::kv_only_mem(1024);
        assert_eq!(store.mode(), KvOnly);
        assert_eq!(store.threshold(), Some(1024));
        assert!(store.large().is_none(), "kv-only has no large tier");

        for len in [0u64, 1023, 1024, 1025, u64::MAX] {
            assert_eq!(
                store.tier_for(len),
                Tier::Kv,
                "kv-only routes every length to its sole tier"
            );
        }
    }

    /// The ADR-012 §6.3 dispatch-coverage shape: `kv = mem` +
    /// `large = local` dual-tier construction works, both tiers
    /// load-bearing, dispatch threshold active — boundary values route
    /// deterministically (at/at-below threshold kv, above large), and
    /// tunability is proven against a non-default threshold.
    #[cfg(feature = "mem")]
    #[tokio::test]
    async fn dual_tier_mem_local_constructs_and_routes() {
        let root = temp_root("dual-tier");
        let store = Store::dual_tier_mem_local(root.path(), 1024)
            .await
            .expect("dual-tier mem+local constructs");
        assert_eq!(store.mode(), DualTier);
        assert_eq!(store.threshold(), Some(1024));
        assert!(store.large().is_some(), "both tiers present");

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "length {len} routes deterministically"
            );
            assert_eq!(
                store.tier_for(len),
                tier,
                "same length ⇒ same tier (the purity invariant, re-checked)"
            );
        }
    }

    /// The shipped default threshold is the constructor-tunable
    /// baseline: dual-tier at `DEFAULT_DISPATCH_THRESHOLD` routes at
    /// the 128 KiB boundary exactly as the pure function does.
    #[cfg(feature = "mem")]
    #[tokio::test]
    async fn dual_tier_routes_at_the_default_threshold() {
        let root = temp_root("dual-tier-default");
        let store = Store::dual_tier_mem_local(root.path(), DEFAULT_DISPATCH_THRESHOLD)
            .await
            .expect("dual-tier constructs at the default threshold");

        for (len, tier) in [
            (131_071, Tier::Kv),
            (131_072, Tier::Kv),
            (131_073, Tier::Large),
        ] {
            assert_eq!(store.tier_for(len), tier, "length {len} at 128 KiB");
        }
    }

    /// The GC state home rides as a field of the store instance
    /// (ADR-010 §2 — the store instance owns GC state) and is
    /// reachable crate-internally: the gc-protocol task's machinery
    /// lands against this handle.
    #[cfg(feature = "mem")]
    #[test]
    fn gc_state_home_rides_on_the_foundation() {
        let store = Store::mem_only();
        let _ = store.gc();
    }

    /// Entry planters straight into the tier engines (bypassing the
    /// put seam, which is a later task): opaque key bytes + value,
    /// exactly what the engines' CAS tables accept. Planting
    /// mismatching content under a key is how the read path's
    /// verification postures are observed.
    async fn plant_kv(store: &Store, key: &[u8], value: &[u8]) {
        match store.kv() {
            KvEngine::Mem(engine) => engine.put(key, value).await.expect("kv plant"),
        }
    }

    async fn plant_large(store: &Store, key: &[u8], value: &[u8]) {
        match store.large() {
            Some(LargeEngine::Local(engine)) => engine.put(key, value).await.expect("large plant"),
            None => panic!("planting into a mode with no large tier"),
        }
    }

    fn key_for(content: &[u8]) -> Key {
        Key::derive(content, content.len() as u64)
    }

    fn sha1_key_for(content: &[u8]) -> Key {
        Key::from(Algorithm::GitBlobSha1.derive(content, content.len() as u64))
    }

    /// The both-tier-miss shapes every read op must resolve to
    /// (store-api.md §Error model; ADR-012 §6.5 ruling 5) — pinned on
    /// all three constructible modes, virgin-store reads as no-ops.
    async fn assert_virgin_read_shapes(store: &Store) {
        let key = key_for(b"nothing ever landed").as_bytes();

        assert!(
            store.get(&key).await.unwrap().is_none(),
            "get's both-tier miss is Ok(None)"
        );
        assert_eq!(store.stat(&key).await.unwrap(), None, "stat misses as None");
        assert_eq!(
            store.read_range(&key, 0, 16).await.unwrap_err(),
            Error::Missing,
            "read_range's both-tier miss is Err(Missing) — the pinned shape difference vs get"
        );
        assert!(
            !store.has(&key).await.unwrap(),
            "has's both-tier miss is false"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "virgin-store list is empty"
        );
    }

    #[tokio::test]
    async fn virgin_store_reads_are_no_ops_dual_tier() {
        let root = temp_root("read-virgin-dual");
        let store = Store::dual_tier_mem_local(root.path(), 1024)
            .await
            .expect("dual-tier constructs");
        assert_virgin_read_shapes(&store).await;
    }

    #[tokio::test]
    async fn virgin_store_reads_are_no_ops_mem_only() {
        let store = Store::mem_only();
        assert_virgin_read_shapes(&store).await;
    }

    #[tokio::test]
    async fn virgin_store_reads_are_no_ops_kv_only() {
        let store = Store::kv_only_mem(64);
        assert_virgin_read_shapes(&store).await;
    }

    /// `get` returns the pinned `(len, stream)` shape from the kv tier:
    /// the tuple carries the metadata length, the `Blob` serves
    /// bounded windows (clamped at EOF) and the ADR-006 range seam.
    #[tokio::test]
    async fn get_returns_the_pinned_len_stream_shape_from_kv() {
        let store = Store::mem_only();
        let content = b"small-tier content";
        let key = key_for(content);
        plant_kv(&store, &key.as_bytes(), content).await;

        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(len, content.len() as u64, "tuple len is the entry length");
        assert_eq!(blob.len(), len);
        assert!(!blob.is_empty());

        let whole = blob.read_at(0, len).await.expect("full window");
        assert_eq!(whole, content, "full window matches");
        assert_eq!(
            blob.read_at(0, len + 17).await.expect("over-read"),
            content,
            "over-read clamps at the entry's end"
        );

        let (slice, digest) = blob.read_range(6, 6).await.expect("ranged window");
        assert_eq!(slice, &content[6..12]);
        assert_eq!(
            digest,
            slice_digest(&content[6..12]),
            "slice digest is SHA-256 over exactly the returned bytes"
        );

        let (full, full_digest) = blob.read_range(0, len).await.expect("full range");
        assert_eq!(full, content);
        assert_eq!(full_digest, slice_digest(content));
    }

    /// The zero-length degenerate: a kv-tier empty entry get's as
    /// `Some((0, blob))`, its windows read empty-never-error, and
    /// `stat` carries exact zero.
    #[tokio::test]
    async fn zero_length_kv_entry_reads_as_empty_blob() {
        let store = Store::mem_only();
        let key = key_for(b"");
        plant_kv(&store, &key.as_bytes(), b"").await;

        let (len, blob) = store.get(&key.as_bytes()).await.unwrap().expect("present");
        assert_eq!(len, 0);
        assert!(blob.is_empty());
        assert_eq!(blob.read_at(0, 8).await.unwrap(), Vec::<u8>::new());

        assert_eq!(
            store.stat(&key.as_bytes()).await.unwrap(),
            Some(EntryMeta {
                len: 0,
                algorithm: key.algorithm()
            })
        );
    }

    /// `get`/`stat`/`read_range`/`has` fall through to the large tier
    /// on a kv miss — consult order, not length routing: a
    /// sub-threshold entry planted only in the large tier is still
    /// found even though `tier_for` routes its length to kv.
    #[tokio::test]
    async fn read_ops_fall_through_kv_miss_to_large() {
        let root = temp_root("read-fall-through");
        let store = Store::dual_tier_mem_local(root.path(), 1024)
            .await
            .expect("dual-tier constructs");
        let content = b"a sub-threshold entry planted directly in the large tier";
        let key = key_for(content);

        assert_eq!(
            store.tier_for(content.len() as u64),
            Tier::Kv,
            "the length *routes* kv — the read still lands via the large tier"
        );
        plant_large(&store, &key.as_bytes(), content).await;

        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("fall-through get succeeds")
            .expect("large-tier entry found via fall-through");
        assert_eq!(len, content.len() as u64);
        assert_eq!(blob.read_at(0, len).await.unwrap(), content);

        assert_eq!(
            store.stat(&key.as_bytes()).await.unwrap(),
            Some(EntryMeta {
                len: content.len() as u64,
                algorithm: key.algorithm()
            }),
            "stat falls through too"
        );
        assert!(
            store.has(&key.as_bytes()).await.unwrap(),
            "has falls through"
        );

        let (slice, digest) = store
            .read_range(&key.as_bytes(), 4, 8)
            .await
            .expect("read_range falls through");
        assert_eq!(slice, &content[4..12]);
        assert_eq!(digest, slice_digest(&slice));
    }

    /// The kv tier is the *first* consult on every key-addressed op:
    /// the same key bytes planted with different (mismatching) content
    /// in both tiers — get/read_range/stat/has must all answer from
    /// the kv tier's entry.
    #[tokio::test]
    async fn kv_tier_is_the_first_consult_on_every_key_addressed_op() {
        let root = temp_root("read-kv-first");
        let store = Store::dual_tier_mem_local(root.path(), 1024)
            .await
            .expect("dual-tier constructs");
        const KV_CONTENT: &[u8] = b"the kv tier's entry";
        let key = key_for(KV_CONTENT);
        plant_kv(&store, &key.as_bytes(), KV_CONTENT).await;
        plant_large(&store, &key.as_bytes(), b"decoy content in the large tier").await;

        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("kv hit wins");
        assert_eq!(
            blob.read_at(0, len).await.unwrap(),
            KV_CONTENT,
            "kv tier first — and its content verifies against the key"
        );

        let (slice, _) = store
            .read_range(&key.as_bytes(), 0, 4)
            .await
            .expect("ranged");
        assert_eq!(
            slice,
            &KV_CONTENT[0..4],
            "read_range also consults kv first"
        );

        assert!(store.has(&key.as_bytes()).await.unwrap());
        assert_eq!(
            store.stat(&key.as_bytes()).await.unwrap(),
            Some(EntryMeta {
                len: KV_CONTENT.len() as u64,
                algorithm: key.algorithm()
            })
        );
    }

    /// Verification-on-get, kv tier (ADR-006 decision 1): a
    /// materialized get hash-checks against the key; mismatch is
    /// `Err(Verification)` and never falls through further. `read_range`
    /// by contrast carries no canonical check — the slice digest is its
    /// whole verification story (ADR-006 decision 2).
    #[tokio::test]
    async fn kv_get_whole_blob_verifies_and_never_falls_through_past_mismatch() {
        let store = Store::mem_only();
        let key = key_for(b"the derived address");
        let bytes = key.as_bytes();
        plant_kv(&store, &bytes, b"content that is not the key").await;

        assert_eq!(
            store.get(&bytes).await.unwrap_err(),
            Error::Verification,
            "kv materialized reads hash-check at get time"
        );
        assert_eq!(
            store.read_range(&bytes, 0, 8).await.unwrap(),
            (b"content ".to_vec(), slice_digest(b"content ")),
            "read_range carries no canonical check — the slice-digest posture"
        );
        assert!(
            store.stat(&bytes).await.unwrap().is_some(),
            "stat is a metadata probe, not a verifier"
        );
    }

    /// Verification-on-get, large tier (the recorded posture's other
    /// half): a streaming get opens metadata + handle only — it does
    /// not re-derive the canonical digest at open. The tier's
    /// integrity story is the put path's whole-blob verification pass
    /// plus the per-range slice digests (ADR-006 decisions 1–2).
    #[tokio::test]
    async fn large_streaming_get_does_not_re_derive_at_open() {
        let root = temp_root("read-large-no-reverify");
        let store = Store::dual_tier_mem_local(root.path(), 8)
            .await
            .expect("dual-tier constructs");
        let key = key_for(b"the derived address");
        plant_large(&store, &key.as_bytes(), b"stored bytes").await;

        assert!(
            store.get(&key.as_bytes()).await.unwrap().is_some(),
            "streaming open never re-derives (mismatching content and all)"
        );
    }

    /// `stat` carries len + algorithm for both algorithms (the
    /// "type" at this layer is the hash algorithm, ADR-002) and is a
    /// `size` ride — the fall-through consult, `None` when absent.
    #[tokio::test]
    async fn stat_carries_len_and_algorithm_from_size_probes() {
        let root = temp_root("read-stat");
        let store = Store::dual_tier_mem_local(root.path(), 1024)
            .await
            .expect("dual-tier constructs");

        let sha256_key = key_for(b"canonical");
        plant_kv(&store, &sha256_key.as_bytes(), b"canonical").await;
        let legacy_key = sha1_key_for(b"legacy");
        plant_large(&store, &legacy_key.as_bytes(), b"legacy").await;

        assert_eq!(
            store.stat(&sha256_key.as_bytes()).await.unwrap(),
            Some(EntryMeta {
                len: 9,
                algorithm: Algorithm::GitBlobSha256
            })
        );
        assert_eq!(
            store
                .stat(&legacy_key.as_bytes())
                .await
                .unwrap()
                .expect("legacy entry present")
                .algorithm,
            Algorithm::GitBlobSha1,
            "the algorithm comes from the key"
        );
        assert_eq!(
            store.stat(&sha256_key.as_bytes()).await.unwrap(),
            store.stat(&sha256_key.as_bytes()).await.unwrap(),
            "stat is a pure probe"
        );
    }

    /// `list` returns the sorted, typed union of both tiers'
    /// enumerations — both algorithms, dedup-free by key identity.
    #[tokio::test]
    async fn list_returns_the_sorted_typed_union_of_both_tiers() {
        let root = temp_root("read-list");
        let store = Store::dual_tier_mem_local(root.path(), 1024)
            .await
            .expect("dual-tier constructs");

        let kv_keys = [
            key_for(b"kv entry one"),
            sha1_key_for(b"kv legacy"),
            key_for(b"kv entry two"),
        ];
        for key in &kv_keys {
            plant_kv(&store, &key.as_bytes(), b"kv bytes").await;
        }
        let large_keys = [key_for(b"large entry one"), key_for(b"large entry two")];
        for key in &large_keys {
            plant_large(&store, &key.as_bytes(), b"large bytes").await;
        }

        let mut expected: Vec<Key> = kv_keys.into_iter().chain(large_keys).collect();
        expected.sort();

        let listed = store.list().await.expect("list succeeds");
        assert_eq!(listed, expected, "complete, typed, both tiers");
        for window in listed.windows(2) {
            assert!(
                window[0] < window[1],
                "output is sorted in the keys' total order"
            );
        }
        for key in &listed {
            assert!(
                store.has(&key.as_bytes()).await.unwrap(),
                "every listed key addresses a real entry"
            );
        }
    }

    /// A malformed pool list is fatal (`Error::KeyInvalid`), never a
    /// silent skip — ADR-003's malformed-list ruling applied at the
    /// facade boundary; the engines own their own body-level checks.
    #[tokio::test]
    async fn list_of_malformed_pool_keys_is_fatal() {
        let store = Store::mem_only();
        plant_kv(&store, &[0x07; 33], b"stray in the table").await;

        assert_eq!(
            store.list().await.unwrap_err(),
            Error::KeyInvalid,
            "malformed pool bytes are fatal on list"
        );
    }

    /// `KeyInvalid` at the facade boundary on every key-addressed
    /// op (typed conversion happens there and only there, ADR-002/003)
    /// — malformed bytes reject on hit paths (a planted entry) and
    /// miss paths alike, before any tier is consulted and never as a
    /// `None`/`false` guess.
    #[tokio::test]
    async fn malformed_key_bytes_reject_keyinvalid_on_every_op() {
        let store = Store::mem_only();
        let anchor = key_for(b"an anchor entry");
        plant_kv(&store, &anchor.as_bytes(), b"anchor").await;

        let key_bytes = anchor.as_bytes();
        let malformed = [
            Vec::new(),                                // empty
            key_bytes[..key_bytes.len() - 1].to_vec(), // truncated digest
            [0x00u8; 33].to_vec(),                     // reserved tag, full width
            [0x03u8; 33].to_vec(),                     // unknown tag
            [0x01u8; 21].to_vec(),                     // right tag, wrong digest length
        ];
        for raw in &malformed {
            assert_eq!(
                store.get(raw).await.unwrap_err(),
                Error::KeyInvalid,
                "get rejects malformed key bytes: {raw:?}"
            );
            assert_eq!(
                store.stat(raw).await.unwrap_err(),
                Error::KeyInvalid,
                "stat rejects malformed key bytes: {raw:?}"
            );
            assert_eq!(
                store.read_range(raw, 0, 1).await.unwrap_err(),
                Error::KeyInvalid,
                "read_range rejects malformed key bytes: {raw:?}"
            );
            assert_eq!(
                store.has(raw).await.unwrap_err(),
                Error::KeyInvalid,
                "has rejects malformed key bytes: {raw:?}"
            );
        }
    }
}