alkblobs 0.2.2

//! The engine-side enum dispatch: the two tier engine enums (the
//! engine-representation design point's closed sets) and the
//! cursor/staging dispatch enums (the sub-enum design points) between
//! the facade and the backends. Everything here is crate-internal —
//! engine types never leak onto the public surface. The per-verb
//! dispatch bodies (the `Store` methods consuming these enums) live
//! with their paths in the sibling files.

use crate::backend::{Backend, ReadCursor, StagingCursor};
#[cfg(any(
    feature = "postgres",
    feature = "sqlite",
    all(feature = "mem", feature = "pg-lo")
))]
use crate::gc::EngineWindowState;
use crate::local::{LocalBackend, LocalCursor, LocalStaging};
#[cfg(feature = "mem")]
use crate::mem::{MemBackend, MemCursor};
#[cfg(feature = "pg-lo")]
use crate::pg_lo::{PgLoBackend, PgLoCursor, PgLoStaging};
#[cfg(feature = "postgres")]
use crate::postgres::{PostgresBackend, PostgresCursor};
#[cfg(feature = "sqlite")]
use crate::sqlite::{SqliteBackend, SqliteCursor};
use crate::Error;
#[cfg(feature = "postgres")]
use std::sync::Arc;

/// **The kv dispatches' empty-build arm** (a build whose feature set
/// enables no kv engine — local-only, pg-lo-only): every kv variant
/// is feature-gated, so [`KvEngine`]/[`KvCursor`] compile with zero
/// variants and every `&self` match needs one wildcard arm to remain
/// exhaustive (rustc treats references as inhabited even over a
/// variant-less enum). The arm is unreachable by construction — every
/// constructor is kv-engine-gated, so `Store` cannot exist in such a
/// build — and closes the dispatch with an error, never a panic (the
/// library code's no-panic convention).
#[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
pub(crate) fn no_kv_engine() -> Error {
    Error::Io(
        "no kv engine: this build enables no kv-tier engine feature, so Store cannot be \
         constructed and the kv dispatch is unreachable by construction"
            .to_owned(),
    )
}

/// 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 — landed, the default),
/// postgres (wave 4 — landed, feature-gated default-off).
///
/// The variant payloads are structurally dissimilar by design — the
/// in-process engine holds one `Arc`, the SQL-backed engines hold
/// their connection postures inline — and every dispatch rides the
/// enum by reference (`&self` matches only), so the size difference
/// costs nothing.
#[allow(clippy::large_enum_variant)]
pub(crate) enum KvEngine {
    /// The contract-reference ephemeral engine (ADR-011 §2).
    #[cfg(feature = "mem")]
    Mem(MemBackend),
    /// The default build's kv engine — the first SQL-backed engine,
    /// the durable delete window's host (ADR-012 §5's one-window-host
    /// rule: the window is the kv engine's when the kv engine is
    /// SQL-backed).
    #[cfg(feature = "sqlite")]
    Sqlite(SqliteBackend),
    /// The second SQL-backed engine — the fleet host (ADR-012 §4) and
    /// a one-window-host alongside sqlite: its constructors run the
    /// GC schema's engine-neutral call (the postgres row's body
    /// creates `gc_pins` + `gc_window`) and its sweeps run the durable
    /// window form on it. Feature `postgres`, default-off. The payload
    /// is an `Arc` shared with the store core's detached pin-renewal
    /// tasks (fleet puts mint one per guard — the task outlives the
    /// `Store`, renewing its row through the engine handle it carries;
    /// `PostgresBackend` itself is not `Clone` — its lease checkout
    /// must not be), so the fleet constructor's one backend serves both
    /// the in-line dispatch and the detached duties.
    #[cfg(feature = "postgres")]
    Postgres(Arc<PostgresBackend>),
}

/// The large-tier engine set. `local` is ungated (dependency-free,
/// always present, ADR-013 §4); pg-lo landed behind feature `pg-lo`
/// (wave 5) — its slotting is exactly the enum-variant-plus-named-
/// constructor pattern the module doc pins.
#[cfg_attr(
    not(any(feature = "mem", feature = "sqlite", feature = "postgres")),
    allow(dead_code)
)]
pub(crate) enum LargeEngine {
    /// The sharded-filesystem engine (ADR-003/008).
    Local(LocalBackend),
    /// The pg-lo large-object engine (ADR-009): postgres Large
    /// Objects hold the bytes, the companion table is the addressing
    /// authority. Feature `pg-lo`, default-off (it does not imply
    /// feature `postgres`, ADR-013 §4).
    #[cfg(feature = "pg-lo")]
    PgLo(PgLoBackend),
}

