Add Database transaction to the Wallet trait

crates/cdk-cli/src/sub_commands/check_pending.rs

@@ -9,8 +9,12 @@ pub async fn check_pending(multi_mint_wallet: &MultiMintWallet) -> Result<()> {
         let mint_url = wallet.mint_url.clone();
         println!("{i}: {mint_url}");
 
+        let mut tx = wallet.localstore.begin_db_transaction().await?;
+
         // Get all pending proofs
-        let pending_proofs = wallet.get_pending_proofs().await?;
+        //
+
+        let pending_proofs = wallet.get_pending_proofs(Some(&mut tx)).await?;
         if pending_proofs.is_empty() {
             println!("No pending proofs found");
             continue;
@@ -24,10 +28,12 @@ pub async fn check_pending(multi_mint_wallet: &MultiMintWallet) -> Result<()> {
         );
 
         // Try to reclaim any proofs that are no longer pending
-        match wallet.reclaim_unspent(pending_proofs).await {
+        match wallet.reclaim_unspent(pending_proofs, &mut tx).await {
             Ok(()) => println!("Successfully reclaimed pending proofs"),
             Err(e) => println!("Error reclaimed pending proofs: {e}"),
         }
+
+        tx.commit().await?;
     }
     Ok(())
 }

crates/cdk-cli/src/sub_commands/list_mint_proofs.rs

@@ -35,7 +35,7 @@ async fn list_proofs(
         }
 
         // Pending proofs
-        let pending_proofs = wallet.get_pending_proofs().await?;
+        let pending_proofs = wallet.get_pending_proofs(None).await?;
         for proof in pending_proofs {
             println!(
                 "| {:8} | {:4} | {:8} | {:64} | {}",

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

@@ -6,7 +6,7 @@ use async_trait::async_trait;
 use cashu::quote_id::QuoteId;
 use cashu::Amount;
 
-use super::Error;
+use super::{DbTransactionFinalizer, Error};
 use crate::mint::{self, MintKeySetInfo, MintQuote as MintMintQuote, Operation};
 use crate::nuts::{
     BlindSignature, BlindedMessage, CurrencyUnit, Id, MeltQuoteState, Proof, Proofs, PublicKey,
@@ -394,19 +394,6 @@ pub trait SagaDatabase {
     ) -> Result<Vec<mint::Saga>, Self::Err>;
 }
 
-#[async_trait]
-/// Commit and Rollback
-pub trait DbTransactionFinalizer {
-    /// Mint Signature Database Error
-    type Err: Into<Error> + From<Error>;
-
-    /// Commits all the changes into the database
-    async fn commit(self: Box<Self>) -> Result<(), Self::Err>;
-
-    /// Rollbacks the write transaction
-    async fn rollback(self: Box<Self>) -> Result<(), Self::Err>;
-}
-
 /// Key-Value Store Transaction trait
 #[async_trait]
 pub trait KVStoreTransaction<'a, Error>: DbTransactionFinalizer<Err = Error> {

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

@@ -7,18 +7,18 @@ mod wallet;
 
 #[cfg(feature = "mint")]
 pub use mint::{
-    Database as MintDatabase, DbTransactionFinalizer as MintDbWriterFinalizer, DynMintDatabase,
-    KVStore as MintKVStore, KVStoreDatabase as MintKVStoreDatabase,
-    KVStoreTransaction as MintKVStoreTransaction, KeysDatabase as MintKeysDatabase,
-    KeysDatabaseTransaction as MintKeyDatabaseTransaction, ProofsDatabase as MintProofsDatabase,
-    ProofsTransaction as MintProofsTransaction, QuotesDatabase as MintQuotesDatabase,
-    QuotesTransaction as MintQuotesTransaction, SignaturesDatabase as MintSignaturesDatabase,
+    Database as MintDatabase, DynMintDatabase, KVStore as MintKVStore,
+    KVStoreDatabase as MintKVStoreDatabase, KVStoreTransaction as MintKVStoreTransaction,
+    KeysDatabase as MintKeysDatabase, KeysDatabaseTransaction as MintKeyDatabaseTransaction,
+    ProofsDatabase as MintProofsDatabase, ProofsTransaction as MintProofsTransaction,
+    QuotesDatabase as MintQuotesDatabase, QuotesTransaction as MintQuotesTransaction,
+    SignaturesDatabase as MintSignaturesDatabase,
     SignaturesTransaction as MintSignatureTransaction, Transaction as MintTransaction,
 };
 #[cfg(all(feature = "mint", feature = "auth"))]
 pub use mint::{DynMintAuthDatabase, MintAuthDatabase, MintAuthTransaction};
 #[cfg(feature = "wallet")]
-pub use wallet::Database as WalletDatabase;
+pub use wallet::{Database as WalletDatabase, DatabaseTransaction as WalletDatabaseTransaction};
 
 /// Data conversion error
 #[derive(thiserror::Error, Debug)]
@@ -203,3 +203,16 @@ impl From<crate::state::Error> for Error {
         }
     }
 }
+
+#[async_trait::async_trait]
+/// Commit and Rollback
+pub trait DbTransactionFinalizer {
+    /// Mint Signature Database Error
+    type Err: Into<Error> + From<Error>;
+
+    /// Commits all the changes into the database
+    async fn commit(self: Box<Self>) -> Result<(), Self::Err>;
+
+    /// Rollbacks the write transaction
+    async fn rollback(self: Box<Self>) -> Result<(), Self::Err>;
+}

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

@@ -6,7 +6,7 @@ use std::fmt::Debug;
 use async_trait::async_trait;
 use cashu::KeySet;
 
-use super::Error;
+use super::{DbTransactionFinalizer, Error};
 use crate::common::ProofInfo;
 use crate::mint_url::MintUrl;
 use crate::nuts::{
@@ -16,78 +16,135 @@ use crate::wallet::{
     self, MintQuote as WalletMintQuote, Transaction, TransactionDirection, TransactionId,
 };
 
-/// Wallet Database trait
+/// Database transaction
+///
+/// This trait encapsulates all the changes to be done in the wallet
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-pub trait Database: Debug {
-    /// Wallet Database Error
-    type Err: Into<Error> + From<Error>;
-
+pub trait DatabaseTransaction<'a, Error>: DbTransactionFinalizer<Err = Error> {
     /// Add Mint to storage
     async fn add_mint(
-        &self,
+        &mut self,
         mint_url: MintUrl,
         mint_info: Option<MintInfo>,
-    ) -> Result<(), Self::Err>;
+    ) -> Result<(), Error>;
+
     /// Remove Mint from storage
-    async fn remove_mint(&self, mint_url: MintUrl) -> Result<(), Self::Err>;
-    /// Get mint from storage
-    async fn get_mint(&self, mint_url: MintUrl) -> Result<Option<MintInfo>, Self::Err>;
-    /// Get all mints from storage
-    async fn get_mints(&self) -> Result<HashMap<MintUrl, Option<MintInfo>>, Self::Err>;
+    async fn remove_mint(&mut self, mint_url: MintUrl) -> Result<(), Error>;
+
     /// Update mint url
     async fn update_mint_url(
-        &self,
+        &mut self,
         old_mint_url: MintUrl,
         new_mint_url: MintUrl,
-    ) -> Result<(), Self::Err>;
+    ) -> Result<(), Error>;
 
     /// Add mint keyset to storage
     async fn add_mint_keysets(
-        &self,
+        &mut self,
         mint_url: MintUrl,
         keysets: Vec<KeySetInfo>,
-    ) -> Result<(), Self::Err>;
+    ) -> Result<(), Error>;
+
+    /// Get mint quote from storage. This function locks the returned minted quote for update
+    async fn get_mint_quote(
+        &mut self,
+        quote_id: &str,
+    ) -> Result<Option<WalletMintQuote>, Self::Err>;
+
+    /// Add mint quote to storage
+    async fn add_mint_quote(&mut self, quote: WalletMintQuote) -> Result<(), Error>;
+
+    /// Remove mint quote from storage
+    async fn remove_mint_quote(&mut self, quote_id: &str) -> Result<(), Error>;
+
+    /// Get melt quote from storage
+    async fn get_melt_quote(&mut self, quote_id: &str) -> Result<Option<wallet::MeltQuote>, Error>;
+
+    /// Add melt quote to storage
+    async fn add_melt_quote(&mut self, quote: wallet::MeltQuote) -> Result<(), Error>;
+
+    /// Remove melt quote from storage
+    async fn remove_melt_quote(&mut self, quote_id: &str) -> Result<(), Error>;
+
+    /// Add [`Keys`] to storage
+    async fn add_keys(&mut self, keyset: KeySet) -> Result<(), Error>;
+
+    /// Remove [`Keys`] from storage
+    async fn remove_keys(&mut self, id: &Id) -> Result<(), Error>;
+
+    /// Get proofs from storage and lock them for update
+    async fn get_proofs(
+        &mut self,
+        mint_url: Option<MintUrl>,
+        unit: Option<CurrencyUnit>,
+        state: Option<Vec<State>>,
+        spending_conditions: Option<Vec<SpendingConditions>>,
+    ) -> Result<Vec<ProofInfo>, Error>;
+
+    /// Update the proofs in storage by adding new proofs or removing proofs by
+    /// their Y value.
+    async fn update_proofs(
+        &mut self,
+        added: Vec<ProofInfo>,
+        removed_ys: Vec<PublicKey>,
+    ) -> Result<(), Error>;
+
+    /// Update proofs state in storage
+    async fn update_proofs_state(&mut self, ys: Vec<PublicKey>, state: State) -> Result<(), Error>;
+
+    /// Atomically increment Keyset counter and return new value
+    async fn increment_keyset_counter(&mut self, keyset_id: &Id, count: u32) -> Result<u32, Error>;
+
+    /// Add transaction to storage
+    async fn add_transaction(&mut self, transaction: Transaction) -> Result<(), Error>;
+
+    /// Remove transaction from storage
+    async fn remove_transaction(&mut self, transaction_id: TransactionId) -> Result<(), Error>;
+}
+
+/// Wallet Database trait
+#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
+#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
+pub trait Database: Debug {
+    /// Wallet Database Error
+    type Err: Into<Error> + From<Error>;
+
+    /// Beings a KV transaction
+    async fn begin_db_transaction<'a>(
+        &'a self,
+    ) -> Result<Box<dyn DatabaseTransaction<'a, Self::Err> + Send + Sync + 'a>, Error>;
+
+    /// Get mint from storage
+    async fn get_mint(&self, mint_url: MintUrl) -> Result<Option<MintInfo>, Self::Err>;
+
+    /// Get all mints from storage
+    async fn get_mints(&self) -> Result<HashMap<MintUrl, Option<MintInfo>>, Self::Err>;
+
     /// Get mint keysets for mint url
     async fn get_mint_keysets(
         &self,
         mint_url: MintUrl,
     ) -> Result<Option<Vec<KeySetInfo>>, Self::Err>;
+
     /// Get mint keyset by id
     async fn get_keyset_by_id(&self, keyset_id: &Id) -> Result<Option<KeySetInfo>, Self::Err>;
 
