feat(store): add async storage backends for PostgreSQL and DynamoDB (v0.9)

Add AsyncStore trait and two feature-gated async backend implementations:

- AsyncStore trait mirrors sync Store with 15 async methods
- PostgresStore (feature: postgres) using sqlx-core + sqlx-postgres
- DynamoStore (feature: dynamodb) with single-table design and GSIs
- Feature flags keep dependencies optional
- Type aliases: BoxAsyncStore, ArcAsyncStore

Technical notes:
- Uses sqlx-core/sqlx-postgres directly to avoid sqlite version conflict
- DynamoDB uses PK=DOC#id, SK prefixes: OP#, SNAP#, MILE#, VV#
- GSI1 for operation lookup, GSI2 for actor-based queries

35 tests passing (31 existing + 4 new DynamoDB key tests).

Author: raffaelschneider

Cargo.toml

@@ -77,6 +77,11 @@ sha2 = "0.10"
 # Storage
 rusqlite = { version = "0.33", features = ["bundled"] }
 
+# Async storage backends (feature-gated in conflux-store)
+# Note: sqlx is declared directly in conflux-store to avoid sqlite version conflicts
+aws-sdk-dynamodb = "1.54"
+aws-config = "1.6"
+
 # CLI
 clap = { version = "4.5", features = ["derive"] }
 

crates/store/Cargo.toml

@@ -8,6 +8,11 @@ repository.workspace = true
 rust-version.workspace = true
 description = "Persistent state storage and operation log for Conflux documents"
 
+[features]
+default = []
+postgres = ["dep:sqlx-core", "dep:sqlx-postgres", "dep:async-trait"]
+dynamodb = ["dep:aws-sdk-dynamodb", "dep:aws-config", "dep:async-trait"]
+
 [dependencies]
 conflux-core = { path = "../core" }
 conflux-schema = { path = "../schema" }
@@ -20,5 +25,18 @@ chrono = { workspace = true }
 uuid = { workspace = true }
 parking_lot = { workspace = true }
 
+# Async (optional, for async backends)
+async-trait = { workspace = true, optional = true }
+
+# PostgreSQL (optional)
+# Note: Using sqlx-core + sqlx-postgres directly to avoid sqlite conflicts with rusqlite
+sqlx-core = { version = "0.8", default-features = false, features = ["_rt-tokio", "chrono", "uuid", "json"], optional = true }
+sqlx-postgres = { version = "0.8", default-features = false, features = ["chrono", "uuid", "json"], optional = true }
+
+# DynamoDB (optional)
+aws-sdk-dynamodb = { workspace = true, optional = true }
+aws-config = { workspace = true, optional = true }
+
 [dev-dependencies]
 insta = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }

crates/store/src/dynamo/keys.rs

