Working on a better database abstraction

After [this question in the chat](https://matrix.to/#/!oJFtttFHGfnTGrIjvD:matrix.cashu.space/$oJFtttFHGfnTGrIjvD:matrix.cashu.space/$I5ZtjJtBM0ctltThDYpoCwClZFlM6PHzf8q2Rjqmso8)
regarding a database transaction within the same function, I realized a few
design flaws in our SQL database abstraction, particularly regarding
transactions.

1. Our upper abstraction got it right, where a transaction is bound with `&mut
   self`, so Rust knows how to handle its lifetime with' async/await'.
2. The raw database does not; instead, it returns &self, and beginning a
   transaction takes &self as well, which is problematic for Rust, but that's not
   all. It is fundamentally wrong. A transaction should take &mut self when
   beginning a transaction, as that connection is bound to a transaction and
   should not be returned to the pool. Currently, that responsibility lies with
   the implementor. If a mistake is made, a transaction could be executed in two
   or more connections.
3. The way a database is bound to our store layer is through a single struct,
   which may or may not internally utilize our connection pool. This is also
   another design flow, in this PR, a connection pool is owned, and to use a
   connection, it should be requested, and that connection is reference with
   mutable when beginning a transaction

crates/cdk-sql-common/Cargo.toml

@@ -23,6 +23,7 @@ cdk-common = { workspace = true, features = ["test"] }
 bitcoin.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
+tokio.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 lightning-invoice.workspace = true

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

@@ -3,11 +3,14 @@ use crate::stmt::query;
 
 /// Migrates the migration generated by `build.rs`
 #[inline(always)]
-pub async fn migrate<C: DatabaseExecutor>(
+pub async fn migrate<C>(
     conn: &C,
     db_prefix: &str,
     migrations: &[(&str, &str, &str)],
-) -> Result<(), cdk_common::database::Error> {
+) -> Result<(), cdk_common::database::Error>
+where
+    C: DatabaseExecutor,
+{
     query(
         r#"
            CREATE TABLE IF NOT EXISTS migrations (

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

@@ -1,10 +1,12 @@
 //! Database traits definition
 
 use std::fmt::Debug;
+use std::marker::PhantomData;
+use std::ops::{Deref, DerefMut};
 
 use cdk_common::database::Error;
 
-use crate::stmt::{Column, Statement};
+use crate::stmt::{query, Column, Statement};
 
 /// Database Executor
 ///
@@ -32,22 +34,170 @@ pub trait DatabaseExecutor: Debug + Sync + Send {
 
 /// Database transaction trait
 #[async_trait::async_trait]
-pub trait DatabaseTransaction<'a>: Debug + DatabaseExecutor + Send + Sync {
+pub trait DatabaseTransaction<DB>
+where
+    DB: DatabaseExecutor,
+{
     /// Consumes the current transaction committing the changes
-    async fn commit(self) -> Result<(), Error>;
+    async fn commit(conn: &mut DB) -> Result<(), Error>;
+
+    /// Begin a transaction
+    async fn begin(conn: &mut DB) -> Result<(), Error>;
 
     /// Consumes the transaction rolling back all changes
-    async fn rollback(self) -> Result<(), Error>;
+    async fn rollback(conn: &mut DB) -> Result<(), Error>;
+}
+
+/// Database connection with a transaction
+#[derive(Debug)]
+pub struct ConnectionWithTransaction<DB, W>
+where
+    DB: DatabaseConnector + 'static,
+    W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
+{
+    inner: Option<W>,
+}
+
+impl<DB, W> ConnectionWithTransaction<DB, W>
+where
+    DB: DatabaseConnector,
+    W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
+{
+    /// Creates a new transaction
+    pub async fn new(mut inner: W) -> Result<Self, Error> {
+        DB::Transaction::begin(inner.deref_mut()).await?;
+        Ok(Self { inner: Some(inner) })
+    }
+
+    /// Commits the transaction consuming it and releasing the connection back to the pool (or
+    /// disconnecting)
+    pub async fn commit(mut self) -> Result<(), Error> {
+        let mut conn = self
+            .inner
+            .take()
+            .ok_or(Error::Internal("Missing connection".to_owned()))?;
+
+        DB::Transaction::commit(&mut conn).await?;
+
+        Ok(())
+    }
+
+    /// Rollback the transaction consuming it and releasing the connection back to the pool (or
+    /// disconnecting)
+    pub async fn rollback(mut self) -> Result<(), Error> {
+        let mut conn = self
+            .inner
+            .take()
+            .ok_or(Error::Internal("Missing connection".to_owned()))?;
+
+        DB::Transaction::rollback(&mut conn).await?;
+
+        Ok(())
+    }
+}
+
+impl<DB, W> Drop for ConnectionWithTransaction<DB, W>
+where
+    DB: DatabaseConnector,
+    W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
+{
+    fn drop(&mut self) {
+        if let Some(mut conn) = self.inner.take() {
+            tokio::spawn(async move {
+                let _ = DB::Transaction::rollback(conn.deref_mut()).await;
+            });
+        }
+    }
+}
+
+#[async_trait::async_trait]
+impl<DB, W> DatabaseExecutor for ConnectionWithTransaction<DB, W>
+where
+    DB: DatabaseConnector,
+    W: Debug + Deref<Target = DB> + DerefMut<Target = DB> + Send + Sync + 'static,
+{
+    fn name() -> &'static str {
+        "Transaction"
+    }
+
+    /// Executes a query and returns the affected rows
+    async fn execute(&self, statement: Statement) -> Result<usize, Error> {
+        self.inner
+            .as_ref()
+            .ok_or(Error::Internal("Missing internal connection".to_owned()))?
+            .execute(statement)
+            .await
+    }
+
+    /// Runs the query and returns the first row or None
+    async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
+        self.inner
+            .as_ref()
+            .ok_or(Error::Internal("Missing internal connection".to_owned()))?
+            .fetch_one(statement)
+            .await
+    }
+
+    /// Runs the query and returns the first row or None
+    async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
+        self.inner
+            .as_ref()
+            .ok_or(Error::Internal("Missing internal connection".to_owned()))?
+            .fetch_all(statement)
+            .await
+    }
+
+    /// Fetches the first row and column from a query
+    async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
+        self.inner
+            .as_ref()
+            .ok_or(Error::Internal("Missing internal connection".to_owned()))?
+            .pluck(statement)
+            .await
+    }
+
+    /// Batch execution
+    async fn batch(&self, statement: Statement) -> Result<(), Error> {
+        self.inner
+            .as_ref()
+            .ok_or(Error::Internal("Missing internal connection".to_owned()))?
+            .batch(statement)
+            .await
+    }
+}
+
+/// Generic transaction handler for SQLite
+pub struct GenericTransactionHandler<W>(PhantomData<W>);
+
+#[async_trait::async_trait]
+impl<W> DatabaseTransaction<W> for GenericTransactionHandler<W>
+where
+    W: DatabaseExecutor,
+{
+    /// Consumes the current transaction committing the changes
+    async fn commit(conn: &mut W) -> Result<(), Error> {
+        query("COMMIT")?.execute(conn).await?;
+        Ok(())
+    }
+
+    /// Begin a transaction
+    async fn begin(conn: &mut W) -> Result<(), Error> {
+        query("BEGIN")?.execute(conn).await?;
+        Ok(())
+    }
+
+    /// Consumes the transaction rolling back all changes
+    async fn rollback(conn: &mut W) -> Result<(), Error> {
+        query("ROLLBACK")?.execute(conn).await?;
+        Ok(())
+    }
 }
 
 /// Database connector
 #[async_trait::async_trait]
 pub trait DatabaseConnector: Debug + DatabaseExecutor + Send + Sync {
-    /// Transaction type for this database connection
-    type Transaction<'a>: DatabaseTransaction<'a>
+    /// Database static trait for the database
+    type Transaction: DatabaseTransaction<Self>
     where
-        Self: 'a;
-
-    /// Begin a new transaction
-    async fn begin(&self) -> Result<Self::Transaction<'_>, Error>;
+        Self: Sized;
 }

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

@@ -1,8 +1,9 @@
 //! SQL Mint Auth
 
 use std::collections::HashMap;
-use std::marker::PhantomData;
+use std::fmt::Debug;
 use std::str::FromStr;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use cdk_common::database::{self, MintAuthDatabase, MintAuthTransaction};
@@ -15,36 +16,43 @@ use tracing::instrument;
 use super::{sql_row_to_blind_signature, sql_row_to_keyset_info, SQLTransaction};
 use crate::column_as_string;
 use crate::common::migrate;
-use crate::database::{DatabaseConnector, DatabaseTransaction};
+use crate::database::{ConnectionWithTransaction, DatabaseConnector};
 use crate::mint::Error;
+use crate::pool::{Pool, PooledResource, ResourceManager};
 use crate::stmt::query;
 
 /// Mint SQL Database
 #[derive(Debug, Clone)]
-pub struct SQLMintAuthDatabase<DB>
+pub struct SQLMintAuthDatabase<DB, RM, C, E>
 where
     DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
-    db: DB,
+    pool: Arc<Pool<RM>>,
 }
 
-impl<DB> SQLMintAuthDatabase<DB>
+impl<DB, RM, C, E> SQLMintAuthDatabase<DB, RM, C, E>
 where
-    DB: DatabaseConnector,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     /// Creates a new instance
     pub async fn new<X>(db: X) -> Result<Self, Error>
     where
-        X: Into<DB>,
+        X: Into<RM::Config>,
     {
-        let db = db.into();
-        Self::migrate(&db).await?;
-        Ok(Self { db })
+        let pool = Pool::new(db.into());
+        Self::migrate(pool.get().map_err(|e| Error::Database(Box::new(e)))?).await?;
+        Ok(Self { pool })
     }
 
     /// Migrate
-    async fn migrate(conn: &DB) -> Result<(), Error> {
-        let tx = conn.begin().await?;
+    async fn migrate(conn: PooledResource<RM>) -> Result<(), Error> {
+        let tx = ConnectionWithTransaction::new(conn).await?;
         migrate(&tx, DB::name(), MIGRATIONS).await?;
         tx.commit().await?;
         Ok(())
@@ -56,9 +64,12 @@ mod migrations;
 
 
 #[async_trait]
-impl<'a, T> MintAuthTransaction<database::Error> for SQLTransaction<'a, T>
+impl<DB, RM, C, E> MintAuthTransaction<database::Error> for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     #[instrument(skip(self))]
     async fn set_active_keyset(&mut self, id: Id) -> Result<(), database::Error> {
@@ -233,9 +244,12 @@ where
 }
 
 #[async_trait]
-impl<DB> MintAuthDatabase for SQLMintAuthDatabase<DB>
+impl<DB, RM, C, E> MintAuthDatabase for SQLMintAuthDatabase<DB, RM, C, E>
 where
-    DB: DatabaseConnector,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = database::Error;
 
@@ -244,12 +258,15 @@ where
     ) -> Result<Box<dyn MintAuthTransaction<database::Error> + Send + Sync + 'a>, database::Error>
     {
         Ok(Box::new(SQLTransaction {
-            inner: self.db.begin().await?,
-            _phantom: PhantomData,
+            inner: ConnectionWithTransaction::new(
+                self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
+            )
+            .await?,
         }))
     }
 
     async fn get_active_keyset_id(&self) -> Result<Option<Id>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -260,13 +277,14 @@ where
                 active = 1;
             "#,
         )?
-        .pluck(&self.db)
+        .pluck(&*conn)
         .await?
         .map(|id| Ok::<_, Error>(column_as_string!(id, Id::from_str, Id::from_bytes)))
         .transpose()?)
     }
 
     async fn get_keyset_info(&self, id: &Id) -> Result<Option<MintKeySetInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"SELECT
                 id,
