Fix payment router recovery indexing using wrong address for comparis…

…on (#7314)

packages/discovery-provider/integration_tests/tasks/test_index_payment_router.py

@@ -37,8 +37,7 @@
 thirdPartyUserBank = "7dw7W4Yv7F1uWb9dVH1CFPm39mePyypuCji2zxcFA556"
 
 # Used as the source wallet for all the mock transactions
-transactionSenderAddress = "G231EZsMoCNBiQKP5quEeAM3oG516Zspirjnh7ywP71i"
-transactionSenderUsdcAccount = "3XmVeZ6M1FYDdUQaNeQZf8dipvtzNP6NVb5xjDkdeiNb"
+transactionSenderOwnerAccount = "HXLN9UWwAjMPgHaFZDfgabT79SmLSdTeu2fUha2xHz9W"
 
 test_entries = {
     "users": [
@@ -258,8 +257,8 @@ def test_process_payment_router_tx_details_transfer_without_purchase(
         assert transaction_record.transaction_type == USDCTransactionType.transfer
         assert transaction_record.method == USDCTransactionMethod.receive
         assert transaction_record.change == 1000000
-        # For transfers, the metadata is the source address
-        assert transaction_record.tx_metadata == transactionSenderUsdcAccount
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
+        assert transaction_record.tx_metadata == transactionSenderOwnerAccount
 
 
 def test_process_payment_router_tx_details_transfer_from_user_bank_without_purchase(
@@ -351,7 +350,7 @@ def test_process_payment_router_tx_details_transfer_recovery(app):
             "method": USDCTransactionMethod.send,
             "change": -1000000,
             "balance": 0,
-            "tx_metadata": transactionSenderUsdcAccount,
+            "tx_metadata": transactionSenderOwnerAccount,
         }
     ]
 
@@ -400,7 +399,7 @@ def test_process_payment_router_tx_details_transfer_partial_recovery(
             "method": USDCTransactionMethod.send,
             "change": -2000000,
             "balance": 0,
-            "tx_metadata": transactionSenderUsdcAccount,
+            "tx_metadata": transactionSenderOwnerAccount,
         }
     ]
 
@@ -420,16 +419,18 @@ def test_process_payment_router_tx_details_transfer_partial_recovery(
             session.query(USDCTransactionsHistory)
             .filter(USDCTransactionsHistory.signature == "existingWithdrawal")
             .filter(USDCTransactionsHistory.user_bank == trackOwnerUserBank)
-            .filter(USDCTransactionsHistory.tx_metadata == transactionSenderUsdcAccount)
+            .filter(
+                USDCTransactionsHistory.tx_metadata == transactionSenderOwnerAccount
+            )
             .first()
         )
         # Recovery transaction was for half of the original amount, expect the difference
         assert transaction_record.change == -1000000
         assert transaction_record.balance == 1000000
 
 
-# Recovery transaction is for more than the original, should index as a new transfer
 def test_process_payment_router_tx_details_transfer_over_recovery(
+    # Recovery transaction is for more than the original, should index as a new transfer
     app,
 ):
     tx_response = mock_valid_transfer_single_recipient_recovery_tx
@@ -451,7 +452,7 @@ def test_process_payment_router_tx_details_transfer_over_recovery(
             "method": USDCTransactionMethod.send,
             "change": -500000,
             "balance": 0,
-            "tx_metadata": transactionSenderUsdcAccount,
+            "tx_metadata": transactionSenderOwnerAccount,
         }
     ]
 
@@ -471,7 +472,9 @@ def test_process_payment_router_tx_details_transfer_over_recovery(
             session.query(USDCTransactionsHistory)
             .filter(USDCTransactionsHistory.signature == "existingWithdrawal")
             .filter(USDCTransactionsHistory.user_bank == trackOwnerUserBank)
-            .filter(USDCTransactionsHistory.tx_metadata == transactionSenderUsdcAccount)
+            .filter(
+                USDCTransactionsHistory.tx_metadata == transactionSenderOwnerAccount
+            )
             .first()
         )
         # Recovery transaction was for more than the original amount, expect original transaction to be unchanged
@@ -482,11 +485,15 @@ def test_process_payment_router_tx_details_transfer_over_recovery(
             session.query(USDCTransactionsHistory)
             .filter(USDCTransactionsHistory.signature == tx_sig_str)
             .filter(USDCTransactionsHistory.user_bank == trackOwnerUserBank)
-            .filter(USDCTransactionsHistory.tx_metadata == transactionSenderUsdcAccount)
+            .filter(
+                USDCTransactionsHistory.tx_metadata == transactionSenderOwnerAccount
+            )
             .first()
         )
 
         assert new_transaction_record is not None
+        assert new_transaction_record.method == USDCTransactionMethod.receive
+        assert new_transaction_record.transaction_type == USDCTransactionType.transfer
         assert new_transaction_record.change == 1000000
         assert new_transaction_record.balance == 1000000
 
@@ -559,8 +566,8 @@ def test_process_payment_router_tx_details_transfer_recovery_address_mismatch(
         assert transaction_record.transaction_type == USDCTransactionType.transfer
         assert transaction_record.method == USDCTransactionMethod.receive
         assert transaction_record.change == 1000000
-        # For transfers, the metadata is the source address
-        assert transaction_record.tx_metadata == transactionSenderUsdcAccount
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
+        assert transaction_record.tx_metadata == transactionSenderOwnerAccount
 
 
 def test_process_payment_router_tx_details_valid_purchase_with_pay_extra(app):
@@ -792,8 +799,8 @@ def test_process_payment_router_tx_details_invalid_purchase_bad_splits(app):
         assert owner_transaction_record.transaction_type == USDCTransactionType.transfer
         assert owner_transaction_record.method == USDCTransactionMethod.receive
         assert owner_transaction_record.change == 1000000
-        # For transfers, the metadata is the source address
-        assert owner_transaction_record.tx_metadata == transactionSenderUsdcAccount
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
+        assert owner_transaction_record.tx_metadata == transactionSenderOwnerAccount
 
         third_party_transaction_record = (
             session.query(USDCTransactionsHistory)
@@ -809,9 +816,9 @@ def test_process_payment_router_tx_details_invalid_purchase_bad_splits(app):
         )
         assert third_party_transaction_record.method == USDCTransactionMethod.receive
         assert third_party_transaction_record.change == 500000
-        # For transfers, the metadata is the source address
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
         assert (
-            third_party_transaction_record.tx_metadata == transactionSenderUsdcAccount
+            third_party_transaction_record.tx_metadata == transactionSenderOwnerAccount
         )
 
 
@@ -860,8 +867,8 @@ def test_process_payment_router_tx_details_invalid_purchase_missing_splits(app):
         assert owner_transaction_record.transaction_type == USDCTransactionType.transfer
         assert owner_transaction_record.method == USDCTransactionMethod.receive
         assert owner_transaction_record.change == 2000000
-        # For transfers, the metadata is the source address
-        assert owner_transaction_record.tx_metadata == transactionSenderUsdcAccount
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
+        assert owner_transaction_record.tx_metadata == transactionSenderOwnerAccount
 
 
 def test_process_payment_router_tx_details_transfer_multiple_users_without_purchase(
@@ -909,8 +916,8 @@ def test_process_payment_router_tx_details_transfer_multiple_users_without_purch
         assert owner_transaction_record.transaction_type == USDCTransactionType.transfer
         assert owner_transaction_record.method == USDCTransactionMethod.receive
         assert owner_transaction_record.change == 1000000
-        # For transfers, the metadata is the source address
-        assert owner_transaction_record.tx_metadata == transactionSenderUsdcAccount
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
+        assert owner_transaction_record.tx_metadata == transactionSenderOwnerAccount
 
         third_party_transaction_record = (
             session.query(USDCTransactionsHistory)
@@ -926,9 +933,9 @@ def test_process_payment_router_tx_details_transfer_multiple_users_without_purch
         )
         assert third_party_transaction_record.method == USDCTransactionMethod.receive
         assert third_party_transaction_record.change == 1000000
-        # For transfers, the metadata is the source address
+        # For transfers, source is the owning wallet unless it's a transfer from a user bank
         assert (
-            third_party_transaction_record.tx_metadata == transactionSenderUsdcAccount
+            third_party_transaction_record.tx_metadata == transactionSenderOwnerAccount
         )
 
 

packages/discovery-provider/src/tasks/index_payment_router.py

@@ -180,6 +180,16 @@ def get_highest_payment_router_tx_slot(session: Session):
     return slot
 
 
+def get_account_owner_from_balance_change(
+    sender_account: str, balance_changes: dict[str, BalanceChange]
+):
+    """Finds the owner of the sender account by looking at the balance changes"""
+    if sender_account in balance_changes:
+        return balance_changes[sender_account]["owner"]
+    else:
+        return None
+
+
 # Cache the latest value committed to DB in redis
 # Used for quick retrieval in health check
 def cache_latest_sol_payment_router_db_tx(redis: Redis, tx):
@@ -397,6 +407,14 @@ def attempt_index_recovery_transfer(
             f"Recovery transfer must have exactly one receiver account. Received: {','.join([a['user_bank_account'] for a in receiver_user_accounts])}"
         )
 
+    sender_account_owner = get_account_owner_from_balance_change(
+        sender_account=sender_account, balance_changes=balance_changes
+    )
+    if sender_account_owner is None:
+        raise Exception(
+            f"Unexpectedly found no owner account for sender account {sender_account}"
+        )
+
     receiver_user_account = receiver_user_accounts[0]
     receiver_bank_account = receiver_user_account["user_bank_account"]
     receiver_balance_change = balance_changes[receiver_bank_account]
@@ -410,7 +428,9 @@ def attempt_index_recovery_transfer(
             and_(
                 USDCTransactionsHistory.user_bank == receiver_bank_account,
                 USDCTransactionsHistory.method == USDCTransactionMethod.send,
-                USDCTransactionsHistory.tx_metadata == sender_account,
+                # The original transaction will be indexed with the owner account
+                # of the original recipient
+                USDCTransactionsHistory.tx_metadata == sender_account_owner,
             )
         )
         .order_by(desc(USDCTransactionsHistory.transaction_created_at))
@@ -457,6 +477,16 @@ def index_transfer(
     timestamp: datetime,
     tx_sig: str,
 ):
+    sender_metadata_address = sender_account
+    # If sending account was a user bank, leave that as the address
+    # Otherwise, map it to an owning solana wallet
+    if sender_user_account is None:
+        sender_metadata_address = (
+            get_account_owner_from_balance_change(
+                sender_account=sender_account, balance_changes=balance_changes
+            )
+            or sender_account
+        )
     for user_account in receiver_user_accounts:
         balance_change = balance_changes[user_account["user_bank_account"]]
         usdc_tx_received = USDCTransactionsHistory(
@@ -468,7 +498,7 @@ def index_transfer(
             transaction_created_at=timestamp,
             change=Decimal(balance_change["change"]),
             balance=Decimal(balance_change["post_balance"]),
-            tx_metadata=sender_account,
+            tx_metadata=sender_metadata_address,
         )
         session.add(usdc_tx_received)
         logger.debug(