Struct Storage 

pub struct Storage { /* private fields */ }

Low-level SQLite storage layer.

Manages the SQLite connection, metadata tables, and per-DynamoDB-table data tables. All SQL lives here — higher layers work with Rust types.

Native-only: this type is the rusqlite-backed backend and is compiled out of backend-neutral builds (for example wasm-sqlite).

Implementations

impl Storage

pub fn new(path: &str) -> Result<Self>

Open a persistent database at the given path.

pub fn with_clock(self, clock: Arc<dyn Clock>) -> Self

Replace the Clock used by the stream and TTL paths. Returns self so callers can chain construction (Storage::memory()?.with_clock(c)).

Default for every constructor is SystemClock. Tests use this to inject a ManualClock.

pub fn memory() -> Result<Self>

Open an in-memory database (for tests and ephemeral use).

pub fn conn(&self) -> &Connection

Get a reference to the underlying connection (for transactions, etc.).

pub fn conn_mut(&mut self) -> &mut Connection

Get a mutable reference to the underlying connection.

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

Insert a row into the _tables metadata table.

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

Get metadata for a table. Returns None if the table doesn’t exist.

Results are cached in memory. The cache is invalidated when metadata is modified via insert_table_metadata, delete_table_metadata, enable_stream, or update_ttl_config.

pub fn delete_table_metadata(&self, table_name: &str) -> Result<bool>

Delete metadata for a table.

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

Update attribute definitions and GSI definitions for a table.

pub fn update_provisioned_throughput( &self, table_name: &str, provisioned_throughput: &str, ) -> Result<()>

Update provisioned throughput for a table.

pub fn clear_provisioned_throughput(&self, table_name: &str) -> Result<()>

Clear provisioned throughput for a table (sets to SQL NULL).

pub fn update_billing_mode( &self, table_name: &str, billing_mode: &str, ) -> Result<()>

Update billing mode for a table.

pub fn update_table_class( &self, table_name: &str, table_class: &str, ) -> Result<()>

Update the table class for a table.

pub fn update_on_demand_throughput( &self, table_name: &str, on_demand_throughput: &str, ) -> Result<()>

Update the on-demand throughput (stored as JSON) for a table.

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

Get tags for a table.

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

Set (merge) tags on a table. New keys overwrite existing keys.

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

Update the deletion protection setting for a table.

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

Remove tags by key from a table.

pub fn list_table_names(&self) -> Result<Vec<String>>

List all table names.

pub fn table_exists(&self, table_name: &str) -> Result<bool>

Check if a table exists in metadata.

pub fn create_data_table(&self, table_name: &str) -> Result<()>

Create a data table for a DynamoDB table.

pub fn drop_data_table(&self, table_name: &str) -> Result<()>

Drop a data table.

pub fn create_gsi_table(&self, table_name: &str, index_name: &str) -> Result<()>

Create a GSI table.

pub fn drop_gsi_table(&self, table_name: &str, index_name: &str) -> Result<()>

Drop a GSI table.

pub 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<()>

Insert an item into a GSI table.

pub fn insert_gsi_items( &self, table_name: &str, index_name: &str, rows: &[GsiItemRow], ) -> Result<()>

Bulk-insert many rows into one GSI table using a single cached prepared statement. Equivalent to calling Self::insert_gsi_item once per row, but reuses the statement across the batch.

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

Delete an item from a GSI table by base table primary key.

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

Query items from a GSI table.

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

Scan all items from a GSI table.

pub fn create_lsi_table(&self, table_name: &str, index_name: &str) -> Result<()>

Create an LSI table for a given base table and index name.

pub fn drop_lsi_table(&self, table_name: &str, index_name: &str) -> Result<()>

Drop an LSI table.

pub 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<()>

Insert an item into an LSI table.

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

Delete an item from an LSI table by base table primary key.

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

Query items from an LSI table.

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

Scan all items from an LSI table.

pub fn begin_transaction(&self) -> Result<()>

Begin an immediate SQLite transaction.

pub fn commit(&self) -> Result<()>

Commit the current transaction.

pub fn rollback(&self) -> Result<()>

Rollback the current transaction.

pub fn enable_bulk_loading(&self) -> Result<()>

Set aggressive PRAGMAs for bulk loading.

Disables fsync, increases cache, and enables memory-mapped I/O. Only safe when data loss on crash is acceptable (e.g., fresh import that can be re-run).

pub fn disable_bulk_loading(&self) -> Result<()>

Restore normal PRAGMAs after bulk loading.

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

Insert or replace an item.

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

Put an item with an explicit hash prefix for parallel scan ordering.

