"""Outbox Processor.""" from __future__ import annotations import uuid from datetime import datetime, timezone from typing import Iterable from lambdacommon.common_config import logger from src.connectors import EventBridgeConnector, Repository from src.enums import AbacusOutboxStatus from src.schemas import AbacusOutbox, EventBridgeEventMetadata, OutboxProcessResponse from src.utils import calc_retry_delta, try_json class OutboxProcessor: """Outbox processor with shared event processing logic. Provides common functionality for publishing events to EventBridge and managing event status in the database. Handles both successful processing and error scenarios with retry scheduling. Idempotency: - Skips events not in PENDING/FAILED status and respects max_retries. - For MySQL iterator: Row-level locks (FOR UPDATE SKIP LOCKED) prevent concurrent processing of the same event. - For Kafka CDC iterator: Kafka redelivery can cause duplicate EventBridge publishes. Downstream consumers MUST implement deduplication using the 'outbox_event_id' field in the event metadata. Transaction Management: Commits status updates (both success and error) immediately with optimistic locking. If update affects 0 rows (another consumer already processed the event), the transaction is rolled back. This ensures proper retry scheduling and prevents stale lock retention. """ def __init__( self, repository: Repository, eventbridge_connector: EventBridgeConnector, ): """Initialize processor. Args: repository: Repository instance for database operations eventbridge_connector: EventBridgeConnector instance for publishing events """ self._repository = repository self._eb_connector = eventbridge_connector def process(self, outbox_events: Iterable[AbacusOutbox]) -> OutboxProcessResponse: """Process outbox events from an iterable. Iterates through all provided events, publishes them to EventBridge, and updates their status in the database. Skips already-completed events (idempotency). Continues processing remaining events even if individual events fail. Args: outbox_events: Iterable of AbacusOutbox events to process Returns: OutboxProcessResponse: Summary with total, processed, skipped, and failed counts Note: Each event's status is committed independently. Failed events are marked as FAILED with retry scheduling. Failures are counted but do not stop processing of remaining events. """ counts = OutboxProcessResponse() for outbox_event in outbox_events: counts.total += 1 if not self._should_process_event(outbox_event): counts.skipped += 1 continue try: self._process_event(outbox_event) counts.processed += 1 except Exception: counts.failed += 1 return counts def _process_event(self, event: AbacusOutbox) -> None: """Process a single outbox event. Publishes the event to EventBridge and updates its status in the database. Assumes the event has already been validated by _should_process_event(). Args: event: The outbox event to process Raises: Exception: Re-raises any exception encountered during processing (EventBridge publish, DB update, or error recording failures) Note: Status updates (both success and error) are committed immediately. Failed events are marked as FAILED with retry scheduling before re-raising the exception. """ event_id = event.abacus_outbox_id logger.info(f'Processing event ID: {event_id}') try: self._publish_to_eventbridge(event) self._update_outbox_completed(event) except Exception as e: logger.error(f'Failed to process event ID {event_id}: {e}') try: self._update_outbox_error(event, str(e)) except Exception as db_err: logger.warning(f'Cannot update event ID {event_id}: {db_err}') raise e from db_err else: raise e def _publish_to_eventbridge(self, event: AbacusOutbox) -> None: """Publish event to EventBridge. Args: event: AbacusOutbox object. """ correlation_id = event.correlation_id or str(uuid.uuid4()) data = try_json(event.details) if event.details else {} event_id = event.abacus_outbox_id event_type = event.event_type target_id = event.target_id target_type = event.target_type created_at = event.created_at.isoformat() processed_at = datetime.now(timezone.utc).isoformat() logger.info( f'Publishing event: correlation_id={correlation_id}, ' f'event_id={event_id}, event_type={event_type}, ' f'target_id={target_id}, target_type={target_type}' ) self._eb_connector.put_event( detail_type=event_type, metadata=EventBridgeEventMetadata( correlation_id=correlation_id, outbox_event_id=event_id, target_id=target_id, target_type=target_type, created_at=created_at, processed_at=processed_at, ), data=data, ) logger.info('Event published') def _should_process_event(self, event: AbacusOutbox) -> bool: """Determine if an event should be processed. Args: event: The outbox event to evaluate Returns: bool: True if event should be processed (PENDING status, or FAILED status with retries remaining). False if event should be skipped (non-processable status like COMPLETED, or FAILED with retries exhausted). """ event_id = event.abacus_outbox_id if event.status == AbacusOutboxStatus.PENDING: return True if event.status != AbacusOutboxStatus.FAILED: logger.info(f'Skipping event ID {event_id}: status "{event.status}"') return False if event.retry_count >= event.max_retries: logger.info(f'Skipping event ID {event_id}: Retries exhausted') return False return True def _update_outbox_completed(self, event: AbacusOutbox) -> None: """Mark event as completed and update database. Args: event: The outbox event to mark as completed. """ event_id = event.abacus_outbox_id processed_at = datetime.now(timezone.utc) status = AbacusOutboxStatus.COMPLETED logger.info(f'Updating status to: {status}') rows_affected = self._repository.update_event_status( event_id=event_id, status=status, processed_at=processed_at, ) if rows_affected == 0: logger.warning(f'Event {event_id} already processed by another consumer') self._repository.rollback() else: self._repository.commit() logger.info('Event status updated') def _update_outbox_error(self, event: AbacusOutbox, error_msg: str) -> None: """Mark event as failed and schedule retry. Args: event: The outbox event that failed. error_msg: The error message to store. """ event_id = event.abacus_outbox_id processed_at = datetime.now(timezone.utc) next_retry_at = processed_at + calc_retry_delta(event.retry_count) status = AbacusOutboxStatus.FAILED logger.info(f'Updating status to: {status}, next retry at: {next_retry_at}') rows_affected = self._repository.update_event_status( event_id=event_id, status=status, error_message=error_msg, next_retry_at=next_retry_at, ) if rows_affected == 0: logger.warning(f'Event {event_id} already processed by another consumer') self._repository.rollback() else: self._repository.commit() logger.info('Event status updated')