Working on a better database abstraction (#931)

* 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

* Improve the abstraction with fewer generics

As suggested by @thesimplekid

* Add BEGIN IMMEDIATE for SQLite

Author: crodas

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;
 }