Add CRUD operations for Lamassu transactions: implement create, retrieve, and update functions for storing processed Lamassu transactions in the database. Enhance migration script to create the necessary database table and update models to include transaction data structures. Integrate transaction storage into the LamassuTransactionProcessor for improved audit and UI display capabilities.

crud.py

@@ -12,7 +12,8 @@ from .models import (
     CreateDepositData, DcaDeposit, UpdateDepositStatusData,
     CreateDcaPaymentData, DcaPayment,
     ClientBalanceSummary,
-    CreateLamassuConfigData, LamassuConfig, UpdateLamassuConfigData
+    CreateLamassuConfigData, LamassuConfig, UpdateLamassuConfigData,
+    CreateLamassuTransactionData, StoredLamassuTransaction
 )
 
 db = Database("ext_myextension")
@@ -439,3 +440,88 @@ async def update_poll_success_time(config_id: str) -> None:
             "updated_at": utc_now
         }
     )
+
+
+# Lamassu Transaction Storage CRUD Operations
+async def create_lamassu_transaction(data: CreateLamassuTransactionData) -> StoredLamassuTransaction:
+    """Store a processed Lamassu transaction"""
+    transaction_id = urlsafe_short_hash()
+    await db.execute(
+        """
+        INSERT INTO myextension.lamassu_transactions 
+        (id, lamassu_transaction_id, fiat_amount, crypto_amount, commission_percentage, 
+         discount, effective_commission, commission_amount_sats, base_amount_sats, 
+         exchange_rate, crypto_code, fiat_code, device_id, transaction_time, processed_at,
+         clients_count, distributions_total_sats)
+        VALUES (:id, :lamassu_transaction_id, :fiat_amount, :crypto_amount, :commission_percentage,
+                :discount, :effective_commission, :commission_amount_sats, :base_amount_sats,
+                :exchange_rate, :crypto_code, :fiat_code, :device_id, :transaction_time, :processed_at,
+                :clients_count, :distributions_total_sats)
+        """,
+        {
+            "id": transaction_id,
+            "lamassu_transaction_id": data.lamassu_transaction_id,
+            "fiat_amount": data.fiat_amount,
+            "crypto_amount": data.crypto_amount,
+            "commission_percentage": data.commission_percentage,
+            "discount": data.discount,
+            "effective_commission": data.effective_commission,
+            "commission_amount_sats": data.commission_amount_sats,
+            "base_amount_sats": data.base_amount_sats,
+            "exchange_rate": data.exchange_rate,
+            "crypto_code": data.crypto_code,
+            "fiat_code": data.fiat_code,
+            "device_id": data.device_id,
+            "transaction_time": data.transaction_time,
+            "processed_at": datetime.now(),
+            "clients_count": 0,  # Will be updated after distributions
+            "distributions_total_sats": 0  # Will be updated after distributions
+        }
+    )
+    return await get_lamassu_transaction(transaction_id)
+
+
+async def get_lamassu_transaction(transaction_id: str) -> Optional[StoredLamassuTransaction]:
+    """Get a stored Lamassu transaction by ID"""
+    return await db.fetchone(
+        "SELECT * FROM myextension.lamassu_transactions WHERE id = :id",
+        {"id": transaction_id},
+        StoredLamassuTransaction,
+    )
+
+
+async def get_lamassu_transaction_by_lamassu_id(lamassu_transaction_id: str) -> Optional[StoredLamassuTransaction]:
+    """Get a stored Lamassu transaction by Lamassu transaction ID"""
+    return await db.fetchone(
+        "SELECT * FROM myextension.lamassu_transactions WHERE lamassu_transaction_id = :lamassu_id",
+        {"lamassu_id": lamassu_transaction_id},
+        StoredLamassuTransaction,
+    )
+
+
+async def get_all_lamassu_transactions() -> List[StoredLamassuTransaction]:
+    """Get all stored Lamassu transactions"""
+    return await db.fetchall(
+        "SELECT * FROM myextension.lamassu_transactions ORDER BY transaction_time DESC",
+        model=StoredLamassuTransaction,
+    )
+
+
+async def update_lamassu_transaction_distribution_stats(
+    transaction_id: str, 
+    clients_count: int, 
+    distributions_total_sats: int
+) -> None:
+    """Update distribution statistics for a Lamassu transaction"""
+    await db.execute(
+        """
+        UPDATE myextension.lamassu_transactions 
+        SET clients_count = :clients_count, distributions_total_sats = :distributions_total_sats 
+        WHERE id = :id
+        """,
+        {
+            "clients_count": clients_count,
+            "distributions_total_sats": distributions_total_sats,
+            "id": transaction_id
+        }
+    )

