dynoxide/storage_backend/mod.rs
1//! Storage backend abstraction.
2//!
3//! Defines the [`StorageBackend`] trait that decouples Dynoxide's data layer
4//! from a specific SQLite binding. The native [`rusqlite`]-backed
5//! [`Storage`](crate::storage::Storage) implements the trait, and the
6//! `wasm-sqlite` build adds [`wasm_backend::WasmBridgeBackend`], which runs the
7//! same SQL against a JS SQLite database over a wasm-bindgen bridge. Both
8//! backends issue identical SQL because they share the builders in
9//! [`sql_builders`].
10//!
11//! The native build consumes the trait monomorphically through `Storage`; the
12//! wasm build consumes it through `WasmBridgeBackend`. The escape hatches
13//! `Storage::conn()` and `Storage::conn_mut()` are not exposed by the trait
14//! and remain native-only.
15//!
16//! # No `Send + Sync` super-trait
17//!
18//! [`Storage`](crate::storage::Storage) carries a `RefCell<HashMap<...>>` for
19//! its metadata cache, so `Storage: !Sync`. A `Send + Sync` super-trait would
20//! refuse the impl on `Storage`. With no dynamic dispatch site in scope,
21//! auto-trait propagation across `.await` is decided per-callsite anyway, so
22//! adding `Send` to the super-trait would not earn any compile-time
23//! guarantee on the futures returned by trait methods.
24
25pub mod clock;
26pub mod error;
27// Internal: the shared SQL contract between the rusqlite and wasm backends.
28// `pub` only because both backend modules consume it across the cfg split; it
29// is not a stable API, hence `#[doc(hidden)]`.
30#[doc(hidden)]
31pub mod sql_builders;
32// The native rusqlite-backed `Storage` exists whenever either SQLite backend
33// feature is on (the crate refuses to build with neither), and the handlers
34// now consume `StorageBackend` through it, so the impl must track the same
35// condition rather than `native-sqlite` alone.
36#[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
37pub mod rusqlite_impl;
38#[cfg(feature = "wasm-sqlite")]
39pub mod wasm_backend;
40
41use crate::storage::{
42 CreateTableMetadata, DatabaseInfo, QueryParams, ScanParams, StreamRecord, TableMetadata,
43 TableStats,
44};
45use crate::types::Tag;
46
47pub use clock::{Clock, ManualClock, SystemClock};
48pub use error::BackendError;
49#[cfg(any(feature = "native-sqlite", feature = "_has-encryption"))]
50pub use error::from_rusqlite;
51#[doc(hidden)]
52pub use sql_builders::SqlParam;
53#[cfg(feature = "wasm-sqlite")]
54pub use wasm_backend::WasmBridgeBackend;
55
56/// One base-table row for a bulk insert via [`StorageBackend::put_base_items`].
57///
58/// Unlike [`StorageBackend::put_item_with_hash`], which preserves any existing
59/// `cached_at` value, the bulk path writes `cached_at` verbatim: this mirrors
60/// the import flow, which sets the timestamp explicitly (or clears it) for
61/// every row it loads.
62#[derive(Debug, Clone)]
63pub struct BaseItemRow {
64 /// Partition key string.
65 pub pk: String,
66 /// Sort key string (empty for tables without a sort key).
67 pub sk: String,
68 /// Serialised item JSON.
69 pub item_json: String,
70 /// Item size in bytes.
71 pub item_size: usize,
72 /// Cache timestamp written verbatim; `None` clears the column.
73 pub cached_at: Option<f64>,
74 /// Hash prefix used for parallel-scan ordering.
75 pub hash_prefix: String,
76}
77
78/// One GSI-table row for a bulk insert via [`StorageBackend::insert_gsi_items`].
79///
80/// The fields mirror the argument order of the single-row
81/// [`StorageBackend::insert_gsi_item`].
82#[derive(Debug, Clone)]
83pub struct GsiItemRow {
84 /// GSI partition key string.
85 pub gsi_pk: String,
86 /// GSI sort key string (empty when the index has no sort key).
87 pub gsi_sk: String,
88 /// Base-table partition key string.
89 pub table_pk: String,
90 /// Base-table sort key string.
91 pub table_sk: String,
92 /// Projected item JSON.
93 pub item_json: String,
94}
95
96/// One index-table write operation, backend-neutral.
97///
98/// The per-write and per-delete GSI/LSI fan-out builds an ordered list of these
99/// and hands it to [`StorageBackend::apply_index_writes`] in a single call. The
100/// default impl replays each op through the matching per-item method, identical
101/// to the per-op loop it replaces; the wasm backend overrides it to collapse the
102/// list into one bridge crossing. Each variant's fields mirror the argument
103/// order of the per-item method it stands in for.
104#[derive(Debug, Clone)]
105pub enum IndexWriteOp {
106 /// Remove this base key's entry from a GSI table.
107 DeleteGsi {
108 table_name: String,
109 index_name: String,
110 table_pk: String,
111 table_sk: String,
112 },
113 /// Insert (or replace) this item's projected entry into a GSI table.
114 InsertGsi {
115 table_name: String,
116 index_name: String,
117 gsi_pk: String,
118 gsi_sk: String,
119 table_pk: String,
120 table_sk: String,
121 item_json: String,
122 },
123 /// Remove this base key's entry from an LSI table.
124 DeleteLsi {
125 table_name: String,
126 index_name: String,
127 base_pk: String,
128 base_sk: String,
129 },
130 /// Insert (or replace) this item's projected entry into an LSI table.
131 InsertLsi {
132 table_name: String,
133 index_name: String,
134 pk: String,
135 sk: String,
136 base_pk: String,
137 base_sk: String,
138 item_json: String,
139 },
140}
141
142/// Backend-neutral storage interface.
143///
144/// Method signatures mirror [`Storage`](crate::storage::Storage)'s public
145/// surface 1:1, with three mechanical transformations:
146///
147/// 1. `Result<T, DynoxideError>` becomes `Result<T, BackendError>`.
148/// 2. `fn` becomes `async fn`.
149/// 3. Filesystem-typed and rusqlite-typed methods are excluded; they remain
150/// on the native [`Storage`](crate::storage::Storage) only.
151///
152/// The trait is not consumed dynamically today. The native
153/// [`Storage`](crate::storage::Storage) and the wasm
154/// [`WasmBridgeBackend`](wasm_backend::WasmBridgeBackend) each implement it
155/// monomorphically.
156///
157/// The `#[allow(async_fn_in_trait)]` reflects the monomorphic-only consumption
158/// model. The lint can be revisited if and when `dyn StorageBackend` becomes
159/// a real callsite.
160#[allow(async_fn_in_trait)]
161pub trait StorageBackend {
162 // -----------------------------------------------------------------------
163 // Capabilities
164 // -----------------------------------------------------------------------
165
166 /// Wall-clock access for the stream and TTL paths.
167 ///
168 /// Sync because reading the clock is not I/O. The native backend returns
169 /// its injected [`Clock`]; the wasm SQLite backend supplies its own.
170 fn clock(&self) -> &dyn Clock;
171
172 /// Whether this backend can record and serve DynamoDB Streams.
173 ///
174 /// A backend that answers `false` still gets its stream methods called
175 /// nowhere on the happy path: the actions consult this before mutating, so
176 /// a request that needs streams is refused before anything is created,
177 /// rather than half-applied and then failed. Defaults to `true`; the wasm
178 /// backend answers `false` until a delivery mechanism exists.
179 fn supports_streams(&self) -> bool {
180 true
181 }
182
183 /// Whether this backend stores resource tags. Same contract as
184 /// [`supports_streams`](Self::supports_streams): consulted before a
185 /// mutation that would need `set_tags`, so a tagged `CreateTable` on a
186 /// backend without tags is refused whole rather than creating the table
187 /// and then failing.
188 fn supports_tags(&self) -> bool {
189 true
190 }
191
192 // -----------------------------------------------------------------------
193 // Table metadata
194 // -----------------------------------------------------------------------
195
196 async fn insert_table_metadata(&self, m: &CreateTableMetadata<'_>) -> Result<(), BackendError>;
197
198 async fn get_table_metadata(
199 &self,
200 table_name: &str,
201 ) -> Result<Option<TableMetadata>, BackendError>;
202
203 async fn delete_table_metadata(&self, table_name: &str) -> Result<bool, BackendError>;
204
205 async fn update_table_metadata(
206 &self,
207 table_name: &str,
208 attribute_definitions: &str,
209 gsi_definitions: Option<&str>,
210 ) -> Result<(), BackendError>;
211
212 async fn update_provisioned_throughput(
213 &self,
214 table_name: &str,
215 provisioned_throughput: &str,
216 ) -> Result<(), BackendError>;
217
218 async fn clear_provisioned_throughput(&self, table_name: &str) -> Result<(), BackendError>;
219
220 async fn update_billing_mode(
221 &self,
222 table_name: &str,
223 billing_mode: &str,
224 ) -> Result<(), BackendError>;
225
226 async fn update_table_class(
227 &self,
228 table_name: &str,
229 table_class: &str,
230 ) -> Result<(), BackendError>;
231
232 /// Store the serialised on-demand throughput ceilings. Implementations
233 /// must store the payload opaquely: the `clear_on_demand_throughput`
234 /// default passes a JSON `null` through this method, which every reader
235 /// treats as absent, so parsing or rejecting the payload here would break
236 /// that default.
237 async fn update_on_demand_throughput(
238 &self,
239 table_name: &str,
240 on_demand_throughput: &str,
241 ) -> Result<(), BackendError>;
242
243 /// Remove any stored on-demand throughput ceilings. The default stores a
244 /// JSON `null`, which every reader treats as absent, so existing backend
245 /// implementations keep working without changes; the in-tree backends
246 /// override this to clear the underlying value properly.
247 async fn clear_on_demand_throughput(&self, table_name: &str) -> Result<(), BackendError> {
248 self.update_on_demand_throughput(table_name, "null").await
249 }
250
251 async fn get_tags(&self, table_name: &str) -> Result<Vec<Tag>, BackendError>;
252
253 async fn set_tags(&self, table_name: &str, new_tags: &[Tag]) -> Result<(), BackendError>;
254
255 async fn update_deletion_protection(
256 &self,
257 table_name: &str,
258 enabled: bool,
259 ) -> Result<(), BackendError>;
260
261 async fn remove_tags(&self, table_name: &str, keys: &[String]) -> Result<(), BackendError>;
262
263 async fn list_table_names(&self) -> Result<Vec<String>, BackendError>;
264
265 async fn table_exists(&self, table_name: &str) -> Result<bool, BackendError>;
266
267 // -----------------------------------------------------------------------
268 // Dynamic data tables (DDL)
269 // -----------------------------------------------------------------------
270
271 async fn create_data_table(&self, table_name: &str) -> Result<(), BackendError>;
272
273 async fn drop_data_table(&self, table_name: &str) -> Result<(), BackendError>;
274
275 async fn create_gsi_table(
276 &self,
277 table_name: &str,
278 index_name: &str,
279 ) -> Result<(), BackendError>;
280
281 async fn drop_gsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError>;
282
283 async fn create_lsi_table(
284 &self,
285 table_name: &str,
286 index_name: &str,
287 ) -> Result<(), BackendError>;
288
289 async fn drop_lsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError>;
290
291 // -----------------------------------------------------------------------
292 // GSI item operations
293 // -----------------------------------------------------------------------
294
295 #[allow(clippy::too_many_arguments)]
296 async fn insert_gsi_item(
297 &self,
298 table_name: &str,
299 index_name: &str,
300 gsi_pk: &str,
301 gsi_sk: &str,
302 table_pk: &str,
303 table_sk: &str,
304 item_json: &str,
305 ) -> Result<(), BackendError>;
306
307 /// Bulk-insert many rows into one GSI table.
308 ///
309 /// Batch-shaped so a backend can amortise per-row round-trips (the native
310 /// backend reuses a single cached prepared statement). Used by the GSI
311 /// backfill path; the per-row [`insert_gsi_item`](Self::insert_gsi_item)
312 /// covers single writes during normal fan-out.
313 async fn insert_gsi_items(
314 &self,
315 table_name: &str,
316 index_name: &str,
317 rows: &[GsiItemRow],
318 ) -> Result<(), BackendError>;
319
320 async fn delete_gsi_item(
321 &self,
322 table_name: &str,
323 index_name: &str,
324 table_pk: &str,
325 table_sk: &str,
326 ) -> Result<(), BackendError>;
327
328 async fn query_gsi_items(
329 &self,
330 table_name: &str,
331 index_name: &str,
332 gsi_pk: &str,
333 params: &QueryParams<'_>,
334 ) -> Result<Vec<(String, String, String)>, BackendError>;
335
336 async fn scan_gsi_items(
337 &self,
338 table_name: &str,
339 index_name: &str,
340 params: &ScanParams<'_>,
341 ) -> Result<Vec<(String, String, String)>, BackendError>;
342
343 // -----------------------------------------------------------------------
344 // LSI item operations
345 // -----------------------------------------------------------------------
346
347 #[allow(clippy::too_many_arguments)]
348 async fn insert_lsi_item(
349 &self,
350 table_name: &str,
351 index_name: &str,
352 pk: &str,
353 sk: &str,
354 base_pk: &str,
355 base_sk: &str,
356 item_json: &str,
357 ) -> Result<(), BackendError>;
358
359 async fn delete_lsi_item(
360 &self,
361 table_name: &str,
362 index_name: &str,
363 base_pk: &str,
364 base_sk: &str,
365 ) -> Result<(), BackendError>;
366
367 async fn query_lsi_items(
368 &self,
369 table_name: &str,
370 index_name: &str,
371 pk: &str,
372 params: &QueryParams<'_>,
373 ) -> Result<Vec<(String, String, String)>, BackendError>;
374
375 async fn scan_lsi_items(
376 &self,
377 table_name: &str,
378 index_name: &str,
379 params: &ScanParams<'_>,
380 ) -> Result<Vec<(String, String, String)>, BackendError>;
381
382 // -----------------------------------------------------------------------
383 // Index write fan-out
384 // -----------------------------------------------------------------------
385
386 /// Apply an ordered batch of GSI/LSI write operations.
387 ///
388 /// The GSI/LSI maintenance helpers build the list and call this once per
389 /// fan-out instead of invoking the per-item methods one at a time. The
390 /// default impl replays each op through the matching per-item method in
391 /// order, so a backend that does not override it behaves exactly as the
392 /// per-op loop did. The wasm backend overrides this to issue the whole list
393 /// in a single bridge crossing.
394 ///
395 /// Owns no transaction: the caller's open transaction supplies atomicity, so
396 /// a mid-batch failure is rolled back by that caller. An empty list does no
397 /// work.
398 async fn apply_index_writes(&self, ops: &[IndexWriteOp]) -> Result<(), BackendError> {
399 for op in ops {
400 match op {
401 IndexWriteOp::DeleteGsi {
402 table_name,
403 index_name,
404 table_pk,
405 table_sk,
406 } => {
407 self.delete_gsi_item(table_name, index_name, table_pk, table_sk)
408 .await?;
409 }
410 IndexWriteOp::InsertGsi {
411 table_name,
412 index_name,
413 gsi_pk,
414 gsi_sk,
415 table_pk,
416 table_sk,
417 item_json,
418 } => {
419 self.insert_gsi_item(
420 table_name, index_name, gsi_pk, gsi_sk, table_pk, table_sk, item_json,
421 )
422 .await?;
423 }
424 IndexWriteOp::DeleteLsi {
425 table_name,
426 index_name,
427 base_pk,
428 base_sk,
429 } => {
430 self.delete_lsi_item(table_name, index_name, base_pk, base_sk)
431 .await?;
432 }
433 IndexWriteOp::InsertLsi {
434 table_name,
435 index_name,
436 pk,
437 sk,
438 base_pk,
439 base_sk,
440 item_json,
441 } => {
442 self.insert_lsi_item(
443 table_name, index_name, pk, sk, base_pk, base_sk, item_json,
444 )
445 .await?;
446 }
447 }
448 }
449 Ok(())
450 }
451
452 // -----------------------------------------------------------------------
453 // Transactions
454 // -----------------------------------------------------------------------
455
456 async fn begin_transaction(&self) -> Result<(), BackendError>;
457 async fn commit(&self) -> Result<(), BackendError>;
458 async fn rollback(&self) -> Result<(), BackendError>;
459
460 // -----------------------------------------------------------------------
461 // Bulk-loading PRAGMAs
462 // -----------------------------------------------------------------------
463
464 async fn enable_bulk_loading(&self) -> Result<(), BackendError>;
465 async fn disable_bulk_loading(&self) -> Result<(), BackendError>;
466
467 // -----------------------------------------------------------------------
468 // Item CRUD
469 // -----------------------------------------------------------------------
470
471 async fn put_item(
472 &self,
473 table_name: &str,
474 pk: &str,
475 sk: &str,
476 item_json: &str,
477 item_size: usize,
478 ) -> Result<Option<String>, BackendError>;
479
480 #[allow(clippy::too_many_arguments)]
481 async fn put_item_with_hash(
482 &self,
483 table_name: &str,
484 pk: &str,
485 sk: &str,
486 item_json: &str,
487 item_size: usize,
488 hash_prefix: &str,
489 ) -> Result<Option<String>, BackendError>;
490
491 /// Bulk-insert many base-table rows in one call (`INSERT OR REPLACE`).
492 ///
493 /// Batch-shaped so a backend can amortise per-row round-trips (the native
494 /// backend reuses a single cached prepared statement). Used by the import
495 /// path. Writes `cached_at` verbatim from each [`BaseItemRow`]; see the
496 /// note there for how this differs from
497 /// [`put_item_with_hash`](Self::put_item_with_hash).
498 async fn put_base_items(
499 &self,
500 table_name: &str,
501 rows: &[BaseItemRow],
502 ) -> Result<(), BackendError>;
503
504 async fn get_item(
505 &self,
506 table_name: &str,
507 pk: &str,
508 sk: &str,
509 ) -> Result<Option<String>, BackendError>;
510
511 async fn get_partition_size(&self, table_name: &str, pk: &str) -> Result<i64, BackendError>;
512
513 async fn get_lsi_partition_size(
514 &self,
515 table_name: &str,
516 index_name: &str,
517 pk: &str,
518 ) -> Result<i64, BackendError>;
519
520 async fn delete_item(
521 &self,
522 table_name: &str,
523 pk: &str,
524 sk: &str,
525 ) -> Result<Option<String>, BackendError>;
526
527 async fn query_items(
528 &self,
529 table_name: &str,
530 pk: &str,
531 params: &QueryParams<'_>,
532 ) -> Result<Vec<(String, String, String)>, BackendError>;
533
534 async fn scan_items(
535 &self,
536 table_name: &str,
537 params: &ScanParams<'_>,
538 ) -> Result<Vec<(String, String, String)>, BackendError>;
539
540 async fn count_items(&self, table_name: &str) -> Result<i64, BackendError>;
541
542 // -----------------------------------------------------------------------
543 // Introspection
544 // -----------------------------------------------------------------------
545
546 async fn db_size_bytes(&self) -> Result<u64, BackendError>;
547 async fn table_count(&self) -> Result<usize, BackendError>;
548 async fn table_stats(&self) -> Result<Vec<TableStats>, BackendError>;
549 async fn database_info(&self) -> Result<DatabaseInfo, BackendError>;
550 async fn vacuum(&self) -> Result<(), BackendError>;
551
552 // -----------------------------------------------------------------------
553 // Streams
554 // -----------------------------------------------------------------------
555
556 async fn enable_stream(
557 &self,
558 table_name: &str,
559 view_type: &str,
560 label: &str,
561 ) -> Result<(), BackendError>;
562
563 async fn disable_stream(&self, table_name: &str) -> Result<(), BackendError>;
564
565 #[allow(clippy::too_many_arguments)]
566 async fn insert_stream_record(
567 &self,
568 table_name: &str,
569 event_name: &str,
570 keys_json: &str,
571 new_image: Option<&str>,
572 old_image: Option<&str>,
573 sequence_number: &str,
574 shard_id: &str,
575 created_at: i64,
576 ) -> Result<(), BackendError>;
577
578 #[allow(clippy::too_many_arguments)]
579 async fn insert_stream_record_with_identity(
580 &self,
581 table_name: &str,
582 event_name: &str,
583 keys_json: &str,
584 new_image: Option<&str>,
585 old_image: Option<&str>,
586 sequence_number: &str,
587 shard_id: &str,
588 created_at: i64,
589 user_identity: Option<&str>,
590 ) -> Result<(), BackendError>;
591
592 async fn next_stream_sequence_number(&self, table_name: &str) -> Result<i64, BackendError>;
593
594 async fn get_stream_records(
595 &self,
596 table_name: &str,
597 shard_id: &str,
598 after_sequence: i64,
599 limit: usize,
600 ) -> Result<Vec<StreamRecord>, BackendError>;
601
602 async fn list_stream_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError>;
603
604 // -----------------------------------------------------------------------
605 // TTL operations
606 // -----------------------------------------------------------------------
607
608 async fn update_ttl_config(
609 &self,
610 table_name: &str,
611 attribute_name: Option<&str>,
612 enabled: bool,
613 ) -> Result<(), BackendError>;
614
615 async fn list_ttl_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError>;
616
617 async fn get_shard_sequence_range(
618 &self,
619 table_name: &str,
620 shard_id: &str,
621 ) -> Result<(Option<String>, Option<String>), BackendError>;
622
623 // -----------------------------------------------------------------------
624 // Cache tracking
625 // -----------------------------------------------------------------------
626
627 async fn touch_cached_at(
628 &self,
629 table_name: &str,
630 pk: &str,
631 sk: &str,
632 timestamp: f64,
633 ) -> Result<(), BackendError>;
634
635 async fn get_lru_items(
636 &self,
637 table_name: &str,
638 limit: usize,
639 ) -> Result<Vec<(String, String, i64)>, BackendError>;
640}