fix: wallet sql db internal txs

Author: thesimplekid

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

@@ -746,6 +746,7 @@ where
         removed_ys: Vec<PublicKey>,
     ) -> Result<(), database::Error> {
         let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        let tx = ConnectionWithTransaction::new(conn).await?;
 
         for proof in added {
             query(
@@ -803,7 +804,7 @@ where
                 "dleq_r",
                 proof.proof.dleq.as_ref().map(|dleq| dleq.r.to_secret_bytes().to_vec()),
             )
-            .execute(&*conn)
+            .execute(&tx)
             .await?;
         }
 
@@ -813,10 +814,12 @@ where
                     "ys",
                     removed_ys.iter().map(|y| y.to_bytes().to_vec()).collect(),
                 )
-                .execute(&*conn)
+                .execute(&tx)
                 .await?;
         }
 
+        tx.commit().await?;
+
         Ok(())
     }
 
@@ -906,6 +909,7 @@ where
         new_mint_url: MintUrl,
     ) -> Result<(), database::Error> {
         let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        let tx = ConnectionWithTransaction::new(conn).await?;
         let tables = ["mint_quote", "proof"];
 
         for table in &tables {
@@ -918,10 +922,12 @@ where
             ))?
             .bind("new_mint_url", new_mint_url.to_string())
             .bind("old_mint_url", old_mint_url.to_string())
-            .execute(&*conn)
+            .execute(&tx)
             .await?;
         }
 
+        tx.commit().await?;
+
         Ok(())
     }
 
@@ -1093,6 +1099,7 @@ where
         keysets: Vec<KeySetInfo>,
     ) -> Result<(), database::Error> {
         let conn = self.pool.get().map_err(|e| Error::Database(Box::new(e)))?;
+        let tx = ConnectionWithTransaction::new(conn).await?;
 
         for keyset in keysets {
             query(
@@ -1113,10 +1120,12 @@ where
             .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(&*conn)
+            .execute(&tx)
             .await?;
         }
 
+        tx.commit().await?;
+
         Ok(())
     }