Add PostgreSQL support

Cargo.toml

@@ -54,8 +54,9 @@ cdk-fake-wallet = { path = "./crates/cdk-fake-wallet", version = "=0.11.0" }
 cdk-payment-processor = { path = "./crates/cdk-payment-processor", default-features = true, version = "=0.11.0" }
 cdk-mint-rpc = { path = "./crates/cdk-mint-rpc", version = "=0.11.0" }
 cdk-redb = { path = "./crates/cdk-redb", default-features = true, version = "=0.11.0" }
-cdk-sqlite = { path = "./crates/cdk-sqlite", default-features = true, version = "=0.11.0" }
 cdk-sql-base = { path = "./crates/cdk-sql-base", default-features = true, version = "=0.11.0" }
+cdk-sqlite = { path = "./crates/cdk-sqlite", default-features = true, version = "=0.11.0" }
+cdk-postgres = { path = "./crates/cdk-postgres", default-features = true, version = "=0.11.0" }
 cdk-signatory = { path = "./crates/cdk-signatory", version = "=0.11.0", default-features = false }
 clap = { version = "4.5.31", features = ["derive"] }
 ciborium = { version = "0.2.2", default-features = false, features = ["std"] }

crates/cdk-common/src/database/mint/test.rs

@@ -225,7 +225,7 @@ macro_rules! mint_db_test {
         mint_db_test!(test_remove_spent_proofs, $make_db_fn);
     };
     ($name:ident, $make_db_fn:ident) => {
-        #[tokio::test]
+        #[tokio::test(flavor = "multi_thread", worker_threads = 10)]
         async fn $name() {
             $crate::database::mint::test::$name($make_db_fn().await).await;
         }

crates/cdk-integration-tests/Cargo.toml

@@ -25,6 +25,7 @@ cdk-cln = { workspace = true }
 cdk-lnd = { workspace = true }
 cdk-axum = { workspace = true }
 cdk-sqlite = { workspace = true }
+cdk-postgres = { workspace = true }
 cdk-redb = { workspace = true }
 cdk-fake-wallet = { workspace = true }
 futures = { workspace = true, default-features = false, features = [

crates/cdk-integration-tests/src/init_pure_tests.rs

@@ -23,6 +23,7 @@ use cdk::util::unix_time;
 use cdk::wallet::{AuthWallet, MintConnector, Wallet, WalletBuilder};
 use cdk::{Amount, Error, Mint};
 use cdk_fake_wallet::FakeWallet;
+use cdk_postgres::MintPgDatabase;
 use tokio::sync::{Notify, RwLock};
 use tracing_subscriber::EnvFilter;
 use uuid::Uuid;
@@ -177,6 +178,15 @@ pub async fn create_and_start_test_mint() -> Result<Mint> {
         "memory" => MintBuilder::new()
             .with_localstore(Arc::new(cdk_sqlite::mint::memory::empty().await?))
             .with_keystore(Arc::new(cdk_sqlite::mint::memory::empty().await?)),
+        "pgsql" => {
+            let connection_str = env::var("CDK_PGSQL").expect("Postgres connection URL");
+
+            MintBuilder::new()
+                .with_localstore(Arc::new(
+                    MintPgDatabase::new(connection_str.as_str()).await?,
+                ))
+                .with_keystore(Arc::new(cdk_sqlite::mint::memory::empty().await?))
+        }
         _ => {
             // Create a temporary directory for SQLite database
             let temp_dir = create_temp_dir("cdk-test-sqlite-mint")?;

crates/cdk-mintd/Cargo.toml

@@ -23,7 +23,7 @@ sqlcipher = ["cdk-sqlite/sqlcipher"]
 # MSRV is not committed to with swagger enabled
 swagger = ["cdk-axum/swagger", "dep:utoipa", "dep:utoipa-swagger-ui"]
 redis = ["cdk-axum/redis"]
-auth = ["cdk/auth", "cdk-sqlite/auth"]
+auth = ["cdk/auth", "cdk-sqlite/auth", "cdk-postgres/auth"]
 
 [dependencies]
 anyhow.workspace = true
@@ -35,6 +35,7 @@ cdk = { workspace = true, features = [
 cdk-sqlite = { workspace = true, features = [
     "mint",
 ] }
+cdk-postgres = { workspace = true, features = ["mint"] }
 cdk-cln = { workspace = true, optional = true }
 cdk-lnbits = { workspace = true, optional = true }
 cdk-lnd = { workspace = true, optional = true }

crates/cdk-mintd/src/config.rs

@@ -190,6 +190,7 @@ pub struct GrpcProcessor {
 pub enum DatabaseEngine {
     #[default]
     Sqlite,
+    PgSql,
 }
 
 impl std::str::FromStr for DatabaseEngine {
@@ -198,6 +199,7 @@ impl std::str::FromStr for DatabaseEngine {
     fn from_str(s: &str) -> Result<Self, Self::Err> {
         match s.to_lowercase().as_str() {
             "sqlite" => Ok(DatabaseEngine::Sqlite),
+            "pgsql" => Ok(DatabaseEngine::PgSql),
             _ => Err(format!("Unknown database engine: {s}")),
         }
     }

crates/cdk-mintd/src/main.rs

@@ -45,6 +45,7 @@ use cdk_mintd::cli::CLIArgs;
 use cdk_mintd::config::{self, DatabaseEngine, LnBackend};
 use cdk_mintd::env_vars::ENV_WORK_DIR;
 use cdk_mintd::setup::LnBackendSetup;
+use cdk_postgres::MintPgDatabase;
 #[cfg(feature = "auth")]
 use cdk_sqlite::mint::MintSqliteAuthDatabase;
 use cdk_sqlite::MintSqliteDatabase;
@@ -202,6 +203,13 @@ async fn setup_database(
             let keystore: Arc<dyn MintKeysDatabase<Err = cdk_database::Error> + Send + Sync> = db;
             Ok((localstore, keystore))
         }
+        DatabaseEngine::PgSql => {
+            let conn_str = "".to_owned();
+            let db = Arc::new(MintPgDatabase::new(conn_str.as_str()).await?);
+            let localstore: Arc<dyn MintDatabase<cdk_database::Error> + Send + Sync> = db.clone();
+            let keystore: Arc<dyn MintKeysDatabase<Err = cdk_database::Error> + Send + Sync> = db;
+            Ok((localstore, keystore))
+        }
     }
 }
 
@@ -518,6 +526,9 @@ async fn setup_authentication(
                 let sqlite_db = MintSqliteAuthDatabase::new(sql_db_path).await?;
                 Arc::new(sqlite_db)
             }
+            DatabaseEngine::PgSql => {
+                todo!()
+            }
         };
 
         mint_builder = mint_builder.with_auth_localstore(auth_localstore.clone());

crates/cdk-postgres/Cargo.toml

@@ -0,0 +1,35 @@
+[package]
+name = "cdk-postgres"
+version.workspace = true
+edition.workspace = true
+authors = ["CDK Developers"]
+description = "PostgreSQL storage backend for CDK"
+license.workspace = true
+homepage = "https://github.com/cashubtc/cdk"
+repository = "https://github.com/cashubtc/cdk.git"
+rust-version.workspace = true                            # MSRV
+readme = "README.md"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[features]
+default = ["mint", "wallet", "auth"]
+mint = ["cdk-common/mint", "cdk-sql-base/mint"]
+wallet = ["cdk-common/wallet", "cdk-sql-base/wallet"]
+auth = ["cdk-common/auth", "cdk-sql-base/auth"]
+
+[dependencies]
+async-trait.workspace = true
+cdk-common = { workspace = true, features = ["test"] }
+bitcoin.workspace = true
+cdk-sql-base = { workspace = true }
+thiserror.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread"] }
+tracing.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+lightning-invoice.workspace = true
+uuid.workspace = true
+tokio-postgres = "0.7.13"
+futures-util = "0.3.31"
+postgres-native-tls = "0.5.1"
+once_cell.workspace = true

crates/cdk-postgres/src/db.rs

@@ -0,0 +1,135 @@
+use cdk_common::database::Error;
+use cdk_sql_base::stmt::{Column, Statement};
+use futures_util::{pin_mut, TryStreamExt};
+use tokio_postgres::Client;
+
+use crate::value::PgValue;
+
+#[inline(always)]
+pub async fn pg_batch(conn: &Client, statement: Statement) -> Result<(), Error> {
+    let (sql, _placeholder_values) = statement.to_sql()?;
+
+    conn.batch_execute(&sql)
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))
+}
+
+#[inline(always)]
+pub async fn pg_execute(conn: &Client, statement: Statement) -> Result<usize, Error> {
+    let (sql, placeholder_values) = statement.to_sql()?;
+    let prepared_statement = conn
+        .prepare(&sql)
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    conn.execute_raw(
+        &prepared_statement,
+        placeholder_values
+            .iter()
+            .map(|x| x.into())
+            .collect::<Vec<PgValue>>(),
+    )
+    .await
+    .map_err(|e| Error::Database(Box::new(e)))
+    .map(|x| x as usize)
+}
+
+#[inline(always)]
+pub async fn pg_fetch_one(
+    conn: &Client,
+    statement: Statement,
+) -> Result<Option<Vec<Column>>, Error> {
+    let (sql, placeholder_values) = statement.to_sql()?;
+    let prepared_statement = conn
+        .prepare(&sql)
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    let stream = conn
+        .query_raw(
+            &prepared_statement,
+            placeholder_values
+                .iter()
+                .map(|x| x.into())
+                .collect::<Vec<PgValue>>(),
+        )
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    pin_mut!(stream);
+
+    Ok(stream
+        .try_next()
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?
+        .map(|row| {
+            (0..row.len())
+                .map(|i| row.get::<_, PgValue>(i).into())
+                .collect::<Vec<_>>()
+        }))
+}
+
+#[inline(always)]
+pub async fn pg_fetch_all(conn: &Client, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
+    let (sql, placeholder_values) = statement.to_sql()?;
+    let prepared_statement = conn
+        .prepare(&sql)
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    let stream = conn
+        .query_raw(
+            &prepared_statement,
+            placeholder_values
+                .iter()
+                .map(|x| x.into())
+                .collect::<Vec<PgValue>>(),
+        )
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    pin_mut!(stream);
+
+    let mut rows = vec![];
+    while let Some(row) = stream
+        .try_next()
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?
+    {
+        rows.push(
+            (0..row.len())
+                .map(|i| row.get::<_, PgValue>(i).into())
+                .collect::<Vec<_>>(),
+        );
+    }
+
+    Ok(rows)
+}
+
+#[inline(always)]
+pub async fn gn_pluck(conn: &Client, statement: Statement) -> Result<Option<Column>, Error> {
+    let (sql, placeholder_values) = statement.to_sql()?;
+    let prepared_statement = conn
+        .prepare(&sql)
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    let stream = conn
+        .query_raw(
+            &prepared_statement,
+            placeholder_values
+                .iter()
+                .map(|x| x.into())
+                .collect::<Vec<PgValue>>(),
+        )
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?;
+
+    pin_mut!(stream);
+
+    Ok(stream
+        .try_next()
+        .await
+        .map_err(|e| Error::Database(Box::new(e)))?
+        .map(|row| row.get::<_, PgValue>(0).into()))
+}

crates/cdk-postgres/src/lib.rs

@@ -0,0 +1,434 @@
+use std::fmt::Debug;
+use std::marker::PhantomData;
+use std::sync::atomic::AtomicBool;
+use std::sync::{Arc, OnceLock};
+use std::time::Duration;
+
+use cdk_common::database::Error;
+use cdk_sql_base::database::{DatabaseConnector, DatabaseExecutor, DatabaseTransaction};
+use cdk_sql_base::mint::SQLMintAuthDatabase;
+use cdk_sql_base::pool::{Pool, PooledResource, ResourceManager};
+use cdk_sql_base::stmt::{Column, Statement};
+use cdk_sql_base::{SQLMintDatabase, SQLWalletDatabase};
+use db::{gn_pluck, pg_batch, pg_execute, pg_fetch_all, pg_fetch_one};
+use tokio::sync::{Mutex, Notify};
+use tokio::time::timeout;
+use tokio_postgres::{connect, Client, Error as PgError, NoTls};
+
+mod db;
+mod value;
+
+#[derive(Debug)]
+pub struct PgConnectionPool;
+
+#[derive(Clone)]
+pub enum SslMode {
+    NoTls(NoTls),
+    NativeTls(postgres_native_tls::MakeTlsConnector),
+}
+
+impl Default for SslMode {
+    fn default() -> Self {
+        SslMode::NoTls(NoTls {})
+    }
+}
+
+impl Debug for SslMode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let debug_text = match self {
+            Self::NoTls(_) => "NoTls",
+            Self::NativeTls(_) => "NativeTls",
+        };
+
+        write!(f, "SslMode::{debug_text}")
+    }
+}
+
+/// Postgres configuration
+#[derive(Clone, Debug)]
+pub struct PgConfig {
+    url: String,
+    tls: SslMode,
+}
+
+/// A simple wrapper for the async connect, this would trigger the `connect` in another tokio task
+/// that would eventually resolve
+#[derive(Debug)]
+pub struct FutureConnect {
+    timeout: Duration,
+    error: Arc<Mutex<Option<cdk_common::database::Error>>>,
+    result: Arc<OnceLock<Client>>,
+    notify: Arc<Notify>,
+}
+
+impl FutureConnect {
+    /// Creates a new instance
+    pub fn new(config: PgConfig, timeout: Duration, still_valid: Arc<AtomicBool>) -> Self {
+        let failed = Arc::new(Mutex::new(None));
+        let result = Arc::new(OnceLock::new());
+        let notify = Arc::new(Notify::new());
+        let error_clone = failed.clone();
+        let result_clone = result.clone();
+        let notify_clone = notify.clone();
+
+        tokio::spawn(async move {
+            match config.tls {
+                SslMode::NoTls(tls) => {
+                    let (client, connection) = match connect(&config.url, tls).await {
+                        Ok((client, connection)) => (client, connection),
+                        Err(err) => {
+                            *error_clone.lock().await =
+                                Some(cdk_common::database::Error::Database(Box::new(err)));
+                            still_valid.store(false, std::sync::atomic::Ordering::Release);
+                            notify_clone.notify_waiters();
+                            return;
+                        }
+                    };
+
+                    tokio::spawn(async move {
+                        let _ = connection.await;
+                        still_valid.store(false, std::sync::atomic::Ordering::Release);
+                    });
+
+                    let _ = result_clone.set(client);
+                    notify_clone.notify_waiters();
+                }
+                SslMode::NativeTls(tls) => {
+                    let (client, connection) = match connect(&config.url, tls).await {
+                        Ok((client, connection)) => (client, connection),
+                        Err(err) => {
+                            *error_clone.lock().await =
+                                Some(cdk_common::database::Error::Database(Box::new(err)));
+                            still_valid.store(false, std::sync::atomic::Ordering::Release);
+                            notify_clone.notify_waiters();
+                            return;
+                        }
+                    };
+
+                    tokio::spawn(async move {
+                        let _ = connection.await;
+                        still_valid.store(false, std::sync::atomic::Ordering::Release);
+                    });
+
+                    let _ = result_clone.set(client);
+                    notify_clone.notify_waiters();
+                }
+            }
+        });
+
+        Self {
+            error: failed,
+            timeout,
+            result,
+            notify,
+        }
+    }
+
+    /// Gets the wrapped instance or the connection error. The connection is returned as reference,
+    /// and the actual error is returned once, next times a generic error would be returned
+    pub async fn client(&self) -> Result<&Client, cdk_common::database::Error> {
+        if let Some(client) = self.result.get() {
+            return Ok(client);
+        }
+
+        if let Some(error) = self.error.lock().await.take() {
+            return Err(error);
+        }
+
+        if timeout(self.timeout, self.notify.notified()).await.is_err() {
+            return Err(cdk_common::database::Error::Internal("Timeout".to_owned()));
+        }
+
+        // Check result again
+        if let Some(client) = self.result.get() {
+            Ok(client)
+        } else {
+            if let Some(error) = self.error.lock().await.take() {
+                Err(error)
+            } else {
+                Err(cdk_common::database::Error::Internal(
+                    "Failed connection".to_owned(),
+                ))
+            }
+        }
+    }
+}
+
+impl ResourceManager for PgConnectionPool {
+    type Config = PgConfig;
+
+    type Resource = FutureConnect;
+
+    type Error = PgError;
+
+    fn new_resource(
+        config: &Self::Config,
+        still_valid: Arc<AtomicBool>,
+        timeout: Duration,
+    ) -> Result<Self::Resource, cdk_sql_base::pool::Error<Self::Error>> {
+        Ok(FutureConnect::new(config.to_owned(), timeout, still_valid))
+    }
+}
+
+#[derive(Debug)]
+pub struct CdkPostgres {
+    pool: Arc<Pool<PgConnectionPool>>,
+}
+
+impl From<&str> for CdkPostgres {
+    fn from(value: &str) -> Self {
+        let config = PgConfig {
+            url: value.to_owned(),
+            tls: Default::default(),
+        };
+        let pool = Pool::<PgConnectionPool>::new(config, 10, Duration::from_secs(10));
+        CdkPostgres { pool }
+    }
+}
+
+pub struct CdkPostgresTx<'a> {
+    conn: Option<PooledResource<PgConnectionPool>>,
+    done: bool,
+    _phantom: PhantomData<&'a ()>,
+}
+
+impl<'a> Drop for CdkPostgresTx<'a> {
+    fn drop(&mut self) {
+        if let Some(conn) = self.conn.take() {
+            if !self.done {
+                tokio::spawn(async move {
+                    let _ = conn
+                        .client()
+                        .await
+                        .expect("client")
+                        .batch_execute("ROLLBACK")
+                        .await;
+                });
+            }
+        }
+    }
+}
+
+impl<'a> Debug for CdkPostgresTx<'a> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "PgTx")
+    }
+}
+
+#[async_trait::async_trait]
+impl DatabaseConnector for CdkPostgres {
+    type Transaction<'a> = CdkPostgresTx<'a>;
+
+    async fn begin(&self) -> Result<Self::Transaction<'_>, Error> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
+        conn.client()
+            .await?
+            .batch_execute("BEGIN TRANSACTION")
+            .await
+            .map_err(|e| Error::Database(Box::new(e)))?;
+
+        Ok(Self::Transaction {
+            conn: Some(conn),
+            done: false,
+            _phantom: PhantomData,
+        })
+    }
+}
+
+#[async_trait::async_trait]
+impl<'a> DatabaseTransaction<'a> for CdkPostgresTx<'a> {
+    async fn commit(mut self) -> Result<(), Error> {
+        self.conn
+            .as_ref()
+            .ok_or(Error::Internal("Missing connection".to_owned()))?
+            .client()
+            .await?
+            .batch_execute("COMMIT")
+            .await
+            .map_err(|e| Error::Database(Box::new(e)))?;
+        self.done = true;
+        Ok(())
+    }
+
+    async fn rollback(mut self) -> Result<(), Error> {
+        self.conn
+            .as_ref()
+            .ok_or(Error::Internal("Missing connection".to_owned()))?
+            .client()
+            .await?
+            .batch_execute("ROLLBACK")
+            .await
+            .map_err(|e| Error::Database(Box::new(e)))?;
+        self.done = true;
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl<'a> DatabaseExecutor for CdkPostgresTx<'a> {
+    fn name() -> &'static str {
+        "postgres"
+    }
+
+    async fn execute(&self, statement: Statement) -> Result<usize, Error> {
+        pg_execute(
+            self.conn
+                .as_ref()
+                .ok_or(Error::Internal("Missing connection".to_owned()))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
+        pg_fetch_one(
+            self.conn
+                .as_ref()
+                .ok_or(Error::Internal("Missing connection".to_owned()))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
+        pg_fetch_all(
+            self.conn
+                .as_ref()
+                .ok_or(Error::Internal("Missing connection".to_owned()))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
+        gn_pluck(
+            self.conn
+                .as_ref()
+                .ok_or(Error::Internal("Missing connection".to_owned()))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn batch(&self, statement: Statement) -> Result<(), Error> {
+        pg_batch(
+            self.conn
+                .as_ref()
+                .ok_or(Error::Internal("Missing connection".to_owned()))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+}
+
+#[async_trait::async_trait]
+impl DatabaseExecutor for CdkPostgres {
+    fn name() -> &'static str {
+        "postgres"
+    }
+
+    async fn execute(&self, statement: Statement) -> Result<usize, Error> {
+        pg_execute(
+            self.pool
+                .get()
+                .map_err(|e| Error::Database(Box::new(e)))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
+        pg_fetch_one(
+            self.pool
+                .get()
+                .map_err(|e| Error::Database(Box::new(e)))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
+        pg_fetch_all(
+            self.pool
+                .get()
+                .map_err(|e| Error::Database(Box::new(e)))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
+        gn_pluck(
+            self.pool
+                .get()
+                .map_err(|e| Error::Database(Box::new(e)))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+
+    async fn batch(&self, statement: Statement) -> Result<(), Error> {
+        pg_batch(
+            self.pool
+                .get()
+                .map_err(|e| Error::Database(Box::new(e)))?
+                .client()
+                .await?,
+            statement,
+        )
+        .await
+    }
+}
+
+/// Mint DB implementation with PostgreSQL
+pub type MintPgDatabase = SQLMintDatabase<CdkPostgres>;
+
+/// Mint Auth database with Postgres
+#[cfg(feature = "auth")]
+pub type MintPgAuthDatabase = SQLMintAuthDatabase<CdkPostgres>;
+
+/// Mint DB implementation with PostgresSQL
+pub type WalletPgDatabase = SQLWalletDatabase<CdkPostgres>;
+
+#[cfg(test)]
+mod test {
+    use cdk_common::mint_db_test;
+    use once_cell::sync::Lazy;
+    use tokio::sync::Mutex;
+
+    use super::*;
+
+    static MIGRATION_LOCK: Lazy<Mutex<()>> = Lazy::new(|| Mutex::new(()));
+
+    async fn provide_db() -> MintPgDatabase {
+        let m = MIGRATION_LOCK.lock().await;
+        let db_url = std::env::var("DATABASE_URL")
+            .unwrap_or("host=localhost user=test password=test dbname=testdb port=5433".to_owned());
+        let db = MintPgDatabase::new(db_url.as_str())
+            .await
+            .expect("database");
+        drop(m);
+        db
+    }
+
+    mint_db_test!(provide_db);
+}

crates/cdk-postgres/src/value.rs

@@ -0,0 +1,130 @@
+use std::fmt::Debug;
+
+use cdk_sql_base::value::Value;
+use tokio_postgres::types::{self, FromSql, ToSql};
+
+#[derive(Debug)]
+pub enum PgValue<'a> {
+    Null,
+    Integer(i64),
+    Real(f64),
+    Text(&'a str),
+    Blob(&'a [u8]),
+}
+
+impl<'a> From<&'a Value> for PgValue<'a> {
+    fn from(value: &'a Value) -> Self {
+        match value {
+            Value::Blob(b) => PgValue::Blob(b),
+            Value::Text(text) => PgValue::Text(text.as_str()),
+            Value::Null => PgValue::Null,
+            Value::Integer(i) => PgValue::Integer(*i),
+            Value::Real(r) => PgValue::Real(*r),
+        }
+    }
+}
+
+impl<'a> From<PgValue<'a>> for Value {
+    fn from(val: PgValue<'a>) -> Self {
+        match val {
+            PgValue::Blob(value) => Value::Blob(value.to_owned()),
+            PgValue::Text(value) => Value::Text(value.to_owned()),
+            PgValue::Null => Value::Null,
+            PgValue::Integer(n) => Value::Integer(n),
+            PgValue::Real(r) => Value::Real(r),
+        }
+    }
+}
+
+impl<'a> FromSql<'a> for PgValue<'a> {
+    fn accepts(_ty: &types::Type) -> bool {
+        true
+    }
+
+    fn from_sql(
+        ty: &types::Type,
+        raw: &'a [u8],
+    ) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        Ok(match *ty {
+            types::Type::VARCHAR | types::Type::TEXT | types::Type::BPCHAR | types::Type::NAME => {
+                PgValue::Text(<&str as FromSql>::from_sql(ty, raw)?)
+            }
+            types::Type::BOOL => PgValue::Integer(if <bool as FromSql>::from_sql(ty, raw)? {
+                1
+            } else {
+                0
+            }),
+            types::Type::INT2 | types::Type::INT4 | types::Type::INT8 => {
+                PgValue::Integer(<i64 as FromSql>::from_sql(ty, raw)?)
+            }
+            types::Type::BIT_ARRAY | types::Type::BYTEA | types::Type::UNKNOWN => {
+                PgValue::Blob(<&[u8] as FromSql>::from_sql(ty, raw)?)
+            }
+            _ => panic!("Unsupported type {ty:?}"),
+        })
+    }
+
+    fn from_sql_null(_ty: &types::Type) -> Result<Self, Box<dyn std::error::Error + Sync + Send>> {
+        Ok(PgValue::Null)
+    }
+}
+
+impl ToSql for PgValue<'_> {
+    fn to_sql(
+        &self,
+        ty: &types::Type,
+        out: &mut types::private::BytesMut,
+    ) -> Result<types::IsNull, Box<dyn std::error::Error + Sync + Send>>
+    where
+        Self: Sized,
+    {
+        match self {
+            PgValue::Blob(blob) => blob.to_sql(ty, out),
+            PgValue::Text(text) => text.to_sql(ty, out),
+            PgValue::Null => Ok(types::IsNull::Yes),
+            PgValue::Real(r) => r.to_sql(ty, out),
+            PgValue::Integer(i) => match *ty {
+                types::Type::BOOL => (*i != 0).to_sql(ty, out),
+                types::Type::INT2 => (*i as i16).to_sql(ty, out),
+                types::Type::INT4 => (*i as i32).to_sql(ty, out),
+                _ => i.to_sql_checked(ty, out),
+            },
+        }
+    }
+
+    fn accepts(_ty: &types::Type) -> bool
+    where
+        Self: Sized,
+    {
+        true
+    }
+
+    fn encode_format(&self, ty: &types::Type) -> types::Format {
+        match self {
+            PgValue::Blob(blob) => blob.encode_format(ty),
+            PgValue::Text(text) => text.encode_format(ty),
+            PgValue::Null => types::Format::Text,
+            PgValue::Real(r) => r.encode_format(ty),
+            PgValue::Integer(i) => i.encode_format(ty),
+        }
+    }
+
+    fn to_sql_checked(
+        &self,
+        ty: &types::Type,
+        out: &mut types::private::BytesMut,
+    ) -> Result<types::IsNull, Box<dyn std::error::Error + Sync + Send>> {
+        match self {
+            PgValue::Blob(blob) => blob.to_sql_checked(ty, out),
+            PgValue::Text(text) => text.to_sql_checked(ty, out),
+            PgValue::Null => Ok(types::IsNull::Yes),
+            PgValue::Real(r) => r.to_sql_checked(ty, out),
+            PgValue::Integer(i) => match *ty {
+                types::Type::BOOL => (*i != 0).to_sql_checked(ty, out),
+                types::Type::INT2 => (*i as i16).to_sql_checked(ty, out),
+                types::Type::INT4 => (*i as i32).to_sql_checked(ty, out),
+                _ => i.to_sql_checked(ty, out),
+            },
+        }
+    }
+}

crates/cdk-postgres/start_db_for_test.sh

@@ -0,0 +1,26 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+CONTAINER_NAME="rust-test-pg"
+DB_USER="test"
+DB_PASS="test"
+DB_NAME="testdb"
+DB_PORT="5433"
+
+echo "Starting fresh PostgreSQL container..."
+docker run -d --rm \
+  --name "${CONTAINER_NAME}" \
+  -e POSTGRES_USER="${DB_USER}" \
+  -e POSTGRES_PASSWORD="${DB_PASS}" \
+  -e POSTGRES_DB="${DB_NAME}" \
+  -p ${DB_PORT}:5432 \
+  postgres:16
+
+echo "Waiting for PostgreSQL to be ready and database '${DB_NAME}' to exist..."
+until docker exec -e PGPASSWORD="${DB_PASS}" "${CONTAINER_NAME}" \
+    psql -U "${DB_USER}" -d "${DB_NAME}" -c "SELECT 1;" >/dev/null 2>&1; do
+  sleep 0.5
+done
+
+export DATABASE_URL="host=localhost user=${DB_USER} password=${DB_PASS} dbname=${DB_NAME} port=${DB_PORT}"
+

crates/cdk-sql-base/build.rs

@@ -18,16 +18,18 @@ fn main() {
         let dest_path = parent.join("migrations.rs");
         let mut out_file = File::create(&dest_path).expect("Failed to create migrations.rs");
 
+        let skip_name = migration_path.to_str().unwrap_or_default().len();
+
         writeln!(out_file, "/// @generated").unwrap();
         writeln!(out_file, "/// Auto-generated by build.rs").unwrap();
         writeln!(out_file, "pub static MIGRATIONS: &[(&str, &str)] = &[").unwrap();
 
         for path in &files {
-            let name = path.file_name().unwrap().to_string_lossy();
+            let rel_name = &path.to_str().unwrap().replace("\\", "/")[skip_name + 1..]; // for Windows
             let rel_path = &path.to_str().unwrap().replace("\\", "/")[skip_path..]; // for Windows
             writeln!(
                 out_file,
-                "    (\"{name}\", include_str!(r#\".{rel_path}\"#)),"
+                "    (\"{rel_name}\", include_str!(r#\".{rel_path}\"#)),"
             )
             .unwrap();
         }

crates/cdk-sql-base/run_test.sh

@@ -1,38 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-CONTAINER_NAME="rust-test-pg"
-DB_USER="test"
-DB_PASS="test"
-DB_NAME="testdb"
-DB_PORT="5433"
-DB_URL="postgres://${DB_USER}:${DB_PASS}@localhost:${DB_PORT}/${DB_NAME}"
-
-cleanup() {
-  echo "Cleaning up..."
-  docker stop "${CONTAINER_NAME}" >/dev/null 2>&1 || true
-  docker rm "${CONTAINER_NAME}" >/dev/null 2>&1 || true
-}
-
-trap cleanup EXIT INT TERM
-
-echo "Starting fresh PostgreSQL container..."
-docker run -d --rm \
-  --name "${CONTAINER_NAME}" \
-  -e POSTGRES_USER="${DB_USER}" \
-  -e POSTGRES_PASSWORD="${DB_PASS}" \
-  -e POSTGRES_DB="${DB_NAME}" \
-  -p ${DB_PORT}:5432 \
-  -v "${PWD}/.docker/pg-init.sql:/docker-entrypoint-initdb.d/init.sql:ro" \
-  postgres:16
-
-echo "Waiting for PostgreSQL to be ready..."
-until docker exec "${CONTAINER_NAME}" pg_isready -U "${DB_USER}" >/dev/null 2>&1; do
-  sleep 0.5
-done
-
-export DATABASE_URL="${DB_URL}"
-
-echo "Running cargo tests..."
-cargo test
-

crates/cdk-sql-base/src/common.rs

@@ -5,6 +5,7 @@ use crate::stmt::query;
 #[inline(always)]
 pub async fn migrate<C: DatabaseExecutor>(
     conn: &C,
+    db_prefix: &str,
     migrations: &[(&str, &str)],
 ) -> Result<(), cdk_common::database::Error> {
     query(
@@ -18,26 +19,14 @@ pub async fn migrate<C: DatabaseExecutor>(
     .execute(conn)
     .await?;
 
-    /*if query(
-        r#"select count(*) from sqlite_master where name = '_sqlx_migrations'"#,
-        [],
-        |row| row.get::<_, i32>(0),
-    )? == 1
-    {
-        tx.execute_batch(
-            r#"
-        INSERT INTO migrations
-        SELECT
-            version || '_' ||  REPLACE(description, ' ', '_') || '.sql',
-            execution_time
-        FROM _sqlx_migrations;
-        DROP TABLE _sqlx_migrations;
-        "#,
-        )?;
-    }*/
-
     // Apply each migration if it hasn’t been applied yet
     for (name, sql) in migrations {
+        if let Some((prefix, _)) = name.split_once(['/', '\\']) {
+            if prefix != db_prefix {
+                continue;
+            }
+        }
+
         let is_missing = query("SELECT name FROM migrations WHERE name = :name")?
             .bind("name", name)
             .pluck(conn)

crates/cdk-sql-base/src/database.rs

@@ -11,6 +11,9 @@ use crate::stmt::{Column, Statement};
 /// This trait defines the expectations of a database execution
 #[async_trait::async_trait]
 pub trait DatabaseExecutor: Debug + Sync + Send {
+    /// Database driver name
+    fn name() -> &'static str;
+
     /// Executes a query and returns the affected rows
     async fn execute(&self, statement: Statement) -> Result<usize, Error>;
 

crates/cdk-sql-base/src/mint/auth/migrations.rs

@@ -1,6 +1,5 @@
 /// @generated
 /// Auto-generated by build.rs
 pub static MIGRATIONS: &[(&str, &str)] = &[
-    ("20250109143347_init.sql", include_str!(r#"./migrations/20250109143347_init.sql"#)),
-    ("20250109143347_init.sql", include_str!(r#"./migrations/sqlite/20250109143347_init.sql"#)),
+    ("sqlite/20250109143347_init.sql", include_str!(r#"./migrations/sqlite/20250109143347_init.sql"#)),
 ];

crates/cdk-sql-base/src/mint/auth/mod.rs

@@ -45,7 +45,7 @@ where
     /// Migrate
     async fn migrate(conn: &DB) -> Result<(), Error> {
         let tx = conn.begin().await?;
-        migrate(&tx, MIGRATIONS).await?;
+        migrate(&tx, DB::name(), MIGRATIONS).await?;
         tx.commit().await?;
         Ok(())
     }

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

@@ -1,46 +1,26 @@
 /// @generated
 /// Auto-generated by build.rs
 pub static MIGRATIONS: &[(&str, &str)] = &[
-    ("20240612124932_init.sql", include_str!(r#"./migrations/20240612124932_init.sql"#)),
-    ("20240618195700_quote_state.sql", include_str!(r#"./migrations/20240618195700_quote_state.sql"#)),
-    ("20240626092101_nut04_state.sql", include_str!(r#"./migrations/20240626092101_nut04_state.sql"#)),
-    ("20240703122347_request_lookup_id.sql", include_str!(r#"./migrations/20240703122347_request_lookup_id.sql"#)),
-    ("20240710145043_input_fee.sql", include_str!(r#"./migrations/20240710145043_input_fee.sql"#)),
-    ("20240711183109_derivation_path_index.sql", include_str!(r#"./migrations/20240711183109_derivation_path_index.sql"#)),
-    ("20240718203721_allow_unspent.sql", include_str!(r#"./migrations/20240718203721_allow_unspent.sql"#)),
-    ("20240811031111_update_mint_url.sql", include_str!(r#"./migrations/20240811031111_update_mint_url.sql"#)),
-    ("20240919103407_proofs_quote_id.sql", include_str!(r#"./migrations/20240919103407_proofs_quote_id.sql"#)),
-    ("20240923153640_melt_requests.sql", include_str!(r#"./migrations/20240923153640_melt_requests.sql"#)),
-    ("20240930101140_dleq_for_sigs.sql", include_str!(r#"./migrations/20240930101140_dleq_for_sigs.sql"#)),
-    ("20241108093102_mint_mint_quote_pubkey.sql", include_str!(r#"./migrations/20241108093102_mint_mint_quote_pubkey.sql"#)),
-    ("20250103201327_amount_to_pay_msats.sql", include_str!(r#"./migrations/20250103201327_amount_to_pay_msats.sql"#)),
-    ("20250129200912_remove_mint_url.sql", include_str!(r#"./migrations/20250129200912_remove_mint_url.sql"#)),
-    ("20250129230326_add_config_table.sql", include_str!(r#"./migrations/20250129230326_add_config_table.sql"#)),
-    ("20250307213652_keyset_id_as_foreign_key.sql", include_str!(r#"./migrations/20250307213652_keyset_id_as_foreign_key.sql"#)),
-    ("20250406091754_mint_time_of_quotes.sql", include_str!(r#"./migrations/20250406091754_mint_time_of_quotes.sql"#)),
-    ("20250406093755_mint_created_time_signature.sql", include_str!(r#"./migrations/20250406093755_mint_created_time_signature.sql"#)),
-    ("20250415093121_drop_keystore_foreign.sql", include_str!(r#"./migrations/20250415093121_drop_keystore_foreign.sql"#)),
-    ("20250626120251_rename_blind_message_y_to_b.sql", include_str!(r#"./migrations/20250626120251_rename_blind_message_y_to_b.sql"#)),
-    ("20250710212391_init.sql", include_str!(r#"./migrations/postgres/20250710212391_init.sql"#)),
-    ("1_fix_sqlx_migration.sql", include_str!(r#"./migrations/sqlite/1_fix_sqlx_migration.sql"#)),
-    ("20240612124932_init.sql", include_str!(r#"./migrations/sqlite/20240612124932_init.sql"#)),
-    ("20240618195700_quote_state.sql", include_str!(r#"./migrations/sqlite/20240618195700_quote_state.sql"#)),
-    ("20240626092101_nut04_state.sql", include_str!(r#"./migrations/sqlite/20240626092101_nut04_state.sql"#)),
-    ("20240703122347_request_lookup_id.sql", include_str!(r#"./migrations/sqlite/20240703122347_request_lookup_id.sql"#)),
-    ("20240710145043_input_fee.sql", include_str!(r#"./migrations/sqlite/20240710145043_input_fee.sql"#)),
-    ("20240711183109_derivation_path_index.sql", include_str!(r#"./migrations/sqlite/20240711183109_derivation_path_index.sql"#)),
-    ("20240718203721_allow_unspent.sql", include_str!(r#"./migrations/sqlite/20240718203721_allow_unspent.sql"#)),
-    ("20240811031111_update_mint_url.sql", include_str!(r#"./migrations/sqlite/20240811031111_update_mint_url.sql"#)),
-    ("20240919103407_proofs_quote_id.sql", include_str!(r#"./migrations/sqlite/20240919103407_proofs_quote_id.sql"#)),
-    ("20240923153640_melt_requests.sql", include_str!(r#"./migrations/sqlite/20240923153640_melt_requests.sql"#)),
-    ("20240930101140_dleq_for_sigs.sql", include_str!(r#"./migrations/sqlite/20240930101140_dleq_for_sigs.sql"#)),
-    ("20241108093102_mint_mint_quote_pubkey.sql", include_str!(r#"./migrations/sqlite/20241108093102_mint_mint_quote_pubkey.sql"#)),
-    ("20250103201327_amount_to_pay_msats.sql", include_str!(r#"./migrations/sqlite/20250103201327_amount_to_pay_msats.sql"#)),
-    ("20250129200912_remove_mint_url.sql", include_str!(r#"./migrations/sqlite/20250129200912_remove_mint_url.sql"#)),
-    ("20250129230326_add_config_table.sql", include_str!(r#"./migrations/sqlite/20250129230326_add_config_table.sql"#)),
-    ("20250307213652_keyset_id_as_foreign_key.sql", include_str!(r#"./migrations/sqlite/20250307213652_keyset_id_as_foreign_key.sql"#)),
-    ("20250406091754_mint_time_of_quotes.sql", include_str!(r#"./migrations/sqlite/20250406091754_mint_time_of_quotes.sql"#)),
-    ("20250406093755_mint_created_time_signature.sql", include_str!(r#"./migrations/sqlite/20250406093755_mint_created_time_signature.sql"#)),
-    ("20250415093121_drop_keystore_foreign.sql", include_str!(r#"./migrations/sqlite/20250415093121_drop_keystore_foreign.sql"#)),
-    ("20250626120251_rename_blind_message_y_to_b.sql", include_str!(r#"./migrations/sqlite/20250626120251_rename_blind_message_y_to_b.sql"#)),
+    ("postgres/20250710212391_init.sql", include_str!(r#"./migrations/postgres/20250710212391_init.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"#)),
+    ("sqlite/20240626092101_nut04_state.sql", include_str!(r#"./migrations/sqlite/20240626092101_nut04_state.sql"#)),
+    ("sqlite/20240703122347_request_lookup_id.sql", include_str!(r#"./migrations/sqlite/20240703122347_request_lookup_id.sql"#)),
+    ("sqlite/20240710145043_input_fee.sql", include_str!(r#"./migrations/sqlite/20240710145043_input_fee.sql"#)),
+    ("sqlite/20240711183109_derivation_path_index.sql", include_str!(r#"./migrations/sqlite/20240711183109_derivation_path_index.sql"#)),
+    ("sqlite/20240718203721_allow_unspent.sql", include_str!(r#"./migrations/sqlite/20240718203721_allow_unspent.sql"#)),
+    ("sqlite/20240811031111_update_mint_url.sql", include_str!(r#"./migrations/sqlite/20240811031111_update_mint_url.sql"#)),
+    ("sqlite/20240919103407_proofs_quote_id.sql", include_str!(r#"./migrations/sqlite/20240919103407_proofs_quote_id.sql"#)),
+    ("sqlite/20240923153640_melt_requests.sql", include_str!(r#"./migrations/sqlite/20240923153640_melt_requests.sql"#)),
+    ("sqlite/20240930101140_dleq_for_sigs.sql", include_str!(r#"./migrations/sqlite/20240930101140_dleq_for_sigs.sql"#)),
+    ("sqlite/20241108093102_mint_mint_quote_pubkey.sql", include_str!(r#"./migrations/sqlite/20241108093102_mint_mint_quote_pubkey.sql"#)),
+    ("sqlite/20250103201327_amount_to_pay_msats.sql", include_str!(r#"./migrations/sqlite/20250103201327_amount_to_pay_msats.sql"#)),
+    ("sqlite/20250129200912_remove_mint_url.sql", include_str!(r#"./migrations/sqlite/20250129200912_remove_mint_url.sql"#)),
+    ("sqlite/20250129230326_add_config_table.sql", include_str!(r#"./migrations/sqlite/20250129230326_add_config_table.sql"#)),
+    ("sqlite/20250307213652_keyset_id_as_foreign_key.sql", include_str!(r#"./migrations/sqlite/20250307213652_keyset_id_as_foreign_key.sql"#)),
+    ("sqlite/20250406091754_mint_time_of_quotes.sql", include_str!(r#"./migrations/sqlite/20250406091754_mint_time_of_quotes.sql"#)),
+    ("sqlite/20250406093755_mint_created_time_signature.sql", include_str!(r#"./migrations/sqlite/20250406093755_mint_created_time_signature.sql"#)),
+    ("sqlite/20250415093121_drop_keystore_foreign.sql", include_str!(r#"./migrations/sqlite/20250415093121_drop_keystore_foreign.sql"#)),
+    ("sqlite/20250626120251_rename_blind_message_y_to_b.sql", include_str!(r#"./migrations/sqlite/20250626120251_rename_blind_message_y_to_b.sql"#)),
 ];

crates/cdk-sql-base/src/mint/migrations/postgres/20250710212391_init.sql

@@ -0,0 +1,82 @@
+CREATE TABLE keyset (
+  id TEXT PRIMARY KEY, unit TEXT NOT NULL,
+  active BOOL NOT NULL, valid_from INTEGER NOT NULL,
+  valid_to INTEGER, derivation_path TEXT NOT NULL,
+  max_order INTEGER NOT NULL, input_fee_ppk INTEGER,
+  derivation_path_index INTEGER
+);
+CREATE TABLE mint_quote (
+  id TEXT PRIMARY KEY,
+  amount INTEGER NOT NULL,
+  unit TEXT NOT NULL,
+  request TEXT NOT NULL,
+  expiry INTEGER NOT NULL,
+  state TEXT CHECK (
+    state IN (
+      'UNPAID', 'PENDING', 'PAID', 'ISSUED'
+    )
+  ) NOT NULL DEFAULT 'UNPAID',
+  request_lookup_id TEXT,
+  pubkey TEXT,
+  created_time INTEGER NOT NULL DEFAULT 0,
+  paid_time INTEGER,
+  issued_time INTEGER
+);
+CREATE TABLE melt_quote (
+  id TEXT PRIMARY KEY,
+  unit TEXT NOT NULL,
+  amount INTEGER NOT NULL,
+  request TEXT NOT NULL,
+  fee_reserve INTEGER NOT NULL,
+  expiry INTEGER NOT NULL,
+  state TEXT CHECK (
+    state IN ('UNPAID', 'PENDING', 'PAID')
+  ) NOT NULL DEFAULT 'UNPAID',
+  payment_preimage TEXT,
+  request_lookup_id TEXT,
+  msat_to_pay INTEGER,
+  created_time INTEGER NOT NULL DEFAULT 0,
+  paid_time INTEGER
+);
+CREATE TABLE melt_request (
+  id TEXT PRIMARY KEY, inputs TEXT NOT NULL,
+  outputs TEXT, method TEXT NOT NULL,
+  unit TEXT NOT NULL
+);
+CREATE TABLE config (
+  id TEXT PRIMARY KEY, value TEXT NOT NULL
+);
+CREATE TABLE IF NOT EXISTS "proof" (
+  y BYTEA PRIMARY KEY,
+  amount INTEGER NOT NULL,
+  keyset_id TEXT NOT NULL,
+  secret TEXT NOT NULL,
+  c BYTEA NOT NULL,
+  witness TEXT,
+  state TEXT CHECK (
+    state IN (
+      'SPENT', 'PENDING', 'UNSPENT', 'RESERVED',
+      'UNKNOWN'
+    )
+  ) NOT NULL,
+  quote_id TEXT,
+  created_time INTEGER NOT NULL DEFAULT 0
+);
+CREATE TABLE IF NOT EXISTS "blind_signature" (
+  y BYTEA PRIMARY KEY,
+  amount INTEGER NOT NULL,
+  keyset_id TEXT NOT NULL,
+  c BYTEA NOT NULL,
+  dleq_e TEXT,
+  dleq_s TEXT,
+  quote_id TEXT,
+  created_time INTEGER NOT NULL DEFAULT 0
+);
+CREATE INDEX unit_index ON keyset(unit);
+CREATE INDEX active_index ON keyset(active);
+CREATE INDEX request_index ON mint_quote(request);
+CREATE INDEX expiry_index ON mint_quote(expiry);
+CREATE INDEX melt_quote_state_index ON melt_quote(state);
+CREATE INDEX mint_quote_state_index ON mint_quote(state);
+CREATE UNIQUE INDEX unique_request_lookup_id_mint ON mint_quote(request_lookup_id);
+CREATE UNIQUE INDEX unique_request_lookup_id_melt ON melt_quote(request_lookup_id);

crates/cdk-sql-base/src/mint/migrations/sqlite/1_fix_sqlx_migration.sql

@@ -0,0 +1,20 @@
+-- Migrate `_sqlx_migrations` to our new migration system
+CREATE TABLE IF NOT EXISTS _sqlx_migrations AS
+SELECT
+    '' AS version,
+    '' AS description,
+    0 AS execution_time
+WHERE 0;
+
+INSERT INTO migrations
+SELECT
+    version || '_' ||  REPLACE(description, ' ', '_') || '.sql',
+    execution_time
+FROM _sqlx_migrations
+WHERE EXISTS (
+    SELECT 1
+    FROM sqlite_master
+    WHERE type = 'table' AND name = '_sqlx_migrations'
+);
+
+DROP TABLE _sqlx_migrations;

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

@@ -129,7 +129,7 @@ where
     /// Migrate
     async fn migrate(conn: &DB) -> Result<(), Error> {
         let tx = conn.begin().await?;
-        migrate(&tx, MIGRATIONS).await?;
+        migrate(&tx, DB::name(), MIGRATIONS).await?;
         tx.commit().await?;
         Ok(())
     }
@@ -472,6 +472,7 @@ where
             WHERE
                 id=:id
                 AND state != :state
+            FOR UPDATE
             "#,
         )?
         .bind("id", quote_id.as_hyphenated().to_string())
@@ -547,7 +548,9 @@ where
                 issued_time
             FROM
                 mint_quote
-            WHERE id = :id"#,
+            WHERE id = :id
+            FOR UPDATE
+            "#,
         )?
         .bind("id", quote_id.as_hyphenated().to_string())
         .fetch_one(&self.inner)
@@ -640,6 +643,7 @@ where
                 melt_quote
             WHERE
                 id=:id
+            FOR UPDATE
             "#,
         )?
         .bind("id", quote_id.as_hyphenated().to_string())
@@ -669,7 +673,9 @@ where
                 issued_time
             FROM
                 mint_quote
-            WHERE request = :request"#,
+            WHERE request = :request
+            FOR UPDATE
+            "#,
         )?
         .bind("request", request.to_owned())
         .fetch_one(&self.inner)
@@ -901,7 +907,7 @@ where
 
         // Check any previous proof, this query should return None in order to proceed storing
         // Any result here would error
-        match query(r#"SELECT state FROM proof WHERE y IN (:ys) LIMIT 1"#)?
+        match query(r#"SELECT state FROM proof WHERE y IN (:ys) LIMIT 1 FOR UPDATE"#)?
             .bind_vec(
                 "ys",
                 proofs

crates/cdk-sql-base/src/pool.rs

@@ -30,7 +30,7 @@ pub trait ResourceManager: Debug {
     type Resource: Debug;
 
     /// The configuration that is needed in order to create the resource
-    type Config: Debug;
+    type Config: Clone + Debug;
 
     /// The error the resource may return when creating a new instance
     type Error: Debug;
@@ -39,6 +39,7 @@ pub trait ResourceManager: Debug {
     fn new_resource(
         config: &Self::Config,
         still_valid: Arc<AtomicBool>,
+        timeout: Duration,
     ) -> Result<Self::Resource, Error<Self::Error>>;
 
     /// The object is dropped
@@ -158,7 +159,7 @@ where
                 return Ok(PooledResource {
                     resource: Some((
                         still_valid.clone(),
-                        RM::new_resource(&self.config, still_valid)?,
+                        RM::new_resource(&self.config, still_valid, timeout)?,
                     )),
                     pool: self.clone(),
                 });

crates/cdk-sql-base/src/stmt.rs

@@ -186,10 +186,10 @@ impl Statement {
                         }
                     }
                 }
-                SqlPart::Raw(raw) => Ok(raw.to_string()),
+                SqlPart::Raw(raw) => Ok(raw.trim().to_string()),
             })
             .collect::<Result<Vec<String>, _>>()?
-            .join("");
+            .join(" ");
 
         Ok((sql, placeholder_values))
     }

crates/cdk-sql-base/src/wallet/migrations.rs

@@ -1,36 +1,20 @@
 /// @generated
 /// Auto-generated by build.rs
 pub static MIGRATIONS: &[(&str, &str)] = &[
-    ("20240612132920_init.sql", include_str!(r#"./migrations/20240612132920_init.sql"#)),
-    ("20240618200350_quote_state.sql", include_str!(r#"./migrations/20240618200350_quote_state.sql"#)),
-    ("20240626091921_nut04_state.sql", include_str!(r#"./migrations/20240626091921_nut04_state.sql"#)),
-    ("20240710144711_input_fee.sql", include_str!(r#"./migrations/20240710144711_input_fee.sql"#)),
-    ("20240810214105_mint_icon_url.sql", include_str!(r#"./migrations/20240810214105_mint_icon_url.sql"#)),
-    ("20240810233905_update_mint_url.sql", include_str!(r#"./migrations/20240810233905_update_mint_url.sql"#)),
-    ("20240902151515_icon_url.sql", include_str!(r#"./migrations/20240902151515_icon_url.sql"#)),
-    ("20240902210905_mint_time.sql", include_str!(r#"./migrations/20240902210905_mint_time.sql"#)),
-    ("20241011125207_mint_urls.sql", include_str!(r#"./migrations/20241011125207_mint_urls.sql"#)),
-    ("20241108092756_wallet_mint_quote_secretkey.sql", include_str!(r#"./migrations/20241108092756_wallet_mint_quote_secretkey.sql"#)),
-    ("20250214135017_mint_tos.sql", include_str!(r#"./migrations/20250214135017_mint_tos.sql"#)),
-    ("20250310111513_drop_nostr_last_checked.sql", include_str!(r#"./migrations/20250310111513_drop_nostr_last_checked.sql"#)),
-    ("20250314082116_allow_pending_spent.sql", include_str!(r#"./migrations/20250314082116_allow_pending_spent.sql"#)),
-    ("20250323152040_wallet_dleq_proofs.sql", include_str!(r#"./migrations/20250323152040_wallet_dleq_proofs.sql"#)),
-    ("20250401120000_add_transactions_table.sql", include_str!(r#"./migrations/20250401120000_add_transactions_table.sql"#)),
-    ("20250616144830_add_keyset_expiry.sql", include_str!(r#"./migrations/20250616144830_add_keyset_expiry.sql"#)),
-    ("20240612132920_init.sql", include_str!(r#"./migrations/sqlite/20240612132920_init.sql"#)),
-    ("20240618200350_quote_state.sql", include_str!(r#"./migrations/sqlite/20240618200350_quote_state.sql"#)),
-    ("20240626091921_nut04_state.sql", include_str!(r#"./migrations/sqlite/20240626091921_nut04_state.sql"#)),
-    ("20240710144711_input_fee.sql", include_str!(r#"./migrations/sqlite/20240710144711_input_fee.sql"#)),
-    ("20240810214105_mint_icon_url.sql", include_str!(r#"./migrations/sqlite/20240810214105_mint_icon_url.sql"#)),
-    ("20240810233905_update_mint_url.sql", include_str!(r#"./migrations/sqlite/20240810233905_update_mint_url.sql"#)),
-    ("20240902151515_icon_url.sql", include_str!(r#"./migrations/sqlite/20240902151515_icon_url.sql"#)),
-    ("20240902210905_mint_time.sql", include_str!(r#"./migrations/sqlite/20240902210905_mint_time.sql"#)),
-    ("20241011125207_mint_urls.sql", include_str!(r#"./migrations/sqlite/20241011125207_mint_urls.sql"#)),
-    ("20241108092756_wallet_mint_quote_secretkey.sql", include_str!(r#"./migrations/sqlite/20241108092756_wallet_mint_quote_secretkey.sql"#)),
-    ("20250214135017_mint_tos.sql", include_str!(r#"./migrations/sqlite/20250214135017_mint_tos.sql"#)),
-    ("20250310111513_drop_nostr_last_checked.sql", include_str!(r#"./migrations/sqlite/20250310111513_drop_nostr_last_checked.sql"#)),
-    ("20250314082116_allow_pending_spent.sql", include_str!(r#"./migrations/sqlite/20250314082116_allow_pending_spent.sql"#)),
-    ("20250323152040_wallet_dleq_proofs.sql", include_str!(r#"./migrations/sqlite/20250323152040_wallet_dleq_proofs.sql"#)),
-    ("20250401120000_add_transactions_table.sql", include_str!(r#"./migrations/sqlite/20250401120000_add_transactions_table.sql"#)),
-    ("20250616144830_add_keyset_expiry.sql", include_str!(r#"./migrations/sqlite/20250616144830_add_keyset_expiry.sql"#)),
+    ("sqlite/20240612132920_init.sql", include_str!(r#"./migrations/sqlite/20240612132920_init.sql"#)),
+    ("sqlite/20240618200350_quote_state.sql", include_str!(r#"./migrations/sqlite/20240618200350_quote_state.sql"#)),
+    ("sqlite/20240626091921_nut04_state.sql", include_str!(r#"./migrations/sqlite/20240626091921_nut04_state.sql"#)),
+    ("sqlite/20240710144711_input_fee.sql", include_str!(r#"./migrations/sqlite/20240710144711_input_fee.sql"#)),
+    ("sqlite/20240810214105_mint_icon_url.sql", include_str!(r#"./migrations/sqlite/20240810214105_mint_icon_url.sql"#)),
+    ("sqlite/20240810233905_update_mint_url.sql", include_str!(r#"./migrations/sqlite/20240810233905_update_mint_url.sql"#)),
+    ("sqlite/20240902151515_icon_url.sql", include_str!(r#"./migrations/sqlite/20240902151515_icon_url.sql"#)),
+    ("sqlite/20240902210905_mint_time.sql", include_str!(r#"./migrations/sqlite/20240902210905_mint_time.sql"#)),
+    ("sqlite/20241011125207_mint_urls.sql", include_str!(r#"./migrations/sqlite/20241011125207_mint_urls.sql"#)),
+    ("sqlite/20241108092756_wallet_mint_quote_secretkey.sql", include_str!(r#"./migrations/sqlite/20241108092756_wallet_mint_quote_secretkey.sql"#)),
+    ("sqlite/20250214135017_mint_tos.sql", include_str!(r#"./migrations/sqlite/20250214135017_mint_tos.sql"#)),
+    ("sqlite/20250310111513_drop_nostr_last_checked.sql", include_str!(r#"./migrations/sqlite/20250310111513_drop_nostr_last_checked.sql"#)),
+    ("sqlite/20250314082116_allow_pending_spent.sql", include_str!(r#"./migrations/sqlite/20250314082116_allow_pending_spent.sql"#)),
+    ("sqlite/20250323152040_wallet_dleq_proofs.sql", include_str!(r#"./migrations/sqlite/20250323152040_wallet_dleq_proofs.sql"#)),
+    ("sqlite/20250401120000_add_transactions_table.sql", include_str!(r#"./migrations/sqlite/20250401120000_add_transactions_table.sql"#)),
+    ("sqlite/20250616144830_add_keyset_expiry.sql", include_str!(r#"./migrations/sqlite/20250616144830_add_keyset_expiry.sql"#)),
 ];

crates/cdk-sql-base/src/wallet/mod.rs

@@ -36,14 +36,14 @@ where
     db: T,
 }
 
-impl<T> SQLWalletDatabase<T>
+impl<DB> SQLWalletDatabase<DB>
 where
-    T: DatabaseExecutor,
+    DB: DatabaseExecutor,
 {
     /// Creates a new instance
     pub async fn new<X>(db: X) -> Result<Self, Error>
     where
-        X: Into<T>,
+        X: Into<DB>,
     {
         let db = db.into();
         Self::migrate(&db).await?;
@@ -51,8 +51,8 @@ where
     }
 
     /// Migrate [`WalletSqliteDatabase`]
-    async fn migrate(conn: &T) -> Result<(), Error> {
-        migrate(conn, migrations::MIGRATIONS).await?;
+    async fn migrate(conn: &DB) -> Result<(), Error> {
+        migrate(conn, DB::name(), migrations::MIGRATIONS).await?;
         Ok(())
     }
 }

crates/cdk-sql-base/tests/legacy-sqlx.sql

@@ -5,7 +5,7 @@ CREATE TABLE _sqlx_migrations (
     description TEXT NOT NULL,
     installed_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
     success BOOLEAN NOT NULL,
-    checksum BLOB NOT NULL,
+    checksum BYTEA NOT NULL,
     execution_time BIGINT NOT NULL
 );
 INSERT INTO _sqlx_migrations VALUES(20240612124932,'init','2025-06-13 20:01:04',1,X'42664ceda25b07bca420c2f7480c90334cb8a720203c1b4b8971181d5d3afabda3171aa89c1c0c8a26421eded94b77fa',921834);
@@ -66,21 +66,21 @@ CREATE TABLE config (
     value TEXT NOT NULL
 );
 CREATE TABLE IF NOT EXISTS "proof" (
-    y BLOB PRIMARY KEY,
+    y BYTEA PRIMARY KEY,
     amount INTEGER NOT NULL,
     keyset_id TEXT NOT NULL, -- no FK constraint here
     secret TEXT NOT NULL,
-    c BLOB NOT NULL,
+    c BYTEA NOT NULL,
     witness TEXT,
     state TEXT CHECK (state IN ('SPENT', 'PENDING', 'UNSPENT', 'RESERVED', 'UNKNOWN')) NOT NULL,
     quote_id TEXT,
     created_time INTEGER NOT NULL DEFAULT 0
 );
 CREATE TABLE IF NOT EXISTS "blind_signature" (
-    y BLOB PRIMARY KEY,
+    y BYTEA PRIMARY KEY,
     amount INTEGER NOT NULL,
     keyset_id TEXT NOT NULL,  -- FK removed
-    c BLOB NOT NULL,
+    c BYTEA NOT NULL,
     dleq_e TEXT,
     dleq_s TEXT,
     quote_id TEXT,

crates/cdk-sqlite/Cargo.toml

@@ -25,7 +25,7 @@ bitcoin.workspace = true
 cdk-sql-base = { workspace = true }
 rusqlite = { version = "0.31", features = ["bundled"]}
 thiserror.workspace = true
-tokio.workspace = true
+tokio = { workspace = true, features = ["rt-multi-thread"]}
 tracing.workspace = true
 serde.workspace = true
 serde_json.workspace = true

crates/cdk-sqlite/src/common.rs

@@ -7,7 +7,7 @@ use cdk_sql_base::value::Value;
 use rusqlite::Connection;
 
 /// The config need to create a new SQLite connection
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct Config {
     path: Option<String>,
     password: Option<String>,
@@ -27,6 +27,7 @@ impl ResourceManager for SqliteConnectionManager {
     fn new_resource(
         config: &Self::Config,
         _still_valid: Arc<AtomicBool>,
+        _timeout: Duration,
     ) -> Result<Self::Resource, pool::Error<Self::Error>> {
         let conn = if let Some(path) = config.path.as_ref() {
             Connection::open(path)?

crates/cdk-sqlite/src/mint/async_rusqlite.rs

@@ -1,4 +1,4 @@
-//! Async and concurrent rusqlite
+//! Async, pipelined rusqlite client
 use std::marker::PhantomData;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicUsize, Ordering};
@@ -127,6 +127,7 @@ fn process_query(conn: &Connection, statement: InnerStatement) -> Result<DbRespo
     let start = Instant::now();
     let expected_response = statement.expected_response;
     let (sql, placeholder_values) = statement.to_sql()?;
+    let sql = sql.trim_end_matches("FOR UPDATE");
 
     let mut stmt = conn.prepare_cached(&sql)?;
     for (i, value) in placeholder_values.into_iter().enumerate() {
@@ -470,6 +471,10 @@ impl DatabaseConnector for AsyncRusqlite {
 
 #[async_trait::async_trait]
 impl DatabaseExecutor for AsyncRusqlite {
+    fn name() -> &'static str {
+        "sqlite"
+    }
+
     async fn fetch_one(&self, mut statement: InnerStatement) -> Result<Option<Vec<Column>>, Error> {
         let (sender, receiver) = oneshot::channel();
         statement.expected_response = ExpectedSqlResponse::SingleRow;
@@ -620,6 +625,10 @@ impl<'a> DatabaseTransaction<'a> for Transaction<'a> {
 
 #[async_trait::async_trait]
 impl DatabaseExecutor for Transaction<'_> {
+    fn name() -> &'static str {
+        "sqlite"
+    }
+
     async fn fetch_one(&self, mut statement: InnerStatement) -> Result<Option<Vec<Column>>, Error> {
         let (sender, receiver) = oneshot::channel();
         statement.expected_response = ExpectedSqlResponse::SingleRow;

crates/cdk-sqlite/src/mint/mod.rs

@@ -16,13 +16,49 @@ pub type MintSqliteAuthDatabase = SQLMintAuthDatabase<async_rusqlite::AsyncRusql
 
 #[cfg(test)]
 mod test {
+    use std::fs::remove_file;
+
     use cdk_common::mint_db_test;
+    use cdk_sql_base::stmt::query;
 
     use super::*;
+    use crate::mint::async_rusqlite::AsyncRusqlite;
 
     async fn provide_db() -> MintSqliteDatabase {
         memory::empty().await.unwrap()
     }
 
     mint_db_test!(provide_db);
+
+    #[tokio::test]
+    async fn open_legacy_and_migrate() {
+        let file = format!(
+            "{}/db.sqlite",
+            std::env::temp_dir().to_str().unwrap_or_default()
+        );
+
+        {
+            let _ = remove_file(&file);
+            #[cfg(not(feature = "sqlcipher"))]
+            let conn: AsyncRusqlite = file.as_str().into();
+            #[cfg(feature = "sqlcipher")]
+            let conn: AsyncRusqlite = (file.as_str(), "test".to_owned()).into();
+
+            query(include_str!("../../tests/legacy-sqlx.sql"))
+                .expect("query")
+                .execute(&conn)
+                .await
+                .expect("create former db failed");
+        }
+
+        #[cfg(not(feature = "sqlcipher"))]
+        let conn = MintSqliteDatabase::new(file.as_str()).await;
+
+        #[cfg(feature = "sqlcipher")]
+        let conn = MintSqliteDatabase::new((file.as_str(), "test".to_owned())).await;
+
+        assert!(conn.is_ok(), "Failed with {:?}", conn.unwrap_err());
+
+        let _ = remove_file(&file);
+    }
 }

crates/cdk-sqlite/src/wallet/mod.rs

@@ -41,6 +41,10 @@ impl SimpleAsyncRusqlite {
 
 #[async_trait::async_trait]
 impl DatabaseExecutor for SimpleAsyncRusqlite {
+    fn name() -> &'static str {
+        "sqlite"
+    }
+
     async fn execute(&self, statement: Statement) -> Result<usize, Error> {
         let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut stmt = self

crates/cdk-sqlite/tests/legacy-sqlx.sql

@@ -70,7 +70,7 @@ CREATE TABLE IF NOT EXISTS "proof" (
     amount INTEGER NOT NULL,
     keyset_id TEXT NOT NULL, -- no FK constraint here
     secret TEXT NOT NULL,
-    c BLOB NOT NULL,
+    c BLOBNOT NULL,
     witness TEXT,
     state TEXT CHECK (state IN ('SPENT', 'PENDING', 'UNSPENT', 'RESERVED', 'UNKNOWN')) NOT NULL,
     quote_id TEXT,