""" Developer portal event listener. Subscribes to payment events and dispatches outbound webhooks to merchant endpoints. Auto-discovered on startup by src/events/listener.py. """ import logging from src.events.dispatcher import on_event_async from src.events.base import BaseEvent from src.apps.events.models import ProcessedEvent logger = logging.getLogger(__name__) HANDLED_EVENTS = [ "transaction.completed", "transaction.failed", "transaction.refunded", "payment_request.created", "payment_request.updated", "payment_request.cancelled", ] # Human-readable labels for PaymentRequestStatusTypes numeric values _PR_STATUS_NAMES: dict[int, str] = { 100: "Created", 101: "Updated", 102: "Draft", 103: "Pending", 200: "Waiting", 201: "Upcoming", 202: "Retrying", 203: "Awaiting Approval", 204: "Captured", 300: "Authorised", 400: "Processing", 500: "Paid", 501: "Partially Paid", 502: "Stalled", 600: "Failed", 601: "Overdue", 602: "Uncollectible", 700: "Cancelled", 701: "Declined", 800: "Refunded", 900: "Expired", } def _resolve_customer(db, customer_id: int | None, enriched: dict) -> None: """Populate customer_name and customer_email from the customers table.""" if not customer_id: return try: from sqlalchemy import select from src.apps.customers.models.customer import Customer customer = db.execute( select(Customer).where(Customer.id == customer_id) ).scalar_one_or_none() if customer: name_parts = [p for p in [customer.first_name, customer.last_name] if p] enriched["customer_name"] = " ".join(name_parts) if name_parts else None enriched["customer_email"] = customer.email except Exception as exc: logger.debug("_resolve_customer: lookup failed: %s", exc) def _enrich_transaction_data(db, data: dict) -> dict: """ Enrich transaction event data with human-readable fields: customer_name, customer_email, billing_name, currency, txn_status. Falls back gracefully if the record is not found. """ from sqlalchemy import select enriched = dict(data) transaction_id = data.get("transaction_id") if transaction_id: try: from src.apps.transactions.models.transaction import Transaction txn = db.execute( select(Transaction).where(Transaction.id == transaction_id) ).scalar_one_or_none() if txn: enriched["txn_id"] = txn.txn_id or enriched.get("txn_id") enriched["txn_status"] = str(txn.txn_status.value) if txn.txn_status else None enriched["currency"] = str(txn.currency.value) if txn.currency else "USD" enriched["billing_name"] = txn.billing_name except Exception as exc: logger.debug("_enrich_transaction_data: txn lookup failed: %s", exc) _resolve_customer(db, data.get("customer_id"), enriched) return enriched def _enrich_payment_request_data(db, data: dict) -> dict: """ Enrich payment_request event data with human-readable fields: - status_label: readable name for the numeric status code - customer_name, customer_email - currency: fetched from the payment request if not already present Falls back gracefully if the record is not found. """ from sqlalchemy import select enriched = dict(data) # Fetch currency and reference from payment request record if missing pr_id = data.get("payment_request_id") if pr_id and not enriched.get("currency"): try: from src.apps.payment_requests.models.payment_request import PaymentRequest pr = db.execute( select(PaymentRequest).where(PaymentRequest.id == pr_id) ).scalar_one_or_none() if pr: enriched["currency"] = str(pr.currency.value) if pr.currency else "USD" if not enriched.get("description") and pr.description: enriched["description"] = pr.description except Exception as exc: logger.debug("_enrich_payment_request_data: PR lookup failed: %s", exc) _resolve_customer(db, data.get("customer_id"), enriched) return enriched async def _dispatch_webhooks_for_event(event: BaseEvent, enrich: str = "") -> None: """ Load all active webhook endpoints for this merchant + event type, dispatch to each, record delivery. """ from src.core.database import SessionCelery from sqlalchemy import select from src.apps.developer.models.merchant_webhook_endpoint import MerchantWebhookEndpoint from src.apps.developer.helpers.webhook import dispatch_webhook merchant_id = event.data.get("merchant_id") if not merchant_id: logger.warning("Developer listener: event %s missing merchant_id", event.event_id) return with SessionCelery() as db: # Idempotency check existing = db.execute( select(ProcessedEvent).where( ProcessedEvent.event_id == event.event_id, ProcessedEvent.handler_name == "developer_webhook", ) ).scalar_one_or_none() if existing: logger.debug("Developer listener: event %s already processed", event.event_id) return # Mark as processed processed = ProcessedEvent( event_id=event.event_id, event_type=event.event_type, handler_name="developer_webhook", ) db.add(processed) db.commit() # Enrich with human-readable fields from the DB if enrich == "transaction": data = _enrich_transaction_data(db, event.data) elif enrich == "payment_request": data = _enrich_payment_request_data(db, event.data) else: data = event.data # Find active webhook endpoints subscribed to this event endpoints = db.execute( select(MerchantWebhookEndpoint).where( MerchantWebhookEndpoint.merchant_id == merchant_id, MerchantWebhookEndpoint.is_active == True, ) ).scalars().all() subscribed = [ep for ep in endpoints if event.event_type in (ep.events or [])] for endpoint in subscribed: await dispatch_webhook( db=db, endpoint=endpoint, event_type=event.event_type, event_id=event.event_id, merchant_id=merchant_id, data=data, ) @on_event_async("transaction.completed") async def handle_transaction_completed(event: BaseEvent) -> None: await _dispatch_webhooks_for_event(event, enrich="transaction") @on_event_async("transaction.failed") async def handle_transaction_failed(event: BaseEvent) -> None: await _dispatch_webhooks_for_event(event, enrich="transaction") @on_event_async("transaction.refunded") async def handle_transaction_refunded(event: BaseEvent) -> None: await _dispatch_webhooks_for_event(event, enrich="transaction") @on_event_async("payment_request.created") async def handle_payment_request_created(event: BaseEvent) -> None: await _dispatch_webhooks_for_event(event, enrich="payment_request") @on_event_async("payment_request.updated") async def handle_payment_request_updated(event: BaseEvent) -> None: await _dispatch_webhooks_for_event(event, enrich="payment_request") @on_event_async("payment_request.cancelled") async def handle_payment_request_cancelled(event: BaseEvent) -> None: await _dispatch_webhooks_for_event(event, enrich="payment_request")