@@ -0,0 +1,135 @@
+//! Key builders for DynamoDB single-table design.
+
+use conflux_core::HlcTimestamp;
+use uuid::Uuid;
+
+/// Key prefix constants and builders for the single-table design.
+pub struct Keys;
+
+impl Keys {
+    // --- Partition Key Builders ---
+
+    /// Builds the partition key for a document: `DOC#<document_id>`
+    pub fn doc_pk(document_id: &str) -> String {
+        format!("DOC#{document_id}")
+    }
+
+    /// Builds the GSI1 partition key for operation lookup: `OP#<operation_id>`
+    pub fn operation_gsi1_pk(operation_id: &Uuid) -> String {
+        format!("OP#{operation_id}")
+    }
+
+    /// Builds the GSI2 partition key for actor-based queries: `ACTOR#<document_id>#<actor_id>`
+    pub fn actor_gsi2_pk(document_id: &str, actor_id: &str) -> String {
+        format!("ACTOR#{document_id}#{actor_id}")
+    }
+
+    // --- Sort Key Builders ---
+
+    /// Builds the sort key for an operation: `OP#<hlc_timestamp>#<operation_id>`
+    ///
+    /// Using HLC as the primary sort component ensures causal ordering.
+    pub fn operation_sk(hlc: &HlcTimestamp, operation_id: &Uuid) -> String {
+        format!("OP#{}#{}", hlc, operation_id)
+    }
+
+    /// Builds the sort key for a snapshot: `SNAP#<hlc_timestamp>#<snapshot_id>`
+    pub fn snapshot_sk(hlc: &HlcTimestamp, snapshot_id: &Uuid) -> String {
+        format!("SNAP#{}#{}", hlc, snapshot_id)
+    }
+
+    /// Builds the sort key for a milestone: `MILE#<created_at>#<milestone_id>`
+    ///
+    /// Using created_at as the primary sort component enables newest-first queries.
+    pub fn milestone_sk(created_at: &str, milestone_id: &Uuid) -> String {
+        format!("MILE#{}#{}", created_at, milestone_id)
+    }
+
+    /// The sort key for a version vector: `VV#`
+    pub fn version_vector_sk() -> &'static str {
+        "VV#"
+    }
+
+    // --- Sort Key Prefixes (for range queries) ---
+
+    /// Prefix for operation sort keys.
+    pub const OP_PREFIX: &'static str = "OP#";
+
+    /// Prefix for snapshot sort keys.
+    pub const SNAP_PREFIX: &'static str = "SNAP#";
+
+    /// Prefix for milestone sort keys.
+    pub const MILE_PREFIX: &'static str = "MILE#";
+
+    // --- Parsing ---
+
+    /// Extracts the HLC timestamp from an operation sort key.
+    ///
+    /// Input format: `OP#<hlc>#<uuid>`
+    #[allow(dead_code)]
+    pub fn parse_operation_sk_hlc(sk: &str) -> Option<&str> {
+        let rest = sk.strip_prefix(Self::OP_PREFIX)?;
+        let parts: Vec<&str> = rest.splitn(2, '#').collect();
+        parts.first().copied()
+    }
+
+    /// Extracts the operation ID from an operation sort key.
+    ///
+    /// Input format: `OP#<hlc>#<uuid>`
+    #[allow(dead_code)]
+    pub fn parse_operation_sk_id(sk: &str) -> Option<&str> {
+        let rest = sk.strip_prefix(Self::OP_PREFIX)?;
+        let parts: Vec<&str> = rest.splitn(2, '#').collect();
+        parts.get(1).copied()
+    }
+
+    /// Extracts the HLC timestamp from a snapshot sort key.
+    ///
+    /// Input format: `SNAP#<hlc>#<uuid>`
+    #[allow(dead_code)]
+    pub fn parse_snapshot_sk_hlc(sk: &str) -> Option<&str> {
+        let rest = sk.strip_prefix(Self::SNAP_PREFIX)?;
+        let parts: Vec<&str> = rest.splitn(2, '#').collect();
+        parts.first().copied()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn doc_pk_format() {
+        assert_eq!(Keys::doc_pk("my-doc"), "DOC#my-doc");
+    }
+
+    #[test]
+    fn operation_gsi1_pk_format() {
+        let id = Uuid::nil();
+        assert_eq!(
+            Keys::operation_gsi1_pk(&id),
+            "OP#00000000-0000-0000-0000-000000000000"
+        );
+    }
+
+    #[test]
+    fn actor_gsi2_pk_format() {
+        assert_eq!(
+            Keys::actor_gsi2_pk("doc-1", "alice"),
+            "ACTOR#doc-1#alice"
+        );
+    }
+
+    #[test]
+    fn parse_operation_sk() {
+        let sk = "OP#2024-01-01T00:00:00.000Z/node1/1#550e8400-e29b-41d4-a716-446655440000";
+        assert_eq!(
+            Keys::parse_operation_sk_hlc(sk),
+            Some("2024-01-01T00:00:00.000Z/node1/1")
+        );
+        assert_eq!(
+            Keys::parse_operation_sk_id(sk),
+            Some("550e8400-e29b-41d4-a716-446655440000")
+        );
+    }
+}

crates/store/src/dynamo/mod.rs

@@ -0,0 +1,35 @@
+//! DynamoDB-backed async storage for Conflux.
+//!
+//! This module provides a DynamoDB implementation of the [`AsyncStore`](crate::AsyncStore) trait
+//! using a single-table design optimized for serverless deployments.
+//!
+//! # Table Design
+//!
+//! Uses a single-table design with the following key structure:
+//!
+//! | PK | SK | Type |
+//! |----|----|----|
+//! | `DOC#<doc_id>` | `OP#<hlc>#<op_id>` | Operation |
+//! | `DOC#<doc_id>` | `SNAP#<hlc>#<id>` | Snapshot |
+//! | `DOC#<doc_id>` | `MILE#<created>#<id>` | Milestone |
+//! | `DOC#<doc_id>` | `VV#` | Version Vector |
+//!
+//! # GSIs
+//!
+//! - **GSI1**: `gsi1_pk=OP#<op_id>` — Operation lookup by ID
+//! - **GSI2**: `gsi2_pk=ACTOR#<doc>#<actor>`, `gsi2_sk=<hlc>` — Operations by actor
+//!
+//! # Example
+//!
+//! ```ignore
+//! use conflux_store::DynamoStore;
+//!
+//! let store = DynamoStore::new("conflux-store").await?;
+//! store.create_table_if_not_exists().await?;
+//! store.append_operation("doc-1", &op).await?;
+//! ```
+
+mod keys;
+mod store;
+
+pub use store::DynamoStore;