dynoxide/storage_backend/

mod.rs

//! Storage backend abstraction.
//!
//! Defines the [`StorageBackend`] trait that decouples Dynoxide's data layer
//! from a specific SQLite binding. The native [`rusqlite`]-backed
//! [`Storage`](crate::storage::Storage) implements the trait, and the
//! `wasm-sqlite` build adds `wasm_backend::WasmBridgeBackend`, which runs the
//! same SQL against a JS SQLite database over a wasm-bindgen bridge. Both
//! backends issue identical SQL because they share the builders in
//! [`sql_builders`].
//!
//! The native build consumes the trait monomorphically through `Storage`; the
//! wasm build consumes it through `WasmBridgeBackend`. The escape hatches
//! `Storage::conn()` and `Storage::conn_mut()` are not exposed by the trait
//! and remain native-only.
//!
//! # No `Send + Sync` super-trait
//!
//! [`Storage`](crate::storage::Storage) carries a `RefCell<HashMap<...>>` for
//! its metadata cache, so `Storage: !Sync`. A `Send + Sync` super-trait would
//! refuse the impl on `Storage`. With no dynamic dispatch site in scope,
//! auto-trait propagation across `.await` is decided per-callsite anyway, so
//! adding `Send` to the super-trait would not earn any compile-time
//! guarantee on the futures returned by trait methods.

pub mod clock;
pub mod error;
// Internal: the shared SQL contract between the rusqlite and wasm backends.
// `pub` only because both backend modules consume it across the cfg split; it
// is not a stable API, hence `#[doc(hidden)]`.
#[doc(hidden)]
pub mod sql_builders;
// Schema versions and the migrations a recorded version calls for on open.
pub(crate) mod schema;
// The native rusqlite-backed `Storage` exists whenever either SQLite backend
// feature is on (the crate refuses to build with neither), and the handlers
// now consume `StorageBackend` through it, so the impl must track the same
// condition rather than `native-sqlite` alone.
#[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
pub mod rusqlite_impl;
#[cfg(feature = "wasm-sqlite")]
pub mod wasm_backend;
// Compile-only proof that the trait is satisfiable by a fresh impl. In-crate
// rather than under `tests/`, because a sealed trait cannot be implemented
// from outside.
#[cfg(test)]
mod trait_compile;

use crate::storage::{
    CreateTableMetadata, DatabaseInfo, QueryParams, ScanParams, StreamRecord, TableMetadata,
    TableStats,
};
use crate::types::Tag;

pub use clock::{Clock, ManualClock, SystemClock};
pub use error::BackendError;
#[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
pub use error::from_rusqlite;
#[doc(hidden)]
pub use sql_builders::SqlParam;
#[cfg(feature = "wasm-sqlite")]
pub use wasm_backend::WasmBridgeBackend;

/// One base-table row for a bulk insert via [`StorageBackend::put_base_items`].
///
/// Unlike [`StorageBackend::put_item_with_hash`], which preserves any existing
/// `cached_at` value, the bulk path writes `cached_at` verbatim: this mirrors
/// the import flow, which sets the timestamp explicitly (or clears it) for
/// every row it loads.
#[derive(Debug, Clone)]
pub struct BaseItemRow {
    /// Partition key string.
    pub pk: String,
    /// Sort key string (empty for tables without a sort key).
    pub sk: String,
    /// Serialised item JSON.
    pub item_json: String,
    /// Item size in bytes.
    pub item_size: usize,
    /// Cache timestamp written verbatim; `None` clears the column.
    pub cached_at: Option<f64>,
    /// Hash prefix used for parallel-scan ordering.
    pub hash_prefix: String,
}

/// One GSI-table row for a bulk insert via [`StorageBackend::insert_gsi_items`].
///
/// The fields mirror the argument order of the single-row
/// [`StorageBackend::insert_gsi_item`].
#[derive(Debug, Clone)]
pub struct GsiItemRow {
    /// GSI partition key string.
    pub gsi_pk: String,
    /// GSI sort key string (empty when the index has no sort key).
    pub gsi_sk: String,
    /// Base-table partition key string.
    pub table_pk: String,
    /// Base-table sort key string.
    pub table_sk: String,
    /// Projected item JSON.
    pub item_json: String,
}

/// The DynamoDB size of one stored index entry, read from its projected JSON.
/// Both backends size a non-ALL index this way, so they agree on it.
pub(crate) fn projected_entry_size(item_json: &str) -> Result<i64, BackendError> {
    let item: crate::types::Item = serde_json::from_str(item_json)
        .map_err(|e| BackendError::Other(format!("unreadable index entry: {e}")))?;
    Ok(crate::types::item_size(&item) as i64)
}

/// One vector shadow-table row for a bulk insert via
/// [`StorageBackend::insert_vector_items`].
///
/// The fields mirror the shadow table's column order.
#[derive(Debug, Clone)]
pub struct VectorItemRow {
    /// Base-table partition key string.
    pub table_pk: String,
    /// Base-table sort key string.
    pub table_sk: String,
    /// SearchSchema HASH attribute key string (empty when the schema declares
    /// no HASH element).
    pub hash_value: String,
    /// The f32-truncated vector as a JSON array.
    pub vector_json: String,
    /// INLINE_FILTER attribute values as a JSON object.
    pub filter_json: String,
    /// Projected item JSON.
    pub item_json: String,
    /// The entry's billable size in bytes, by the captured write formula.
    /// Stored so a search sums a column instead of rebuilding every entry it
    /// scanned.
    pub entry_bytes: i64,
}

/// One candidate row loaded from a vector shadow table for scoring by
/// SearchVectors, via [`StorageBackend::query_vector_candidates`].
///
/// Carries no `hash_value`: scoping by HASH value happens in the query
/// itself, so every returned row already matches the scope.
#[derive(Debug, Clone)]
pub struct VectorCandidateRow {
    /// Base-table partition key string.
    pub table_pk: String,
    /// Base-table sort key string.
    pub table_sk: String,
    /// The stored f32-truncated vector as a JSON array.
    pub vector_json: String,
    /// INLINE_FILTER attribute values as a wire-shaped JSON object.
    pub filter_json: String,
    /// The entry's billable size, read back rather than recomputed. The
    /// projected item is deliberately absent: a search scores from the vector
    /// and the filter, and only the rows it returns need the item, so carrying
    /// it here made a scan of a large index materialise every entry to throw
    /// almost all of them away.
    pub entry_bytes: i64,
}

/// One index-table write operation, backend-neutral.
///
/// The per-write and per-delete GSI/LSI fan-out builds an ordered list of these
/// and hands it to [`StorageBackend::apply_index_writes`] in a single call. The
/// default impl replays each op through the matching per-item method, identical
/// to the per-op loop it replaces; the wasm backend overrides it to collapse the
/// list into one bridge crossing. Each variant's fields mirror the argument
/// order of the per-item method it stands in for.
#[derive(Debug, Clone)]
pub enum IndexWriteOp {
    /// Remove this base key's entry from a GSI table.
    DeleteGsi {
        table_name: String,
        index_name: String,
        table_pk: String,
        table_sk: String,
    },
    /// Insert (or replace) this item's projected entry into a GSI table.
    InsertGsi {
        table_name: String,
        index_name: String,
        gsi_pk: String,
        gsi_sk: String,
        table_pk: String,
        table_sk: String,
        item_json: String,
    },
    /// Remove this base key's entry from an LSI table.
    DeleteLsi {
        table_name: String,
        index_name: String,
        base_pk: String,
        base_sk: String,
    },
    /// Insert (or replace) this item's projected entry into an LSI table.
    InsertLsi {
        table_name: String,
        index_name: String,
        pk: String,
        sk: String,
        base_pk: String,
        base_sk: String,
        item_json: String,
    },
    /// Remove this base key's row from a vector index shadow table.
    DeleteVector {
        table_name: String,
        index_name: String,
        table_pk: String,
        table_sk: String,
    },
    /// Insert (or replace) this item's derived row into a vector index shadow
    /// table. Carries the whole [`VectorItemRow`] rather than flattened fields
    /// because the row struct already mirrors the shadow table's column order.
    /// Boxed so the six-string row does not widen the whole enum: every op in
    /// a fan-out pays for the largest variant, and the size should stay
    /// pinned by `InsertGsi` (see the size test below).
    InsertVector {
        table_name: String,
        index_name: String,
        row: Box<VectorItemRow>,
    },
}

/// Restricts [`StorageBackend`] to this crate's own backends.
///
/// The trait's method set tracks engine features rather than a stable
/// interface: vector index support added seven methods to it in a single
/// release. Sealing keeps that growth out of the versioning contract, because
/// a trait nothing downstream can implement cannot be broken by a new method.
/// The trait stays nameable, so [`Database`](crate::Database) and the wasm
/// dispatch bounds are unaffected.
///
/// Unsealing later is a minor, if third-party backends ever earn their place.
/// Sealing after 1.0.0 would not be, which is why it happens before.
mod private {
    pub trait Sealed {}

    #[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
    impl Sealed for crate::storage::Storage {}

    #[cfg(feature = "wasm-sqlite")]
    impl Sealed for super::wasm_backend::WasmBridgeBackend {}
}

/// Backend-neutral storage interface.
///
/// Sealed: implemented by this crate's own backends only. It is public because
/// [`Database`](crate::Database) is generic over it and the wasm dispatch
/// functions take it as a bound, so it can be named and used as a bound from
/// anywhere. Its method set tracks engine features rather than a stable
/// interface, so closing it to outside impls keeps adding one out of the
/// versioning contract. The [versioning policy] carries the reasoning.
///
/// [versioning policy]: https://github.com/nubo-db/dynoxide/blob/main/docs/versioning.md
///
/// Method signatures mirror [`Storage`](crate::storage::Storage)'s public
/// surface, with three mechanical transformations:
///
/// 1. `Result<T, DynoxideError>` becomes `Result<T, BackendError>`.
/// 2. `fn` becomes `async fn`.
/// 3. Filesystem-typed and rusqlite-typed methods are excluded; they remain
///    on the native [`Storage`](crate::storage::Storage) only.
///
/// Methods marked `#[doc(hidden)]` serve this crate's own action handlers,
/// have no public `Storage` counterpart and are not part of the stable API.
///
/// The trait is not consumed dynamically today. The native
/// [`Storage`](crate::storage::Storage) and the wasm
/// `WasmBridgeBackend` each implement it
/// monomorphically.
///
/// The `#[allow(async_fn_in_trait)]` reflects the monomorphic-only consumption
/// model. The lint can be revisited if and when `dyn StorageBackend` becomes
/// a real callsite.
#[allow(async_fn_in_trait)]
pub trait StorageBackend: private::Sealed {
    // -----------------------------------------------------------------------
    // Capabilities
    // -----------------------------------------------------------------------

    /// Wall-clock access for the stream and TTL paths.
    ///
    /// Sync because reading the clock is not I/O. The native backend returns
    /// its injected [`Clock`]; the wasm SQLite backend supplies its own.
    fn clock(&self) -> &dyn Clock;

    /// Whether this backend can record and serve DynamoDB Streams.
    ///
    /// A backend that answers `false` still gets its stream methods called
    /// nowhere on the happy path: the actions consult this before mutating, so
    /// a request that needs streams is refused before anything is created,
    /// rather than half-applied and then failed. Defaults to `true`; the wasm
    /// backend answers `false` until a delivery mechanism exists.
    fn supports_streams(&self) -> bool {
        true
    }

    /// Whether this backend stores resource tags. Same contract as
    /// [`supports_streams`](Self::supports_streams): consulted before a
    /// mutation that would need `set_tags`, so a tagged `CreateTable` on a
    /// backend without tags is refused whole rather than creating the table
    /// and then failing.
    fn supports_tags(&self) -> bool {
        true
    }

    // -----------------------------------------------------------------------
    // Table metadata
    // -----------------------------------------------------------------------

    async fn insert_table_metadata(&self, m: &CreateTableMetadata<'_>) -> Result<(), BackendError>;

    async fn get_table_metadata(
        &self,
        table_name: &str,
    ) -> Result<Option<TableMetadata>, BackendError>;

    async fn delete_table_metadata(&self, table_name: &str) -> Result<bool, BackendError>;

    async fn update_table_metadata(
        &self,
        table_name: &str,
        attribute_definitions: &str,
        gsi_definitions: Option<&str>,
    ) -> Result<(), BackendError>;

    /// Update a table's vector index definitions. The write the
    /// add/delete-vector path of `UpdateTable` issues; `None` clears the
    /// column when no vector indexes remain.
    async fn update_vector_index_definitions(
        &self,
        table_name: &str,
        vector_index_definitions: Option<&str>,
    ) -> Result<(), BackendError>;

    async fn update_provisioned_throughput(
        &self,
        table_name: &str,
        provisioned_throughput: &str,
    ) -> Result<(), BackendError>;

    async fn clear_provisioned_throughput(&self, table_name: &str) -> Result<(), BackendError>;

    async fn update_billing_mode(
        &self,
        table_name: &str,
        billing_mode: &str,
    ) -> Result<(), BackendError>;

    async fn update_table_class(
        &self,
        table_name: &str,
        table_class: &str,
    ) -> Result<(), BackendError>;

    /// Store the serialised on-demand throughput ceilings. Implementations
    /// must store the payload opaquely: the `clear_on_demand_throughput`
    /// default passes a JSON `null` through this method, which every reader
    /// treats as absent, so parsing or rejecting the payload here would break
    /// that default.
    async fn update_on_demand_throughput(
        &self,
        table_name: &str,
        on_demand_throughput: &str,
    ) -> Result<(), BackendError>;

    /// Remove any stored on-demand throughput ceilings. The default stores a
    /// JSON `null`, which every reader treats as absent, so existing backend
    /// implementations keep working without changes; the in-tree backends
    /// override this to clear the underlying value properly.
    async fn clear_on_demand_throughput(&self, table_name: &str) -> Result<(), BackendError> {
        self.update_on_demand_throughput(table_name, "null").await
    }

    async fn get_tags(&self, table_name: &str) -> Result<Vec<Tag>, BackendError>;

    async fn set_tags(&self, table_name: &str, new_tags: &[Tag]) -> Result<(), BackendError>;

    async fn update_deletion_protection(
        &self,
        table_name: &str,
        enabled: bool,
    ) -> Result<(), BackendError>;

    async fn remove_tags(&self, table_name: &str, keys: &[String]) -> Result<(), BackendError>;

    async fn list_table_names(&self) -> Result<Vec<String>, BackendError>;

    async fn table_exists(&self, table_name: &str) -> Result<bool, BackendError>;

    // -----------------------------------------------------------------------
    // Dynamic data tables (DDL)
    // -----------------------------------------------------------------------

    async fn create_data_table(&self, table_name: &str) -> Result<(), BackendError>;

    async fn drop_data_table(&self, table_name: &str) -> Result<(), BackendError>;

    async fn create_gsi_table(
        &self,
        table_name: &str,
        index_name: &str,
    ) -> Result<(), BackendError>;

    async fn drop_gsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError>;

    async fn create_lsi_table(
        &self,
        table_name: &str,
        index_name: &str,
    ) -> Result<(), BackendError>;

    async fn drop_lsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError>;

    /// Create the physical shadow table for a vector index.
    async fn create_vector_table(
        &self,
        table_name: &str,
        index_name: &str,
    ) -> Result<(), BackendError>;

    /// Drop the physical shadow table for a vector index, if it exists.
    async fn drop_vector_table(
        &self,
        table_name: &str,
        index_name: &str,
    ) -> Result<(), BackendError>;

    /// Bulk-insert many rows into one vector shadow table.
    ///
    /// Batch-shaped so a backend can amortise per-row round-trips, mirroring
    /// [`insert_gsi_items`](Self::insert_gsi_items). Used by the vector index
    /// backfill path, and by the default
    /// [`apply_index_writes`](Self::apply_index_writes) with a one-row slice
    /// for live-write maintenance.
    async fn insert_vector_items(
        &self,
        table_name: &str,
        index_name: &str,
        rows: &[VectorItemRow],
    ) -> Result<(), BackendError>;

    /// Delete a vector shadow-table row by base-table primary key. Used by the
    /// live-write maintenance fan-out, mirroring
    /// [`delete_gsi_item`](Self::delete_gsi_item).
    async fn delete_vector_item(
        &self,
        table_name: &str,
        index_name: &str,
        table_pk: &str,
        table_sk: &str,
    ) -> Result<(), BackendError>;

    /// Load the rows SearchVectors scores, optionally scoped to one
    /// SearchSchema HASH value (key-string encoded, matching the write
    /// path's `hash_value` column). Rows come back ordered by base-table
    /// primary key so equal scores tie-break deterministically.
    async fn query_vector_candidates(
        &self,
        table_name: &str,
        index_name: &str,
        hash_value: Option<&str>,
    ) -> Result<Vec<VectorCandidateRow>, BackendError>;

    /// Load the projected entries for the base-table keys a search returns.
    ///
    /// Paired with [`query_vector_candidates`](Self::query_vector_candidates),
    /// which leaves the projected item behind so a scan of a large index does
    /// not materialise every entry only to score it and drop it. `keys` are the
    /// post-`TopK` survivors, so this reads at most `TopK` rows. Returns
    /// `(table_pk, table_sk, item_json)` per row found; a key with no row is
    /// omitted rather than erroring.
    async fn vector_items_for_keys(
        &self,
        table_name: &str,
        index_name: &str,
        keys: &[(String, String)],
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    // -----------------------------------------------------------------------
    // GSI item operations
    // -----------------------------------------------------------------------

    #[allow(clippy::too_many_arguments)]
    async fn insert_gsi_item(
        &self,
        table_name: &str,
        index_name: &str,
        gsi_pk: &str,
        gsi_sk: &str,
        table_pk: &str,
        table_sk: &str,
        item_json: &str,
    ) -> Result<(), BackendError>;

    /// Bulk-insert many rows into one GSI table.
    ///
    /// Batch-shaped so a backend can amortise per-row round-trips (the native
    /// backend reuses a single cached prepared statement). Used by the GSI
    /// backfill path; the per-row [`insert_gsi_item`](Self::insert_gsi_item)
    /// covers single writes during normal fan-out.
    async fn insert_gsi_items(
        &self,
        table_name: &str,
        index_name: &str,
        rows: &[GsiItemRow],
    ) -> Result<(), BackendError>;

    async fn delete_gsi_item(
        &self,
        table_name: &str,
        index_name: &str,
        table_pk: &str,
        table_sk: &str,
    ) -> Result<(), BackendError>;

    async fn query_gsi_items(
        &self,
        table_name: &str,
        index_name: &str,
        gsi_pk: &str,
        params: &QueryParams<'_>,
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    async fn scan_gsi_items(
        &self,
        table_name: &str,
        index_name: &str,
        params: &ScanParams<'_>,
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    // -----------------------------------------------------------------------
    // LSI item operations
    // -----------------------------------------------------------------------

    #[allow(clippy::too_many_arguments)]
    async fn insert_lsi_item(
        &self,
        table_name: &str,
        index_name: &str,
        pk: &str,
        sk: &str,
        base_pk: &str,
        base_sk: &str,
        item_json: &str,
    ) -> Result<(), BackendError>;

    async fn delete_lsi_item(
        &self,
        table_name: &str,
        index_name: &str,
        base_pk: &str,
        base_sk: &str,
    ) -> Result<(), BackendError>;

    async fn query_lsi_items(
        &self,
        table_name: &str,
        index_name: &str,
        pk: &str,
        params: &QueryParams<'_>,
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    async fn scan_lsi_items(
        &self,
        table_name: &str,
        index_name: &str,
        params: &ScanParams<'_>,
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    // -----------------------------------------------------------------------
    // Index write fan-out
    // -----------------------------------------------------------------------

    /// Apply an ordered batch of GSI/LSI write operations.
    ///
    /// The GSI/LSI maintenance helpers build the list and call this once per
    /// fan-out instead of invoking the per-item methods one at a time. The
    /// default impl replays each op through the matching per-item method in
    /// order, so a backend that does not override it behaves exactly as the
    /// per-op loop did. The wasm backend overrides this to issue the whole list
    /// in a single bridge crossing.
    ///
    /// Owns no transaction: the caller's open transaction supplies atomicity, so
    /// a mid-batch failure is rolled back by that caller. An empty list does no
    /// work.
    async fn apply_index_writes(&self, ops: &[IndexWriteOp]) -> Result<(), BackendError> {
        for op in ops {
            match op {
                IndexWriteOp::DeleteGsi {
                    table_name,
                    index_name,
                    table_pk,
                    table_sk,
                } => {
                    self.delete_gsi_item(table_name, index_name, table_pk, table_sk)
                        .await?;
                }
                IndexWriteOp::InsertGsi {
                    table_name,
                    index_name,
                    gsi_pk,
                    gsi_sk,
                    table_pk,
                    table_sk,
                    item_json,
                } => {
                    self.insert_gsi_item(
                        table_name, index_name, gsi_pk, gsi_sk, table_pk, table_sk, item_json,
                    )
                    .await?;
                }
                IndexWriteOp::DeleteLsi {
                    table_name,
                    index_name,
                    base_pk,
                    base_sk,
                } => {
                    self.delete_lsi_item(table_name, index_name, base_pk, base_sk)
                        .await?;
                }
                IndexWriteOp::InsertLsi {
                    table_name,
                    index_name,
                    pk,
                    sk,
                    base_pk,
                    base_sk,
                    item_json,
                } => {
                    self.insert_lsi_item(
                        table_name, index_name, pk, sk, base_pk, base_sk, item_json,
                    )
                    .await?;
                }
                IndexWriteOp::DeleteVector {
                    table_name,
                    index_name,
                    table_pk,
                    table_sk,
                } => {
                    self.delete_vector_item(table_name, index_name, table_pk, table_sk)
                        .await?;
                }
                IndexWriteOp::InsertVector {
                    table_name,
                    index_name,
                    row,
                } => {
                    // The batch-shaped insert with a one-row slice: no separate
                    // per-item method exists because the row struct already
                    // carries the full column set.
                    self.insert_vector_items(
                        table_name,
                        index_name,
                        std::slice::from_ref(row.as_ref()),
                    )
                    .await?;
                }
            }
        }
        Ok(())
    }

    // -----------------------------------------------------------------------
    // Transactions
    // -----------------------------------------------------------------------

    async fn begin_transaction(&self) -> Result<(), BackendError>;
    async fn commit(&self) -> Result<(), BackendError>;
    async fn rollback(&self) -> Result<(), BackendError>;

    // -----------------------------------------------------------------------
    // Schema version
    // -----------------------------------------------------------------------

    /// The schema version the database records, or `None` when it records
    /// none, or one that isn't a number.
    async fn recorded_schema_version(&self) -> Result<Option<u32>, BackendError>;

    /// Record `version` as the database's schema version, replacing any
    /// recorded one.
    async fn record_schema_version(&self, version: u32) -> Result<(), BackendError>;

    // -----------------------------------------------------------------------
    // Bulk-loading PRAGMAs
    // -----------------------------------------------------------------------

    async fn enable_bulk_loading(&self) -> Result<(), BackendError>;
    async fn disable_bulk_loading(&self) -> Result<(), BackendError>;

    // -----------------------------------------------------------------------
    // Item CRUD
    // -----------------------------------------------------------------------

    async fn put_item(
        &self,
        table_name: &str,
        pk: &str,
        sk: &str,
        item_json: &str,
        item_size: usize,
    ) -> Result<Option<String>, BackendError>;

    #[allow(clippy::too_many_arguments)]
    async fn put_item_with_hash(
        &self,
        table_name: &str,
        pk: &str,
        sk: &str,
        item_json: &str,
        item_size: usize,
        hash_prefix: &str,
    ) -> Result<Option<String>, BackendError>;

    /// Bulk-insert many base-table rows in one call (`INSERT OR REPLACE`).
    ///
    /// Batch-shaped so a backend can amortise per-row round-trips (the native
    /// backend reuses a single cached prepared statement). Used by the import
    /// path. Writes `cached_at` verbatim from each [`BaseItemRow`]; see the
    /// note there for how this differs from
    /// [`put_item_with_hash`](Self::put_item_with_hash).
    async fn put_base_items(
        &self,
        table_name: &str,
        rows: &[BaseItemRow],
    ) -> Result<(), BackendError>;

    async fn get_item(
        &self,
        table_name: &str,
        pk: &str,
        sk: &str,
    ) -> Result<Option<String>, BackendError>;

    async fn get_partition_size(&self, table_name: &str, pk: &str) -> Result<i64, BackendError>;

    async fn get_lsi_partition_size(
        &self,
        table_name: &str,
        index_name: &str,
        pk: &str,
    ) -> Result<i64, BackendError>;

    async fn delete_item(
        &self,
        table_name: &str,
        pk: &str,
        sk: &str,
    ) -> Result<Option<String>, BackendError>;

    async fn query_items(
        &self,
        table_name: &str,
        pk: &str,
        params: &QueryParams<'_>,
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    async fn scan_items(
        &self,
        table_name: &str,
        params: &ScanParams<'_>,
    ) -> Result<Vec<(String, String, String)>, BackendError>;

    async fn count_items(&self, table_name: &str) -> Result<i64, BackendError>;

    /// The number of items in a data table and the sum of their stored
    /// `item_size`, which is the DynamoDB item size each write computed.
    async fn item_count_and_size(&self, table_name: &str) -> Result<(i64, i64), BackendError>;

    /// The number of entries in a GSI and the total of their projected item
    /// sizes: its `ItemCount` and `IndexSizeBytes`. `whole_items` is true for
    /// an ALL projection, where each entry is the base item and the base
    /// table's stored size is its size; any other projection sizes each
    /// entry's stored JSON.
    async fn gsi_item_count_and_size(
        &self,
        table_name: &str,
        index_name: &str,
        whole_items: bool,
    ) -> Result<(i64, i64), BackendError>;

    /// [`gsi_item_count_and_size`](Self::gsi_item_count_and_size) for an LSI.
    async fn lsi_item_count_and_size(
        &self,
        table_name: &str,
        index_name: &str,
        whole_items: bool,
    ) -> Result<(i64, i64), BackendError>;

    /// The number of entries in a vector index and the total of their stored
    /// billable sizes: its `ItemCount` and `IndexSizeBytes`.
    async fn vector_item_count_and_size(
        &self,
        table_name: &str,
        index_name: &str,
    ) -> Result<(i64, i64), BackendError>;

    /// Rewrite the stored `item_size` of the given rows, each a
    /// `(pk, sk, item_size)`, leaving the rows otherwise unchanged.
    async fn set_item_sizes(
        &self,
        table_name: &str,
        sizes: &[(String, String, usize)],
    ) -> Result<(), BackendError>;

    // -----------------------------------------------------------------------
    // Introspection
    // -----------------------------------------------------------------------

    async fn db_size_bytes(&self) -> Result<u64, BackendError>;
    async fn table_count(&self) -> Result<usize, BackendError>;
    async fn table_stats(&self) -> Result<Vec<TableStats>, BackendError>;
    async fn database_info(&self) -> Result<DatabaseInfo, BackendError>;
    async fn vacuum(&self) -> Result<(), BackendError>;

    // -----------------------------------------------------------------------
    // Streams
    // -----------------------------------------------------------------------

    async fn enable_stream(
        &self,
        table_name: &str,
        view_type: &str,
        label: &str,
    ) -> Result<(), BackendError>;

    async fn disable_stream(&self, table_name: &str) -> Result<(), BackendError>;

    #[allow(clippy::too_many_arguments)]
    async fn insert_stream_record(
        &self,
        table_name: &str,
        event_name: &str,
        keys_json: &str,
        new_image: Option<&str>,
        old_image: Option<&str>,
        sequence_number: &str,
        shard_id: &str,
        created_at: i64,
    ) -> Result<(), BackendError>;

    #[allow(clippy::too_many_arguments)]
    async fn insert_stream_record_with_identity(
        &self,
        table_name: &str,
        event_name: &str,
        keys_json: &str,
        new_image: Option<&str>,
        old_image: Option<&str>,
        sequence_number: &str,
        shard_id: &str,
        created_at: i64,
        user_identity: Option<&str>,
    ) -> Result<(), BackendError>;

    async fn next_stream_sequence_number(&self, table_name: &str) -> Result<i64, BackendError>;

    async fn get_stream_records(
        &self,
        table_name: &str,
        shard_id: &str,
        after_sequence: i64,
        limit: usize,
    ) -> Result<Vec<StreamRecord>, BackendError>;

    /// As `get_stream_records`, with each record's stored event id. The id is
    /// `None` only on a row an older build wrote since this one opened the file.
    #[doc(hidden)]
    async fn get_stream_records_with_event_ids(
        &self,
        table_name: &str,
        shard_id: &str,
        after_sequence: i64,
        limit: usize,
    ) -> Result<Vec<(StreamRecord, Option<String>)>, BackendError>;

    async fn list_stream_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError>;

    // -----------------------------------------------------------------------
    // TTL operations
    // -----------------------------------------------------------------------

    async fn update_ttl_config(
        &self,
        table_name: &str,
        attribute_name: Option<&str>,
        enabled: bool,
    ) -> Result<(), BackendError>;

    async fn list_ttl_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError>;

    async fn get_shard_sequence_range(
        &self,
        table_name: &str,
        shard_id: &str,
    ) -> Result<(Option<String>, Option<String>), BackendError>;

    // -----------------------------------------------------------------------
    // Cache tracking
    // -----------------------------------------------------------------------

    async fn touch_cached_at(
        &self,
        table_name: &str,
        pk: &str,
        sk: &str,
        timestamp: f64,
    ) -> Result<(), BackendError>;

    async fn get_lru_items(
        &self,
        table_name: &str,
        limit: usize,
    ) -> Result<Vec<(String, String, i64)>, BackendError>;
}

#[cfg(test)]
mod tests {
    use super::IndexWriteOp;

    /// `InsertVector` carries its row boxed so the vector family does not
    /// widen the whole enum: every op in a fan-out list pays for the largest
    /// variant, and before the vector variants arrived that was `InsertGsi`
    /// at seven `String`s plus the discriminant word. Pin the enum to that
    /// size so a future field on the vector row cannot regrow it unnoticed.
    #[test]
    fn index_write_op_stays_at_the_insert_gsi_driven_size() {
        let insert_gsi_driven = 7 * std::mem::size_of::<String>() + std::mem::size_of::<usize>();
        assert!(
            std::mem::size_of::<IndexWriteOp>() <= insert_gsi_driven,
            "IndexWriteOp grew past the InsertGsi-driven size: {} > {}",
            std::mem::size_of::<IndexWriteOp>(),
            insert_gsi_driven
        );
    }
}