@@ -283,13 +301,14 @@ where
                 WHERE id=:id"#,
         )?
         .bind("id", id.to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_keyset_info)
         .transpose()?)
     }
 
     async fn get_keyset_infos(&self) -> Result<Vec<MintKeySetInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"SELECT
                 id,
@@ -305,7 +324,7 @@ where
                 keyset
                 WHERE id=:id"#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_keyset_info)
@@ -313,9 +332,10 @@ where
     }
 
     async fn get_proofs_states(&self, ys: &[PublicKey]) -> Result<Vec<Option<State>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut current_states = query(r#"SELECT y, state FROM proof WHERE y IN (:ys)"#)?
             .bind_vec("ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
-            .fetch_all(&self.db)
+            .fetch_all(&*conn)
             .await?
             .into_iter()
             .map(|row| {
@@ -333,6 +353,7 @@ where
         &self,
         blinded_messages: &[PublicKey],
     ) -> Result<Vec<Option<BlindSignature>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut blinded_signatures = query(
             r#"SELECT
                 keyset_id,
@@ -353,7 +374,7 @@ where
                 .map(|y| y.to_bytes().to_vec())
                 .collect(),
         )
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(|mut row| {
@@ -377,10 +398,11 @@ where
         &self,
         protected_endpoint: ProtectedEndpoint,
     ) -> Result<Option<AuthRequired>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(
             query(r#"SELECT auth FROM protected_endpoints WHERE endpoint = :endpoint"#)?
                 .bind("endpoint", serde_json::to_string(&protected_endpoint)?)
-                .pluck(&self.db)
+                .pluck(&*conn)
                 .await?
                 .map(|auth| {
                     Ok::<_, Error>(column_as_string!(
@@ -396,8 +418,9 @@ where
     async fn get_auth_for_endpoints(
         &self,
     ) -> Result<HashMap<ProtectedEndpoint, Option<AuthRequired>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(r#"SELECT endpoint, auth FROM protected_endpoints"#)?
-            .fetch_all(&self.db)
+            .fetch_all(&*conn)
             .await?
             .into_iter()
             .map(|row| {

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

@@ -9,8 +9,9 @@
 //! clients in a pool and expose them to an asynchronous environment, making them compatible with
 //! Mint.
 use std::collections::HashMap;
-use std::marker::PhantomData;
+use std::fmt::Debug;
 use std::str::FromStr;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use bitcoin::bip32::DerivationPath;
@@ -38,7 +39,8 @@ use tracing::instrument;
 use uuid::Uuid;
 
 use crate::common::migrate;
-use crate::database::{DatabaseConnector, DatabaseExecutor, DatabaseTransaction};
+use crate::database::{ConnectionWithTransaction, DatabaseConnector, DatabaseExecutor};
+use crate::pool::{Pool, PooledResource, ResourceManager};
 use crate::stmt::{query, Column};
 use crate::{
     column_as_nullable_number, column_as_nullable_string, column_as_number, column_as_string,
@@ -57,20 +59,25 @@ pub use auth::SQLMintAuthDatabase;
 
 /// Mint SQL Database
 #[derive(Debug, Clone)]
-pub struct SQLMintDatabase<DB>
+pub struct SQLMintDatabase<DB, RM, C, E>
 where
     DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
-    db: DB,
+    pool: Arc<Pool<RM>>,
 }
 
 /// SQL Transaction Writer
-pub struct SQLTransaction<'a, T>
+pub struct SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
-    inner: T,
-    _phantom: PhantomData<&'a ()>,
+    inner: ConnectionWithTransaction<DB, PooledResource<RM>>,
 }
 
 #[inline(always)]
@@ -115,23 +122,28 @@ where
     Ok(())
 }
 
-impl<DB> SQLMintDatabase<DB>
+impl<DB, RM, C, E> SQLMintDatabase<DB, RM, C, E>
 where
-    DB: DatabaseConnector,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     /// Creates a new instance
     pub async fn new<X>(db: X) -> Result<Self, Error>
     where
-        X: Into<DB>,
+        X: Into<RM::Config>,
     {
-        let db = db.into();
-        Self::migrate(&db).await?;
-        Ok(Self { db })
+        let pool = Pool::new(db.into());
+
+        Self::migrate(pool.get().map_err(|e| Error::Database(Box::new(e)))?).await?;
+
+        Ok(Self { pool })
     }
 
     /// Migrate
-    async fn migrate(conn: &DB) -> Result<(), Error> {
-        let tx = conn.begin().await?;
+    async fn migrate(conn: PooledResource<RM>) -> Result<(), Error> {
+        let tx = ConnectionWithTransaction::new(conn).await?;
         migrate(&tx, DB::name(), MIGRATIONS).await?;
         tx.commit().await?;
         Ok(())
@@ -142,9 +154,10 @@ where
     where
         R: serde::de::DeserializeOwned,
     {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let value = column_as_string!(query(r#"SELECT value FROM config WHERE id = :id LIMIT 1"#)?
             .bind("id", id.to_owned())
-            .pluck(&self.db)
+            .pluck(&*conn)
             .await?
             .ok_or(Error::UnknownQuoteTTL)?);
 
@@ -153,9 +166,12 @@ where
 }
 
 #[async_trait]
-impl<'a, T> database::MintProofsTransaction<'a> for SQLTransaction<'a, T>
+impl<DB, RM, C, E> database::MintProofsTransaction<'_> for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
@@ -267,9 +283,12 @@ where
 }
 
 #[async_trait]
-impl<'a, T> database::MintTransaction<'a, Error> for SQLTransaction<'a, T>
+impl<DB, RM, C, E> database::MintTransaction<'_, Error> for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     async fn set_mint_info(&mut self, mint_info: MintInfo) -> Result<(), Error> {
         Ok(set_to_config(&self.inner, "mint_info", &mint_info).await?)
@@ -281,18 +300,21 @@ where
 }
 
 #[async_trait]
-impl<'a, T> MintDbWriterFinalizer for SQLTransaction<'a, T>
+impl<DB, RM, C, E> MintDbWriterFinalizer for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
     async fn commit(self: Box<Self>) -> Result<(), Error> {
-        Ok(self.inner.commit().await?)
+        self.inner.commit().await
     }
 
     async fn rollback(self: Box<Self>) -> Result<(), Error> {
-        Ok(self.inner.rollback().await?)
+        self.inner.rollback().await
     }
 }
 
@@ -357,9 +379,12 @@ WHERE quote_id=:quote_id
 }
 
 #[async_trait]
-impl<'a, T> MintKeyDatabaseTransaction<'a, Error> for SQLTransaction<'a, T>
+impl<DB, RM, C, E> MintKeyDatabaseTransaction<'_, Error> for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     async fn add_keyset_info(&mut self, keyset: MintKeySetInfo) -> Result<(), Error> {
         query(
@@ -416,9 +441,12 @@ where
 }
 
 #[async_trait]
-impl<DB> MintKeysDatabase for SQLMintDatabase<DB>
+impl<DB, RM, C, E> MintKeysDatabase for SQLMintDatabase<DB, RM, C, E>
 where
-    DB: DatabaseConnector,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
@@ -426,16 +454,19 @@ where
         &'a self,
     ) -> Result<Box<dyn MintKeyDatabaseTransaction<'a, Error> + Send + Sync + 'a>, Error> {
         Ok(Box::new(SQLTransaction {
-            inner: self.db.begin().await?,
-            _phantom: PhantomData,
+            inner: ConnectionWithTransaction::new(
+                self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
+            )
+            .await?,
         }))
     }
 
     async fn get_active_keyset_id(&self, unit: &CurrencyUnit) -> Result<Option<Id>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(
             query(r#" SELECT id FROM keyset WHERE active = 1 AND unit IS :unit"#)?
                 .bind("unit", unit.to_string())
-                .pluck(&self.db)
+                .pluck(&*conn)
                 .await?
                 .map(|id| match id {
                     Column::Text(text) => Ok(Id::from_str(&text)?),
@@ -447,8 +478,9 @@ where
     }
 
     async fn get_active_keysets(&self) -> Result<HashMap<CurrencyUnit, Id>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(r#"SELECT id, unit FROM keyset WHERE active = 1"#)?
-            .fetch_all(&self.db)
+            .fetch_all(&*conn)
             .await?
             .into_iter()
             .map(|row| {
@@ -461,6 +493,7 @@ where
     }
 
     async fn get_keyset_info(&self, id: &Id) -> Result<Option<MintKeySetInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"SELECT
                 id,
@@ -477,13 +510,14 @@ where
                 WHERE id=:id"#,
         )?
         .bind("id", id.to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_keyset_info)
         .transpose()?)
     }
 
     async fn get_keyset_infos(&self) -> Result<Vec<MintKeySetInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"SELECT
                 id,
@@ -499,7 +533,7 @@ where
                 keyset
             "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_keyset_info)
@@ -508,9 +542,12 @@ where
 }
 
 #[async_trait]
-impl<'a, T> MintQuotesTransaction<'a> for SQLTransaction<'a, T>
+impl<DB, RM, C, E> MintQuotesTransaction<'_> for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
@@ -1028,15 +1065,20 @@ VALUES (:quote_id, :amount, :timestamp);
 }
 
 #[async_trait]
-impl<DB> MintQuotesDatabase for SQLMintDatabase<DB>
+impl<DB, RM, C, E> MintQuotesDatabase for SQLMintDatabase<DB, RM, C, E>
 where
     DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
     async fn get_mint_quote(&self, quote_id: &Uuid) -> Result<Option<MintQuote>, Self::Err> {
-        let payments = get_mint_quote_payments(&self.db, quote_id).await?;
-        let issuance = get_mint_quote_issuance(&self.db, quote_id).await?;
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
+        let payments = get_mint_quote_payments(&*conn, quote_id).await?;
+        let issuance = get_mint_quote_issuance(&*conn, quote_id).await?;
 
         Ok(query(
             r#"
@@ -1058,7 +1100,7 @@ where
             WHERE id = :id"#,
         )?
         .bind("id", quote_id.as_hyphenated().to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(|row| sql_row_to_mint_quote(row, payments, issuance))
         .transpose()?)
@@ -1068,6 +1110,7 @@ where
         &self,
         request: &str,
     ) -> Result<Option<MintQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut mint_quote = query(
             r#"
             SELECT
@@ -1088,14 +1131,14 @@ where
             WHERE request = :request"#,
         )?
         .bind("request", request.to_owned())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(|row| sql_row_to_mint_quote(row, vec![], vec![]))
         .transpose()?;
 
         if let Some(quote) = mint_quote.as_mut() {
-            let payments = get_mint_quote_payments(&self.db, &quote.id).await?;
-            let issuance = get_mint_quote_issuance(&self.db, &quote.id).await?;
+            let payments = get_mint_quote_payments(&*conn, &quote.id).await?;
+            let issuance = get_mint_quote_issuance(&*conn, &quote.id).await?;
             quote.issuance = issuance;
             quote.payments = payments;
         }
@@ -1107,6 +1150,7 @@ where
         &self,
         request_lookup_id: &PaymentIdentifier,
     ) -> Result<Option<MintQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut mint_quote = query(
             r#"
             SELECT
@@ -1130,15 +1174,15 @@ where
         )?
         .bind("request_lookup_id", request_lookup_id.to_string())
         .bind("request_lookup_id_kind", request_lookup_id.kind())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(|row| sql_row_to_mint_quote(row, vec![], vec![]))
         .transpose()?;
 
         // TODO: these should use an sql join so they can be done in one query
         if let Some(quote) = mint_quote.as_mut() {
-            let payments = get_mint_quote_payments(&self.db, &quote.id).await?;
-            let issuance = get_mint_quote_issuance(&self.db, &quote.id).await?;
+            let payments = get_mint_quote_payments(&*conn, &quote.id).await?;
+            let issuance = get_mint_quote_issuance(&*conn, &quote.id).await?;
             quote.issuance = issuance;
             quote.payments = payments;
         }
@@ -1147,6 +1191,7 @@ where
     }
 
     async fn get_mint_quotes(&self) -> Result<Vec<MintQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut mint_quotes = query(
             r#"
             SELECT
@@ -1166,15 +1211,15 @@ where
                 mint_quote
             "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(|row| sql_row_to_mint_quote(row, vec![], vec![]))
         .collect::<Result<Vec<_>, _>>()?;
 
         for quote in mint_quotes.as_mut_slice() {
-            let payments = get_mint_quote_payments(&self.db, &quote.id).await?;
-            let issuance = get_mint_quote_issuance(&self.db, &quote.id).await?;
+            let payments = get_mint_quote_payments(&*conn, &quote.id).await?;
+            let issuance = get_mint_quote_issuance(&*conn, &quote.id).await?;
             quote.issuance = issuance;
             quote.payments = payments;
         }
@@ -1183,6 +1228,7 @@ where
     }
 
     async fn get_melt_quote(&self, quote_id: &Uuid) -> Result<Option<mint::MeltQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -1207,13 +1253,14 @@ where
             "#,
         )?
         .bind("id", quote_id.as_hyphenated().to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_melt_quote)
         .transpose()?)
     }
 
     async fn get_melt_quotes(&self) -> Result<Vec<mint::MeltQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -1235,7 +1282,7 @@ where
                 melt_quote
             "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_melt_quote)
@@ -1244,13 +1291,17 @@ where
 }
 
 #[async_trait]
-impl<DB> MintProofsDatabase for SQLMintDatabase<DB>
+impl<DB, RM, C, E> MintProofsDatabase for SQLMintDatabase<DB, RM, C, E>
 where
     DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
     async fn get_proofs_by_ys(&self, ys: &[PublicKey]) -> Result<Vec<Option<Proof>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut proofs = query(
             r#"
             SELECT
@@ -1267,7 +1318,7 @@ where
             "#,
         )?
         .bind_vec("ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(|mut row| {
@@ -1286,6 +1337,7 @@ where
     }
 
     async fn get_proof_ys_by_quote_id(&self, quote_id: &Uuid) -> Result<Vec<PublicKey>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -1301,7 +1353,7 @@ where
             "#,
         )?
         .bind("quote_id", quote_id.as_hyphenated().to_string())
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_proof)
@@ -1310,7 +1362,8 @@ where
     }
 
     async fn get_proofs_states(&self, ys: &[PublicKey]) -> Result<Vec<Option<State>>, Self::Err> {
-        let mut current_states = get_current_states(&self.db, ys).await?;
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        let mut current_states = get_current_states(&*conn, ys).await?;
 
         Ok(ys.iter().map(|y| current_states.remove(y)).collect())
     }
@@ -1319,6 +1372,7 @@ where
         &self,
         keyset_id: &Id,
     ) -> Result<(Proofs, Vec<Option<State>>), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -1335,7 +1389,7 @@ where
             "#,
         )?
         .bind("keyset_id", keyset_id.to_string())
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_proof_with_state)
@@ -1346,9 +1400,12 @@ where
 }
 
 #[async_trait]
-impl<'a, T> MintSignatureTransaction<'a> for SQLTransaction<'a, T>
+impl<DB, RM, C, E> MintSignatureTransaction<'_> for SQLTransaction<DB, RM, C, E>
 where
-    T: DatabaseTransaction<'a>,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
@@ -1436,9 +1493,12 @@ where
 }
 
 #[async_trait]
-impl<DB> MintSignaturesDatabase for SQLMintDatabase<DB>
+impl<DB, RM, C, E> MintSignaturesDatabase for SQLMintDatabase<DB, RM, C, E>
 where
     DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = Error;
 
@@ -1446,6 +1506,7 @@ where
         &self,
         blinded_messages: &[PublicKey],
     ) -> Result<Vec<Option<BlindSignature>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mut blinded_signatures = query(
             r#"SELECT
                 keyset_id,
@@ -1466,7 +1527,7 @@ where
                 .map(|b_| b_.to_bytes().to_vec())
                 .collect(),
         )
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(|mut row| {
@@ -1490,6 +1551,7 @@ where
         &self,
         keyset_id: &Id,
     ) -> Result<Vec<BlindSignature>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -1505,7 +1567,7 @@ where
             "#,
         )?
         .bind("keyset_id", keyset_id.to_string())
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_blind_signature)
@@ -1517,6 +1579,7 @@ where
         &self,
         quote_id: &Uuid,
     ) -> Result<Vec<BlindSignature>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -1532,7 +1595,7 @@ where
             "#,
         )?
         .bind("quote_id", quote_id.to_string())
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_blind_signature)
@@ -1541,16 +1604,21 @@ where
 }
 
 #[async_trait]
-impl<DB> MintDatabase<Error> for SQLMintDatabase<DB>
+impl<DB, RM, C, E> MintDatabase<Error> for SQLMintDatabase<DB, RM, C, E>
 where
-    DB: DatabaseConnector,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     async fn begin_transaction<'a>(
         &'a self,
     ) -> Result<Box<dyn database::MintTransaction<'a, Error> + Send + Sync + 'a>, Error> {
         Ok(Box::new(SQLTransaction {
-            inner: self.db.begin().await?,
-            _phantom: PhantomData,
+            inner: ConnectionWithTransaction::new(
+                self.pool.get().map_err(|e| Error::Database(Box::new(e)))?,
+            )
+            .await?,
         }))
     }
 

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

@@ -10,7 +10,10 @@ use std::time::Duration;
 
 /// Pool error
 #[derive(thiserror::Error, Debug)]
-pub enum Error<E> {
+pub enum Error<E>
+where
+    E: std::error::Error + Send + Sync + 'static,
+{
     /// Mutex Poison Error
     #[error("Internal: PoisonError")]
     Poison,
@@ -24,16 +27,25 @@ pub enum Error<E> {
     Resource(#[from] E),
 }
 
+/// Configuration
+pub trait Config: Clone + Debug + Send + Sync {
+    /// Max resource sizes
+    fn max_size(&self) -> usize;
+
+    /// Default timeout
+    fn default_timeout(&self) -> Duration;
+}
+
 /// Trait to manage resources
 pub trait ResourceManager: Debug {
     /// The resource to be pooled
     type Resource: Debug;
 
     /// The configuration that is needed in order to create the resource
-    type Config: Clone + Debug;
+    type Config: Config;
 
     /// The error the resource may return when creating a new instance
-    type Error: Debug;
+    type Error: Debug + std::error::Error + Send + Sync + 'static;
 
     /// Creates a new resource with a given config.
     ///
@@ -72,6 +84,15 @@ where
     pool: Arc<Pool<RM>>,
 }
 
+impl<RM> Debug for PooledResource<RM>
+where
+    RM: ResourceManager,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Resource: {:?}", self.resource)
+    }
+}
+
 impl<RM> Drop for PooledResource<RM>
 where
     RM: ResourceManager,
@@ -113,14 +134,14 @@ where
     RM: ResourceManager,
 {
     /// Creates a new pool
-    pub fn new(config: RM::Config, max_size: usize, default_timeout: Duration) -> Arc<Self> {
+    pub fn new(config: RM::Config) -> Arc<Self> {
         Arc::new(Self {
+            default_timeout: config.default_timeout(),
+            max_size: config.max_size(),
             config,
             queue: Default::default(),
             in_use: Default::default(),
             waiter: Default::default(),
-            default_timeout,
-            max_size,
         })
     }
 

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

@@ -1,7 +1,9 @@
 //! SQLite Wallet Database
 
 use std::collections::HashMap;
+use std::fmt::Debug;
 use std::str::FromStr;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use cdk_common::common::ProofInfo;
@@ -17,7 +19,8 @@ use cdk_common::{
 use tracing::instrument;
 
 use crate::common::migrate;
-use crate::database::DatabaseExecutor;
+use crate::database::{ConnectionWithTransaction, DatabaseConnector, DatabaseExecutor};
+use crate::pool::{Pool, PooledResource, ResourceManager};
 use crate::stmt::{query, Column};
 use crate::{
     column_as_binary, column_as_nullable_binary, column_as_nullable_number,
@@ -29,36 +32,49 @@ mod migrations;
 
 /// Wallet SQLite Database
 #[derive(Debug, Clone)]
-pub struct SQLWalletDatabase<T>
+pub struct SQLWalletDatabase<DB, RM, C, E>
 where
-    T: DatabaseExecutor,
+    DB: DatabaseConnector,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E>,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
-    db: T,
+    pool: Arc<Pool<RM>>,
 }
 
-impl<DB> SQLWalletDatabase<DB>
+impl<DB, RM, C, E> SQLWalletDatabase<DB, RM, C, E>
 where
-    DB: DatabaseExecutor,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     /// Creates a new instance
     pub async fn new<X>(db: X) -> Result<Self, Error>
     where
-        X: Into<DB>,
+        X: Into<RM::Config>,
     {
-        let db = db.into();
-        Self::migrate(&db).await?;
-        Ok(Self { db })
+        let pool = Pool::new(db.into());
+        Self::migrate(pool.get().map_err(|e| Error::Database(Box::new(e)))?).await?;
+
+        Ok(Self { pool })
     }
 
     /// Migrate [`WalletSqliteDatabase`]
-    async fn migrate(conn: &DB) -> Result<(), Error> {
-        migrate(conn, DB::name(), migrations::MIGRATIONS).await?;
+    async fn migrate(conn: PooledResource<RM>) -> Result<(), Error> {
+        let tx = ConnectionWithTransaction::new(conn).await?;
+        migrate(&tx, DB::name(), migrations::MIGRATIONS).await?;
         // Update any existing keys with missing keyset_u32 values
-        Self::add_keyset_u32(conn).await?;
+        Self::add_keyset_u32(&tx).await?;
+        tx.commit().await?;
+
         Ok(())
     }
 
-    async fn add_keyset_u32(conn: &DB) -> Result<(), Error> {
+    async fn add_keyset_u32<T>(conn: &T) -> Result<(), Error>
+    where
+        T: DatabaseExecutor,
+    {
         // First get the keysets where keyset_u32 on key is null
         let keys_without_u32: Vec<Vec<Column>> = query(
             r#"
@@ -126,14 +142,19 @@ where
 }
 
 #[async_trait]
-impl<T> WalletDatabase for SQLWalletDatabase<T>
+impl<DB, RM, C, E> WalletDatabase for SQLWalletDatabase<DB, RM, C, E>
 where
-    T: DatabaseExecutor,
+    DB: DatabaseConnector + 'static,
+    RM: ResourceManager<Resource = DB, Config = C, Error = E> + 'static,
+    C: Debug + Clone + Send + Sync,
+    E: Debug + std::error::Error + Send + Sync + 'static,
 {
     type Err = database::Error;
 
     #[instrument(skip(self))]
     async fn get_melt_quotes(&self) -> Result<Vec<wallet::MeltQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         Ok(query(
             r#"
               SELECT
@@ -149,7 +170,7 @@ where
                   melt_quote
               "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_melt_quote)
@@ -212,6 +233,8 @@ where
             ),
         };
 
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         query(
             r#"
 INSERT INTO mint
@@ -253,7 +276,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
         .bind("motd", motd)
         .bind("mint_time", time.map(|v| v as i64))
         .bind("tos_url", tos_url)
-        .execute(&self.db)
+        .execute(&*conn)
         .await?;
 
         Ok(())
@@ -261,9 +284,11 @@ ON CONFLICT(mint_url) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn remove_mint(&self, mint_url: MintUrl) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         query(r#"DELETE FROM mint WHERE mint_url=:mint_url"#)?
             .bind("mint_url", mint_url.to_string())
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
 
         Ok(())
@@ -271,6 +296,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn get_mint(&self, mint_url: MintUrl) -> Result<Option<MintInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -292,7 +318,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
             "#,
         )?
         .bind("mint_url", mint_url.to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_mint_info)
         .transpose()?)
@@ -300,6 +326,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn get_mints(&self) -> Result<HashMap<MintUrl, Option<MintInfo>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
                 SELECT
@@ -320,7 +347,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
                     mint
                 "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(|mut row| {
@@ -340,6 +367,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
         old_mint_url: MintUrl,
         new_mint_url: MintUrl,
     ) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let tables = ["mint_quote", "proof"];
 
         for table in &tables {
@@ -352,7 +380,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
             ))?
             .bind("new_mint_url", new_mint_url.to_string())
             .bind("old_mint_url", old_mint_url.to_string())
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
         }
 
@@ -365,6 +393,8 @@ ON CONFLICT(mint_url) DO UPDATE SET
         mint_url: MintUrl,
         keysets: Vec<KeySetInfo>,
     ) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         for keyset in keysets {
             query(
                 r#"
@@ -384,7 +414,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
             .bind("input_fee_ppk", keyset.input_fee_ppk as i64)
             .bind("final_expiry", keyset.final_expiry.map(|v| v as i64))
             .bind("keyset_u32", u32::from(keyset.id))
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
         }
 
@@ -396,6 +426,8 @@ ON CONFLICT(mint_url) DO UPDATE SET
         &self,
         mint_url: MintUrl,
     ) -> Result<Option<Vec<KeySetInfo>>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         let keysets = query(
             r#"
             SELECT
@@ -410,7 +442,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
             "#,
         )?
         .bind("mint_url", mint_url.to_string())
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_keyset)
@@ -424,6 +456,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
 
     #[instrument(skip(self), fields(keyset_id = %keyset_id))]
     async fn get_keyset_by_id(&self, keyset_id: &Id) -> Result<Option<KeySetInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -438,7 +471,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
             "#,
         )?
         .bind("id", keyset_id.to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_keyset)
         .transpose()?)
@@ -446,6 +479,7 @@ ON CONFLICT(mint_url) DO UPDATE SET
 
     #[instrument(skip_all)]
     async fn add_mint_quote(&self, quote: MintQuote) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query(
             r#"
 INSERT INTO mint_quote
@@ -477,13 +511,14 @@ ON CONFLICT(id) DO UPDATE SET
         .bind("payment_method", quote.payment_method.to_string())
         .bind("amount_issued", quote.amount_issued.to_i64())
         .bind("amount_paid", quote.amount_paid.to_i64())
-        .execute(&self.db).await?;
+        .execute(&*conn).await?;
 
         Ok(())
     }
 
     #[instrument(skip(self))]
     async fn get_mint_quote(&self, quote_id: &str) -> Result<Option<MintQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -505,7 +540,7 @@ ON CONFLICT(id) DO UPDATE SET
             "#,
         )?
         .bind("id", quote_id.to_string())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_mint_quote)
         .transpose()?)
@@ -513,6 +548,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn get_mint_quotes(&self) -> Result<Vec<MintQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -528,7 +564,7 @@ ON CONFLICT(id) DO UPDATE SET
                 mint_quote
             "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .map(sql_row_to_mint_quote)
@@ -537,9 +573,10 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn remove_mint_quote(&self, quote_id: &str) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query(r#"DELETE FROM mint_quote WHERE id=:id"#)?
             .bind("id", quote_id.to_string())
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
 
         Ok(())
@@ -547,6 +584,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip_all)]
     async fn add_melt_quote(&self, quote: wallet::MeltQuote) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query(
             r#"
 INSERT INTO melt_quote
@@ -570,7 +608,7 @@ ON CONFLICT(id) DO UPDATE SET
         .bind("fee_reserve", u64::from(quote.fee_reserve) as i64)
         .bind("state", quote.state.to_string())
         .bind("expiry", quote.expiry as i64)
-        .execute(&self.db)
+        .execute(&*conn)
         .await?;
 
         Ok(())
@@ -578,6 +616,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn get_melt_quote(&self, quote_id: &str) -> Result<Option<wallet::MeltQuote>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -596,7 +635,7 @@ ON CONFLICT(id) DO UPDATE SET
             "#,
         )?
         .bind("id", quote_id.to_owned())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_melt_quote)
         .transpose()?)
@@ -604,9 +643,10 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn remove_melt_quote(&self, quote_id: &str) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query(r#"DELETE FROM melt_quote WHERE id=:id"#)?
             .bind("id", quote_id.to_owned())
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
 
         Ok(())
@@ -614,6 +654,8 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip_all)]
     async fn add_keys(&self, keyset: KeySet) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         // Recompute ID for verification
         keyset.verify_id()?;
 
@@ -631,7 +673,7 @@ ON CONFLICT(id) DO UPDATE SET
             serde_json::to_string(&keyset.keys).map_err(Error::from)?,
         )
         .bind("keyset_u32", u32::from(keyset.id))
-        .execute(&self.db)
+        .execute(&*conn)
         .await?;
 
         Ok(())
@@ -639,6 +681,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self), fields(keyset_id = %keyset_id))]
     async fn get_keys(&self, keyset_id: &Id) -> Result<Option<Keys>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -648,7 +691,7 @@ ON CONFLICT(id) DO UPDATE SET
             "#,
         )?
         .bind("id", keyset_id.to_string())
