fix: duplicate tx when importing completed tx (#7064)

Description
---
When importing a completed transaction (not yet broadcasted), the wallet
creates a duplicate of this transaction while scanning UTXOs.

Motivation and Context
---
Check if the corresponding transaction hasn’t been imported. Use
existing tx's `tx_id` instead of creating a duplicate record.

How Has This Been Tested?
---
Manually

What process can a PR reviewer use to test or verify this change?
---
1. Send one-sided tx via non-existing base node
2. Export tx
3. Imported tx to another wallet instance
4. Run this wallet instance and wait until tx is mined.

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Enhanced wallet output recovery by linking recovered outputs to
existing transactions using address-based transaction queries.
- Added the ability to filter completed transactions by source and
destination addresses in transaction history queries.

- **Tests**
- Introduced new tests for querying completed transactions filtered by
addresses.
- Updated test setups to include the transaction service handle for
improved output recovery support.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: mmrrnn

base_layer/wallet/src/output_manager_service/mod.rs

@@ -62,6 +62,7 @@ use crate::{
         service::OutputManagerService,
         storage::database::{OutputManagerBackend, OutputManagerDatabase},
     },
+    transaction_service::handle::TransactionServiceHandle,
     utxo_scanner_service::handle::UtxoScannerHandle,
 };
 
@@ -126,6 +127,7 @@ where
             let connectivity = handles.expect_handle::<WalletConnectivityHandle>();
             let key_manager = handles.expect_handle::<TKeyManagerInterface>();
             let utxo_scanner_handle = handles.expect_handle::<UtxoScannerHandle>();
+            let transaction_service_handle = handles.expect_handle::<TransactionServiceHandle>();
 
             let service = OutputManagerService::new(
                 config,
@@ -140,6 +142,7 @@ where
                 connectivity,
                 key_manager,
                 utxo_scanner_handle,
+                transaction_service_handle,
             )
             .await
             .expect("Could not initialize Output Manager Service")

base_layer/wallet/src/output_manager_service/recovery/standard_outputs_recoverer.rs

@@ -42,34 +42,47 @@ use tari_crypto::keys::SecretKey;
 use tari_script::{inputs, script, ExecutionStack, Opcode, TariScript};
 use tari_utilities::hex::Hex;
 
