From dc35cae44eeb05caa3eed81d6d0f20257a9db5ee Mon Sep 17 00:00:00 2001 From: padreug Date: Fri, 20 Jun 2025 00:49:46 +0200 Subject: [PATCH] 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 | 88 +++++++++++++++++++++++++++++++++++++++- migrations.py | 29 +++++++++++++ models.py | 37 +++++++++++++++++ transaction_processor.py | 67 +++++++++++++++++++++++++++++- 4 files changed, 218 insertions(+), 3 deletions(-) diff --git a/crud.py b/crud.py index cc3384e..d94b19b 100644 --- a/crud.py +++ b/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 + } + ) diff --git a/migrations.py b/migrations.py index 39379a3..470734b 100644 --- a/migrations.py +++ b/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 + ); + """ + ) diff --git a/models.py b/models.py index 77dac12..60656bf 100644 --- a/models.py +++ b/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 diff --git a/transaction_processor.py b/transaction_processor.py index 1db7fad..00142b2 100644 --- a/transaction_processor.py +++ b/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: