feat(cdk): add generic key-value store functionality for mint databases (#1022)

* feat(cdk): add generic key-value store functionality for mint databases

Implements a comprehensive KV store system with transaction support,
namespace-based organization, and validation for mint databases.

- Add KVStoreDatabase and KVStoreTransaction traits for generic storage
- Include namespace and key validation with ASCII character restrictions
- Add database migrations for kv_store table in SQLite and PostgreSQL
- Implement comprehensive test suite for KV store functionality
- Integrate KV store traits into existing Database and Transaction bounds

crates/cdk-sql-common/src/mint/migrations.rs

@@ -3,6 +3,7 @@
 pub static MIGRATIONS: &[(&str, &str, &str)] = &[
     ("postgres", "1_initial.sql", include_str!(r#"./migrations/postgres/1_initial.sql"#)),
     ("postgres", "2_remove_request_lookup_kind_constraints.sql", include_str!(r#"./migrations/postgres/2_remove_request_lookup_kind_constraints.sql"#)),
+    ("postgres", "3_add_kv_store.sql", include_str!(r#"./migrations/postgres/3_add_kv_store.sql"#)),
     ("sqlite", "1_fix_sqlx_migration.sql", include_str!(r#"./migrations/sqlite/1_fix_sqlx_migration.sql"#)),
     ("sqlite", "20240612124932_init.sql", include_str!(r#"./migrations/sqlite/20240612124932_init.sql"#)),
     ("sqlite", "20240618195700_quote_state.sql", include_str!(r#"./migrations/sqlite/20240618195700_quote_state.sql"#)),
@@ -27,4 +28,5 @@ pub static MIGRATIONS: &[(&str, &str, &str)] = &[
     ("sqlite", "20250706101057_bolt12.sql", include_str!(r#"./migrations/sqlite/20250706101057_bolt12.sql"#)),
     ("sqlite", "20250812132015_drop_melt_request.sql", include_str!(r#"./migrations/sqlite/20250812132015_drop_melt_request.sql"#)),
     ("sqlite", "20250819200000_remove_request_lookup_kind_constraints.sql", include_str!(r#"./migrations/sqlite/20250819200000_remove_request_lookup_kind_constraints.sql"#)),
+    ("sqlite", "20250901090000_add_kv_store.sql", include_str!(r#"./migrations/sqlite/20250901090000_add_kv_store.sql"#)),
 ];

crates/cdk-sql-common/src/mint/migrations/postgres/3_add_kv_store.sql

@@ -0,0 +1,18 @@
+-- Add kv_store table for generic key-value storage
+CREATE TABLE IF NOT EXISTS kv_store (
+    primary_namespace TEXT NOT NULL,
+    secondary_namespace TEXT NOT NULL,
+    key TEXT NOT NULL,
+    value BYTEA NOT NULL,
+    created_time BIGINT NOT NULL,
+    updated_time BIGINT NOT NULL,
+    PRIMARY KEY (primary_namespace, secondary_namespace, key)
+);
+
+-- Index for efficient listing of keys by namespace
+CREATE INDEX IF NOT EXISTS idx_kv_store_namespaces 
+ON kv_store (primary_namespace, secondary_namespace);
+
+-- Index for efficient querying by update time
+CREATE INDEX IF NOT EXISTS idx_kv_store_updated_time 
+ON kv_store (updated_time);

crates/cdk-sql-common/src/mint/migrations/sqlite/20250901090000_add_kv_store.sql

@@ -0,0 +1,18 @@
+-- Add kv_store table for generic key-value storage
+CREATE TABLE IF NOT EXISTS kv_store (
+    primary_namespace TEXT NOT NULL,
+    secondary_namespace TEXT NOT NULL,
+    key TEXT NOT NULL,
+    value BLOB NOT NULL,
+    created_time INTEGER NOT NULL,
+    updated_time INTEGER NOT NULL,
+    PRIMARY KEY (primary_namespace, secondary_namespace, key)
+);
+
+-- Index for efficient listing of keys by namespace
+CREATE INDEX IF NOT EXISTS idx_kv_store_namespaces 
+ON kv_store (primary_namespace, secondary_namespace);
+
+-- Index for efficient querying by update time
+CREATE INDEX IF NOT EXISTS idx_kv_store_updated_time 
+ON kv_store (updated_time);

crates/cdk-sql-common/src/mint/mod.rs

@@ -16,6 +16,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use bitcoin::bip32::DerivationPath;
 use cdk_common::common::QuoteTTL;
+use cdk_common::database::mint::validate_kvstore_params;
 use cdk_common::database::{
     self, ConversionError, Error, MintDatabase, MintDbWriterFinalizer, MintKeyDatabaseTransaction,
     MintKeysDatabase, MintProofsDatabase, MintQuotesDatabase, MintQuotesTransaction,
@@ -1582,6 +1583,224 @@ where
     }
 }
 
+#[async_trait]
+impl<RM> database::MintKVStoreTransaction<'_, Error> for SQLTransaction<RM>
+where
+    RM: DatabasePool + 'static,
+{
+    async fn kv_read(
+        &mut self,
+        primary_namespace: &str,
+        secondary_namespace: &str,
+        key: &str,
+    ) -> Result<Option<Vec<u8>>, Error> {
+        // Validate parameters according to KV store requirements
+        validate_kvstore_params(primary_namespace, secondary_namespace, key)?;
+        Ok(query(
+            r#"
+            SELECT value
+            FROM kv_store
+            WHERE primary_namespace = :primary_namespace
+            AND secondary_namespace = :secondary_namespace
+            AND key = :key
+            "#,
+        )?
+        .bind("primary_namespace", primary_namespace.to_owned())
+        .bind("secondary_namespace", secondary_namespace.to_owned())
+        .bind("key", key.to_owned())
+        .pluck(&self.inner)
+        .await?
+        .and_then(|col| match col {
+            Column::Blob(data) => Some(data),
+            _ => None,
+        }))
+    }
+
+    async fn kv_write(
+        &mut self,
+        primary_namespace: &str,
+        secondary_namespace: &str,
+        key: &str,
+        value: &[u8],
+    ) -> Result<(), Error> {
+        // Validate parameters according to KV store requirements
+        validate_kvstore_params(primary_namespace, secondary_namespace, key)?;
+
+        let current_time = unix_time();
+
+        query(
+            r#"
+            INSERT INTO kv_store
+            (primary_namespace, secondary_namespace, key, value, created_time, updated_time)
+            VALUES (:primary_namespace, :secondary_namespace, :key, :value, :created_time, :updated_time)
+            ON CONFLICT(primary_namespace, secondary_namespace, key)
+            DO UPDATE SET
+                value = excluded.value,
+                updated_time = excluded.updated_time
+            "#,
+        )?
+        .bind("primary_namespace", primary_namespace.to_owned())
+        .bind("secondary_namespace", secondary_namespace.to_owned())
+        .bind("key", key.to_owned())
+        .bind("value", value.to_vec())
+        .bind("created_time", current_time as i64)
+        .bind("updated_time", current_time as i64)
+        .execute(&self.inner)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn kv_remove(
+        &mut self,
+        primary_namespace: &str,
+        secondary_namespace: &str,
+        key: &str,
+    ) -> Result<(), Error> {
+        // Validate parameters according to KV store requirements
+        validate_kvstore_params(primary_namespace, secondary_namespace, key)?;
+        query(
+            r#"
+            DELETE FROM kv_store
+            WHERE primary_namespace = :primary_namespace
+            AND secondary_namespace = :secondary_namespace
+            AND key = :key
+            "#,
+        )?
+        .bind("primary_namespace", primary_namespace.to_owned())
+        .bind("secondary_namespace", secondary_namespace.to_owned())
+        .bind("key", key.to_owned())
+        .execute(&self.inner)
+        .await?;
+
+        Ok(())
+    }
+
+    async fn kv_list(
+        &mut self,
+        primary_namespace: &str,
+        secondary_namespace: &str,
+    ) -> Result<Vec<String>, Error> {
+        // Validate namespace parameters according to KV store requirements
+        cdk_common::database::mint::validate_kvstore_string(primary_namespace)?;
+        cdk_common::database::mint::validate_kvstore_string(secondary_namespace)?;
+
+        // Check empty namespace rules
+        if primary_namespace.is_empty() && !secondary_namespace.is_empty() {
+            return Err(Error::KVStoreInvalidKey(
+                "If primary_namespace is empty, secondary_namespace must also be empty".to_string(),
+            ));
+        }
+        Ok(query(
+            r#"
+            SELECT key
+            FROM kv_store
+            WHERE primary_namespace = :primary_namespace
+            AND secondary_namespace = :secondary_namespace
+            ORDER BY key
+            "#,
+        )?
+        .bind("primary_namespace", primary_namespace.to_owned())
+        .bind("secondary_namespace", secondary_namespace.to_owned())
+        .fetch_all(&self.inner)
+        .await?
+        .into_iter()
+        .map(|row| Ok(column_as_string!(&row[0])))
+        .collect::<Result<Vec<_>, Error>>()?)
+    }
+}
+
+#[async_trait]
+impl<RM> database::MintKVStoreDatabase for SQLMintDatabase<RM>
+where
+    RM: DatabasePool + 'static,
+{
+    type Err = Error;
+
+    async fn kv_read(
+        &self,
+        primary_namespace: &str,
+        secondary_namespace: &str,
+        key: &str,
+    ) -> Result<Option<Vec<u8>>, Error> {
+        // Validate parameters according to KV store requirements
+        validate_kvstore_params(primary_namespace, secondary_namespace, key)?;
+
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        Ok(query(
+            r#"
+            SELECT value
+            FROM kv_store
+            WHERE primary_namespace = :primary_namespace
+            AND secondary_namespace = :secondary_namespace
+            AND key = :key
+            "#,
+        )?
+        .bind("primary_namespace", primary_namespace.to_owned())
+        .bind("secondary_namespace", secondary_namespace.to_owned())
+        .bind("key", key.to_owned())
+        .pluck(&*conn)
+        .await?
+        .and_then(|col| match col {
+            Column::Blob(data) => Some(data),
+            _ => None,
+        }))
+    }
+
+    async fn kv_list(
+        &self,
+        primary_namespace: &str,
+        secondary_namespace: &str,
+    ) -> Result<Vec<String>, Error> {
+        // Validate namespace parameters according to KV store requirements
+        cdk_common::database::mint::validate_kvstore_string(primary_namespace)?;
+        cdk_common::database::mint::validate_kvstore_string(secondary_namespace)?;
+
+        // Check empty namespace rules
+        if primary_namespace.is_empty() && !secondary_namespace.is_empty() {
+            return Err(Error::KVStoreInvalidKey(
+                "If primary_namespace is empty, secondary_namespace must also be empty".to_string(),
+            ));
+        }
+
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        Ok(query(
+            r#"
+            SELECT key
+            FROM kv_store
+            WHERE primary_namespace = :primary_namespace
+            AND secondary_namespace = :secondary_namespace
+            ORDER BY key
+            "#,
+        )?
+        .bind("primary_namespace", primary_namespace.to_owned())
+        .bind("secondary_namespace", secondary_namespace.to_owned())
+        .fetch_all(&*conn)
+        .await?
+        .into_iter()
+        .map(|row| Ok(column_as_string!(&row[0])))
+        .collect::<Result<Vec<_>, Error>>()?)
+    }
+}
+
+#[async_trait]
+impl<RM> database::MintKVStore for SQLMintDatabase<RM>
+where
+    RM: DatabasePool + 'static,
+{
+    async fn begin_transaction<'a>(
+        &'a self,
+    ) -> Result<Box<dyn database::MintKVStoreTransaction<'a, Self::Err> + Send + Sync + 'a>, Error>
+    {
+        Ok(Box::new(SQLTransaction {
+            inner: ConnectionWithTransaction::new(
+                self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
+            )
+            .await?,
+        }))
+    }
+}
+
 #[async_trait]
 impl<RM> MintDatabase<Error> for SQLMintDatabase<RM>
 where