/// **The large-arm shape over multiple large engines (named design
/// point): a `LargeCursor` sub-enum behind [`AnyCursor`]'s one large
/// arm — the recorded [`KvCursor`] sub-enum pattern's large-tier
/// twin, on the same bounded-memory framing as the kv side.** The
/// alternative — one `AnyCursor` variant per large engine — would
/// copy the large streaming arm into the tier dispatch per engine,
/// and the tier dispatch's variant count must stay bounded so
/// `Blob`-tier consumers (the verification posture's arm shapes in
/// `Store::get`) stay a two-arm tier dispatch while large engines
/// multiply inside this one. Recorded for the review gate and landed
/// with wave 5: pg-lo slotted one arm in here (exactly the shape the
/// design point named, no reshaping).
pub(crate) enum LargeCursor {
    /// The pread body over the committed file (ADR-003/008).
    Local(LocalCursor),
    /// The descriptorless window body over the companion row —
    /// `lo_get` windows (ADR-009).
    #[cfg(feature = "pg-lo")]
    PgLo(PgLoCursor),
}

impl ReadCursor for LargeCursor {
    fn len(&self) -> u64 {
        match self {
            Self::Local(cursor) => cursor.len(),
            #[cfg(feature = "pg-lo")]
            Self::PgLo(cursor) => cursor.len(),
        }
    }

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

/// **The kv-arm shape over multiple kv engines (named design point):
/// a `KvCursor` sub-enum behind [`AnyCursor`]'s one kv arm.** The
/// alternative — one `AnyCursor` variant per kv engine — would copy
/// the kv materializing arm (and its consumers: the verification-on-
/// get body, `Blob`'s drive) per engine without adding a behavior;
/// the sub-enum keeps [`AnyCursor`] a two-arm tier dispatch (kv /
/// large) while kv engines multiply inside this one. Recorded for the
/// review gate and landed with wave 4: postgres slotted one arm in
/// here (exactly the predicted shape, no reshaping).
pub(crate) enum KvCursor {
    /// The contract-reference ephemeral body (ADR-011 §2).
    #[cfg(feature = "mem")]
    Mem(MemCursor),
    /// The default build's materializing body: the row's value loaded
    /// once, reads in-memory slices (the sqlite engine's record).
    #[cfg(feature = "sqlite")]
    Sqlite(SqliteCursor),
    /// The postgres body — the same materializing shape, the value
    /// loaded once from the engine's pooled row (the postgres
    /// engine's record).
    #[cfg(feature = "postgres")]
    Postgres(PostgresCursor),
}

#[cfg_attr(
    not(any(feature = "mem", feature = "sqlite", feature = "postgres")),
    allow(unused_variables)
)]
impl ReadCursor for KvCursor {
    fn len(&self) -> u64 {
        match self {
            #[cfg(feature = "mem")]
            Self::Mem(cursor) => cursor.len(),
            #[cfg(feature = "postgres")]
            Self::Postgres(cursor) => cursor.len(),
            #[cfg(feature = "sqlite")]
            Self::Sqlite(cursor) => cursor.len(),
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => 0,
        }
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        match self {
            #[cfg(feature = "mem")]
            Self::Mem(cursor) => cursor.read_at(offset, len).await,
            #[cfg(feature = "postgres")]
            Self::Postgres(cursor) => cursor.read_at(offset, len).await,
            #[cfg(feature = "sqlite")]
            Self::Sqlite(cursor) => cursor.read_at(offset, len).await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }
}

/// 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`). The per-tier body families live
/// behind the sub-enums — [`KvCursor`] for the kv arm, [`LargeCursor`]
/// for the large arm (both the recorded design points).
#[cfg_attr(
    not(any(feature = "mem", feature = "sqlite", feature = "postgres")),
    allow(dead_code)
)]
pub(crate) enum AnyCursor {
    /// The kv body family: the whole value materialized in memory
    /// (bounded by tier policy).
    Kv(KvCursor),
    /// The large body family: streaming readers over the committed
    /// home, behind the large-tier sub-enum.
    Large(LargeCursor),
}