-        .pluck(&self.db)
+        .pluck(&*conn)
         .await?
         .map(|keys| {
             let keys = column_as_string!(keys);
@@ -659,9 +702,10 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn remove_keys(&self, id: &Id) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query(r#"DELETE FROM key WHERE id = :id"#)?
             .bind("id", id.to_string())
-            .pluck(&self.db)
+            .pluck(&*conn)
             .await?;
 
         Ok(())
@@ -672,6 +716,10 @@ ON CONFLICT(id) DO UPDATE SET
         added: Vec<ProofInfo>,
         removed_ys: Vec<PublicKey>,
     ) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
+        let tx = ConnectionWithTransaction::new(conn).await?;
+
         // TODO: Use a transaction for all these operations
         for proof in added {
             query(
@@ -729,7 +777,7 @@ ON CONFLICT(id) DO UPDATE SET
                 "dleq_r",
                 proof.proof.dleq.as_ref().map(|dleq| dleq.r.to_secret_bytes().to_vec()),
             )
-            .execute(&self.db).await?;
+            .execute(&tx).await?;
         }
 
         query(r#"DELETE FROM proof WHERE y IN (:ys)"#)?
@@ -737,9 +785,11 @@ ON CONFLICT(id) DO UPDATE SET
                 "ys",
                 removed_ys.iter().map(|y| y.to_bytes().to_vec()).collect(),
             )
-            .execute(&self.db)
+            .execute(&tx)
             .await?;
 
+        tx.commit().await?;
+
         Ok(())
     }
 
@@ -751,6 +801,7 @@ ON CONFLICT(id) DO UPDATE SET
         state: Option<Vec<State>>,
         spending_conditions: Option<Vec<SpendingConditions>>,
     ) -> Result<Vec<ProofInfo>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -770,7 +821,7 @@ ON CONFLICT(id) DO UPDATE SET
             FROM proof
         "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .filter_map(|row| {
@@ -786,10 +837,11 @@ ON CONFLICT(id) DO UPDATE SET
     }
 
     async fn update_proofs_state(&self, ys: Vec<PublicKey>, state: State) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query("UPDATE proof SET state = :state WHERE y IN (:ys)")?
             .bind_vec("ys", ys.iter().map(|y| y.to_bytes().to_vec()).collect())
             .bind("state", state.to_string())
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
 
         Ok(())
@@ -797,6 +849,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self), fields(keyset_id = %keyset_id))]
     async fn increment_keyset_counter(&self, keyset_id: &Id, count: u32) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         query(
             r#"
             UPDATE keyset
@@ -806,7 +859,7 @@ ON CONFLICT(id) DO UPDATE SET
         )?
         .bind("count", count)
         .bind("id", keyset_id.to_string())
-        .execute(&self.db)
+        .execute(&*conn)
         .await?;
 
         Ok(())
@@ -814,6 +867,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self), fields(keyset_id = %keyset_id))]
     async fn get_keyset_counter(&self, keyset_id: &Id) -> Result<Option<u32>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -825,7 +879,7 @@ ON CONFLICT(id) DO UPDATE SET
             "#,
         )?
         .bind("id", keyset_id.to_string())
-        .pluck(&self.db)
+        .pluck(&*conn)
         .await?
         .map(|n| Ok::<_, Error>(column_as_number!(n)))
         .transpose()?)
@@ -833,6 +887,7 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn add_transaction(&self, transaction: Transaction) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         let mint_url = transaction.mint_url.to_string();
         let direction = transaction.direction.to_string();
         let unit = transaction.unit.to_string();
@@ -876,7 +931,7 @@ ON CONFLICT(id) DO UPDATE SET
             "metadata",
             serde_json::to_string(&transaction.metadata).map_err(Error::from)?,
         )
-        .execute(&self.db)
+        .execute(&*conn)
         .await?;
 
         Ok(())
@@ -887,6 +942,7 @@ ON CONFLICT(id) DO UPDATE SET
         &self,
         transaction_id: TransactionId,
     ) -> Result<Option<Transaction>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
         Ok(query(
             r#"
             SELECT
@@ -906,7 +962,7 @@ ON CONFLICT(id) DO UPDATE SET
             "#,
         )?
         .bind("id", transaction_id.as_slice().to_vec())
-        .fetch_one(&self.db)
+        .fetch_one(&*conn)
         .await?
         .map(sql_row_to_transaction)
         .transpose()?)
@@ -919,6 +975,8 @@ ON CONFLICT(id) DO UPDATE SET
         direction: Option<TransactionDirection>,
         unit: Option<CurrencyUnit>,
     ) -> Result<Vec<Transaction>, Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         Ok(query(
             r#"
             SELECT
@@ -935,7 +993,7 @@ ON CONFLICT(id) DO UPDATE SET
                 transactions
             "#,
         )?
-        .fetch_all(&self.db)
+        .fetch_all(&*conn)
         .await?
         .into_iter()
         .filter_map(|row| {
@@ -952,9 +1010,11 @@ ON CONFLICT(id) DO UPDATE SET
 
     #[instrument(skip(self))]
     async fn remove_transaction(&self, transaction_id: TransactionId) -> Result<(), Self::Err> {
+        let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+
         query(r#"DELETE FROM transactions WHERE id=:id"#)?
             .bind("id", transaction_id.as_slice().to_vec())
-            .execute(&self.db)
+            .execute(&*conn)
             .await?;
 
         Ok(())

crates/cdk-sqlite/src/async_sqlite.rs

@@ -0,0 +1,173 @@
+//! Simple SQLite
+use cdk_common::database::Error;
+use cdk_sql_common::database::{DatabaseConnector, DatabaseExecutor, GenericTransactionHandler};
+use cdk_sql_common::stmt::{Column, SqlPart, Statement};
+use rusqlite::{ffi, CachedStatement, Connection, Error as SqliteError, ErrorCode};
+use tokio::sync::Mutex;
+
+use crate::common::{from_sqlite, to_sqlite};
+
+/// Async Sqlite wrapper
+#[derive(Debug)]
+pub struct AsyncSqlite {
+    inner: Mutex<Connection>,
+}
+
+impl AsyncSqlite {
+    pub fn new(inner: Connection) -> Self {
+        Self {
+            inner: inner.into(),
+        }
+    }
+}
+impl AsyncSqlite {
+    fn get_stmt<'a>(
+        &self,
+        conn: &'a Connection,
+        statement: Statement,
+    ) -> Result<CachedStatement<'a>, Error> {
+        let (sql, placeholder_values) = statement.to_sql()?;
+
+        let new_sql = sql.trim().trim_end_matches("FOR UPDATE");
+
+        let mut stmt = conn
+            .prepare_cached(new_sql)
+            .map_err(|e| Error::Database(Box::new(e)))?;
+
+        for (i, value) in placeholder_values.into_iter().enumerate() {
+            stmt.raw_bind_parameter(i + 1, to_sqlite(value))
+                .map_err(|e| Error::Database(Box::new(e)))?;
+        }
+
+        Ok(stmt)
+    }
+}
+
+#[inline(always)]
+fn to_sqlite_error(err: SqliteError) -> Error {
+    tracing::error!("Failed query with error {:?}", err);
+    if let rusqlite::Error::SqliteFailure(
+        ffi::Error {
+            code,
+            extended_code,
+        },
+        _,
+    ) = err
+    {
+        if code == ErrorCode::ConstraintViolation
+            && (extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
+                || extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
+        {
+            Error::Duplicate
+        } else {
+            Error::Database(Box::new(err))
+        }
+    } else {
+        Error::Database(Box::new(err))
+    }
+}
+
+impl DatabaseConnector for AsyncSqlite {
+    type Transaction = GenericTransactionHandler<Self>;
+}
+
+#[async_trait::async_trait]
+impl DatabaseExecutor for AsyncSqlite {
+    fn name() -> &'static str {
+        "sqlite"
+    }
+
+    async fn execute(&self, statement: Statement) -> Result<usize, Error> {
+        let conn = self.inner.lock().await;
+
+        let mut stmt = self
+            .get_stmt(&conn, statement)
+            .map_err(|e| Error::Database(Box::new(e)))?;
+
+        Ok(stmt.raw_execute().map_err(to_sqlite_error)?)
+    }
+
+    async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
+        let conn = self.inner.lock().await;
+        let mut stmt = self
+            .get_stmt(&conn, statement)
+            .map_err(|e| Error::Database(Box::new(e)))?;
+
+        let columns = stmt.column_count();
+
+        let mut rows = stmt.raw_query();
+        rows.next()
+            .map_err(to_sqlite_error)?
+            .map(|row| {
+                (0..columns)
+                    .map(|i| row.get(i).map(from_sqlite))
+                    .collect::<Result<Vec<_>, _>>()
+            })
+            .transpose()
+            .map_err(to_sqlite_error)
+    }
+
+    async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
+        let conn = self.inner.lock().await;
+        let mut stmt = self
+            .get_stmt(&conn, statement)
+            .map_err(|e| Error::Database(Box::new(e)))?;
+
+        let columns = stmt.column_count();
+
+        let mut rows = stmt.raw_query();
+        let mut results = vec![];
+
+        while let Some(row) = rows.next().map_err(to_sqlite_error)? {
+            results.push(
+                (0..columns)
+                    .map(|i| row.get(i).map(from_sqlite))
+                    .collect::<Result<Vec<_>, _>>()
+                    .map_err(to_sqlite_error)?,
+            )
+        }
+
+        Ok(results)
+    }
+
+    async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
+        let conn = self.inner.lock().await;
+        let mut stmt = self
+            .get_stmt(&conn, statement)
+            .map_err(|e| Error::Database(Box::new(e)))?;
+
+        let mut rows = stmt.raw_query();
+        rows.next()
+            .map_err(to_sqlite_error)?
+            .map(|row| row.get(0usize).map(from_sqlite))
+            .transpose()
+            .map_err(to_sqlite_error)
+    }
+
+    async fn batch(&self, mut statement: Statement) -> Result<(), Error> {
+        let sql = {
+            let part = statement
+                .parts
+                .pop()
+                .ok_or(Error::Internal("Empty SQL".to_owned()))?;
+
+            if !statement.parts.is_empty() || matches!(part, SqlPart::Placeholder(_, _)) {
+                return Err(Error::Internal(
+                    "Invalid usage, batch does not support placeholders".to_owned(),
+                ));
+            }
+
+            if let SqlPart::Raw(sql) = part {
+                sql
+            } else {
+                unreachable!()
+            }
+        };
+
+        self.inner
+            .lock()
+            .await
+            .execute_batch(&sql)
+            .map_err(to_sqlite_error)
+    }
+}

crates/cdk-sqlite/src/common.rs

@@ -1,11 +1,14 @@
+use std::path::PathBuf;
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 use std::time::Duration;
 
-use cdk_sql_common::pool::{self, Pool, ResourceManager};
+use cdk_sql_common::pool::{self, ResourceManager};
 use cdk_sql_common::value::Value;
 use rusqlite::Connection;
 
+use crate::async_sqlite;
+
 /// The config need to create a new SQLite connection
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -13,6 +16,20 @@ pub struct Config {
     password: Option<String>,
 }
 
+impl pool::Config for Config {
+    fn default_timeout(&self) -> Duration {
+        Duration::from_secs(5)
+    }
+
+    fn max_size(&self) -> usize {
+        if self.password.is_none() {
+            1
+        } else {
+            20
+        }
+    }
+}
+
 /// Sqlite connection manager
 #[derive(Debug)]
 pub struct SqliteConnectionManager;
@@ -20,7 +37,7 @@ pub struct SqliteConnectionManager;
 impl ResourceManager for SqliteConnectionManager {
     type Config = Config;
 
-    type Resource = Connection;
+    type Resource = async_sqlite::AsyncSqlite;
 
     type Error = rusqlite::Error;
 
@@ -52,35 +69,58 @@ impl ResourceManager for SqliteConnectionManager {
 
         conn.busy_timeout(Duration::from_secs(10))?;
 
-        Ok(conn)
+        Ok(async_sqlite::AsyncSqlite::new(conn))
     }
 }
 
-/// Create a configured rusqlite connection to a SQLite database.
-/// For SQLCipher support, enable the "sqlcipher" feature and pass a password.
-pub fn create_sqlite_pool(
-    path: &str,
-    password: Option<String>,
-) -> Arc<Pool<SqliteConnectionManager>> {
-    let (config, max_size) = if path.contains(":memory:") {
-        (
+impl From<PathBuf> for Config {
+    fn from(path: PathBuf) -> Self {
+        path.to_str().unwrap_or_default().into()
+    }
+}
+
+impl From<(PathBuf, String)> for Config {
+    fn from((path, password): (PathBuf, String)) -> Self {
+        (path.to_str().unwrap_or_default(), password.as_str()).into()
+    }
+}
+
+impl From<&PathBuf> for Config {
+    fn from(path: &PathBuf) -> Self {
+        path.to_str().unwrap_or_default().into()
+    }
+}
+
+impl From<&str> for Config {
+    fn from(path: &str) -> Self {
+        if path.contains(":memory:") {
             Config {
                 path: None,
-                password,
-            },
-            1,
-        )
-    } else {
-        (
+                password: None,
+            }
+        } else {
             Config {
                 path: Some(path.to_owned()),
-                password,
-            },
-            20,
-        )
-    };
+                password: None,
+            }
+        }
+    }
+}
 
-    Pool::new(config, max_size, Duration::from_secs(10))
+impl From<(&str, &str)> for Config {
+    fn from((path, pass): (&str, &str)) -> Self {
+        if path.contains(":memory:") {
+            Config {
+                path: None,
+                password: Some(pass.to_owned()),
+            }
+        } else {
+            Config {
+                path: Some(path.to_owned()),
+                password: Some(pass.to_owned()),
+            }
+        }
+    }
 }
 
 /// Convert cdk_sql_common::value::Value to rusqlite Value

crates/cdk-sqlite/src/lib.rs

@@ -3,6 +3,7 @@
 #![warn(missing_docs)]
 #![warn(rustdoc::bare_urls)]
 
+mod async_sqlite;
 mod common;
 
 #[cfg(feature = "mint")]

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