-use crate::output_manager_service::{
-    error::{OutputManagerError, OutputManagerStorageError},
-    handle::RecoveredOutput,
-    storage::{
-        database::{OutputManagerBackend, OutputManagerDatabase},
-        models::{DbWalletOutput, KnownOneSidedPaymentScript},
-        OutputSource,
+use crate::{
+    output_manager_service::{
+        error::{OutputManagerError, OutputManagerStorageError},
+        handle::RecoveredOutput,
+        storage::{
+            database::{OutputManagerBackend, OutputManagerDatabase},
+            models::{DbWalletOutput, KnownOneSidedPaymentScript},
+            OutputSource,
+        },
     },
+    transaction_service::{handle::TransactionServiceHandle, storage::models::CompletedTransaction},
 };
 
 const LOG_TARGET: &str = "wallet::output_manager_service::recovery";
 
 pub(crate) struct StandardUtxoRecoverer<TBackend: OutputManagerBackend + 'static, TKeyManagerInterface> {
     master_key_manager: TKeyManagerInterface,
     db: OutputManagerDatabase<TBackend>,
+    transaction_service_handle: TransactionServiceHandle,
 }
 
 impl<TBackend, TKeyManagerInterface> StandardUtxoRecoverer<TBackend, TKeyManagerInterface>
 where
     TBackend: OutputManagerBackend + 'static,
     TKeyManagerInterface: TransactionKeyManagerInterface,
 {
-    pub fn new(master_key_manager: TKeyManagerInterface, db: OutputManagerDatabase<TBackend>) -> Self {
-        Self { master_key_manager, db }
+    pub fn new(
+        master_key_manager: TKeyManagerInterface,
+        db: OutputManagerDatabase<TBackend>,
+        transaction_service_handle: TransactionServiceHandle,
+    ) -> Self {
+        Self {
+            master_key_manager,
+            db,
+            transaction_service_handle,
+        }
     }
 
     /// Attempt to rewind all of the given transaction outputs into key_manager outputs. If they can be rewound then add
     /// them to the database and increment the key manager index
+    #[allow(clippy::too_many_lines)]
     pub async fn scan_and_recover_outputs(
         &mut self,
         outputs: Vec<(TransactionOutput, Option<TxId>)>,
@@ -145,7 +158,33 @@ where
             .await?;
             let tx_id = match tx_id {
                 Some(id) => *id,
-                None => TxId::new_random(),
+                None => {
+                    let mut related_txs: Vec<CompletedTransaction> = Vec::new();
+                    let (source_address, recipient_address) = match &db_output.payment_id {
+                        PaymentId::AddressAndData { sender_address, .. } => (Some(sender_address.clone()), None),
+                        PaymentId::TransactionInfo { recipient_address, .. } => (None, Some(recipient_address.clone())),
+                        _ => (None, None),
+                    };
+
+                    if source_address.is_some() || recipient_address.is_some() {
+                        related_txs = self
+                            .transaction_service_handle
+                            .get_completed_transactions_by_addresses(source_address, recipient_address)
+                            .await
+                            .unwrap_or_default();
+                    }
+
+                    let tx_id = related_txs.iter().find_map(|tx| {
+                        tx.transaction
+                            .body
+                            .outputs()
+                            .iter()
+                            .find(|tx| tx.commitment == db_output.commitment)
+                            .map(|_| tx.tx_id)
+                    });
+
+                    tx_id.unwrap_or_else(TxId::new_random)
+                },
             };
             let output_hex = db_output.commitment.to_hex();
             if let Err(e) = self.db.add_unspent_output_with_tx_id(tx_id, db_output) {

base_layer/wallet/src/output_manager_service/resources.rs

@@ -30,6 +30,7 @@ use crate::{
         handle::OutputManagerEventSender,
         storage::database::OutputManagerDatabase,
     },
+    transaction_service::handle::TransactionServiceHandle,
     utxo_scanner_service::handle::UtxoScannerHandle,
 };
 
@@ -47,4 +48,5 @@ pub(crate) struct OutputManagerResources<TBackend, TWalletConnectivity, TKeyMana
     pub interactive_tari_address: TariAddress,
     pub one_sided_tari_address: TariAddress,
     pub utxo_scanner_handle: UtxoScannerHandle,
+    pub transaction_service_handle: TransactionServiceHandle,
 }

base_layer/wallet/src/output_manager_service/service.rs

@@ -115,6 +115,7 @@ use crate::{
         tasks::TxoValidationTask,
         TRANSACTION_INPUTS_LIMIT,
     },
+    transaction_service::handle::TransactionServiceHandle,
     utxo_scanner_service::handle::{UtxoScannerEvent, UtxoScannerHandle},
 };
 
@@ -156,6 +157,7 @@ where
         connectivity: TWalletConnectivity,
         key_manager: TKeyManagerInterface,
         utxo_scanner_handle: UtxoScannerHandle,
+        transaction_service_handle: TransactionServiceHandle,
     ) -> Result<Self, OutputManagerError> {
         let view_key = key_manager.get_view_key().await?;
         let spend_key = key_manager.get_spend_key().await?;
@@ -186,6 +188,7 @@ where
             one_sided_tari_address,
             interactive_tari_address,
             utxo_scanner_handle,
+            transaction_service_handle,
         };
 
         Ok(Self {
@@ -486,12 +489,14 @@ where
                 .await
                 .map(OutputManagerResponse::Transaction),
 
-            OutputManagerRequest::ScanForRecoverableOutputs(outputs) => {
-                StandardUtxoRecoverer::new(self.resources.key_manager.clone(), self.resources.db.clone())
-                    .scan_and_recover_outputs(outputs)
-                    .await
-                    .map(OutputManagerResponse::RewoundOutputs)
-            },
+            OutputManagerRequest::ScanForRecoverableOutputs(outputs) => StandardUtxoRecoverer::new(
+                self.resources.key_manager.clone(),
+                self.resources.db.clone(),
+                self.resources.transaction_service_handle.clone(),
+            )
+            .scan_and_recover_outputs(outputs)
+            .await
+            .map(OutputManagerResponse::RewoundOutputs),
             OutputManagerRequest::ScanOutputs(outputs) => self
                 .scan_outputs_for_one_sided_payments(outputs)
                 .await

base_layer/wallet/src/transaction_service/handle.rs

@@ -84,6 +84,10 @@ pub enum TransactionServiceRequest {
         block_hash: Option<FixedHash>,
         block_height: Option<u64>,
     },
+    GetCompletedTransactionsByAddresses {
+        source_address: Option<TariAddress>,
+        destination_address: Option<TariAddress>,
+    },
     GetCancelledPendingInboundTransactions,
     GetCancelledPendingOutboundTransactions,
     GetCancelledCompletedTransactions,
@@ -213,6 +217,7 @@ impl fmt::Display for TransactionServiceRequest {
             Self::GetPendingInboundTransactions => write!(f, "GetPendingInboundTransactions"),
             Self::GetPendingOutboundTransactions => write!(f, "GetPendingOutboundTransactions"),
             Self::GetCompletedTransactions { .. } => write!(f, "GetCompletedTransactions"),
+            Self::GetCompletedTransactionsByAddresses { .. } => write!(f, "GetCompletedTransactionsByAddresses"),
             Self::ImportTransaction(tx) => write!(f, "ImportTransaction: {:?}", tx),
             Self::GetCancelledPendingInboundTransactions => write!(f, "GetCancelledPendingInboundTransactions"),
             Self::GetCancelledPendingOutboundTransactions => write!(f, "GetCancelledPendingOutboundTransactions"),
@@ -984,6 +989,24 @@ impl TransactionServiceHandle {
         }
     }
 
+    pub async fn get_completed_transactions_by_addresses(
+        &mut self,
+        source_address: Option<TariAddress>,
+        destination_address: Option<TariAddress>,
+    ) -> Result<Vec<CompletedTransaction>, TransactionServiceError> {
+        match self
+            .handle
+            .call(TransactionServiceRequest::GetCompletedTransactionsByAddresses {
+                source_address,
+                destination_address,
+            })
+            .await??
+        {
+            TransactionServiceResponse::CompletedTransactions(c) => Ok(c),
+            _ => Err(TransactionServiceError::UnexpectedApiResponse),
+        }
+    }
+
     pub async fn get_cancelled_completed_transactions(
         &mut self,
     ) -> Result<Vec<CompletedTransaction>, TransactionServiceError> {

base_layer/wallet/src/transaction_service/service.rs

@@ -899,6 +899,13 @@ where
                 self.db
                     .get_completed_transactions(payment_id, block_hash, block_height)?,
             )),
+            TransactionServiceRequest::GetCompletedTransactionsByAddresses {
+                source_address,
+                destination_address,
+            } => Ok(TransactionServiceResponse::CompletedTransactions(
+                self.db
+                    .get_completed_transactions_by_addresses(source_address, destination_address)?,
+            )),
             TransactionServiceRequest::GetCancelledPendingInboundTransactions => {
                 Ok(TransactionServiceResponse::PendingInboundTransactions(
                     self.db.get_cancelled_pending_inbound_transactions()?,

base_layer/wallet/src/transaction_service/storage/database.rs

@@ -162,6 +162,11 @@ pub trait TransactionBackend: Send + Sync + Clone {
         block_hash: Option<FixedHash>,
         block_height: Option<u64>,
     ) -> Result<Vec<CompletedTransaction>, TransactionStorageError>;
+    fn find_completed_transactions_filter_addresses(
+        &self,
+        source_address: Option<TariAddress>,
+        destination_address: Option<TariAddress>,
+    ) -> Result<Vec<CompletedTransaction>, TransactionStorageError>;
 }
 
 #[derive(Clone, PartialEq)]
@@ -611,6 +616,15 @@ where T: TransactionBackend + 'static
         self.get_completed_transactions_by_cancelled(payment_id, false, block_hash, block_height)
     }
 
+    pub fn get_completed_transactions_by_addresses(
+        &self,
+        source_address: Option<TariAddress>,
+        destination_address: Option<TariAddress>,
+    ) -> Result<Vec<CompletedTransaction>, TransactionStorageError> {
+        self.db
+            .find_completed_transactions_filter_addresses(source_address, destination_address)
+    }
+
     pub fn get_cancelled_completed_transactions(&self) -> Result<Vec<CompletedTransaction>, TransactionStorageError> {
         self.get_completed_transactions_by_cancelled(None, true, None, None)
     }

base_layer/wallet/src/transaction_service/storage/sqlite_db.rs

@@ -1048,6 +1048,31 @@ impl TransactionBackend for TransactionServiceSqliteDatabase {
         Ok(coinbases)
     }
 
+    fn find_completed_transactions_filter_addresses(
+        &self,
+        source_address: Option<TariAddress>,
+        destination_address: Option<TariAddress>,
+    ) -> Result<Vec<CompletedTransaction>, TransactionStorageError> {
+        let mut conn = self.database_connection.get_pooled_connection()?;
+        let cipher = acquire_read_lock!(self.cipher);
+
+        let mut query = completed_transactions::table.into_boxed();
+        if let Some(source_address) = source_address {
+            query = query.filter(completed_transactions::source_address.eq(source_address.to_vec()));
+        }
+        if let Some(destination_address) = destination_address {
+            query = query.filter(completed_transactions::destination_address.eq(destination_address.to_vec()));
+        }
+
+        query
+            .load::<CompletedTransactionSql>(&mut conn)?
+            .into_iter()
+            .map(|ct: CompletedTransactionSql| {
+                CompletedTransaction::try_from(ct, &cipher).map_err(TransactionStorageError::from)
+            })
+            .collect::<Result<Vec<CompletedTransaction>, TransactionStorageError>>()
+    }
+
     fn find_completed_transactions_filter_payment_id_block_hash(
         &self,
         payment_id: Option<Vec<u8>>,

base_layer/wallet/tests/output_manager_service_tests/service.rs

@@ -182,6 +182,7 @@ async fn setup_output_manager_service<T: OutputManagerBackend + 'static>(
         wallet_connectivity_mock.clone(),
         key_manager.clone(),
         scanner_handle,
+        ts_handle.clone(),
     )
     .await
     .unwrap();
@@ -251,6 +252,7 @@ pub async fn setup_oms_with_bn_state<T: OutputManagerBackend + 'static>(
         connectivity,
         key_manager.clone(),
         scanner_handle,
+        ts_handle.clone(),
     )
     .await
     .unwrap();