impl ReadCursor for AnyCursor {
    fn len(&self) -> u64 {
        match self {
            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 {
            Self::Kv(cursor) => cursor.read_at(offset, len).await,
            Self::Large(cursor) => cursor.read_at(offset, len).await,
        }
    }
}

/// The write path's staging-cursor dispatch (ADR-010 §1: the store core
/// programs the abstraction): the large tier's stage bodies behind one
/// crate-internal value, the put seam's twin of [`AnyCursor`]. The kv
/// tier never stages on the facade path (the whole-value put is its
/// settled form); a kv-degenerate staged body variant arrives only if
/// the conformance suite demands the staged form through the facade.
/// Crate-internal — engine cursors never leak onto the public surface.
/// The stage bodies live behind [`LargeStaging`] (the sub-enum design
/// point's staging-side twin).
pub(crate) enum AnyStaging {
    /// The large body family: stage bodies behind the large-tier
    /// sub-enum.
    Large(LargeStaging),
}

impl ReadCursor for AnyStaging {
    fn len(&self) -> u64 {
        match self {
            Self::Large(cursor) => cursor.len(),
        }
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        match self {
            Self::Large(cursor) => cursor.read_at(offset, len).await,
        }
    }
}

/// The staged arm's sub-enum — the [`LargeCursor`]/staging-side twin
/// of the `KvCursor` pattern: the local stage-file body and pg-lo's
/// held-transaction body behind [`AnyStaging`]'s one large arm rather
/// than one `AnyStaging` variant per large engine (the same
/// bounded-variant argument; landed with wave 5). The variant payloads
/// are structurally dissimilar by design (the file-descriptor posture
/// vs the held pooled checkout + tx) and every dispatch rides the enum
/// by value-consumption or `&self`, so the size difference costs
/// nothing — the recorded `KvEngine` enum's allow posture.
#[allow(clippy::large_enum_variant)]
pub(crate) enum LargeStaging {
    /// The stage-file body under the stage-hygiene protocol,
    /// published by commit-rename (ADR-003/008).
    Local(LocalStaging),
    /// The held-transaction body: the stage lives in uncommitted LO
    /// pages, published by the companion-row insert + COMMIT
    /// (ADR-009).
    #[cfg(feature = "pg-lo")]
    PgLo(PgLoStaging),
}

impl ReadCursor for LargeStaging {
    fn len(&self) -> u64 {
        match self {
            Self::Local(cursor) => cursor.len(),
            #[cfg(feature = "pg-lo")]
            Self::PgLo(cursor) => cursor.len(),
        }
    }

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

impl StagingCursor for LargeStaging {
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        match self {
            Self::Local(cursor) => cursor.append(chunk).await,
            #[cfg(feature = "pg-lo")]
            Self::PgLo(cursor) => cursor.append(chunk).await,
        }
    }

    async fn commit(self, key: &[u8]) -> Result<(), Error> {
        match self {
            Self::Local(cursor) => cursor.commit(key).await,
            #[cfg(feature = "pg-lo")]
            Self::PgLo(cursor) => cursor.commit(key).await,
        }
    }

    async fn discard(self) {
        match self {
            Self::Local(cursor) => cursor.discard().await,
            #[cfg(feature = "pg-lo")]
            Self::PgLo(cursor) => cursor.discard().await,
        }
    }
}

impl StagingCursor for AnyStaging {
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        match self {
            Self::Large(cursor) => cursor.append(chunk).await,
        }
    }

    async fn commit(self, key: &[u8]) -> Result<(), Error> {
        match self {
            Self::Large(cursor) => cursor.commit(key).await,
        }
    }

    async fn discard(self) {
        match self {
            Self::Large(cursor) => cursor.discard().await,
        }
    }
}

#[cfg_attr(
    not(any(feature = "mem", feature = "sqlite", feature = "postgres")),
    allow(unused_variables)
)]
impl KvEngine {
    pub(crate) async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(engine) => engine.put(key, value).await,
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine.put(key, value).await,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine.put(key, value).await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }

    pub(crate) async fn get(&self, key: &[u8]) -> Result<Option<AnyCursor>, Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(engine) => Ok(engine
                .get(key)
                .await?
                .map(|c| AnyCursor::Kv(KvCursor::Mem(c)))),
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => Ok(engine
                .get(key)
                .await?
                .map(|c| AnyCursor::Kv(KvCursor::Postgres(c)))),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => Ok(engine
                .get(key)
                .await?
                .map(|c| AnyCursor::Kv(KvCursor::Sqlite(c)))),
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }

    /// The GC schema's engine-neutral kv-host arm (ADR-010 §2's
    /// companion seam; ADR-012 §5's applicability table): a SQL-backed
    /// kv engine creates the tables its applicability row names —
    /// sqlite's body creates `gc_window` only, postgres's creates
    /// `gc_pins` + `gc_window` (the fleet host's both-tables row) — and
    /// an in-process kv engine is structurally nothing (the
    /// local/mem row of the table: those engines execute the window
    /// protocol inline, ADR-005, and carry no GC schema). The
    /// store-core construction call is
    /// [`ensure_gc_schema_at_construction`] — the one-window-host
    /// rule's dispatch, which consults this method on the SQL-backed
    /// kv arms and the large tier only when the kv engine is not
    /// SQL-backed.
    // consumer: `ensure_gc_schema_at_construction`'s SQL-backed kv
    // arms (features `sqlite`/`postgres`); a build whose kv set holds
    // only mem compiles the closed dispatch unread.
    #[allow(dead_code)]
    pub(crate) async fn ensure_gc_schema(&self) -> Result<(), Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => Ok(()),
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine.ensure_gc_schema().await,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine.ensure_gc_schema().await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }

    pub(crate) async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(engine) => engine.has(key).await,
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine.has(key).await,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine.has(key).await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }

    pub(crate) async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(engine) => engine.size(key).await,
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine.size(key).await,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine.size(key).await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }

    pub(crate) async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(engine) => engine.delete(key).await,
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine.delete(key).await,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine.delete(key).await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }

    pub(crate) async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        match self {
            #[cfg(feature = "mem")]
            KvEngine::Mem(engine) => engine.list().await,
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine.list().await,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine.list().await,
            #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
            _ => Err(no_kv_engine()),
        }
    }
}

impl LargeEngine {
    /// The staged form's handout (ADR-010 §1): the put seam's large path
    /// streams into this cursor while hashing; the key resolves at the
    /// staging cursor's commit.
    pub(crate) async fn put_staged(&self) -> Result<AnyStaging, Error> {
        match self {
            LargeEngine::Local(engine) => Ok(engine
                .put_staged()
                .await
                .map(|c| AnyStaging::Large(LargeStaging::Local(c)))?),
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => Ok(engine
                .put_staged()
                .await
                .map(|c| AnyStaging::Large(LargeStaging::PgLo(c)))?),
        }
    }

    pub(crate) async fn get(&self, key: &[u8]) -> Result<Option<AnyCursor>, Error> {
        match self {
            LargeEngine::Local(engine) => Ok(engine
                .get(key)
                .await?
                .map(|c| AnyCursor::Large(LargeCursor::Local(c)))),
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => Ok(engine
                .get(key)
                .await?
                .map(|c| AnyCursor::Large(LargeCursor::PgLo(c)))),
        }
    }

    pub(crate) async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        match self {
            LargeEngine::Local(engine) => engine.has(key).await,
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => engine.has(key).await,
        }
    }

    pub(crate) async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        match self {
            LargeEngine::Local(engine) => engine.size(key).await,
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => engine.size(key).await,
        }
    }

    pub(crate) async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        match self {
            LargeEngine::Local(engine) => engine.delete(key).await,
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => engine.delete(key).await,
        }
    }

    pub(crate) async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        match self {
            LargeEngine::Local(engine) => engine.list().await,
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => engine.list().await,
        }
    }

    /// The large tier's arm of the one-window-host rule's construction
    /// call: the pg-lo host's `ensure_gc_schema` (creating `gc_window`
    /// in the pg-lo schema), consulted by
    /// [`ensure_gc_schema_at_construction`] only when the kv engine is
    /// not SQL-backed. No arm exists for `local` — it is
    /// seam-excluded by construction (ADR-010/012 §5).
    #[cfg(feature = "mem")]
    // consumer: the construction dispatch's large-host arm
    // (kv = mem + large = pg-lo); a pg-lo-less mem build compiles the
    // local arm only, structurally nothing.
    #[allow(dead_code)]
    pub(crate) async fn ensure_gc_schema(&self) -> Result<(), Error> {
        match self {
            #[cfg(feature = "pg-lo")]
            LargeEngine::PgLo(engine) => engine.ensure_gc_schema().await,
            LargeEngine::Local(_) => Ok(()),
        }
    }
}