pub fn put_base_items( &self, table_name: &str, rows: &[BaseItemRow], ) -> Result<()>

Bulk-insert many base-table rows using a single cached prepared statement (INSERT OR REPLACE). Unlike Self::put_item_with_hash, which preserves any existing cached_at, this writes cached_at verbatim from each row, matching the import flow’s semantics.

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

Get a single item by primary key.

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

Return the total item_size for all items sharing the given partition key.

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

Return the total size of LSI items for a given partition key. LSI items are stored as JSON text, so we use length(item_json).

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

Delete an item by primary key. Returns the old item_json if it existed.

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

Query items by partition key with optional sort key condition.

sk_condition is a SQL fragment like AND sk > ? with sk_params providing values. Returns (items, has_more) where items is a vec of (pk, sk, item_json).

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

Scan items from a table with pagination.

Returns (pk, sk, item_json) tuples ordered by hash_prefix for dynalite-compatible parallel scan behaviour.

pub fn count_items(&self, table_name: &str) -> Result<i64>

Count items in a table.

pub fn db_path(&self) -> Option<String>

Get the database file path, or None for in-memory databases.

pub fn db_size_bytes(&self) -> Result<u64>

Get the total database size in bytes.

pub fn table_count(&self) -> Result<usize>

Count the number of DynamoDB tables.

pub fn table_stats(&self) -> Result<Vec<TableStats>>

Get per-table statistics: name, item count, and approximate size in bytes.

Uses a single query per table (COUNT + SUM combined) instead of separate queries.

pub fn database_info(&self) -> Result<DatabaseInfo>

Get combined database info in a single call for atomic consistency.

Returns all introspection data that get_database_info tools need without releasing the lock between queries.

pub fn vacuum_into(&self, path: &str) -> Result<()>

Create a snapshot of the database by copying it to the given path. Uses VACUUM INTO which works for both in-memory and file-backed databases.

pub fn vacuum(&self) -> Result<()>

Run VACUUM to compact the database file in place.

pub fn restore_from(&mut self, path: &str) -> Result<()>

Restore the database from a snapshot file using SQLite’s backup API. This replaces the current database contents with the snapshot contents. Works for both in-memory and file-backed databases.

pub fn backup_to_memory(&self) -> Result<Connection>

Backup the current database to a new in-memory SQLite connection.

Used for in-memory snapshot storage — the returned connection holds a complete copy of the database without touching the filesystem.

pub fn restore_from_connection(&mut self, source: &Connection) -> Result<()>

Restore the database from another SQLite connection using the backup API.

Replaces the current database contents with the source connection’s contents. Invalidates the metadata cache.

pub fn connection_size_bytes(conn: &Connection) -> Result<u64>

Get the database size in bytes for an arbitrary connection.

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

Enable streams on a table.

pub fn disable_stream(&self, table_name: &str) -> Result<()>

Disable streams on a table.

pub 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<()>

Insert a stream record.

pub 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<()>

Insert a stream record with optional user identity (for TTL deletions).

pub fn next_stream_sequence_number(&self, table_name: &str) -> Result<i64>

Get the next sequence number for a table’s stream.

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

Get stream records for a shard starting after a given sequence number.

pub fn list_stream_enabled_tables(&self) -> Result<Vec<TableMetadata>>

List tables that have streams enabled.

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

Update TTL configuration for a table.

pub fn list_ttl_enabled_tables(&self) -> Result<Vec<TableMetadata>>

List tables that have TTL enabled.

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

Get the min and max sequence numbers for a shard.

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

Update the cached_at timestamp for a single item.

pub fn get_lru_items( &self, table_name: &str, limit: usize, ) -> Result<Vec<(String, String, i64)>>

Get items ordered by cached_at (oldest first) for LRU eviction.

Returns (pk, sk, item_size) tuples. Items with NULL cached_at are excluded (they were never cached from a remote source).

Trait Implementations

impl StorageBackend for Storage

fn clock(&self) -> &dyn Clock

Wall-clock access for the stream and TTL paths.

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>

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>

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>

async fn insert_gsi_items( &self, table_name: &str, index_name: &str, rows: &[GsiItemRow], ) -> Result<(), BackendError>

Bulk-insert many rows into one GSI table.

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>

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>

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

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

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

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

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

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

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>

async fn put_base_items( &self, table_name: &str, rows: &[BaseItemRow], ) -> Result<(), BackendError>

Bulk-insert many base-table rows in one call (INSERT OR REPLACE).

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>

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>

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>

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>

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>

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>

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>

async fn apply_index_writes( &self, ops: &[IndexWriteOp], ) -> Result<(), BackendError>

Apply an ordered batch of GSI/LSI write operations.