@@ -1,727 +0,0 @@
-//! Async, pipelined rusqlite client
-use std::marker::PhantomData;
-use std::path::PathBuf;
-use std::sync::atomic::{AtomicUsize, Ordering};
-use std::sync::{mpsc as std_mpsc, Arc, Mutex};
-use std::thread::spawn;
-use std::time::Instant;
-
-use cdk_common::database::Error;
-use cdk_sql_common::database::{DatabaseConnector, DatabaseExecutor, DatabaseTransaction};
-use cdk_sql_common::pool::{self, Pool, PooledResource};
-use cdk_sql_common::stmt::{Column, ExpectedSqlResponse, Statement as InnerStatement};
-use cdk_sql_common::ConversionError;
-use rusqlite::{ffi, Connection, ErrorCode, TransactionBehavior};
-use tokio::sync::{mpsc, oneshot};
-
-use crate::common::{create_sqlite_pool, from_sqlite, to_sqlite, SqliteConnectionManager};
-
-/// The number of queued SQL statements before it start failing
-const SQL_QUEUE_SIZE: usize = 10_000;
-/// How many ms is considered a slow query, and it'd be logged for further debugging
-const SLOW_QUERY_THRESHOLD_MS: u128 = 20;
-/// How many SQLite parallel connections can be used to read things in parallel
-const WORKING_THREAD_POOL_SIZE: usize = 5;
-
-#[derive(Debug, Clone)]
-pub struct AsyncRusqlite {
-    sender: mpsc::Sender<DbRequest>,
-    inflight_requests: Arc<AtomicUsize>,
-}
-
-impl From<PathBuf> for AsyncRusqlite {
-    fn from(value: PathBuf) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
-    }
-}
-
-impl From<&str> for AsyncRusqlite {
-    fn from(value: &str) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(value, None))
-    }
-}
-
-impl From<(&str, &str)> for AsyncRusqlite {
-    fn from((value, pass): (&str, &str)) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(value, Some(pass.to_owned())))
-    }
-}
-
-impl From<(PathBuf, &str)> for AsyncRusqlite {
-    fn from((value, pass): (PathBuf, &str)) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(
-            value.to_str().unwrap_or_default(),
-            Some(pass.to_owned()),
-        ))
-    }
-}
-
-impl From<(&str, String)> for AsyncRusqlite {
-    fn from((value, pass): (&str, String)) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(value, Some(pass)))
-    }
-}
-
-impl From<(PathBuf, String)> for AsyncRusqlite {
-    fn from((value, pass): (PathBuf, String)) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(
-            value.to_str().unwrap_or_default(),
-            Some(pass),
-        ))
-    }
-}
-
-impl From<&PathBuf> for AsyncRusqlite {
-    fn from(value: &PathBuf) -> Self {
-        AsyncRusqlite::new(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
-    }
-}
-
-/// Internal request for the database thread
-#[derive(Debug)]
-enum DbRequest {
-    Sql(InnerStatement, oneshot::Sender<DbResponse>),
-    Begin(oneshot::Sender<DbResponse>),
-    Commit(oneshot::Sender<DbResponse>),
-    Rollback(oneshot::Sender<DbResponse>),
-}
-
-#[derive(Debug)]
-enum DbResponse {
-    Transaction(mpsc::Sender<DbRequest>),
-    AffectedRows(usize),
-    Pluck(Option<Column>),
-    Row(Option<Vec<Column>>),
-    Rows(Vec<Vec<Column>>),
-    Error(SqliteError),
-    Unexpected,
-    Ok,
-}
-
-#[derive(thiserror::Error, Debug)]
-enum SqliteError {
-    #[error(transparent)]
-    Sqlite(#[from] rusqlite::Error),
-
-    #[error(transparent)]
-    Inner(#[from] Error),
-
-    #[error(transparent)]
-    Pool(#[from] pool::Error<rusqlite::Error>),
-
-    /// Duplicate entry
-    #[error("Duplicate")]
-    Duplicate,
-
-    #[error(transparent)]
-    Conversion(#[from] ConversionError),
-}
-
-impl From<SqliteError> for Error {
-    fn from(val: SqliteError) -> Self {
-        match val {
-            SqliteError::Duplicate => Error::Duplicate,
-            SqliteError::Conversion(e) => e.into(),
-            o => Error::Internal(o.to_string()),
-        }
-    }
-}
-
-/// Process a query
-#[inline(always)]
-fn process_query(conn: &Connection, statement: InnerStatement) -> Result<DbResponse, SqliteError> {
-    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() {
-        stmt.raw_bind_parameter(i + 1, to_sqlite(value))?;
-    }
-
-    let columns = stmt.column_count();
-
-    let to_return = match expected_response {
-        ExpectedSqlResponse::AffectedRows => DbResponse::AffectedRows(stmt.raw_execute()?),
-        ExpectedSqlResponse::Batch => {
-            conn.execute_batch(sql)?;
-            DbResponse::Ok
-        }
-        ExpectedSqlResponse::ManyRows => {
-            let mut rows = stmt.raw_query();
-            let mut results = vec![];
-
-            while let Some(row) = rows.next()? {
-                results.push(
-                    (0..columns)
-                        .map(|i| row.get(i).map(from_sqlite))
-                        .collect::<Result<Vec<_>, _>>()?,
-                )
-            }
-
-            DbResponse::Rows(results)
-        }
-        ExpectedSqlResponse::Pluck => {
-            let mut rows = stmt.raw_query();
-            DbResponse::Pluck(
-                rows.next()?
-                    .map(|row| row.get(0usize).map(from_sqlite))
-                    .transpose()?,
-            )
-        }
-        ExpectedSqlResponse::SingleRow => {
-            let mut rows = stmt.raw_query();
-            let row = rows
-                .next()?
-                .map(|row| {
-                    (0..columns)
-                        .map(|i| row.get(i).map(from_sqlite))
-                        .collect::<Result<Vec<_>, _>>()
-                })
-                .transpose()?;
-            DbResponse::Row(row)
-        }
-    };
-
-    let duration = start.elapsed();
-
-    if duration.as_millis() > SLOW_QUERY_THRESHOLD_MS {
-        tracing::warn!("[SLOW QUERY] Took {} ms: {}", duration.as_millis(), sql);
-    }
-
-    Ok(to_return)
-}
-
-/// Spawns N number of threads to execute SQL statements
-///
-/// Enable parallelism with a pool of threads.
-///
-/// There is a main thread, which receives SQL requests and routes them to a worker thread from a
-/// fixed-size pool.
-///
-/// By doing so, SQLite does synchronization, and Rust will only intervene when a transaction is
-/// executed. Transactions are executed in the main thread.
-fn rusqlite_spawn_worker_threads(
-    inflight_requests: Arc<AtomicUsize>,
-    threads: usize,
-) -> std_mpsc::Sender<(
-    PooledResource<SqliteConnectionManager>,
-    InnerStatement,
-    oneshot::Sender<DbResponse>,
-)> {
-    let (sender, receiver) = std_mpsc::channel::<(
-        PooledResource<SqliteConnectionManager>,
-        InnerStatement,
-        oneshot::Sender<DbResponse>,
-    )>();
-    let receiver = Arc::new(Mutex::new(receiver));
-
-    for _ in 0..threads {
-        let rx = receiver.clone();
-        let inflight_requests = inflight_requests.clone();
-        spawn(move || loop {
-            while let Ok((conn, sql, reply_to)) = rx.lock().expect("failed to acquire").recv() {
-                let result = process_query(&conn, sql);
-                let _ = match result {
-                    Ok(ok) => reply_to.send(ok),
-                    Err(err) => {
-                        tracing::error!("Failed query with error {:?}", err);
-                        let err = if let SqliteError::Sqlite(rusqlite::Error::SqliteFailure(
-                            ffi::Error {
-                                code,
-                                extended_code,
-                            },
-                            _,
-                        )) = &err
-                        {
-                            if *code == ErrorCode::ConstraintViolation
-                                && (*extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
-                                    || *extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
-                            {
-                                SqliteError::Duplicate
-                            } else {
-                                err
-                            }
-                        } else {
-                            err
-                        };
-
-                        reply_to.send(DbResponse::Error(err))
-                    }
-                };
-                drop(conn);
-                inflight_requests.fetch_sub(1, Ordering::Relaxed);
-            }
-        });
-    }
-    sender
-}
-
-/// # Rusqlite main worker
-///
-/// This function takes ownership of a pool of connections to SQLite, executes SQL statements, and
-/// returns the results or number of affected rows to the caller. All communications are done
-/// through channels. This function is synchronous, but a thread pool exists to execute queries, and
-/// SQLite will coordinate data access. Transactions are executed in the main and it takes ownership
-/// of the main thread until it is finalized
-///
-/// This is meant to be called in their thread, as it will not exit the loop until the communication
-/// channel is closed.
-fn rusqlite_worker_manager(
-    mut receiver: mpsc::Receiver<DbRequest>,
-    pool: Arc<Pool<SqliteConnectionManager>>,
-    inflight_requests: Arc<AtomicUsize>,
-) {
-    let send_sql_to_thread =
-        rusqlite_spawn_worker_threads(inflight_requests.clone(), WORKING_THREAD_POOL_SIZE);
-
-    let mut tx_id: usize = 0;
-
-    while let Some(request) = receiver.blocking_recv() {
-        inflight_requests.fetch_add(1, Ordering::Relaxed);
-        match request {
-            DbRequest::Sql(statement, reply_to) => {
-                let conn = match pool.get() {
-                    Ok(conn) => conn,
-                    Err(err) => {
-                        tracing::error!("Failed to acquire a pool connection: {:?}", err);
-                        inflight_requests.fetch_sub(1, Ordering::Relaxed);
-                        let _ = reply_to.send(DbResponse::Error(err.into()));
-                        continue;
-                    }
-                };
-
-                let _ = send_sql_to_thread.send((conn, statement, reply_to));
-                continue;
-            }
-            DbRequest::Begin(reply_to) => {
-                let (sender, mut receiver) = mpsc::channel(SQL_QUEUE_SIZE);
-                let mut conn = match pool.get() {
-                    Ok(conn) => conn,
-                    Err(err) => {
-                        tracing::error!("Failed to acquire a pool connection: {:?}", err);
-                        inflight_requests.fetch_sub(1, Ordering::Relaxed);
-                        let _ = reply_to.send(DbResponse::Error(err.into()));
-                        continue;
-                    }
-                };
-
-                let tx = match conn.transaction_with_behavior(TransactionBehavior::Immediate) {
-                    Ok(tx) => tx,
-                    Err(err) => {
-                        tracing::error!("Failed to begin a transaction: {:?}", err);
-                        inflight_requests.fetch_sub(1, Ordering::Relaxed);
-                        let _ = reply_to.send(DbResponse::Error(err.into()));
-                        continue;
-                    }
-                };
-
-                // Transaction has begun successfully, send the `sender` back to the caller
-                // and wait for statements to execute. On `Drop` the wrapper transaction
-                // should send a `rollback`.
-                let _ = reply_to.send(DbResponse::Transaction(sender));
-
-                tx_id += 1;
-
-                // We intentionally handle the transaction hijacking the main loop, there is
-                // no point is queueing more operations for SQLite, since transaction have
-                // exclusive access. In other database implementation this block of code
-                // should be sent to their own thread to allow concurrency
-                loop {
-                    let request = if let Some(request) = receiver.blocking_recv() {
-                        request
-                    } else {
-                        // If the receiver loop is broken (i.e no more `senders` are active) and no
-                        // `Commit` statement has been sent, this will trigger a `Rollback`
-                        // automatically
-                        tracing::trace!("Tx {}: Transaction rollback on drop", tx_id);
-                        let _ = tx.rollback();
-                        break;
-                    };
-
-                    match request {
-                        DbRequest::Commit(reply_to) => {
-                            tracing::trace!("Tx {}: Commit", tx_id);
-                            let _ = reply_to.send(match tx.commit() {
-                                Ok(()) => DbResponse::Ok,
-                                Err(err) => {
-                                    tracing::error!("Failed commit {:?}", err);
-                                    DbResponse::Error(err.into())
-                                }
-                            });
-                            break;
-                        }
-                        DbRequest::Rollback(reply_to) => {
-                            tracing::trace!("Tx {}: Rollback", tx_id);
-                            let _ = reply_to.send(match tx.rollback() {
-                                Ok(()) => DbResponse::Ok,
-                                Err(err) => {
-                                    tracing::error!("Failed rollback {:?}", err);
-                                    DbResponse::Error(err.into())
-                                }
-                            });
-                            break;
-                        }
-                        DbRequest::Begin(reply_to) => {
-                            let _ = reply_to.send(DbResponse::Unexpected);
-                        }
-                        DbRequest::Sql(statement, reply_to) => {
-                            tracing::trace!("Tx {}: SQL {:?}", tx_id, statement);
-                            let _ = match process_query(&tx, statement) {
-                                Ok(ok) => reply_to.send(ok),
-                                Err(err) => {
-                                    tracing::error!(
-                                        "Tx {}: Failed query with error {:?}",
-                                        tx_id,
-                                        err
-                                    );
-                                    let err = if let SqliteError::Sqlite(
-                                        rusqlite::Error::SqliteFailure(
-                                            ffi::Error {
-                                                code,
-                                                extended_code,
-                                            },
-                                            _,
-                                        ),
-                                    ) = &err
-                                    {
-                                        if *code == ErrorCode::ConstraintViolation
-                                            && (*extended_code == ffi::SQLITE_CONSTRAINT_PRIMARYKEY
-                                                || *extended_code == ffi::SQLITE_CONSTRAINT_UNIQUE)
-                                        {
-                                            SqliteError::Duplicate
-                                        } else {
-                                            err
-                                        }
-                                    } else {
-                                        err
-                                    };
-                                    reply_to.send(DbResponse::Error(err))
-                                }
-                            };
-                        }
-                    }
-                }
-
-                drop(conn);
-            }
-            DbRequest::Commit(reply_to) => {
-                let _ = reply_to.send(DbResponse::Unexpected);
-            }
-            DbRequest::Rollback(reply_to) => {
-                let _ = reply_to.send(DbResponse::Unexpected);
-            }
-        }
-
-        // If wasn't a `continue` the transaction is done by reaching this code, and we should
-        // decrease the inflight_request counter
-        inflight_requests.fetch_sub(1, Ordering::Relaxed);
-    }
-}
-
-impl AsyncRusqlite {
-    /// Creates a new Async Rusqlite wrapper.
-    pub fn new(pool: Arc<Pool<SqliteConnectionManager>>) -> Self {
-        let (sender, receiver) = mpsc::channel(SQL_QUEUE_SIZE);
-        let inflight_requests = Arc::new(AtomicUsize::new(0));
-        let inflight_requests_for_thread = inflight_requests.clone();
-        spawn(move || {
-            rusqlite_worker_manager(receiver, pool, inflight_requests_for_thread);
-        });
-
-        Self {
-            sender,
-            inflight_requests,
-        }
-    }
-
-    fn get_queue_sender(&self) -> &mpsc::Sender<DbRequest> {
-        &self.sender
-    }
-
-    /// Show how many inflight requests
-    #[allow(dead_code)]
-    pub fn inflight_requests(&self) -> usize {
-        self.inflight_requests.load(Ordering::Relaxed)
-    }
-}
-
-#[async_trait::async_trait]
-impl DatabaseConnector for AsyncRusqlite {
-    type Transaction<'a> = Transaction<'a>;
-
-    /// Begins a transaction
-    ///
-    /// If the transaction is Drop it will trigger a rollback operation
-    async fn begin(&self) -> Result<Self::Transaction<'_>, Error> {
-        let (sender, receiver) = oneshot::channel();
-        self.sender
-            .send(DbRequest::Begin(sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Transaction(db_sender) => Ok(Transaction {
-                db_sender,
-                _marker: PhantomData,
-            }),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-}
-
-#[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;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Row(row) => Ok(row),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn batch(&self, mut statement: InnerStatement) -> Result<(), Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::Batch;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Ok => Ok(()),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn fetch_all(&self, mut statement: InnerStatement) -> Result<Vec<Vec<Column>>, Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::ManyRows;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Rows(row) => Ok(row),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn execute(&self, mut statement: InnerStatement) -> Result<usize, Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::AffectedRows;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::AffectedRows(total) => Ok(total),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn pluck(&self, mut statement: InnerStatement) -> Result<Option<Column>, Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::Pluck;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Pluck(value) => Ok(value),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-}
-
-/// Database transaction
-#[derive(Debug)]
-pub struct Transaction<'conn> {
-    db_sender: mpsc::Sender<DbRequest>,
-    _marker: PhantomData<&'conn ()>,
-}
-
-impl Transaction<'_> {
-    fn get_queue_sender(&self) -> &mpsc::Sender<DbRequest> {
-        &self.db_sender
-    }
-}
-
-impl Drop for Transaction<'_> {
-    fn drop(&mut self) {
-        let (sender, _) = oneshot::channel();
-        let _ = self.db_sender.try_send(DbRequest::Rollback(sender));
-    }
-}
-
-#[async_trait::async_trait]
-impl<'a> DatabaseTransaction<'a> for Transaction<'a> {
-    async fn commit(self) -> Result<(), Error> {
-        let (sender, receiver) = oneshot::channel();
-        self.db_sender
-            .send(DbRequest::Commit(sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Ok => Ok(()),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn rollback(self) -> Result<(), Error> {
-        let (sender, receiver) = oneshot::channel();
-        self.db_sender
-            .send(DbRequest::Rollback(sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Ok => Ok(()),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-}
-
-#[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;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Row(row) => Ok(row),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn batch(&self, mut statement: InnerStatement) -> Result<(), Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::Batch;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Ok => Ok(()),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn fetch_all(&self, mut statement: InnerStatement) -> Result<Vec<Vec<Column>>, Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::ManyRows;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Rows(row) => Ok(row),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn execute(&self, mut statement: InnerStatement) -> Result<usize, Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::AffectedRows;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::AffectedRows(total) => Ok(total),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-
-    async fn pluck(&self, mut statement: InnerStatement) -> Result<Option<Column>, Error> {
-        let (sender, receiver) = oneshot::channel();
-        statement.expected_response = ExpectedSqlResponse::Pluck;
-        self.get_queue_sender()
-            .send(DbRequest::Sql(statement, sender))
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?;
-
-        match receiver
-            .await
-            .map_err(|_| Error::Internal("Communication".to_owned()))?
-        {
-            DbResponse::Pluck(value) => Ok(value),
-            DbResponse::Error(err) => Err(err.into()),
-            _ => Err(Error::InvalidDbResponse),
-        }
-    }
-}

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

@@ -3,26 +3,33 @@
 use cdk_sql_common::mint::SQLMintAuthDatabase;
 use cdk_sql_common::SQLMintDatabase;
 
-mod async_rusqlite;
+use crate::async_sqlite;
+use crate::common::{Config, SqliteConnectionManager};
 
 pub mod memory;
 
 /// Mint SQLite implementation with rusqlite
-pub type MintSqliteDatabase = SQLMintDatabase<async_rusqlite::AsyncRusqlite>;
+pub type MintSqliteDatabase =
+    SQLMintDatabase<async_sqlite::AsyncSqlite, SqliteConnectionManager, Config, rusqlite::Error>;
 
 /// Mint Auth database with rusqlite
 #[cfg(feature = "auth")]
-pub type MintSqliteAuthDatabase = SQLMintAuthDatabase<async_rusqlite::AsyncRusqlite>;
+pub type MintSqliteAuthDatabase = SQLMintAuthDatabase<
+    async_sqlite::AsyncSqlite,
+    SqliteConnectionManager,
+    Config,
+    rusqlite::Error,
+>;
 
 #[cfg(test)]
 mod test {
     use std::fs::remove_file;
 
     use cdk_common::mint_db_test;
+    use cdk_sql_common::pool::Pool;
     use cdk_sql_common::stmt::query;
 
     use super::*;
-    use crate::mint::async_rusqlite::AsyncRusqlite;
 
     async fn provide_db() -> MintSqliteDatabase {
         memory::empty().await.unwrap()
@@ -40,13 +47,17 @@ mod test {
         {
             let _ = remove_file(&file);
             #[cfg(not(feature = "sqlcipher"))]
-            let conn: AsyncRusqlite = file.as_str().into();
+            let config: Config = file.as_str().into();
             #[cfg(feature = "sqlcipher")]
-            let conn: AsyncRusqlite = (file.as_str(), "test".to_owned()).into();
+            let config: Config = (file.as_str(), "test").into();
+
+            let pool = Pool::<SqliteConnectionManager>::new(config);
+
+            let conn = pool.get().expect("valid connection");
 
             query(include_str!("../../tests/legacy-sqlx.sql"))
                 .expect("query")
-                .execute(&conn)
+                .execute(&*conn)
                 .await
                 .expect("create former db failed");
         }
@@ -55,7 +66,7 @@ mod test {
         let conn = MintSqliteDatabase::new(file.as_str()).await;
 
         #[cfg(feature = "sqlcipher")]
-        let conn = MintSqliteDatabase::new((file.as_str(), "test".to_owned())).await;
+        let conn = MintSqliteDatabase::new((file.as_str(), "test")).await;
 
         assert!(conn.is_ok(), "Failed with {:?}", conn.unwrap_err());
 

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

@@ -1,195 +1,15 @@
 //! SQLite Wallet Database
 
-use std::path::PathBuf;
-use std::sync::Arc;
-
-use cdk_common::database::Error;
-use cdk_sql_common::database::DatabaseExecutor;
-use cdk_sql_common::pool::{Pool, PooledResource};
-use cdk_sql_common::stmt::{Column, SqlPart, Statement};
 use cdk_sql_common::SQLWalletDatabase;
-use rusqlite::CachedStatement;
 
-use crate::common::{create_sqlite_pool, from_sqlite, to_sqlite, SqliteConnectionManager};
+use crate::async_sqlite;
+use crate::common::{Config, SqliteConnectionManager};
 
 pub mod memory;
 
-/// Simple Sqlite wapper, since the wallet may not need rusqlite with concurrency, a shared instance
-/// may be enough
-#[derive(Debug)]
-pub struct SimpleAsyncRusqlite(Arc<Pool<SqliteConnectionManager>>);
-
-impl SimpleAsyncRusqlite {
-    fn get_stmt<'a>(
-        &self,
-        conn: &'a PooledResource<SqliteConnectionManager>,
-        statement: Statement,
-    ) -> Result<CachedStatement<'a>, Error> {
-        let (sql, placeholder_values) = statement.to_sql()?;
-        let mut stmt = conn
-            .prepare_cached(&sql)
-            .map_err(|e| Error::Database(Box::new(e)))?;
-
-        for (i, value) in placeholder_values.into_iter().enumerate() {
-            stmt.raw_bind_parameter(i + 1, to_sqlite(value))
-                .map_err(|e| Error::Database(Box::new(e)))?;
-        }
-
-        Ok(stmt)
-    }
-}
-
-#[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
-            .get_stmt(&conn, statement)
-            .map_err(|e| Error::Database(Box::new(e)))?;
-
-        Ok(stmt
-            .raw_execute()
-            .map_err(|e| Error::Database(Box::new(e)))?)
-    }
-
-    async fn fetch_one(&self, statement: Statement) -> Result<Option<Vec<Column>>, Error> {
-        let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
-        let mut stmt = self
-            .get_stmt(&conn, statement)
-            .map_err(|e| Error::Database(Box::new(e)))?;
-
-        let columns = stmt.column_count();
-
-        let mut rows = stmt.raw_query();
-        rows.next()
-            .map_err(|e| Error::Database(Box::new(e)))?
-            .map(|row| {
-                (0..columns)
-                    .map(|i| row.get(i).map(from_sqlite))
-                    .collect::<Result<Vec<_>, _>>()
-            })
-            .transpose()
-            .map_err(|e| Error::Database(Box::new(e)))
-    }
-
-    async fn fetch_all(&self, statement: Statement) -> Result<Vec<Vec<Column>>, Error> {
-        let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
-        let mut stmt = self
-            .get_stmt(&conn, statement)
-            .map_err(|e| Error::Database(Box::new(e)))?;
-
-        let columns = stmt.column_count();
-
-        let mut rows = stmt.raw_query();
-        let mut results = vec![];
-
-        while let Some(row) = rows.next().map_err(|e| Error::Database(Box::new(e)))? {
-            results.push(
-                (0..columns)
-                    .map(|i| row.get(i).map(from_sqlite))
-                    .collect::<Result<Vec<_>, _>>()
-                    .map_err(|e| Error::Database(Box::new(e)))?,
-            )
-        }
-
-        Ok(results)
-    }
-
-    async fn pluck(&self, statement: Statement) -> Result<Option<Column>, Error> {
-        let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
-        let mut stmt = self
-            .get_stmt(&conn, statement)
-            .map_err(|e| Error::Database(Box::new(e)))?;
-
-        let mut rows = stmt.raw_query();
-        rows.next()
-            .map_err(|e| Error::Database(Box::new(e)))?
-            .map(|row| row.get(0usize).map(from_sqlite))
-            .transpose()
-            .map_err(|e| Error::Database(Box::new(e)))
-    }
-
-    async fn batch(&self, mut statement: Statement) -> Result<(), Error> {
-        let conn = self.0.get().map_err(|e| Error::Database(Box::new(e)))?;
-
-        let sql = {
-            let part = statement
-                .parts
-                .pop()
-                .ok_or(Error::Internal("Empty SQL".to_owned()))?;
-
-            if !statement.parts.is_empty() || matches!(part, SqlPart::Placeholder(_, _)) {
-                return Err(Error::Internal(
-                    "Invalid usage, batch does not support placeholders".to_owned(),
-                ));
-            }
-
-            if let SqlPart::Raw(sql) = part {
-                sql
-            } else {
-                unreachable!()
-            }
-        };
-
-        conn.execute_batch(&sql)
-            .map_err(|e| Error::Database(Box::new(e)))
-    }
-}
-
-impl From<PathBuf> for SimpleAsyncRusqlite {
-    fn from(value: PathBuf) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
-    }
-}
-
-impl From<&str> for SimpleAsyncRusqlite {
-    fn from(value: &str) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(value, None))
-    }
-}
-
-impl From<(&str, &str)> for SimpleAsyncRusqlite {
-    fn from((value, pass): (&str, &str)) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(value, Some(pass.to_owned())))
-    }
-}
-
-impl From<(PathBuf, &str)> for SimpleAsyncRusqlite {
-    fn from((value, pass): (PathBuf, &str)) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(
-            value.to_str().unwrap_or_default(),
-            Some(pass.to_owned()),
-        ))
-    }
-}
-
-impl From<(&str, String)> for SimpleAsyncRusqlite {
-    fn from((value, pass): (&str, String)) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(value, Some(pass)))
-    }
-}
-
-impl From<(PathBuf, String)> for SimpleAsyncRusqlite {
-    fn from((value, pass): (PathBuf, String)) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(
-            value.to_str().unwrap_or_default(),
-            Some(pass),
-        ))
-    }
-}
-
-impl From<&PathBuf> for SimpleAsyncRusqlite {
-    fn from(value: &PathBuf) -> Self {
-        SimpleAsyncRusqlite(create_sqlite_pool(value.to_str().unwrap_or_default(), None))
-    }
-}
-
 /// Mint SQLite implementation with rusqlite
-pub type WalletSqliteDatabase = SQLWalletDatabase<SimpleAsyncRusqlite>;
+pub type WalletSqliteDatabase =
+    SQLWalletDatabase<async_sqlite::AsyncSqlite, SqliteConnectionManager, Config, rusqlite::Error>;
 
 #[cfg(test)]
 mod tests {
@@ -330,7 +150,7 @@ mod tests {
 
         // Test PaymentMethod variants
         let mint_url = MintUrl::from_str("https://example.com").unwrap();
-        let payment_methods = vec![
+        let payment_methods = [
             PaymentMethod::Bolt11,
             PaymentMethod::Bolt12,
             PaymentMethod::Custom("custom".to_string()),