dynoxide-rs 0.11.1

//! 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;
// 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;

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,
}

/// 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,
    },
}

/// Backend-neutral storage interface.
///
/// Method signatures mirror [`Storage`](crate::storage::Storage)'s public
/// surface 1:1, 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.
///
/// The trait is not consumed dynamically today. The native
/// [`Storage`](crate::storage::Storage) and the wasm
/// [`WasmBridgeBackend`](wasm_backend::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 {
    // -----------------------------------------------------------------------
    // 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;

    // -----------------------------------------------------------------------
    // 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>;

    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>;

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

    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>;

    // -----------------------------------------------------------------------
    // 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?;
                }
            }
        }
        Ok(())
    }

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

    async fn begin_transaction(&self) -> Result<(), BackendError>;
    async fn commit(&self) -> Result<(), BackendError>;
    async fn rollback(&self) -> 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>;

    // -----------------------------------------------------------------------
    // 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>;

    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>;
}