migrations.py

@@ -212,3 +212,32 @@ async def m011_add_commission_wallet_to_lamassu_config(db):
         ADD COLUMN commission_wallet_id TEXT;
         """
     )
+
+
+async def m012_create_lamassu_transactions_table(db):
+    """
+    Create table to store processed Lamassu transactions for audit and UI display.
+    """
+    await db.execute(
+        f"""
+        CREATE TABLE myextension.lamassu_transactions (
+            id TEXT PRIMARY KEY NOT NULL,
+            lamassu_transaction_id TEXT NOT NULL UNIQUE,
+            fiat_amount INTEGER NOT NULL,
+            crypto_amount INTEGER NOT NULL,
+            commission_percentage REAL NOT NULL,
+            discount REAL NOT NULL DEFAULT 0.0,
+            effective_commission REAL NOT NULL,
+            commission_amount_sats INTEGER NOT NULL,
+            base_amount_sats INTEGER NOT NULL,
+            exchange_rate REAL NOT NULL,
+            crypto_code TEXT NOT NULL DEFAULT 'BTC',
+            fiat_code TEXT NOT NULL DEFAULT 'GTQ',
+            device_id TEXT,
+            transaction_time TIMESTAMP NOT NULL,
+            processed_at TIMESTAMP NOT NULL DEFAULT {db.timestamp_now},
+            clients_count INTEGER NOT NULL DEFAULT 0,
+            distributions_total_sats INTEGER NOT NULL DEFAULT 0
+        );
+        """
+    )

models.py

@@ -102,6 +102,43 @@ class LamassuTransaction(BaseModel):
     timestamp: datetime
 
 
+# Lamassu Transaction Storage Models
+class CreateLamassuTransactionData(BaseModel):
+    lamassu_transaction_id: str
+    fiat_amount: int
+    crypto_amount: int
+    commission_percentage: float
+    discount: float = 0.0
+    effective_commission: float
+    commission_amount_sats: int
+    base_amount_sats: int
+    exchange_rate: float
+    crypto_code: str = "BTC"
+    fiat_code: str = "GTQ"
+    device_id: Optional[str] = None
+    transaction_time: datetime
+
+
+class StoredLamassuTransaction(BaseModel):
+    id: str
+    lamassu_transaction_id: str
+    fiat_amount: int
+    crypto_amount: int
+    commission_percentage: float
+    discount: float
+    effective_commission: float
+    commission_amount_sats: int
+    base_amount_sats: int
+    exchange_rate: float
+    crypto_code: str
+    fiat_code: str
+    device_id: Optional[str]
+    transaction_time: datetime
+    processed_at: datetime
+    clients_count: int  # Number of clients who received distributions
+    distributions_total_sats: int  # Total sats distributed to clients
+
+
 # Lamassu Configuration Models
 class CreateLamassuConfigData(BaseModel):
     host: str

transaction_processor.py

@@ -35,9 +35,11 @@ from .crud import (
     update_config_test_result,
     update_poll_start_time,
     update_poll_success_time,
-    update_dca_payment_status
+    update_dca_payment_status,
+    create_lamassu_transaction,
+    update_lamassu_transaction_distribution_stats
 )
-from .models import CreateDcaPaymentData, LamassuTransaction, DcaClient
+from .models import CreateDcaPaymentData, LamassuTransaction, DcaClient, CreateLamassuTransactionData
 
 
 class LamassuTransactionProcessor:
@@ -746,6 +748,54 @@ class LamassuTransactionProcessor:
         except Exception as e:
             logger.error(f"Error updating payment status for {payment_id}: {e}")
 
+    async def store_lamassu_transaction(self, transaction: Dict[str, Any]) -> Optional[str]:
+        """Store the Lamassu transaction in our database for audit and UI"""
+        try:
+            # Extract and validate transaction data
+            crypto_atoms = transaction.get("crypto_amount", 0)
+            fiat_amount = transaction.get("fiat_amount", 0)
+            commission_percentage = transaction.get("commission_percentage") or 0.0
+            discount = transaction.get("discount") or 0.0
+            
+            # Calculate commission metrics
+            if commission_percentage > 0:
+                effective_commission = commission_percentage * (100 - discount) / 100
+                base_crypto_atoms = int(crypto_atoms / (1 + effective_commission))
+                commission_amount_sats = crypto_atoms - base_crypto_atoms
+            else:
+                effective_commission = 0.0
+                base_crypto_atoms = crypto_atoms
+                commission_amount_sats = 0
+            
+            # Calculate exchange rate
+            exchange_rate = base_crypto_atoms / fiat_amount if fiat_amount > 0 else 0
+            
+            # Create transaction data
+            transaction_data = CreateLamassuTransactionData(
+                lamassu_transaction_id=transaction["transaction_id"],
+                fiat_amount=fiat_amount,
+                crypto_amount=crypto_atoms,
+                commission_percentage=commission_percentage,
+                discount=discount,
+                effective_commission=effective_commission,
+                commission_amount_sats=commission_amount_sats,
+                base_amount_sats=base_crypto_atoms,
+                exchange_rate=exchange_rate,
+                crypto_code=transaction.get("crypto_code", "BTC"),
+                fiat_code=transaction.get("fiat_code", "GTQ"),
+                device_id=transaction.get("device_id"),
+                transaction_time=transaction.get("transaction_time")
+            )
+            
+            # Store in database
+            stored_transaction = await create_lamassu_transaction(transaction_data)
+            logger.info(f"Stored Lamassu transaction {transaction['transaction_id']} in database")
+            return stored_transaction.id
+            
+        except Exception as e:
+            logger.error(f"Error storing Lamassu transaction {transaction.get('transaction_id', 'unknown')}: {e}")
+            return None
+
     async def send_commission_payment(self, transaction: Dict[str, Any], commission_amount_sats: int) -> bool:
         """Send commission to the configured commission wallet"""
         try:
@@ -819,6 +869,9 @@ class LamassuTransactionProcessor:
                 logger.error(f"Failed to credit source wallet for transaction {transaction_id} - skipping distribution")
                 return
             
+            # Store the transaction in our database for audit and UI
+            stored_transaction = await self.store_lamassu_transaction(transaction)
+            
             # Calculate distribution amounts
             distributions = await self.calculate_distribution_amounts(transaction)
             
@@ -845,6 +898,16 @@ class LamassuTransactionProcessor:
             if commission_amount_sats > 0:
                 await self.send_commission_payment(transaction, commission_amount_sats)
             
+            # Update distribution statistics in stored transaction
+            if stored_transaction:
+                clients_count = len(distributions)
+                distributions_total_sats = sum(dist["sats_amount"] for dist in distributions.values())
+                await update_lamassu_transaction_distribution_stats(
+                    stored_transaction, 
+                    clients_count, 
+                    distributions_total_sats
+                )
+            
             logger.info(f"Successfully processed transaction {transaction_id}")
             
         except Exception as e: