[C-4512] Implement purchase vendor tracking

packages/discovery-provider/ddl/functions/handle_usdc_purchase.sql

@@ -18,7 +18,8 @@ begin
                 'seller_user_id', new.seller_user_id,
                 'amount', new.amount,
                 'extra_amount', new.extra_amount,
-                'content_id', new.content_id
+                'content_id', new.content_id,
+                'vendor', new.vendor
               )
           ),
           (

packages/discovery-provider/ddl/migrations/0077_purchase_vendor.sql

@@ -0,0 +1,3 @@
+begin;
+    alter table usdc_purchases add column if not exists vendor varchar;
+end;

packages/discovery-provider/default_config.ini

@@ -40,6 +40,7 @@ user_bank_min_slot = 0
 payment_router_min_slot = 0
 user_bank_program_address = Ewkv3JahEFRKkcJmpoKB7pXbnUHwjAyXiwEo4ZY2rezQ
 payment_router_program_address = paytYpX3LPN98TAeen6bFFeraGSuWnomZmCXjAsoqPa
+coinflow_program_address = FD1amxhTsDpwzoVX41dxp2ygAESURV2zdUACzxM1Dfw9
 waudio_mint = 9LzCMqDgTKYz9Drzqnpgee3SGa89up3a247ypMj2xrqM
 usdc_mint = EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v
 rewards_manager_program_address = DDZDcYdQFEMwcu2Mwo75yGFjJ1mUQyyXLWzhZLEVFcei

packages/discovery-provider/integration_tests/notifications/test_usdc_purchase.py

@@ -59,6 +59,7 @@ def test_usdc_purchase_notification(app):
             "amount": 1000,
             "extra_amount": 0,
             "content_id": 100,
+            "vendor": None,
         }
         assert seller_notifications[0].slot == 4
         assert buyer_notifications[0].user_ids == [1]

packages/discovery-provider/integration_tests/queries/test_notifications/test_usdc_purchase_notification.py

@@ -60,6 +60,7 @@ def test_get_usdc_purchase_notifications(app):
                 "amount": 1000000,
                 "extra_amount": 0,
                 "content_id": 1,
+                "vendor": None,
             }
 
             args = {

packages/discovery-provider/src/models/users/usdc_purchase.py

@@ -17,6 +17,11 @@ class PurchaseAccessType(str, enum.Enum):
     download = "download"
 
 
+class PurchaseVendor(str, enum.Enum):
+    user_bank = "user_bank"
+    coinflow = "coinflow"
+
+
 class USDCPurchase(Base, RepresentableMixin):
     __tablename__ = "usdc_purchases"
 
@@ -32,6 +37,7 @@ class USDCPurchase(Base, RepresentableMixin):
     city = Column(String, nullable=True)
     region = Column(String, nullable=True)
     country = Column(String, nullable=True)
+    vendor = Column(Enum(PurchaseVendor), nullable=True)
 
     created_at = Column(
         DateTime, nullable=False, index=True, server_default=text("CURRENT_TIMESTAMP")

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

@@ -27,6 +27,7 @@
 from src.models.users.usdc_purchase import (
     PurchaseAccessType,
     PurchaseType,
+    PurchaseVendor,
     USDCPurchase,
 )
 from src.models.users.usdc_transactions_history import (
@@ -77,6 +78,7 @@
 
 # Populate values used in indexing from config
 PAYMENT_ROUTER_ADDRESS = shared_config["solana"]["payment_router_program_address"]
+COINFLOW_PROGRAM_ADDRESS = shared_config["solana"]["coinflow_program_address"]
 WAUDIO_MINT = shared_config["solana"]["waudio_mint"]
 USDC_MINT = shared_config["solana"]["usdc_mint"]
 
@@ -431,6 +433,7 @@ def index_purchase(
     slot: int,
     timestamp: datetime,
     tx_sig: str,
+    vendor: PurchaseVendor,
 ):
     # Detect "pay extra" amount (difference between total balance change across
     # all recipients and the purchase price)
@@ -451,6 +454,7 @@ def index_purchase(
         city=geo_metadata.get("city") if geo_metadata else None,
         region=geo_metadata.get("region") if geo_metadata else None,
         country=geo_metadata.get("country") if geo_metadata else None,
+        vendor=vendor,
     )
     logger.debug(
         f"index_payment_router.py | tx: {tx_sig} | Creating usdc_purchase for purchase {usdc_purchase}"
@@ -567,6 +571,7 @@ def validate_and_index_usdc_transfers(
     slot: int,
     timestamp: datetime,
     tx_sig: str,
+    vendor: PurchaseVendor,
     challenge_event_bus: ChallengeEventBus,
 ):
     """Checks if the transaction is a valid purchase and if so creates the purchase record. Otherwise, indexes a transfer."""
@@ -589,6 +594,7 @@ def validate_and_index_usdc_transfers(
             slot=slot,
             timestamp=timestamp,
             tx_sig=tx_sig,
+            vendor=vendor,
         )
 
         # dispatch audio matching challenge events
@@ -660,6 +666,7 @@ def process_route_instruction(
     slot: int,
     challenge_event_bus: ChallengeEventBus,
     timestamp: datetime,
+    vendor: PurchaseVendor,
 ):
     # Route instructions have a varying number of receiver accounts but they are always
     # at the end
@@ -761,6 +768,7 @@ def process_route_instruction(
             slot=slot,
             timestamp=timestamp,
             tx_sig=tx_sig,
+            vendor=vendor,
             challenge_event_bus=challenge_event_bus,
         )
 
@@ -801,6 +809,9 @@ def process_payment_router_tx_details(
     has_route_instruction = has_log(meta, "Program log: Instruction: Route")
 
     instruction = get_valid_instruction(tx_message, meta, PAYMENT_ROUTER_ADDRESS)
+    contains_conflow_instruction = get_valid_instruction(
+        tx_message, meta, COINFLOW_PROGRAM_ADDRESS
+    )
     if instruction is None:
         logger.error(
             f"index_payment_router.py | tx: {tx_sig} |  No Valid instruction found"
@@ -818,6 +829,11 @@ def process_payment_router_tx_details(
             slot=result.slot,
             challenge_event_bus=challenge_event_bus,
             timestamp=timestamp,
+            vendor=(
+                PurchaseVendor.coinflow
+                if contains_conflow_instruction
+                else PurchaseVendor.user_bank
+            ),
         )
 
 

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

@@ -29,6 +29,7 @@
 from src.models.users.usdc_purchase import (
     PurchaseAccessType,
     PurchaseType,
+    PurchaseVendor,
     USDCPurchase,
 )
 from src.models.users.usdc_transactions_history import (
@@ -460,6 +461,7 @@ def index_purchase(
         content_type=purchase_metadata["type"],
         content_id=purchase_metadata["id"],
         access=purchase_metadata["access"],
+        vendor=PurchaseVendor.user_bank,
     )
     logger.debug(
         f"index_user_bank.py | Creating usdc_purchase for purchase {usdc_purchase}"