Use BDK events in update_payment_store instead of scanning all transactions

Replace the full transaction list scan in `update_payment_store` with
handling of BDK's `WalletEvent` stream during sync. This leverages the
new events in BDK 2.2, reduces redundant work, and prepares the
foundation for reliable RBF/CPFP tracking via `WalletEvent::TxReplaced`

Author: Camillarhi

src/wallet/mod.rs

@@ -12,6 +12,7 @@ use std::sync::{Arc, Mutex};
 
 use bdk_chain::spk_client::{FullScanRequest, SyncRequest};
 use bdk_wallet::descriptor::ExtendedDescriptor;
+use bdk_wallet::event::WalletEvent;
 #[allow(deprecated)]
 use bdk_wallet::SignOptions;
 use bdk_wallet::{Balance, KeychainKind, PersistedWallet, Update};
@@ -49,8 +50,10 @@ use crate::config::Config;
 use crate::fee_estimator::{ConfirmationTarget, FeeEstimator, OnchainFeeEstimator};
 use crate::logger::{log_debug, log_error, log_info, log_trace, LdkLogger, Logger};
 use crate::payment::store::ConfirmationStatus;
-use crate::payment::{PaymentDetails, PaymentDirection, PaymentStatus};
-use crate::types::{Broadcaster, PaymentStore};
+use crate::payment::{
+	PaymentDetails, PaymentDirection, PaymentKind, PaymentStatus, PendingPaymentDetails,
+};
+use crate::types::{Broadcaster, PaymentStore, PendingPaymentStore};
 use crate::Error;
 
 pub(crate) enum OnchainSendAmount {
@@ -71,18 +74,28 @@ pub(crate) struct Wallet {
 	payment_store: Arc<PaymentStore>,
 	config: Arc<Config>,
 	logger: Arc<Logger>,
+	pending_payment_store: Arc<PendingPaymentStore>,
 }
 
 impl Wallet {
 	pub(crate) fn new(
 		wallet: bdk_wallet::PersistedWallet<KVStoreWalletPersister>,
 		wallet_persister: KVStoreWalletPersister, broadcaster: Arc<Broadcaster>,
 		fee_estimator: Arc<OnchainFeeEstimator>, payment_store: Arc<PaymentStore>,
-		config: Arc<Config>, logger: Arc<Logger>,
+		config: Arc<Config>, logger: Arc<Logger>, pending_payment_store: Arc<PendingPaymentStore>,
 	) -> Self {
 		let inner = Mutex::new(wallet);
 		let persister = Mutex::new(wallet_persister);
-		Self { inner, persister, broadcaster, fee_estimator, payment_store, config, logger }
+		Self {
+			inner,
+			persister,
+			broadcaster,
+			fee_estimator,
+			payment_store,
+			config,
+			logger,
+			pending_payment_store,
+		}
 	}
 
 	pub(crate) fn get_full_scan_request(&self) -> FullScanRequest<KeychainKind> {
@@ -114,15 +127,15 @@ impl Wallet {
 
 	pub(crate) fn apply_update(&self, update: impl Into<Update>) -> Result<(), Error> {
 		let mut locked_wallet = self.inner.lock().unwrap();
-		match locked_wallet.apply_update(update) {
-			Ok(()) => {
+		match locked_wallet.apply_update_events(update) {
+			Ok(events) => {
 				let mut locked_persister = self.persister.lock().unwrap();
 				locked_wallet.persist(&mut locked_persister).map_err(|e| {
 					log_error!(self.logger, "Failed to persist wallet: {}", e);
 					Error::PersistenceFailed
 				})?;
 
-				self.update_payment_store(&mut *locked_wallet).map_err(|e| {
+				self.update_payment_store(&mut *locked_wallet, events).map_err(|e| {
 					log_error!(self.logger, "Failed to update payment store: {}", e);
 					Error::PersistenceFailed
 				})?;
@@ -167,42 +180,159 @@ impl Wallet {
 
 	fn update_payment_store<'a>(
 		&self, locked_wallet: &'a mut PersistedWallet<KVStoreWalletPersister>,
+		mut events: Vec<WalletEvent>,
 	) -> Result<(), Error> {
-		for wtx in locked_wallet.transactions() {
-			let id = PaymentId(wtx.tx_node.txid.to_byte_array());
-			let txid = wtx.tx_node.txid;
-			let (payment_status, confirmation_status) = match wtx.chain_position {
-				bdk_chain::ChainPosition::Confirmed { anchor, .. } => {
-					let confirmation_height = anchor.block_id.height;
+		if events.is_empty() {
+			return Ok(());
+		}
+
+		// Sort events to ensure proper sequencing for data consistency:
+		// 1. TXReplaced (0) before TxUnconfirmed (1) - Critical for RBF handling
+		//    When a transaction is replaced via RBF, both events fire. Processing
+		//    TXReplaced first stores the replaced transaction, allowing TxUnconfirmed
+		//    to detect and skip duplicate payment record creation.
+		// 2. TxConfirmed (2) before ChainTipChanged (3) - Ensures height accuracy
+		//    ChainTipChanged updates block height. Processing TxConfirmed first ensures
+		//    it references the correct height for confirmation depth calculations.
+		// 3. Other events follow in deterministic order for predictable processing
+		if events.len() > 1 {
+			events.sort_by_key(|e| match e {
+				WalletEvent::TxReplaced { .. } => 0,
+				WalletEvent::TxUnconfirmed { .. } => 1,
+				WalletEvent::TxConfirmed { .. } => 2,
+				WalletEvent::ChainTipChanged { .. } => 3,
+				WalletEvent::TxDropped { .. } => 4,
+				_ => 5,
+			});
+		}
+
+		for event in events {
+			match event {
+				WalletEvent::TxConfirmed { txid, tx, block_time, .. } => {
 					let cur_height = locked_wallet.latest_checkpoint().height();
+					let confirmation_height = block_time.block_id.height;
 					let payment_status = if cur_height >= confirmation_height + ANTI_REORG_DELAY - 1
 					{
 						PaymentStatus::Succeeded
 					} else {
 						PaymentStatus::Pending
 					};
+
 					let confirmation_status = ConfirmationStatus::Confirmed {
-						block_hash: anchor.block_id.hash,
+						block_hash: block_time.block_id.hash,
 						height: confirmation_height,
-						timestamp: anchor.confirmation_time,
+						timestamp: block_time.confirmation_time,
 					};
-					(payment_status, confirmation_status)
+
+					let payment_id = self
+						.find_payment_by_txid(txid)
+						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
+
+					let payment = self.create_payment_from_tx(
+						locked_wallet,
+						txid,
+						payment_id,
+						&tx,
+						payment_status,
+						confirmation_status,
+					);
+
+					let pending_payment =
+						self.create_pending_payment_from_tx(payment.clone(), Vec::new());
+
+					self.payment_store.insert_or_update(payment)?;
+					self.pending_payment_store.insert_or_update(pending_payment)?;
 				},
-				bdk_chain::ChainPosition::Unconfirmed { .. } => {
-					(PaymentStatus::Pending, ConfirmationStatus::Unconfirmed)
+				WalletEvent::ChainTipChanged { new_tip, .. } => {
+					// Get all payments that are Pending with Confirmed status
+					let pending_payments: Vec<PendingPaymentDetails> =
+						self.pending_payment_store.list_filter(|p| {
+							p.details.status == PaymentStatus::Pending
+								&& matches!(
+									p.details.kind,
+									PaymentKind::Onchain {
+										status: ConfirmationStatus::Confirmed { .. },
+										..
+									}
+								)
+						});
+
+					for mut payment in pending_payments {
+						if let PaymentKind::Onchain {
+							status: ConfirmationStatus::Confirmed { height, .. },
+							..
+						} = payment.details.kind
+						{
+							let payment_id = payment.details.id;
+							if new_tip.height >= height + ANTI_REORG_DELAY - 1 {
+								payment.details.status = PaymentStatus::Succeeded;
+								self.payment_store.insert_or_update(payment.details)?;
+								self.pending_payment_store.remove(&payment_id)?;
+							}
+						}
+					}
 				},
-			};
-
-			let payment = self.create_payment_from_tx(
-				locked_wallet,
-				txid,
-				id,
-				&wtx.tx_node.tx,
-				payment_status,
-				confirmation_status,
-			);
+				WalletEvent::TxUnconfirmed { txid, tx, old_block_time: None } => {
+					let payment_id = self
+						.find_payment_by_txid(txid)
+						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
+
+					let payment = self.create_payment_from_tx(
+						locked_wallet,
+						txid,
+						payment_id,
+						&tx,
+						PaymentStatus::Pending,
+						ConfirmationStatus::Unconfirmed,
+					);
+					let pending_payment =
+						self.create_pending_payment_from_tx(payment.clone(), Vec::new());
+					self.payment_store.insert_or_update(payment)?;
+					self.pending_payment_store.insert_or_update(pending_payment)?;
+				},
+				WalletEvent::TxReplaced { txid, conflicts, tx, .. } => {
+					let payment_id = self
+						.find_payment_by_txid(txid)
+						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
+
+					// Collect all conflict txids
+					let conflict_txids: Vec<Txid> =
+						conflicts.iter().map(|(_, conflict_txid)| *conflict_txid).collect();
+
+					let payment = self.create_payment_from_tx(
+						locked_wallet,
+						txid,
+						payment_id,
+						&tx,
+						PaymentStatus::Pending,
+						ConfirmationStatus::Unconfirmed,
+					);
+					let pending_payment_details = self
+						.create_pending_payment_from_tx(payment.clone(), conflict_txids.clone());
 
-			self.payment_store.insert_or_update(payment)?;
+					self.pending_payment_store.insert_or_update(pending_payment_details)?;
+				},
+				WalletEvent::TxDropped { txid, tx } => {
+					let payment_id = self
+						.find_payment_by_txid(txid)
+						.unwrap_or_else(|| PaymentId(txid.to_byte_array()));
+					let payment = self.create_payment_from_tx(
+						locked_wallet,
+						txid,
+						payment_id,
+						&tx,
+						PaymentStatus::Pending,
+						ConfirmationStatus::Unconfirmed,
+					);
+					let pending_payment =
+						self.create_pending_payment_from_tx(payment.clone(), Vec::new());
+					self.payment_store.insert_or_update(payment)?;
+					self.pending_payment_store.insert_or_update(pending_payment)?;
+				},
+				_ => {
+					continue;
+				},
+			};
 		}
 
 		Ok(())
@@ -793,27 +923,60 @@ impl Wallet {
 		// here to determine the `PaymentKind`, but that's not really satisfactory, so
 		// we're punting on it until we can come up with a better solution.
 
-		let kind = crate::payment::PaymentKind::Onchain { txid, status: confirmation_status };
+		let kind = PaymentKind::Onchain { txid, status: confirmation_status };
 
 		let fee = locked_wallet.calculate_fee(tx).unwrap_or(Amount::ZERO);
 		let (sent, received) = locked_wallet.sent_and_received(tx);
+		let fee_sat = fee.to_sat();
+
 		let (direction, amount_msat) = if sent > received {
-			let direction = PaymentDirection::Outbound;
-			let amount_msat = Some(
-				sent.to_sat().saturating_sub(fee.to_sat()).saturating_sub(received.to_sat()) * 1000,
-			);
-			(direction, amount_msat)
+			(
+				PaymentDirection::Outbound,
+				Some(
+					(sent.to_sat().saturating_sub(fee_sat).saturating_sub(received.to_sat()))
+						* 1000,
+				),
+			)
 		} else {
-			let direction = PaymentDirection::Inbound;
-			let amount_msat = Some(
-				received.to_sat().saturating_sub(sent.to_sat().saturating_sub(fee.to_sat())) * 1000,
-			);
-			(direction, amount_msat)
+			(
+				PaymentDirection::Inbound,
+				Some(
+					received.to_sat().saturating_sub(sent.to_sat().saturating_sub(fee_sat)) * 1000,
+				),
+			)
 		};
 
-		let fee_paid_msat = Some(fee.to_sat() * 1000);
+		PaymentDetails::new(
+			payment_id,
+			kind,
+			amount_msat,
+			Some(fee_sat * 1000),
+			direction,
+			payment_status,
+		)
+	}
+
+	fn create_pending_payment_from_tx(
+		&self, payment: PaymentDetails, conflicting_txids: Vec<Txid>,
+	) -> PendingPaymentDetails {
+		PendingPaymentDetails::new(payment, conflicting_txids)
+	}
+
+	fn find_payment_by_txid(&self, target_txid: Txid) -> Option<PaymentId> {
+		let direct_payment_id = PaymentId(target_txid.to_byte_array());
+		if self.pending_payment_store.contains_key(&direct_payment_id) {
+			return Some(direct_payment_id);
+		}
+
+		if let Some(replaced_details) = self
+			.pending_payment_store
+			.list_filter(|p| p.conflicting_txids.contains(&target_txid))
+			.first()
+		{
+			return Some(replaced_details.details.id);
+		}
 
-		PaymentDetails::new(payment_id, kind, amount_msat, fee_paid_msat, direction, payment_status)
+		None
 	}
 }
 
@@ -843,9 +1006,9 @@ impl Listen for Wallet {
 			);
 		}
 
-		match locked_wallet.apply_block(block, height) {
-			Ok(()) => {
-				if let Err(e) = self.update_payment_store(&mut *locked_wallet) {
+		match locked_wallet.apply_block_events(block, height) {
+			Ok(events) => {
+				if let Err(e) = self.update_payment_store(&mut *locked_wallet, events) {
 					log_error!(self.logger, "Failed to update payment store: {}", e);
 					return;
 				}