-    /// Add mint quote to storage
-    async fn add_mint_quote(&self, quote: WalletMintQuote) -> Result<(), Self::Err>;
     /// Get mint quote from storage
     async fn get_mint_quote(&self, quote_id: &str) -> Result<Option<WalletMintQuote>, Self::Err>;
+
     /// Get mint quotes from storage
     async fn get_mint_quotes(&self) -> Result<Vec<WalletMintQuote>, Self::Err>;
-    /// Remove mint quote from storage
-    async fn remove_mint_quote(&self, quote_id: &str) -> Result<(), Self::Err>;
 
-    /// Add melt quote to storage
-    async fn add_melt_quote(&self, quote: wallet::MeltQuote) -> Result<(), Self::Err>;
     /// Get melt quote from storage
     async fn get_melt_quote(&self, quote_id: &str) -> Result<Option<wallet::MeltQuote>, Self::Err>;
+
     /// Get melt quotes from storage
     async fn get_melt_quotes(&self) -> Result<Vec<wallet::MeltQuote>, Self::Err>;
-    /// Remove melt quote from storage
-    async fn remove_melt_quote(&self, quote_id: &str) -> Result<(), Self::Err>;
 
-    /// Add [`Keys`] to storage
-    async fn add_keys(&self, keyset: KeySet) -> Result<(), Self::Err>;
     /// Get [`Keys`] from storage
     async fn get_keys(&self, id: &Id) -> Result<Option<Keys>, Self::Err>;
-    /// Remove [`Keys`] from storage
-    async fn remove_keys(&self, id: &Id) -> Result<(), Self::Err>;
 
-    /// Update the proofs in storage by adding new proofs or removing proofs by
-    /// their Y value.
-    async fn update_proofs(
-        &self,
-        added: Vec<ProofInfo>,
-        removed_ys: Vec<PublicKey>,
-    ) -> Result<(), Self::Err>;
     /// Get proofs from storage
     async fn get_proofs(
         &self,
@@ -96,6 +153,7 @@ pub trait Database: Debug {
         state: Option<Vec<State>>,
         spending_conditions: Option<Vec<SpendingConditions>>,
     ) -> Result<Vec<ProofInfo>, Self::Err>;
+
     /// Get balance
     async fn get_balance(
         &self,
@@ -103,19 +161,13 @@ pub trait Database: Debug {
         unit: Option<CurrencyUnit>,
         state: Option<Vec<State>>,
     ) -> Result<u64, Self::Err>;
-    /// Update proofs state in storage
-    async fn update_proofs_state(&self, ys: Vec<PublicKey>, state: State) -> Result<(), Self::Err>;
 
-    /// Atomically increment Keyset counter and return new value
-    async fn increment_keyset_counter(&self, keyset_id: &Id, count: u32) -> Result<u32, Self::Err>;
-
-    /// Add transaction to storage
-    async fn add_transaction(&self, transaction: Transaction) -> Result<(), Self::Err>;
     /// Get transaction from storage
     async fn get_transaction(
         &self,
         transaction_id: TransactionId,
     ) -> Result<Option<Transaction>, Self::Err>;
+
     /// List transactions from storage
     async fn list_transactions(
         &self,
@@ -123,6 +175,4 @@ pub trait Database: Debug {
         direction: Option<TransactionDirection>,
         unit: Option<CurrencyUnit>,
     ) -> Result<Vec<Transaction>, Self::Err>;
-    /// Remove transaction from storage
-    async fn remove_transaction(&self, transaction_id: TransactionId) -> Result<(), Self::Err>;
 }

crates/cdk-ffi/src/wallet.rs

@@ -82,7 +82,7 @@ impl Wallet {
 
     /// Get mint info
     pub async fn get_mint_info(&self) -> Result<Option<MintInfo>, FfiError> {
-        let info = self.inner.fetch_mint_info().await?;
+        let info = self.inner.fetch_mint_info(None).await?;
         Ok(info.map(Into::into))
     }
 
@@ -271,9 +271,9 @@ impl Wallet {
         for state in states {
             let proofs = match state {
                 ProofState::Unspent => self.inner.get_unspent_proofs().await?,
-                ProofState::Pending => self.inner.get_pending_proofs().await?,
+                ProofState::Pending => self.inner.get_pending_proofs(None).await?,
                 ProofState::Reserved => self.inner.get_reserved_proofs().await?,
-                ProofState::PendingSpent => self.inner.get_pending_spent_proofs().await?,
+                ProofState::PendingSpent => self.inner.get_pending_spent_proofs(None).await?,
                 ProofState::Spent => {
                     // CDK doesn't have a method to get spent proofs directly
                     // They are removed from the database when spent
@@ -351,7 +351,7 @@ impl Wallet {
 
     /// Refresh keysets from the mint
     pub async fn refresh_keysets(&self) -> Result<Vec<KeySetInfo>, FfiError> {
-        let keysets = self.inner.refresh_keysets().await?;
+        let keysets = self.inner.refresh_keysets(None).await?;
         Ok(keysets.into_iter().map(Into::into).collect())
     }
 

crates/cdk-integration-tests/tests/bolt12.rs

@@ -325,7 +325,11 @@ async fn test_regtest_bolt12_mint_extra() -> Result<()> {
     // Create a single-use BOLT12 quote
     let mint_quote = wallet.mint_bolt12_quote(None, None).await?;
 
-    let state = wallet.mint_bolt12_quote_state(&mint_quote.id).await?;
+    let mut tx = wallet.localstore.begin_db_transaction().await?;
+    let state = wallet
+        .mint_bolt12_quote_state(&mint_quote.id, &mut tx)
+        .await?;
+    tx.commit().await?;
 
     assert_eq!(state.amount_paid, Amount::ZERO);
     assert_eq!(state.amount_issued, Amount::ZERO);
@@ -436,10 +440,12 @@ async fn test_attempt_to_mint_unpaid() {
         .await
         .unwrap();
 
+    let mut tx = wallet.localstore.begin_db_transaction().await?;
     let state = wallet
-        .mint_bolt12_quote_state(&mint_quote.id)
+        .mint_bolt12_quote_state(&mint_quote.id, &mut tx)
         .await
         .unwrap();
+    tx.commit().await?;
 
     assert!(state.amount_paid == Amount::ZERO);
 

crates/cdk-integration-tests/tests/fake_auth.rs

@@ -277,7 +277,7 @@ async fn test_mint_blind_auth() {
         .seed(Mnemonic::generate(12).unwrap().to_seed_normalized(""))
         .build()
         .expect("Wallet");
-    let mint_info = wallet.fetch_mint_info().await.unwrap().unwrap();
+    let mint_info = wallet.fetch_mint_info(None).await.unwrap().unwrap();
 
     let (access_token, _) = get_access_token(&mint_info).await;
 
@@ -355,7 +355,7 @@ async fn test_swap_with_auth() {
         .seed(Mnemonic::generate(12).unwrap().to_seed_normalized(""))
         .build()
         .expect("Wallet");
-    let mint_info = wallet.fetch_mint_info().await.unwrap().unwrap();
+    let mint_info = wallet.fetch_mint_info(None).await.unwrap().unwrap();
     let (access_token, _) = get_access_token(&mint_info).await;
 
     wallet.set_cat(access_token).await.unwrap();
@@ -384,7 +384,7 @@ async fn test_swap_with_auth() {
         .expect("Could not swap");
 
     let check_spent = wallet
-        .check_proofs_spent(proofs.clone())
+        .check_proofs_spent(proofs.clone(), None)
         .await
         .expect("Could not check proofs");
 
@@ -410,7 +410,7 @@ async fn test_melt_with_auth() {
         .expect("Wallet");
 
     let mint_info = wallet
-        .fetch_mint_info()
+        .fetch_mint_info(None)
         .await
         .expect("Mint info not found")
         .expect("Mint info not found");
@@ -452,7 +452,7 @@ async fn test_mint_auth_over_max() {
     let wallet = Arc::new(wallet);
 
     let mint_info = wallet
-        .fetch_mint_info()
+        .fetch_mint_info(None)
         .await
         .expect("Mint info not found")
         .expect("Mint info not found");
@@ -490,7 +490,7 @@ async fn test_reuse_auth_proof() {
         .seed(Mnemonic::generate(12).unwrap().to_seed_normalized(""))
         .build()
         .expect("Wallet");
-    let mint_info = wallet.fetch_mint_info().await.unwrap().unwrap();
+    let mint_info = wallet.fetch_mint_info(None).await.unwrap().unwrap();
 
     let (access_token, _) = get_access_token(&mint_info).await;
 
@@ -542,7 +542,7 @@ async fn test_melt_with_invalid_auth() {
         .seed(Mnemonic::generate(12).unwrap().to_seed_normalized(""))
         .build()
         .expect("Wallet");
-    let mint_info = wallet.fetch_mint_info().await.unwrap().unwrap();
+    let mint_info = wallet.fetch_mint_info(None).await.unwrap().unwrap();
 
     let (access_token, _) = get_access_token(&mint_info).await;
 
@@ -607,7 +607,7 @@ async fn test_refresh_access_token() {
         .expect("Wallet");
 
     let mint_info = wallet
-        .fetch_mint_info()
+        .fetch_mint_info(None)
         .await
         .expect("mint info")
         .expect("could not get mint info");
@@ -663,7 +663,7 @@ async fn test_invalid_refresh_token() {
         .expect("Wallet");
 
     let mint_info = wallet
-        .fetch_mint_info()
+        .fetch_mint_info(None)
         .await
         .expect("mint info")
         .expect("could not get mint info");

crates/cdk-integration-tests/tests/fake_wallet.rs

@@ -128,7 +128,7 @@ async fn test_fake_melt_payment_fail() {
 
     // The mint should have unset proofs from pending since payment failed
     let all_proof = wallet.get_unspent_proofs().await.unwrap();
-    let states = wallet.check_proofs_spent(all_proof).await.unwrap();
+    let states = wallet.check_proofs_spent(all_proof, None).await.unwrap();
     for state in states {
         assert!(state.state == State::Unspent);
     }
@@ -749,7 +749,7 @@ async fn test_fake_mint_multiple_unit_swap() {
         None,
     )
     .expect("failed to create usd wallet");
-    wallet_usd.refresh_keysets().await.unwrap();
+    wallet_usd.refresh_keysets(None).await.unwrap();
 
     let mint_quote = wallet_usd.mint_quote(100.into(), None).await.unwrap();
 

crates/cdk-integration-tests/tests/happy_path_mint_wallet.rs

@@ -338,7 +338,7 @@ async fn test_restore() {
 
     let proofs = wallet.get_unspent_proofs().await.unwrap();
 
-    let states = wallet.check_proofs_spent(proofs).await.unwrap();
+    let states = wallet.check_proofs_spent(proofs, None).await.unwrap();
 
     for state in states {
         if state.state != State::Spent {

crates/cdk-integration-tests/tests/integration_tests_pure.rs

@@ -99,7 +99,7 @@ async fn test_swap_to_send() {
         HashSet::<_, RandomState>::from_iter(token_proofs.ys().expect("Failed to get ys")),
         HashSet::from_iter(
             wallet_alice
-                .get_pending_spent_proofs()
+                .get_pending_spent_proofs(None)
                 .await
                 .expect("Failed to get pending spent proofs")
                 .ys()
@@ -765,10 +765,16 @@ async fn test_mint_change_with_fee_melt() {
         .await
         .unwrap();
 
+    let mut tx = wallet_alice
+        .localstore
+        .begin_db_transaction()
+        .await
+        .unwrap();
     let w = wallet_alice
-        .melt_proofs(&melt_quote.id, proofs)
+        .melt_proofs_with_metadata(&melt_quote.id, proofs, HashMap::new(), &mut tx)
         .await
         .unwrap();
+    tx.commit().await.unwrap();
 
     assert_eq!(w.change.unwrap().total_amount().unwrap(), 97.into());
 }

crates/cdk-integration-tests/tests/test_fees.rs

@@ -108,10 +108,14 @@ async fn test_fake_melt_change_in_quote() {
     let proofs_total = proofs.total_amount().unwrap();
 
     let fee = wallet.get_proofs_fee(&proofs).await.unwrap();
+
+    let mut tx = wallet.localstore.begin_db_transaction().await.unwrap();
     let melt = wallet
-        .melt_proofs(&melt_quote.id, proofs.clone())
+        .melt_proofs_with_metadata(&melt_quote.id, proofs, HashMap::new())
         .await
         .unwrap();
+    tx.commit().await.unwrap();
+
     let change = melt.change.unwrap().total_amount().unwrap();
     let idk = proofs.total_amount().unwrap() - Amount::from(invoice_amount) - change;
 

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

@@ -17,7 +17,7 @@ use async_trait::async_trait;
 use bitcoin::bip32::DerivationPath;
 use cdk_common::database::mint::{validate_kvstore_params, SagaDatabase, SagaTransaction};
 use cdk_common::database::{
-    self, ConversionError, Error, MintDatabase, MintDbWriterFinalizer, MintKeyDatabaseTransaction,
+    self, ConversionError, DbTransactionFinalizer, Error, MintDatabase, MintKeyDatabaseTransaction,
     MintKeysDatabase, MintProofsDatabase, MintQuotesDatabase, MintQuotesTransaction,
     MintSignatureTransaction, MintSignaturesDatabase,
 };
@@ -280,7 +280,7 @@ impl<RM> database::MintTransaction<'_, Error> for SQLTransaction<RM> where RM: D
 {}
 
 #[async_trait]
-impl<RM> MintDbWriterFinalizer for SQLTransaction<RM>
+impl<RM> DbTransactionFinalizer for SQLTransaction<RM>
 where
     RM: DatabasePool + 'static,
 {

crates/cdk/examples/auth_wallet.rs

@@ -37,7 +37,7 @@ async fn main() -> Result<(), Error> {
     let wallet = Wallet::new(mint_url, unit, Arc::new(localstore), seed, None)?;
 
     let mint_info = wallet
-        .fetch_mint_info()
+        .fetch_mint_info(None)
         .await
         .expect("mint info")
         .expect("could not get mint info");

crates/cdk/examples/proof-selection.rs

@@ -52,7 +52,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Select proofs to send
     let amount = Amount::from(64);
     let active_keyset_ids = wallet
-        .refresh_keysets()
+        .refresh_keysets(None)
         .await?
         .active()
         .map(|keyset| keyset.id)

crates/cdk/src/wallet/auth/auth_wallet.rs

@@ -19,6 +19,7 @@ use crate::nuts::{
 };
 use crate::types::ProofInfo;
 use crate::wallet::mint_connector::AuthHttpClient;
+use crate::wallet::Tx;
 use crate::{Amount, Error, OidcClient};
 
 /// JWT Claims structure for decoding tokens
@@ -181,7 +182,9 @@ impl AuthWallet {
 
             keys.verify_id()?;
 
-            self.localstore.add_keys(keys.clone()).await?;
+            let mut tx = self.localstore.begin_db_transaction().await?;
+            tx.add_keys(keys.clone()).await?;
+            tx.commit().await?;
 
             keys.keys
         };
@@ -195,8 +198,11 @@ impl AuthWallet {
     /// goes online to refresh keysets from the mint and updates the local database.
     /// This is the main method for getting auth keysets in operations that can work offline
     /// but will fall back to online if needed.
-    #[instrument(skip(self))]
-    pub async fn load_mint_keysets(&self) -> Result<Vec<KeySetInfo>, Error> {
+    #[instrument(skip(self, tx))]
+    pub async fn load_mint_keysets(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Vec<KeySetInfo>, Error> {
         match self
             .localstore
             .get_mint_keysets(self.mint_url.clone())
@@ -207,7 +213,7 @@ impl AuthWallet {
                     keysets_info.unit(CurrencyUnit::Sat).cloned().collect();
                 if auth_keysets.is_empty() {
                     // If we don't have any auth keysets, fetch them from the mint
-                    let keysets = self.refresh_keysets().await?;
+                    let keysets = self.refresh_keysets(tx).await?;
                     Ok(keysets)
                 } else {
                     Ok(auth_keysets)
@@ -215,7 +221,7 @@ impl AuthWallet {
             }
             None => {
                 // If we don't have any keysets, fetch them from the mint
-                let keysets = self.refresh_keysets().await?;
+                let keysets = self.refresh_keysets(tx).await?;
                 Ok(keysets)
             }
         }
@@ -227,15 +233,24 @@ impl AuthWallet {
     /// It updates the local database with the fetched keysets and ensures we have keys for all keysets.
     /// Returns only the keysets with Auth currency unit. This is used when operations need the most
     /// up-to-date keyset information and are willing to go online.
-    #[instrument(skip(self))]
-    pub async fn refresh_keysets(&self) -> Result<Vec<KeySetInfo>, Error> {
+    #[instrument(skip(self, tx))]
+    pub async fn refresh_keysets(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Vec<KeySetInfo>, Error> {
         let keysets_response = self.client.get_mint_blind_auth_keysets().await?;
         let keysets = keysets_response.keysets;
 
         // Update local store with keysets
-        self.localstore
-            .add_mint_keysets(self.mint_url.clone(), keysets.clone())
-            .await?;
+        if let Some(tx) = tx {
+            tx.add_mint_keysets(self.mint_url.clone(), keysets.clone())
+                .await?;
+        } else {
+            let mut tx = self.localstore.begin_db_transaction().await?;
+            tx.add_mint_keysets(self.mint_url.clone(), keysets.clone())
+                .await?;
+            tx.commit().await?;
+        }
 
         // Filter for auth keysets
         let auth_keysets = keysets
@@ -260,9 +275,12 @@ impl AuthWallet {
     /// This method always goes online to refresh keysets from the mint and then returns
     /// the first active keyset found. Use this when you need the most up-to-date
     /// keyset information for blind auth operations.
-    #[instrument(skip(self))]
-    pub async fn fetch_active_keyset(&self) -> Result<KeySetInfo, Error> {
-        let auth_keysets = self.refresh_keysets().await?;
+    #[instrument(skip(self, tx))]
+    pub async fn fetch_active_keyset(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<KeySetInfo, Error> {
+        let auth_keysets = self.refresh_keysets(tx).await?;
         let keyset = auth_keysets.first().ok_or(Error::NoActiveKeyset)?;
         Ok(keyset.clone())
     }
@@ -271,11 +289,16 @@ impl AuthWallet {
     ///
     /// Returns auth proofs from the local database that are in the Unspent state.
     /// This is an offline operation that does not contact the mint.
-    #[instrument(skip(self))]
-    pub async fn get_unspent_auth_proofs(&self) -> Result<Vec<AuthProof>, Error> {
-        Ok(self
-            .localstore
-            .get_proofs(
+    ///
+    /// If a DB Transaction is passed as an argument the selected proofs are locked for update
+    /// within this DBTransaction
+    #[instrument(skip(self, tx))]
+    pub async fn get_unspent_auth_proofs(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Vec<AuthProof>, Error> {
+        Ok(if let Some(tx) = tx {
+            tx.get_proofs(
                 Some(self.mint_url.clone()),
                 Some(CurrencyUnit::Auth),
                 Some(vec![State::Unspent]),
@@ -284,7 +307,20 @@ impl AuthWallet {
             .await?
             .into_iter()
             .map(|p| p.proof.try_into())
-            .collect::<Result<Vec<AuthProof>, _>>()?)
+            .collect::<Result<Vec<AuthProof>, _>>()?
+        } else {
+            self.localstore
+                .get_proofs(
+                    Some(self.mint_url.clone()),
+                    Some(CurrencyUnit::Auth),
+                    Some(vec![State::Unspent]),
+                    None,
+                )
+                .await?
+                .into_iter()
+                .map(|p| p.proof.try_into())
+                .collect::<Result<Vec<AuthProof>, _>>()?
+        })
     }
 
     /// Check if and what kind of auth is required for a method
@@ -298,18 +334,20 @@ impl AuthWallet {
     /// Get Auth Token
     #[instrument(skip(self))]
     pub async fn get_blind_auth_token(&self) -> Result<Option<BlindAuthToken>, Error> {
-        let unspent = self.get_unspent_auth_proofs().await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        let unspent = self.get_unspent_auth_proofs(Some(&mut tx)).await?;
 
         let auth_proof = match unspent.first() {
             Some(proof) => {
-                self.localstore
-                    .update_proofs(vec![], vec![proof.y()?])
-                    .await?;
+                tx.update_proofs(vec![], vec![proof.y()?]).await?;
                 proof
             }
             None => return Ok(None),
         };
 
+        tx.commit().await?;
+
         Ok(Some(BlindAuthToken {
             auth_proof: auth_proof.clone(),
         }))
@@ -394,13 +432,13 @@ impl AuthWallet {
         }
 
         let keysets = self
-            .load_mint_keysets()
+            .load_mint_keysets(None)
             .await?
             .into_iter()
             .map(|x| (x.id, x))
             .collect::<HashMap<_, _>>();
 
-        let active_keyset_id = self.fetch_active_keyset().await?.id;
+        let active_keyset_id = self.fetch_active_keyset(None).await?.id;
         let fee_and_amounts = (
             keysets
                 .get(&active_keyset_id)
@@ -467,7 +505,9 @@ impl AuthWallet {
             .collect::<Result<Vec<ProofInfo>, _>>()?;
 
         // Add new proofs to store
-        self.localstore.update_proofs(proof_infos, vec![]).await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        tx.update_proofs(proof_infos, vec![]).await?;
+        tx.commit().await?;
 
         Ok(proofs)
     }
@@ -476,7 +516,7 @@ impl AuthWallet {
     #[instrument(skip(self))]
     pub async fn total_blind_auth_balance(&self) -> Result<Amount, Error> {
         Ok(Amount::from(
-            self.get_unspent_auth_proofs().await?.len() as u64
+            self.get_unspent_auth_proofs(None).await?.len() as u64,
         ))
     }
 }

crates/cdk/src/wallet/auth/mod.rs

@@ -30,7 +30,7 @@ impl Wallet {
             .await
             .as_ref()
             .ok_or(Error::AuthSettingsUndefined)?
-            .get_unspent_auth_proofs()
+            .get_unspent_auth_proofs(None)
             .await
     }
 

crates/cdk/src/wallet/balance.rs

@@ -23,7 +23,7 @@ impl Wallet {
     /// Total pending balance
     #[instrument(skip(self))]
     pub async fn total_pending_balance(&self) -> Result<Amount, Error> {
-        Ok(self.get_pending_proofs().await?.total_amount()?)
+        Ok(self.get_pending_proofs(None).await?.total_amount()?)
     }
 
     /// Total reserved balance

crates/cdk/src/wallet/issue/issue_bolt11.rs

@@ -52,7 +52,9 @@ impl Wallet {
         let mint_url = self.mint_url.clone();
         let unit = self.unit.clone();
 
-        self.refresh_keysets().await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        self.refresh_keysets(Some(&mut tx)).await?;
 
         // If we have a description, we check that the mint supports it.
         if description.is_some() {
@@ -94,7 +96,8 @@ impl Wallet {
             Some(secret_key),
         );
 
-        self.localstore.add_mint_quote(quote.clone()).await?;
+        tx.add_mint_quote(quote.clone()).await?;
+        tx.commit().await?;
 
         Ok(quote)
     }
@@ -107,12 +110,15 @@ impl Wallet {
     ) -> Result<MintQuoteBolt11Response<String>, Error> {
         let response = self.client.get_mint_quote_status(quote_id).await?;
 
-        match self.localstore.get_mint_quote(quote_id).await? {
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        match tx.get_mint_quote(quote_id).await? {
             Some(quote) => {
                 let mut quote = quote;
 
                 quote.state = response.state;
-                self.localstore.add_mint_quote(quote).await?;
+                tx.add_mint_quote(quote).await?;
+                tx.commit().await?;
             }
             None => {
                 tracing::info!("Quote mint {} unknown", quote_id);
@@ -137,7 +143,9 @@ impl Wallet {
                     .await?;
                 total_amount += proofs.total_amount()?;
             } else if mint_quote.expiry.le(&unix_time()) {
-                self.localstore.remove_mint_quote(&mint_quote.id).await?;
+                let mut tx = self.localstore.begin_db_transaction().await?;
+                tx.remove_mint_quote(&mint_quote.id).await?;
+                tx.commit().await?;
             }
         }
         Ok(total_amount)
@@ -196,10 +204,11 @@ impl Wallet {
         amount_split_target: SplitTarget,
         spending_conditions: Option<SpendingConditions>,
     ) -> Result<Proofs, Error> {
-        self.refresh_keysets().await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        self.refresh_keysets(Some(&mut tx)).await?;
 
-        let quote_info = self
-            .localstore
+        let quote_info = tx
             .get_mint_quote(quote_id)
             .await?
             .ok_or(Error::UnknownQuote)?;
@@ -221,7 +230,7 @@ impl Wallet {
             tracing::warn!("Attempting to mint with expired quote.");
         }
 
-        let active_keyset_id = self.fetch_active_keyset().await?.id;
+        let active_keyset_id = self.fetch_active_keyset(Some(&mut tx)).await?.id;
         let fee_and_amounts = self
             .get_keyset_fees_and_amounts_by_id(active_keyset_id)
             .await?;
@@ -247,8 +256,7 @@ impl Wallet {
                 );
 
                 // Atomically get the counter range we need
-                let new_counter = self
-                    .localstore
+                let new_counter = tx
                     .increment_keyset_counter(&active_keyset_id, num_secrets)
                     .await?;
 
@@ -277,12 +285,14 @@ impl Wallet {
 
         let mint_res = self.client.post_mint(request).await?;
 
-        let keys = self.load_keyset_keys(active_keyset_id).await?;
+        let keys = self
+            .load_keyset_keys(active_keyset_id, Some(&mut tx))
+            .await?;
 
         // Verify the signature DLEQ is valid
         {
             for (sig, premint) in mint_res.signatures.iter().zip(&premint_secrets.secrets) {
-                let keys = self.load_keyset_keys(sig.keyset_id).await?;
+                let keys = self.load_keyset_keys(sig.keyset_id, Some(&mut tx)).await?;
                 let key = keys.amount_key(sig.amount).ok_or(Error::AmountKey)?;
                 match sig.verify_dleq(key, premint.blinded_message.blinded_secret) {
                     Ok(_) | Err(nut12::Error::MissingDleqProof) => (),
@@ -299,7 +309,7 @@ impl Wallet {
         )?;
 
         // Remove filled quote from store
-        self.localstore.remove_mint_quote(&quote_info.id).await?;
+        tx.remove_mint_quote(&quote_info.id).await?;
 
         let proof_infos = proofs
             .iter()
@@ -314,25 +324,26 @@ impl Wallet {
             .collect::<Result<Vec<ProofInfo>, _>>()?;
 
         // Add new proofs to store
-        self.localstore.update_proofs(proof_infos, vec![]).await?;
+        tx.update_proofs(proof_infos, vec![]).await?;
 
         // Add transaction to store
-        self.localstore
-            .add_transaction(Transaction {
-                mint_url: self.mint_url.clone(),
-                direction: TransactionDirection::Incoming,
-                amount: proofs.total_amount()?,
-                fee: Amount::ZERO,
-                unit: self.unit.clone(),
-                ys: proofs.ys()?,
-                timestamp: unix_time,
-                memo: None,
-                metadata: HashMap::new(),
-                quote_id: Some(quote_id.to_string()),
-                payment_request: Some(quote_info.request),
-                payment_proof: None,
-            })
-            .await?;
+        tx.add_transaction(Transaction {
+            mint_url: self.mint_url.clone(),
+            direction: TransactionDirection::Incoming,
+            amount: proofs.total_amount()?,
+            fee: Amount::ZERO,
+            unit: self.unit.clone(),
+            ys: proofs.ys()?,
+            timestamp: unix_time,
+            memo: None,
+            metadata: HashMap::new(),
+            quote_id: Some(quote_id.to_string()),
+            payment_request: Some(quote_info.request),
+            payment_proof: None,
+        })
+        .await?;
+
+        tx.commit().await?;
 
         Ok(proofs)
     }

crates/cdk/src/wallet/issue/issue_bolt12.rs

@@ -15,7 +15,7 @@ use crate::nuts::{
 };
 use crate::types::ProofInfo;
 use crate::util::unix_time;
-use crate::wallet::MintQuote;
+use crate::wallet::{MintQuote, Tx};
 use crate::{Amount, Error, Wallet};
 
 impl Wallet {
@@ -29,7 +29,9 @@ impl Wallet {
         let mint_url = self.mint_url.clone();
         let unit = &self.unit;
 
-        self.refresh_keysets().await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        self.refresh_keysets(Some(&mut tx)).await?;
 
         // If we have a description, we check that the mint supports it.
         if description.is_some() {
@@ -71,7 +73,8 @@ impl Wallet {
             Some(secret_key),
         );
 
-        self.localstore.add_mint_quote(quote.clone()).await?;
+        tx.add_mint_quote(quote.clone()).await?;
+        tx.commit().await?;
 
         Ok(quote)
     }
@@ -85,9 +88,11 @@ impl Wallet {
         amount_split_target: SplitTarget,
         spending_conditions: Option<SpendingConditions>,
     ) -> Result<Proofs, Error> {
-        self.refresh_keysets().await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        self.refresh_keysets(Some(&mut tx)).await?;
 
-        let quote_info = self.localstore.get_mint_quote(quote_id).await?;
+        let quote_info = tx.get_mint_quote(quote_id).await?;
 
         let quote_info = if let Some(quote) = quote_info {
             if quote.expiry.le(&unix_time()) && quote.expiry.ne(&0) {
@@ -99,7 +104,7 @@ impl Wallet {
             return Err(Error::UnknownQuote);
         };
 
-        let active_keyset_id = self.fetch_active_keyset().await?.id;
+        let active_keyset_id = self.fetch_active_keyset(Some(&mut tx)).await?.id;
         let fee_and_amounts = self
             .get_keyset_fees_and_amounts_by_id(active_keyset_id)
             .await?;
@@ -109,7 +114,7 @@ impl Wallet {
             None => {
                 // If an amount it not supplied with check the status of the quote
                 // The mint will tell us how much can be minted
-                let state = self.mint_bolt12_quote_state(quote_id).await?;
+                let state = self.mint_bolt12_quote_state(quote_id, &mut tx).await?;
 
                 state.amount_paid - state.amount_issued
             }
@@ -140,8 +145,7 @@ impl Wallet {
                 );
 
                 // Atomically get the counter range we need
-                let new_counter = self
-                    .localstore
+                let new_counter = tx
                     .increment_keyset_counter(&active_keyset_id, num_secrets)
                     .await?;
 
@@ -173,12 +177,14 @@ impl Wallet {
 
         let mint_res = self.client.post_mint(request).await?;
 
-        let keys = self.load_keyset_keys(active_keyset_id).await?;
+        let keys = self
+            .load_keyset_keys(active_keyset_id, Some(&mut tx))
+            .await?;
 
         // Verify the signature DLEQ is valid
         {
             for (sig, premint) in mint_res.signatures.iter().zip(&premint_secrets.secrets) {
-                let keys = self.load_keyset_keys(sig.keyset_id).await?;
+                let keys = self.load_keyset_keys(sig.keyset_id, Some(&mut tx)).await?;
                 let key = keys.amount_key(sig.amount).ok_or(Error::AmountKey)?;
                 match sig.verify_dleq(key, premint.blinded_message.blinded_secret) {
                     Ok(_) | Err(nut12::Error::MissingDleqProof) => (),
@@ -202,7 +208,7 @@ impl Wallet {
             .ok_or(Error::UnpaidQuote)?;
         quote_info.amount_issued += proofs.total_amount()?;
 
-        self.localstore.add_mint_quote(quote_info.clone()).await?;
+        tx.add_mint_quote(quote_info.clone()).await?;
 
         let proof_infos = proofs
             .iter()
@@ -217,44 +223,46 @@ impl Wallet {
             .collect::<Result<Vec<ProofInfo>, _>>()?;
 
         // Add new proofs to store
-        self.localstore.update_proofs(proof_infos, vec![]).await?;
+        tx.update_proofs(proof_infos, vec![]).await?;
 
         // Add transaction to store
-        self.localstore
-            .add_transaction(Transaction {
-                mint_url: self.mint_url.clone(),
-                direction: TransactionDirection::Incoming,
-                amount: proofs.total_amount()?,
-                fee: Amount::ZERO,
-                unit: self.unit.clone(),
-                ys: proofs.ys()?,
-                timestamp: unix_time(),
-                memo: None,
-                metadata: HashMap::new(),
-                quote_id: Some(quote_id.to_string()),
-                payment_request: Some(quote_info.request),
-                payment_proof: None,
-            })
-            .await?;
+        tx.add_transaction(Transaction {
+            mint_url: self.mint_url.clone(),
+            direction: TransactionDirection::Incoming,
+            amount: proofs.total_amount()?,
+            fee: Amount::ZERO,
+            unit: self.unit.clone(),
+            ys: proofs.ys()?,
+            timestamp: unix_time(),
+            memo: None,
+            metadata: HashMap::new(),
+            quote_id: Some(quote_id.to_string()),
+            payment_request: Some(quote_info.request),
+            payment_proof: None,
+        })
+        .await?;
+
+        tx.commit().await?;
 
         Ok(proofs)
     }
 
     /// Check mint quote status
-    #[instrument(skip(self, quote_id))]
+    #[instrument(skip(self, quote_id, tx))]
     pub async fn mint_bolt12_quote_state(
         &self,
         quote_id: &str,
+        tx: &mut Tx<'_, '_>,
     ) -> Result<MintQuoteBolt12Response<String>, Error> {
         let response = self.client.get_mint_quote_bolt12_status(quote_id).await?;
 
-        match self.localstore.get_mint_quote(quote_id).await? {
+        match tx.get_mint_quote(quote_id).await? {
             Some(quote) => {
                 let mut quote = quote;
                 quote.amount_issued = response.amount_issued;
                 quote.amount_paid = response.amount_paid;
 
-                self.localstore.add_mint_quote(quote).await?;
+                tx.add_mint_quote(quote).await?;
             }
             None => {
                 tracing::info!("Quote mint {} unknown", quote_id);

crates/cdk/src/wallet/keysets.rs

@@ -7,13 +7,19 @@ use tracing::instrument;
 use crate::nuts::{Id, KeySetInfo, Keys};
 use crate::{Error, Wallet};
 
+use super::Tx;
+
 impl Wallet {
     /// Load keys for mint keyset
     ///
     /// Returns keys from local database if they are already stored.
     /// If keys are not found locally, goes online to query the mint for the keyset and stores the [`Keys`] in local database.
-    #[instrument(skip(self))]
-    pub async fn load_keyset_keys(&self, keyset_id: Id) -> Result<Keys, Error> {
+    #[instrument(skip(self, tx))]
+    pub async fn load_keyset_keys(
+        &self,
+        keyset_id: Id,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Keys, Error> {
         let keys = if let Some(keys) = self.localstore.get_keys(&keyset_id).await? {
             keys
         } else {
@@ -27,7 +33,13 @@ impl Wallet {
 
             keys.verify_id()?;
 
-            self.localstore.add_keys(keys.clone()).await?;
+            if let Some(tx) = tx {
+                tx.add_keys(keys.clone()).await?;
+            } else {
+                let mut tx = self.localstore.begin_db_transaction().await?;
+                tx.add_keys(keys.clone()).await?;
+                tx.commit().await?;
+            }
 
             keys.keys
         };
@@ -51,7 +63,7 @@ impl Wallet {
             Some(keysets_info) => Ok(keysets_info),
             None => {
                 // If we don't have any keysets, fetch them from the mint
-                let keysets = self.refresh_keysets().await?;
+                let keysets = self.refresh_keysets(None).await?;
                 Ok(keysets)
             }
         }
@@ -80,26 +92,37 @@ impl Wallet {
     /// It updates the local database with the fetched keysets and ensures we have keys
     /// for all active keysets. This is used when operations need the most up-to-date
     /// keyset information and are willing to go online.
-    #[instrument(skip(self))]
-    pub async fn refresh_keysets(&self) -> Result<KeySetInfos, Error> {
+    #[instrument(skip(self, tx))]
+    pub async fn refresh_keysets(&self, tx: Option<&mut Tx<'_, '_>>) -> Result<KeySetInfos, Error> {
         tracing::debug!("Refreshing keysets and ensuring we have keys");
-        let _ = self.fetch_mint_info().await?;
+        let _ = self.fetch_mint_info(None).await?;
 
         // Fetch all current keysets from mint
         let keysets_response = self.client.get_mint_keysets().await?;
         let all_keysets = keysets_response.keysets;
 
         // Update local storage with keyset info
-        self.localstore
-            .add_mint_keysets(self.mint_url.clone(), all_keysets.clone())
-            .await?;
+        let mut tx = tx;
+        if let Some(tx) = tx.as_mut() {
+            tx.add_mint_keysets(self.mint_url.clone(), all_keysets.clone())
+                .await?;
+        } else {
+            let mut tx = self.localstore.begin_db_transaction().await?;
+            tx.add_mint_keysets(self.mint_url.clone(), all_keysets.clone())
+                .await?;
+            tx.commit().await?;
+        }
 
         // Filter for active keysets matching our unit
         let keysets: KeySetInfos = all_keysets.unit(self.unit.clone()).cloned().collect();
 
         // Ensure we have keys for all active keysets
         for keyset in &keysets {
-            self.load_keyset_keys(keyset.id).await?;
+            if let Some(tx) = tx.as_mut() {
+                self.load_keyset_keys(keyset.id, Some(*tx)).await?;
+            } else {
+                self.load_keyset_keys(keyset.id, None).await?;
+            }
         }
 
         Ok(keysets)
@@ -110,9 +133,12 @@ impl Wallet {
     /// This method always goes online to refresh keysets from the mint and then returns
     /// the active keyset with the minimum input fees. Use this when you need the most
     /// up-to-date keyset information for operations.
-    #[instrument(skip(self))]
-    pub async fn fetch_active_keyset(&self) -> Result<KeySetInfo, Error> {
-        self.refresh_keysets()
+    #[instrument(skip(self, tx))]
+    pub async fn fetch_active_keyset(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<KeySetInfo, Error> {
+        self.refresh_keysets(tx)
             .await?
             .active()
             .min_by_key(|k| k.input_fee_ppk)
@@ -158,7 +184,7 @@ impl Wallet {
                 keyset.id,
                 (
                     keyset.input_fee_ppk,
-                    self.load_keyset_keys(keyset.id)
+                    self.load_keyset_keys(keyset.id, None)
                         .await?
                         .iter()
                         .map(|(amount, _)| amount.to_u64())

crates/cdk/src/wallet/melt/melt_bolt11.rs

@@ -15,7 +15,7 @@ use crate::nuts::{
 };
 use crate::types::{Melted, ProofInfo};
 use crate::util::unix_time;
-use crate::wallet::MeltQuote;
+use crate::wallet::{MeltQuote, Tx};
 use crate::{ensure_cdk, Amount, Error, Wallet};
 
 impl Wallet {
@@ -49,7 +49,7 @@ impl Wallet {
         request: String,
         options: Option<MeltOptions>,
     ) -> Result<MeltQuote, Error> {
-        self.refresh_keysets().await?;
+        self.refresh_keysets(None).await?;
 
         let invoice = Bolt11Invoice::from_str(&request)?;
 
@@ -91,7 +91,9 @@ impl Wallet {
             payment_method: PaymentMethod::Bolt11,
         };
 
-        self.localstore.add_melt_quote(quote.clone()).await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        tx.add_melt_quote(quote.clone()).await?;
+        tx.commit().await?;
 
         Ok(quote)
     }
@@ -103,50 +105,54 @@ impl Wallet {
         quote_id: &str,
     ) -> Result<MeltQuoteBolt11Response<String>, Error> {
         let response = self.client.get_melt_quote_status(quote_id).await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
 
-        match self.localstore.get_melt_quote(quote_id).await? {
+        match tx.get_melt_quote(quote_id).await? {
             Some(quote) => {
                 let mut quote = quote;
 
                 if let Err(e) = self
-                    .add_transaction_for_pending_melt(&quote, &response)
+                    .add_transaction_for_pending_melt(&quote, &response, &mut tx)
                     .await
                 {
                     tracing::error!("Failed to add transaction for pending melt: {}", e);
                 }
 
                 quote.state = response.state;
-                self.localstore.add_melt_quote(quote).await?;
+                tx.add_melt_quote(quote).await?;
             }
             None => {
                 tracing::info!("Quote melt {} unknown", quote_id);
             }
         }
 
-        Ok(response)
-    }
+        tx.commit().await?;
 
-    /// Melt specific proofs
-    #[instrument(skip(self, proofs))]
-    pub async fn melt_proofs(&self, quote_id: &str, proofs: Proofs) -> Result<Melted, Error> {
-        self.melt_proofs_with_metadata(quote_id, proofs, HashMap::new())
-            .await
+        Ok(response)
     }
 
     /// Melt specific proofs
-    #[instrument(skip(self, proofs))]
+    #[instrument(skip(self, proofs, tx))]
     pub async fn melt_proofs_with_metadata(
         &self,
         quote_id: &str,
         proofs: Proofs,
         metadata: HashMap<String, String>,
+        tx: &mut Tx<'_, '_>,
     ) -> Result<Melted, Error> {
-        let quote_info = self
-            .localstore
+        let quote_info = tx
             .get_melt_quote(quote_id)
             .await?
             .ok_or(Error::UnknownQuote)?;
 
+        let active_keyset_id = self.fetch_active_keyset(Some(tx)).await?.id;
+
+        let active_keys = self
+            .localstore
+            .get_keys(&active_keyset_id)
+            .await?
+            .ok_or(Error::NoActiveKeyset)?;
+
         ensure_cdk!(
             quote_info.expiry.gt(&unix_time()),
             Error::ExpiredQuote(quote_info.expiry, unix_time())
@@ -158,11 +164,7 @@ impl Wallet {
         }
 
         let ys = proofs.ys()?;
-        self.localstore
-            .update_proofs_state(ys, State::Pending)
-            .await?;
-
-        let active_keyset_id = self.fetch_active_keyset().await?.id;
+        tx.update_proofs_state(ys, State::Pending).await?;
 
         let change_amount = proofs_total - quote_info.amount;
 
@@ -181,8 +183,7 @@ impl Wallet {
             );
 
             // Atomically get the counter range we need
-            let new_counter = self
-                .localstore
+            let new_counter = tx
                 .increment_keyset_counter(&active_keyset_id, num_secrets)
                 .await?;
 
@@ -211,18 +212,12 @@ impl Wallet {
                 tracing::error!("Could not melt: {}", err);
                 tracing::info!("Checking status of input proofs.");
 
-                self.reclaim_unspent(proofs).await?;
+                self.reclaim_unspent(proofs, tx).await?;
 
                 return Err(err);
             }
         };
 
-        let active_keys = self
-            .localstore
-            .get_keys(&active_keyset_id)
-            .await?
-            .ok_or(Error::NoActiveKeyset)?;
-
         let change_proofs = match melt_response.change {
             Some(change) => {
                 let num_change_proof = change.len();
@@ -280,30 +275,27 @@ impl Wallet {
             None => Vec::new(),
         };
 
-        self.localstore.remove_melt_quote(&quote_info.id).await?;
+        tx.remove_melt_quote(&quote_info.id).await?;
 
         let deleted_ys = proofs.ys()?;
-        self.localstore
-            .update_proofs(change_proof_infos, deleted_ys)
-            .await?;
+        tx.update_proofs(change_proof_infos, deleted_ys).await?;
 
         // Add transaction to store
-        self.localstore
-            .add_transaction(Transaction {
-                mint_url: self.mint_url.clone(),
-                direction: TransactionDirection::Outgoing,
-                amount: melted.amount,
-                fee: melted.fee_paid,
-                unit: self.unit.clone(),
-                ys: proofs.ys()?,
-                timestamp: unix_time(),
-                memo: None,
-                metadata,
-                quote_id: Some(quote_id.to_string()),
-                payment_request: Some(quote_info.request),
-                payment_proof: payment_preimage,
-            })
-            .await?;
+        tx.add_transaction(Transaction {
+            mint_url: self.mint_url.clone(),
+            direction: TransactionDirection::Outgoing,
+            amount: melted.amount,
+            fee: melted.fee_paid,
+            unit: self.unit.clone(),
+            ys: proofs.ys()?,
+            timestamp: unix_time(),
+            memo: None,
+            metadata,
+            quote_id: Some(quote_id.to_string()),
+            payment_request: Some(quote_info.request),
+            payment_proof: payment_preimage,
+        })
+        .await?;
 
         Ok(melted)
     }
@@ -374,8 +366,8 @@ impl Wallet {
         quote_id: &str,
         metadata: HashMap<String, String>,
     ) -> Result<Melted, Error> {
-        let quote_info = self
-            .localstore
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        let quote_info = tx
             .get_melt_quote(quote_id)
             .await?
             .ok_or(Error::UnknownQuote)?;
@@ -390,7 +382,7 @@ impl Wallet {
         let available_proofs = self.get_unspent_proofs().await?;
 
         let active_keyset_ids = self
-            .refresh_keysets()
+            .refresh_keysets(Some(&mut tx))
             .await?
             .into_iter()
             .map(|k| k.id)
@@ -422,7 +414,12 @@ impl Wallet {
             input_proofs.extend_from_slice(&new_proofs);
         }
 
-        self.melt_proofs_with_metadata(quote_id, input_proofs, metadata)
-            .await
+        let melted = self
+            .melt_proofs_with_metadata(quote_id, input_proofs, metadata, &mut tx)
+            .await?;
+
+        tx.commit().await?;
+
+        Ok(melted)
     }
 }

crates/cdk/src/wallet/melt/melt_bolt12.rs

@@ -61,7 +61,9 @@ impl Wallet {
             payment_method: PaymentMethod::Bolt12,
         };
 
-        self.localstore.add_melt_quote(quote.clone()).await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        tx.add_melt_quote(quote.clone()).await?;
+        tx.commit().await?;
 
         Ok(quote)
     }
@@ -74,25 +76,29 @@ impl Wallet {
     ) -> Result<MeltQuoteBolt11Response<String>, Error> {
         let response = self.client.get_melt_bolt12_quote_status(quote_id).await?;
 
-        match self.localstore.get_melt_quote(quote_id).await? {
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        match tx.get_melt_quote(quote_id).await? {
             Some(quote) => {
                 let mut quote = quote;
 
                 if let Err(e) = self
-                    .add_transaction_for_pending_melt(&quote, &response)
+                    .add_transaction_for_pending_melt(&quote, &response, &mut tx)
                     .await
                 {
                     tracing::error!("Failed to add transaction for pending melt: {}", e);
                 }
 
                 quote.state = response.state;
-                self.localstore.add_melt_quote(quote).await?;
+                tx.add_melt_quote(quote).await?;
             }
             None => {
                 tracing::info!("Quote melt {} unknown", quote_id);
             }
         }
 
+        tx.commit().await?;
+
         Ok(response)
     }
 }

crates/cdk/src/wallet/melt/mod.rs

@@ -7,6 +7,8 @@ use tracing::instrument;
 
 use crate::Wallet;
 
+use super::Tx;
+
 #[cfg(all(feature = "bip353", not(target_arch = "wasm32")))]
 mod melt_bip353;
 mod melt_bolt11;
@@ -48,6 +50,7 @@ impl Wallet {
         &self,
         quote: &MeltQuote,
         response: &MeltQuoteBolt11Response<String>,
+        tx: &mut Tx<'_, '_>,
     ) -> Result<(), Error> {
         if quote.state != response.state {
             tracing::info!(
@@ -57,29 +60,28 @@ impl Wallet {
                 response.state
             );
             if response.state == MeltQuoteState::Paid {
-                let pending_proofs = self.get_pending_proofs().await?;
+                let pending_proofs = self.get_pending_proofs(Some(tx)).await?;
                 let proofs_total = pending_proofs.total_amount().unwrap_or_default();
                 let change_total = response.change_amount().unwrap_or_default();
 
-                self.localstore
-                    .add_transaction(Transaction {
-                        mint_url: self.mint_url.clone(),
-                        direction: TransactionDirection::Outgoing,
-                        amount: response.amount,
-                        fee: proofs_total
-                            .checked_sub(response.amount)
-                            .and_then(|amt| amt.checked_sub(change_total))
-                            .unwrap_or_default(),
-                        unit: quote.unit.clone(),
-                        ys: pending_proofs.ys()?,
-                        timestamp: unix_time(),
-                        memo: None,
-                        metadata: HashMap::new(),
-                        quote_id: Some(quote.id.clone()),
-                        payment_request: Some(quote.request.clone()),
-                        payment_proof: response.payment_preimage.clone(),
-                    })
-                    .await?;
+                tx.add_transaction(Transaction {
+                    mint_url: self.mint_url.clone(),
+                    direction: TransactionDirection::Outgoing,
+                    amount: response.amount,
+                    fee: proofs_total
+                        .checked_sub(response.amount)
+                        .and_then(|amt| amt.checked_sub(change_total))
+                        .unwrap_or_default(),
+                    unit: quote.unit.clone(),
+                    ys: pending_proofs.ys()?,
+                    timestamp: unix_time(),
+                    memo: None,
+                    metadata: HashMap::new(),
+                    quote_id: Some(quote.id.clone()),
+                    payment_request: Some(quote.request.clone()),
+                    payment_proof: response.payment_preimage.clone(),
+                })
+                .await?;
             }
         }
         Ok(())

crates/cdk/src/wallet/mod.rs

@@ -5,7 +5,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 
 use cdk_common::amount::FeeAndAmounts;
-use cdk_common::database::{self, WalletDatabase};
+use cdk_common::database::{self, WalletDatabase, WalletDatabaseTransaction};
 use cdk_common::subscription::WalletParams;
 use getrandom::getrandom;
 use subscription::{ActiveSubscription, SubscriptionManager};
@@ -71,6 +71,9 @@ pub use types::{MeltQuote, MintQuote, SendKind};
 
 use crate::nuts::nut00::ProofsMethods;
 
+pub(crate) type Tx<'a, 'b> =
+    Box<dyn WalletDatabaseTransaction<'a, database::Error> + Send + Sync + 'b>;
+
 /// CDK Wallet
 ///
 /// The CDK [`Wallet`] is a high level cashu wallet.
@@ -249,9 +252,10 @@ impl Wallet {
     #[instrument(skip(self))]
     pub async fn update_mint_url(&mut self, new_mint_url: MintUrl) -> Result<(), Error> {
         // Update the mint URL in the wallet DB
-        self.localstore
-            .update_mint_url(self.mint_url.clone(), new_mint_url.clone())
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        tx.update_mint_url(self.mint_url.clone(), new_mint_url.clone())
             .await?;
+        tx.commit().await?;
 
         // Update the mint URL in the wallet struct field
         self.mint_url = new_mint_url;
@@ -260,8 +264,11 @@ impl Wallet {
     }
 
     /// Query mint for current mint information
-    #[instrument(skip(self))]
-    pub async fn fetch_mint_info(&self) -> Result<Option<MintInfo>, Error> {
+    #[instrument(skip(self, tx))]
+    pub async fn fetch_mint_info(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Option<MintInfo>, Error> {
         match self.client.get_mint_info().await {
             Ok(mint_info) => {
                 // If mint provides time make sure it is accurate
@@ -314,9 +321,15 @@ impl Wallet {
                     }
                 }
 
-                self.localstore
-                    .add_mint(self.mint_url.clone(), Some(mint_info.clone()))
-                    .await?;
+                if let Some(tx) = tx {
+                    tx.add_mint(self.mint_url.clone(), Some(mint_info.clone()))
+                        .await?;
+                } else {
+                    let mut tx = self.localstore.begin_db_transaction().await?;
+                    tx.add_mint(self.mint_url.clone(), Some(mint_info.clone()))
+                        .await?;
+                    tx.commit().await?;
+                };
 
                 tracing::trace!("Mint info updated for {}", self.mint_url);
 
@@ -399,7 +412,7 @@ impl Wallet {
             .await?
             .is_none()
         {
-            self.fetch_mint_info().await?;
+            self.fetch_mint_info(None).await?;
         }
 
         let keysets = self.load_mint_keysets().await?;
@@ -407,7 +420,8 @@ impl Wallet {
         let mut restored_value = Amount::ZERO;
 
         for keyset in keysets {
-            let keys = self.load_keyset_keys(keyset.id).await?;
+            let mut tx = self.localstore.begin_db_transaction().await?;
+            let keys = self.load_keyset_keys(keyset.id, Some(&mut tx)).await?;
             let mut empty_batch = 0;
             let mut start_counter = 0;
 
@@ -458,11 +472,12 @@ impl Wallet {
 
                 tracing::debug!("Restored {} proofs", proofs.len());
 
-                self.localstore
-                    .increment_keyset_counter(&keyset.id, proofs.len() as u32)
+                tx.increment_keyset_counter(&keyset.id, proofs.len() as u32)
                     .await?;
 
-                let states = self.check_proofs_spent(proofs.clone()).await?;
+                let states = self
+                    .check_proofs_spent(proofs.clone(), Some(&mut tx))
+                    .await?;
 
                 let unspent_proofs: Vec<Proof> = proofs
                     .iter()
@@ -486,13 +501,13 @@ impl Wallet {
                     })
                     .collect::<Result<Vec<ProofInfo>, _>>()?;
 
-                self.localstore
-                    .update_proofs(unspent_proofs, vec![])
-                    .await?;
+                tx.update_proofs(unspent_proofs, vec![]).await?;
 
                 empty_batch = 0;
                 start_counter += 100;
             }
+
+            tx.commit().await?;
         }
         Ok(restored_value)
     }
@@ -656,7 +671,7 @@ impl Wallet {
             let mint_pubkey = match keys_cache.get(&proof.keyset_id) {
                 Some(keys) => keys.amount_key(proof.amount),
                 None => {
-                    let keys = self.load_keyset_keys(proof.keyset_id).await?;
+                    let keys = self.load_keyset_keys(proof.keyset_id, None).await?;
 
                     let key = keys.amount_key(proof.amount);
                     keys_cache.insert(proof.keyset_id, keys);

crates/cdk/src/wallet/multi_mint_wallet.rs

@@ -1173,8 +1173,15 @@ impl MultiMintWallet {
 
         let mut amount_received = Amount::ZERO;
 
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
         match wallet
-            .receive_proofs(proofs, opts.receive_options, token_data.memo().clone())
+            .receive_proofs(
+                &mut tx,
+                proofs,
+                opts.receive_options,
+                token_data.memo().clone(),
+            )
             .await
         {
             Ok(amount) => {
@@ -1190,6 +1197,8 @@ impl MultiMintWallet {
             }
         }
 
+        tx.commit().await?;
+
         drop(wallets);
 
         // If we should transfer to a trusted mint, do so now
@@ -1680,7 +1689,7 @@ impl MultiMintWallet {
             mint_url: mint_url.to_string(),
         })?;
 
-        wallet.fetch_mint_info().await
+        wallet.fetch_mint_info(None).await
     }
 }
 

crates/cdk/src/wallet/proofs.rs

@@ -14,30 +14,36 @@ use crate::nuts::{
 use crate::types::ProofInfo;
 use crate::{ensure_cdk, Amount, Error, Wallet};
 
+use super::Tx;
+
 impl Wallet {
     /// Get unspent proofs for mint
     #[instrument(skip(self))]
     pub async fn get_unspent_proofs(&self) -> Result<Proofs, Error> {
-        self.get_proofs_with(Some(vec![State::Unspent]), None).await
+        self.get_proofs_with(Some(vec![State::Unspent]), None, None)
+            .await
     }
 
     /// Get pending [`Proofs`]
-    #[instrument(skip(self))]
-    pub async fn get_pending_proofs(&self) -> Result<Proofs, Error> {
-        self.get_proofs_with(Some(vec![State::Pending]), None).await
+    #[instrument(skip(self, tx))]
+    pub async fn get_pending_proofs(&self, tx: Option<&mut Tx<'_, '_>>) -> Result<Proofs, Error> {
+        self.get_proofs_with(Some(vec![State::Pending]), None, tx)
+            .await
     }
 
     /// Get reserved [`Proofs`]
     #[instrument(skip(self))]
     pub async fn get_reserved_proofs(&self) -> Result<Proofs, Error> {
-        self.get_proofs_with(Some(vec![State::Reserved]), None)
+        self.get_proofs_with(Some(vec![State::Reserved]), None, None)
             .await
     }
 
     /// Get pending spent [`Proofs`]
-    #[instrument(skip(self))]
-    pub async fn get_pending_spent_proofs(&self) -> Result<Proofs, Error> {
-        self.get_proofs_with(Some(vec![State::PendingSpent]), None)
+    pub async fn get_pending_spent_proofs(
+        &self,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Proofs, Error> {
+        self.get_proofs_with(Some(vec![State::PendingSpent]), None, tx)
             .await
     }
 
@@ -46,10 +52,10 @@ impl Wallet {
         &self,
         state: Option<Vec<State>>,
         spending_conditions: Option<Vec<SpendingConditions>>,
+        tx: Option<&mut Tx<'_, '_>>,
     ) -> Result<Proofs, Error> {
-        Ok(self
-            .localstore
-            .get_proofs(
+        Ok(if let Some(tx) = tx {
+            tx.get_proofs(
                 Some(self.mint_url.clone()),
                 Some(self.unit.clone()),
                 state,
@@ -58,23 +64,37 @@ impl Wallet {
             .await?
             .into_iter()
             .map(|p| p.proof)
-            .collect())
+            .collect()
+        } else {
+            self.localstore
+                .get_proofs(
+                    Some(self.mint_url.clone()),
+                    Some(self.unit.clone()),
+                    state,
+                    spending_conditions,
+                )
+                .await?
+                .into_iter()
+                .map(|p| p.proof)
+                .collect()
+        })
     }
 
     /// Return proofs to unspent allowing them to be selected and spent
     #[instrument(skip(self))]
     pub async fn unreserve_proofs(&self, ys: Vec<PublicKey>) -> Result<(), Error> {
-        Ok(self
-            .localstore
-            .update_proofs_state(ys, State::Unspent)
-            .await?)
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        tx.update_proofs_state(ys, State::Unspent).await?;
+
+        Ok(tx.commit().await?)
     }
 
     /// Reclaim unspent proofs
     ///
     /// Checks the stats of [`Proofs`] swapping for a new [`Proof`] if unspent
-    #[instrument(skip(self, proofs))]
-    pub async fn reclaim_unspent(&self, proofs: Proofs) -> Result<(), Error> {
+    #[instrument(skip(self, proofs, tx))]
+    pub async fn reclaim_unspent(&self, proofs: Proofs, tx: &mut Tx<'_, '_>) -> Result<(), Error> {
         let proof_ys = proofs.ys()?;
 
         let transaction_id = TransactionId::new(proof_ys.clone());
@@ -94,7 +114,7 @@ impl Wallet {
         self.swap(None, SplitTarget::default(), unspent, None, false)
             .await?;
 
-        match self.localstore.remove_transaction(transaction_id).await {
+        match tx.remove_transaction(transaction_id).await {
             Ok(_) => (),
             Err(e) => {
                 tracing::warn!("Failed to remove transaction: {:?}", e);
@@ -105,8 +125,12 @@ impl Wallet {
     }
 
     /// NUT-07 Check the state of a [`Proof`] with the mint
-    #[instrument(skip(self, proofs))]
-    pub async fn check_proofs_spent(&self, proofs: Proofs) -> Result<Vec<ProofState>, Error> {
+    #[instrument(skip(self, proofs, tx))]
+    pub async fn check_proofs_spent(
+        &self,
+        proofs: Proofs,
+        tx: Option<&mut Tx<'_, '_>>,
+    ) -> Result<Vec<ProofState>, Error> {
         let spendable = self
             .client
             .post_check_state(CheckStateRequest { ys: proofs.ys()? })
@@ -121,7 +145,13 @@ impl Wallet {
             })
             .collect();
 
-        self.localstore.update_proofs(vec![], spent_ys).await?;
+        if let Some(tx) = tx {
+            tx.update_proofs(vec![], spent_ys).await?;
+        } else {
+            let mut tx = self.localstore.begin_db_transaction().await?;
+            tx.update_proofs(vec![], spent_ys).await?;
+            tx.commit().await?;
+        }
 
         Ok(spendable.states)
     }
@@ -131,8 +161,9 @@ impl Wallet {
     pub async fn check_all_pending_proofs(&self) -> Result<Amount, Error> {
         let mut balance = Amount::ZERO;
 
-        let proofs = self
-            .localstore
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
+        let proofs = tx
             .get_proofs(
                 Some(self.mint_url.clone()),
                 Some(self.unit.clone()),
@@ -146,7 +177,10 @@ impl Wallet {
         }
 
         let states = self
-            .check_proofs_spent(proofs.clone().into_iter().map(|p| p.proof).collect())
+            .check_proofs_spent(
+                proofs.clone().into_iter().map(|p| p.proof).collect(),
+                Some(&mut tx),
+            )
             .await?;
 
         // Both `State::Pending` and `State::Unspent` should be included in the pending
@@ -165,15 +199,16 @@ impl Wallet {
 
         let amount = Amount::try_sum(pending_proofs.iter().map(|p| p.proof.amount))?;
 
-        self.localstore
-            .update_proofs(
-                vec![],
-                non_pending_proofs.into_iter().map(|p| p.y).collect(),
-            )
-            .await?;
+        tx.update_proofs(
+            vec![],
+            non_pending_proofs.into_iter().map(|p| p.y).collect(),
+        )
+        .await?;
 
         balance += amount;
 
+        tx.commit().await?;
+
         Ok(balance)
     }
 

crates/cdk/src/wallet/receive.rs

@@ -17,22 +17,25 @@ use crate::types::ProofInfo;
 use crate::util::hex;
 use crate::{ensure_cdk, Amount, Error, Wallet, SECP256K1};
 
+use super::Tx;
+
 impl Wallet {
     /// Receive proofs
     #[instrument(skip_all)]
     pub async fn receive_proofs(
         &self,
+        tx: &mut Tx<'_, '_>,
         proofs: Proofs,
         opts: ReceiveOptions,
         memo: Option<String>,
     ) -> Result<Amount, Error> {
         let mint_url = &self.mint_url;
 
-        self.refresh_keysets().await?;
+        self.refresh_keysets(Some(tx)).await?;
 
-        let active_keyset_id = self.fetch_active_keyset().await?.id;
+        let active_keyset_id = self.fetch_active_keyset(Some(tx)).await?.id;
 
-        let keys = self.load_keyset_keys(active_keyset_id).await?;
+        let keys = self.load_keyset_keys(active_keyset_id, Some(tx)).await?;
 
         let mut proofs = proofs;
 
@@ -60,7 +63,7 @@ impl Wallet {
         for proof in &mut proofs {
             // Verify that proof DLEQ is valid
             if proof.dleq.is_some() {
-                let keys = self.load_keyset_keys(proof.keyset_id).await?;
+                let keys = self.load_keyset_keys(proof.keyset_id, Some(tx)).await?;
                 let key = keys.amount_key(proof.amount).ok_or(Error::AmountKey)?;
                 proof.verify_dleq(key)?;
             }
@@ -112,12 +115,10 @@ impl Wallet {
             .into_iter()
             .map(|p| ProofInfo::new(p, self.mint_url.clone(), State::Pending, self.unit.clone()))
             .collect::<Result<Vec<ProofInfo>, _>>()?;
-        self.localstore
-            .update_proofs(proofs_info.clone(), vec![])
-            .await?;
+        tx.update_proofs(proofs_info.clone(), vec![]).await?;
 
         let mut pre_swap = self
-            .create_swap(None, opts.amount_split_target, proofs, None, false)
+            .create_swap(tx, None, opts.amount_split_target, proofs, None, false)
             .await?;
 
         if sig_flag.eq(&SigFlag::SigAll) {
@@ -138,8 +139,7 @@ impl Wallet {
             &keys,
         )?;
 
-        self.localstore
-            .increment_keyset_counter(&active_keyset_id, recv_proofs.len() as u32)
+        tx.increment_keyset_counter(&active_keyset_id, recv_proofs.len() as u32)
             .await?;
 
         let total_amount = recv_proofs.total_amount()?;
@@ -148,30 +148,29 @@ impl Wallet {
             .into_iter()
             .map(|proof| ProofInfo::new(proof, mint_url.clone(), State::Unspent, self.unit.clone()))
             .collect::<Result<Vec<ProofInfo>, _>>()?;
-        self.localstore
-            .update_proofs(
-                recv_proof_infos,
-                proofs_info.into_iter().map(|p| p.y).collect(),
-            )
-            .await?;
+
+        tx.update_proofs(
+            recv_proof_infos,
+            proofs_info.into_iter().map(|p| p.y).collect(),
+        )
+        .await?;
 
         // Add transaction to store
-        self.localstore
-            .add_transaction(Transaction {
-                mint_url: self.mint_url.clone(),
-                direction: TransactionDirection::Incoming,
-                amount: total_amount,
-                fee: proofs_amount - total_amount,
-                unit: self.unit.clone(),
-                ys: proofs_ys,
-                timestamp: unix_time(),
-                memo,
-                metadata: opts.metadata,
-                quote_id: None,
-                payment_request: None,
-                payment_proof: None,
-            })
-            .await?;
+        tx.add_transaction(Transaction {
+            mint_url: self.mint_url.clone(),
+            direction: TransactionDirection::Incoming,
+            amount: total_amount,
+            fee: proofs_amount - total_amount,
+            unit: self.unit.clone(),
+            ys: proofs_ys,
+            timestamp: unix_time(),
+            memo,
+            metadata: opts.metadata,
+            quote_id: None,
+            payment_request: None,
+            payment_proof: None,
+        })
+        .await?;
 
         Ok(total_amount)
     }
@@ -221,10 +220,14 @@ impl Wallet {
 
         ensure_cdk!(self.mint_url == token.mint_url()?, Error::IncorrectMint);
 
+        let mut tx = self.localstore.begin_db_transaction().await?;
+
         let amount = self
-            .receive_proofs(proofs, opts, token.memo().clone())
+            .receive_proofs(&mut tx, proofs, opts, token.memo().clone())
             .await?;
 
+        tx.commit().await?;
+
         Ok(amount)
     }
 

crates/cdk/src/wallet/send.rs

@@ -33,7 +33,7 @@ impl Wallet {
 
         // If online send check mint for current keysets fees
         if opts.send_kind.is_online() {
-            if let Err(e) = self.refresh_keysets().await {
+            if let Err(e) = self.refresh_keysets(None).await {
                 tracing::error!("Error refreshing keysets: {:?}. Using stored keysets", e);
             }
         }
@@ -46,6 +46,7 @@ impl Wallet {
             .get_proofs_with(
                 Some(vec![State::Unspent]),
                 opts.conditions.clone().map(|c| vec![c]),
+                None,
             )
             .await?;
 
@@ -128,6 +129,7 @@ impl Wallet {
         proofs: Proofs,
         force_swap: bool,
     ) -> Result<PreparedSend, Error> {
+        let wallet = self.clone();
         // Split amount with fee if necessary
         let active_keyset_id = self.get_active_keyset().await?.id;
         let fee_and_amounts = self
@@ -152,10 +154,12 @@ impl Wallet {
         tracing::debug!("Send amounts: {:?}", send_amounts);
         tracing::debug!("Send fee: {:?}", send_fee);
 
+        let mut tx = wallet.localstore.begin_db_transaction().await?;
+
         // Reserve proofs
-        self.localstore
-            .update_proofs_state(proofs.ys()?, State::Reserved)
+        tx.update_proofs_state(proofs.ys()?, State::Reserved)
             .await?;
+        tx.commit().await?;
 
         // Check if proofs are exact send amount (and does not exceed max_proofs)
         let mut exact_proofs = proofs.total_amount()? == amount + send_fee;
@@ -190,7 +194,7 @@ impl Wallet {
 
         // Return prepared send
         Ok(PreparedSend {
-            wallet: self.clone(),
+            wallet,
             amount,
             options: opts,
             proofs_to_swap,
@@ -262,8 +266,10 @@ impl PreparedSend {
         let total_send_fee = self.fee();
         let mut proofs_to_send = self.proofs_to_send;
 
+        let mut tx = self.wallet.localstore.begin_db_transaction().await?;
+
         // Get active keyset ID
-        let active_keyset_id = self.wallet.fetch_active_keyset().await?.id;
+        let active_keyset_id = self.wallet.fetch_active_keyset(Some(&mut tx)).await?.id;
         tracing::debug!("Active keyset ID: {:?}", active_keyset_id);
 
         // Get keyset fees
@@ -311,9 +317,11 @@ impl PreparedSend {
             .get_proofs_with(
                 Some(vec![State::Reserved, State::Unspent]),
                 self.options.conditions.clone().map(|c| vec![c]),
+                None,
             )
             .await?
             .ys()?;
+
         if proofs_to_send
             .ys()?
             .iter()
@@ -328,9 +336,7 @@ impl PreparedSend {
             "Updating proofs state to pending spent: {:?}",
             proofs_to_send.ys()?
         );
-        self.wallet
-            .localstore
-            .update_proofs_state(proofs_to_send.ys()?, State::PendingSpent)
+        tx.update_proofs_state(proofs_to_send.ys()?, State::PendingSpent)
             .await?;
 
         // Include token memo
@@ -338,23 +344,23 @@ impl PreparedSend {
         let memo = send_memo.and_then(|m| if m.include_memo { Some(m.memo) } else { None });
 
         // Add transaction to store
-        self.wallet
-            .localstore
-            .add_transaction(Transaction {
-                mint_url: self.wallet.mint_url.clone(),
-                direction: TransactionDirection::Outgoing,
-                amount: self.amount,
-                fee: total_send_fee,
-                unit: self.wallet.unit.clone(),
-                ys: proofs_to_send.ys()?,
-                timestamp: unix_time(),
-                memo: memo.clone(),
-                metadata: self.options.metadata,
-                quote_id: None,
-                payment_request: None,
-                payment_proof: None,
-            })
-            .await?;
+        tx.add_transaction(Transaction {
+            mint_url: self.wallet.mint_url.clone(),
+            direction: TransactionDirection::Outgoing,
+            amount: self.amount,
+            fee: total_send_fee,
+            unit: self.wallet.unit.clone(),
+            ys: proofs_to_send.ys()?,
+            timestamp: unix_time(),
+            memo: memo.clone(),
+            metadata: self.options.metadata,
+            quote_id: None,
+            payment_request: None,
+            payment_proof: None,
+        })
+        .await?;
+
+        tx.commit().await?;
 
         // Create and return token
         Ok(Token::new(
@@ -368,6 +374,7 @@ impl PreparedSend {
     /// Cancel the prepared send
     pub async fn cancel(self) -> Result<(), Error> {
         tracing::info!("Cancelling prepared send");
+        let mut tx = self.wallet.localstore.begin_db_transaction().await?;
 
         // Double-check proofs state
         let reserved_proofs = self.wallet.get_reserved_proofs().await?.ys()?;
@@ -380,11 +387,11 @@ impl PreparedSend {
             return Err(Error::UnexpectedProofState);
         }
 
-        self.wallet
-            .localstore
-            .update_proofs_state(self.proofs().ys()?, State::Unspent)
+        tx.update_proofs_state(self.proofs().ys()?, State::Unspent)
             .await?;
 
+        tx.commit().await?;
+
         Ok(())
     }
 }

crates/cdk/src/wallet/swap.rs

@@ -10,6 +10,8 @@ use crate::nuts::{
 use crate::types::ProofInfo;
 use crate::{ensure_cdk, Amount, Error, Wallet};
 
+use super::Tx;
+
 impl Wallet {
     /// Swap
     #[instrument(skip(self, input_proofs))]
@@ -21,7 +23,8 @@ impl Wallet {
         spending_conditions: Option<SpendingConditions>,
         include_fees: bool,
     ) -> Result<Option<Proofs>, Error> {
-        self.refresh_keysets().await?;
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        self.refresh_keysets(Some(&mut tx)).await?;
 
         tracing::info!("Swapping");
         let mint_url = &self.mint_url;
@@ -29,6 +32,7 @@ impl Wallet {
 
         let pre_swap = self
             .create_swap(
+                &mut tx,
                 amount,
                 amount_split_target.clone(),
                 input_proofs.clone(),
@@ -134,9 +138,10 @@ impl Wallet {
             .map(|proof| proof.y())
             .collect::<Result<Vec<PublicKey>, _>>()?;
 
-        self.localstore
-            .update_proofs(added_proofs, deleted_ys)
-            .await?;
+        tx.update_proofs(added_proofs, deleted_ys).await?;
+
+        tx.commit().await?;
+
         Ok(send_proofs)
     }
 
@@ -148,8 +153,8 @@ impl Wallet {
         conditions: Option<SpendingConditions>,
         include_fees: bool,
     ) -> Result<Proofs, Error> {
-        let available_proofs = self
-            .localstore
+        let mut tx = self.localstore.begin_db_transaction().await?;
+        let available_proofs = tx
             .get_proofs(
                 Some(self.mint_url.clone()),
                 Some(self.unit.clone()),
@@ -170,7 +175,7 @@ impl Wallet {
         ensure_cdk!(proofs_sum >= amount, Error::InsufficientFunds);
 
         let active_keyset_ids = self
-            .refresh_keysets()
+            .refresh_keysets(Some(&mut tx))
             .await?
             .active()
             .map(|k| k.id)
@@ -185,21 +190,27 @@ impl Wallet {
             true,
         )?;
 
-        self.swap(
-            Some(amount),
-            SplitTarget::default(),
-            proofs,
-            conditions,
-            include_fees,
-        )
-        .await?
-        .ok_or(Error::InsufficientFunds)
+        let to_return = self
+            .swap(
+                Some(amount),
+                SplitTarget::default(),
+                proofs,
+                conditions,
+                include_fees,
+            )
+            .await?
+            .ok_or(Error::InsufficientFunds)?;
+
+        tx.commit().await?;
+
+        Ok(to_return)
     }
 
     /// Create Swap Payload
-    #[instrument(skip(self, proofs))]
+    #[instrument(skip(self, tx, proofs))]
     pub async fn create_swap(
         &self,
+        tx: &mut Tx<'_, '_>,
         amount: Option<Amount>,
         amount_split_target: SplitTarget,
         proofs: Proofs,
@@ -207,15 +218,13 @@ impl Wallet {
         include_fees: bool,
     ) -> Result<PreSwap, Error> {
         tracing::info!("Creating swap");
-        let active_keyset_id = self.fetch_active_keyset().await?.id;
+        let active_keyset_id = self.fetch_active_keyset(Some(tx)).await?.id;
 
         // Desired amount is either amount passed or value of all proof
         let proofs_total = proofs.total_amount()?;
 
         let ys: Vec<PublicKey> = proofs.ys()?;
-        self.localstore
-            .update_proofs_state(ys, State::Reserved)
-            .await?;
+        tx.update_proofs_state(ys, State::Reserved).await?;
 
         let fee = self.get_proofs_fee(&proofs).await?;
 
@@ -297,8 +306,7 @@ impl Wallet {
                 total_secrets_needed
             );
 
-            let new_counter = self
-                .localstore
+            let new_counter = tx
                 .increment_keyset_counter(&active_keyset_id, total_secrets_needed)
                 .await?;
 

crates/cdk/src/wallet/transactions.rs

@@ -41,8 +41,10 @@ impl Wallet {
             return Err(Error::InvalidTransactionDirection);
         }
 
+        let mut db_tx = self.localstore.begin_db_transaction().await?;
+
         let pending_spent_proofs = self
-            .get_pending_spent_proofs()
+            .get_pending_spent_proofs(Some(&mut db_tx))
             .await?
             .into_iter()
             .filter(|p| match p.y() {
@@ -51,7 +53,11 @@ impl Wallet {
             })
             .collect::<Vec<_>>();
 
-        self.reclaim_unspent(pending_spent_proofs).await?;
+        self.reclaim_unspent(pending_spent_proofs, &mut db_tx)
+            .await?;
+
+        db_tx.commit().await?;
+
         Ok(())
     }
 }