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 // -----------------------------------------------------------------------
173 // Table metadata
174 // -----------------------------------------------------------------------
175
176 async fn insert_table_metadata(&self, m: &CreateTableMetadata<'_>) -> Result<(), BackendError>;
177
178 async fn get_table_metadata(
179 &self,
180 table_name: &str,
181 ) -> Result<Option<TableMetadata>, BackendError>;
182
183 async fn delete_table_metadata(&self, table_name: &str) -> Result<bool, BackendError>;
184
185 async fn update_table_metadata(
186 &self,
187 table_name: &str,
188 attribute_definitions: &str,
189 gsi_definitions: Option<&str>,
190 ) -> Result<(), BackendError>;
191
192 async fn update_provisioned_throughput(
193 &self,
194 table_name: &str,
195 provisioned_throughput: &str,
196 ) -> Result<(), BackendError>;
197
198 async fn clear_provisioned_throughput(&self, table_name: &str) -> Result<(), BackendError>;
199
200 async fn update_billing_mode(
201 &self,
202 table_name: &str,
203 billing_mode: &str,
204 ) -> Result<(), BackendError>;
205
206 async fn update_table_class(
207 &self,
208 table_name: &str,
209 table_class: &str,
210 ) -> Result<(), BackendError>;
211
212 /// Store the serialised on-demand throughput ceilings. Implementations
213 /// must store the payload opaquely: the `clear_on_demand_throughput`
214 /// default passes a JSON `null` through this method, which every reader
215 /// treats as absent, so parsing or rejecting the payload here would break
216 /// that default.
217 async fn update_on_demand_throughput(
218 &self,
219 table_name: &str,
220 on_demand_throughput: &str,
221 ) -> Result<(), BackendError>;
222
223 /// Remove any stored on-demand throughput ceilings. The default stores a
224 /// JSON `null`, which every reader treats as absent, so existing backend
225 /// implementations keep working without changes; the in-tree backends
226 /// override this to clear the underlying value properly.
227 async fn clear_on_demand_throughput(&self, table_name: &str) -> Result<(), BackendError> {
228 self.update_on_demand_throughput(table_name, "null").await
229 }
230
231 async fn get_tags(&self, table_name: &str) -> Result<Vec<Tag>, BackendError>;
232
233 async fn set_tags(&self, table_name: &str, new_tags: &[Tag]) -> Result<(), BackendError>;
234
235 async fn update_deletion_protection(
236 &self,
237 table_name: &str,
238 enabled: bool,
239 ) -> Result<(), BackendError>;
240
241 async fn remove_tags(&self, table_name: &str, keys: &[String]) -> Result<(), BackendError>;
242
243 async fn list_table_names(&self) -> Result<Vec<String>, BackendError>;
244
245 async fn table_exists(&self, table_name: &str) -> Result<bool, BackendError>;
246
247 // -----------------------------------------------------------------------
248 // Dynamic data tables (DDL)
249 // -----------------------------------------------------------------------
250
251 async fn create_data_table(&self, table_name: &str) -> Result<(), BackendError>;
252
253 async fn drop_data_table(&self, table_name: &str) -> Result<(), BackendError>;
254
255 async fn create_gsi_table(
256 &self,
257 table_name: &str,
258 index_name: &str,
259 ) -> Result<(), BackendError>;
260
261 async fn drop_gsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError>;
262
263 async fn create_lsi_table(
264 &self,
265 table_name: &str,
266 index_name: &str,
267 ) -> Result<(), BackendError>;
268
269 async fn drop_lsi_table(&self, table_name: &str, index_name: &str) -> Result<(), BackendError>;
270
271 // -----------------------------------------------------------------------
272 // GSI item operations
273 // -----------------------------------------------------------------------
274
275 #[allow(clippy::too_many_arguments)]
276 async fn insert_gsi_item(
277 &self,
278 table_name: &str,
279 index_name: &str,
280 gsi_pk: &str,
281 gsi_sk: &str,
282 table_pk: &str,
283 table_sk: &str,
284 item_json: &str,
285 ) -> Result<(), BackendError>;
286
287 /// Bulk-insert many rows into one GSI table.
288 ///
289 /// Batch-shaped so a backend can amortise per-row round-trips (the native
290 /// backend reuses a single cached prepared statement). Used by the GSI
291 /// backfill path; the per-row [`insert_gsi_item`](Self::insert_gsi_item)
292 /// covers single writes during normal fan-out.
293 async fn insert_gsi_items(
294 &self,
295 table_name: &str,
296 index_name: &str,
297 rows: &[GsiItemRow],
298 ) -> Result<(), BackendError>;
299
300 async fn delete_gsi_item(
301 &self,
302 table_name: &str,
303 index_name: &str,
304 table_pk: &str,
305 table_sk: &str,
306 ) -> Result<(), BackendError>;
307
308 async fn query_gsi_items(
309 &self,
310 table_name: &str,
311 index_name: &str,
312 gsi_pk: &str,
313 params: &QueryParams<'_>,
314 ) -> Result<Vec<(String, String, String)>, BackendError>;
315
316 async fn scan_gsi_items(
317 &self,
318 table_name: &str,
319 index_name: &str,
320 params: &ScanParams<'_>,
321 ) -> Result<Vec<(String, String, String)>, BackendError>;
322
323 // -----------------------------------------------------------------------
324 // LSI item operations
325 // -----------------------------------------------------------------------
326
327 #[allow(clippy::too_many_arguments)]
328 async fn insert_lsi_item(
329 &self,
330 table_name: &str,
331 index_name: &str,
332 pk: &str,
333 sk: &str,
334 base_pk: &str,
335 base_sk: &str,
336 item_json: &str,
337 ) -> Result<(), BackendError>;
338
339 async fn delete_lsi_item(
340 &self,
341 table_name: &str,
342 index_name: &str,
343 base_pk: &str,
344 base_sk: &str,
345 ) -> Result<(), BackendError>;
346
347 async fn query_lsi_items(
348 &self,
349 table_name: &str,
350 index_name: &str,
351 pk: &str,
352 params: &QueryParams<'_>,
353 ) -> Result<Vec<(String, String, String)>, BackendError>;
354
355 async fn scan_lsi_items(
356 &self,
357 table_name: &str,
358 index_name: &str,
359 params: &ScanParams<'_>,
360 ) -> Result<Vec<(String, String, String)>, BackendError>;
361
362 // -----------------------------------------------------------------------
363 // Index write fan-out
364 // -----------------------------------------------------------------------
365
366 /// Apply an ordered batch of GSI/LSI write operations.
367 ///
368 /// The GSI/LSI maintenance helpers build the list and call this once per
369 /// fan-out instead of invoking the per-item methods one at a time. The
370 /// default impl replays each op through the matching per-item method in
371 /// order, so a backend that does not override it behaves exactly as the
372 /// per-op loop did. The wasm backend overrides this to issue the whole list
373 /// in a single bridge crossing.
374 ///
375 /// Owns no transaction: the caller's open transaction supplies atomicity, so
376 /// a mid-batch failure is rolled back by that caller. An empty list does no
377 /// work.
378 async fn apply_index_writes(&self, ops: &[IndexWriteOp]) -> Result<(), BackendError> {
379 for op in ops {
380 match op {
381 IndexWriteOp::DeleteGsi {
382 table_name,
383 index_name,
384 table_pk,
385 table_sk,
386 } => {
387 self.delete_gsi_item(table_name, index_name, table_pk, table_sk)
388 .await?;
389 }
390 IndexWriteOp::InsertGsi {
391 table_name,
392 index_name,
393 gsi_pk,
394 gsi_sk,
395 table_pk,
396 table_sk,
397 item_json,
398 } => {
399 self.insert_gsi_item(
400 table_name, index_name, gsi_pk, gsi_sk, table_pk, table_sk, item_json,
401 )
402 .await?;
403 }
404 IndexWriteOp::DeleteLsi {
405 table_name,
406 index_name,
407 base_pk,
408 base_sk,
409 } => {
410 self.delete_lsi_item(table_name, index_name, base_pk, base_sk)
411 .await?;
412 }
413 IndexWriteOp::InsertLsi {
414 table_name,
415 index_name,
416 pk,
417 sk,
418 base_pk,
419 base_sk,
420 item_json,
421 } => {
422 self.insert_lsi_item(
423 table_name, index_name, pk, sk, base_pk, base_sk, item_json,
424 )
425 .await?;
426 }
427 }
428 }
429 Ok(())
430 }
431
432 // -----------------------------------------------------------------------
433 // Transactions
434 // -----------------------------------------------------------------------
435
436 async fn begin_transaction(&self) -> Result<(), BackendError>;
437 async fn commit(&self) -> Result<(), BackendError>;
438 async fn rollback(&self) -> Result<(), BackendError>;
439
440 // -----------------------------------------------------------------------
441 // Bulk-loading PRAGMAs
442 // -----------------------------------------------------------------------
443
444 async fn enable_bulk_loading(&self) -> Result<(), BackendError>;
445 async fn disable_bulk_loading(&self) -> Result<(), BackendError>;
446
447 // -----------------------------------------------------------------------
448 // Item CRUD
449 // -----------------------------------------------------------------------
450
451 async fn put_item(
452 &self,
453 table_name: &str,
454 pk: &str,
455 sk: &str,
456 item_json: &str,
457 item_size: usize,
458 ) -> Result<Option<String>, BackendError>;
459
460 #[allow(clippy::too_many_arguments)]
461 async fn put_item_with_hash(
462 &self,
463 table_name: &str,
464 pk: &str,
465 sk: &str,
466 item_json: &str,
467 item_size: usize,
468 hash_prefix: &str,
469 ) -> Result<Option<String>, BackendError>;
470
471 /// Bulk-insert many base-table rows in one call (`INSERT OR REPLACE`).
472 ///
473 /// Batch-shaped so a backend can amortise per-row round-trips (the native
474 /// backend reuses a single cached prepared statement). Used by the import
475 /// path. Writes `cached_at` verbatim from each [`BaseItemRow`]; see the
476 /// note there for how this differs from
477 /// [`put_item_with_hash`](Self::put_item_with_hash).
478 async fn put_base_items(
479 &self,
480 table_name: &str,
481 rows: &[BaseItemRow],
482 ) -> Result<(), BackendError>;
483
484 async fn get_item(
485 &self,
486 table_name: &str,
487 pk: &str,
488 sk: &str,
489 ) -> Result<Option<String>, BackendError>;
490
491 async fn get_partition_size(&self, table_name: &str, pk: &str) -> Result<i64, BackendError>;
492
493 async fn get_lsi_partition_size(
494 &self,
495 table_name: &str,
496 index_name: &str,
497 pk: &str,
498 ) -> Result<i64, BackendError>;
499
500 async fn delete_item(
501 &self,
502 table_name: &str,
503 pk: &str,
504 sk: &str,
505 ) -> Result<Option<String>, BackendError>;
506
507 async fn query_items(
508 &self,
509 table_name: &str,
510 pk: &str,
511 params: &QueryParams<'_>,
512 ) -> Result<Vec<(String, String, String)>, BackendError>;
513
514 async fn scan_items(
515 &self,
516 table_name: &str,
517 params: &ScanParams<'_>,
518 ) -> Result<Vec<(String, String, String)>, BackendError>;
519
520 async fn count_items(&self, table_name: &str) -> Result<i64, BackendError>;
521
522 // -----------------------------------------------------------------------
523 // Introspection
524 // -----------------------------------------------------------------------
525
526 async fn db_size_bytes(&self) -> Result<u64, BackendError>;
527 async fn table_count(&self) -> Result<usize, BackendError>;
528 async fn table_stats(&self) -> Result<Vec<TableStats>, BackendError>;
529 async fn database_info(&self) -> Result<DatabaseInfo, BackendError>;
530 async fn vacuum(&self) -> Result<(), BackendError>;
531
532 // -----------------------------------------------------------------------
533 // Streams
534 // -----------------------------------------------------------------------
535
536 async fn enable_stream(
537 &self,
538 table_name: &str,
539 view_type: &str,
540 label: &str,
541 ) -> Result<(), BackendError>;
542
543 async fn disable_stream(&self, table_name: &str) -> Result<(), BackendError>;
544
545 #[allow(clippy::too_many_arguments)]
546 async fn insert_stream_record(
547 &self,
548 table_name: &str,
549 event_name: &str,
550 keys_json: &str,
551 new_image: Option<&str>,
552 old_image: Option<&str>,
553 sequence_number: &str,
554 shard_id: &str,
555 created_at: i64,
556 ) -> Result<(), BackendError>;
557
558 #[allow(clippy::too_many_arguments)]
559 async fn insert_stream_record_with_identity(
560 &self,
561 table_name: &str,
562 event_name: &str,
563 keys_json: &str,
564 new_image: Option<&str>,
565 old_image: Option<&str>,
566 sequence_number: &str,
567 shard_id: &str,
568 created_at: i64,
569 user_identity: Option<&str>,
570 ) -> Result<(), BackendError>;
571
572 async fn next_stream_sequence_number(&self, table_name: &str) -> Result<i64, BackendError>;
573
574 async fn get_stream_records(
575 &self,
576 table_name: &str,
577 shard_id: &str,
578 after_sequence: i64,
579 limit: usize,
580 ) -> Result<Vec<StreamRecord>, BackendError>;
581
582 async fn list_stream_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError>;
583
584 // -----------------------------------------------------------------------
585 // TTL operations
586 // -----------------------------------------------------------------------
587
588 async fn update_ttl_config(
589 &self,
590 table_name: &str,
591 attribute_name: Option<&str>,
592 enabled: bool,
593 ) -> Result<(), BackendError>;
594
595 async fn list_ttl_enabled_tables(&self) -> Result<Vec<TableMetadata>, BackendError>;
596
597 async fn get_shard_sequence_range(
598 &self,
599 table_name: &str,
600 shard_id: &str,
601 ) -> Result<(Option<String>, Option<String>), BackendError>;
602
603 // -----------------------------------------------------------------------
604 // Cache tracking
605 // -----------------------------------------------------------------------
606
607 async fn touch_cached_at(
608 &self,
609 table_name: &str,
610 pk: &str,
611 sk: &str,
612 timestamp: f64,
613 ) -> Result<(), BackendError>;
614
615 async fn get_lru_items(
616 &self,
617 table_name: &str,
618 limit: usize,
619 ) -> Result<Vec<(String, String, i64)>, BackendError>;
620}