from datetime import timedelta from typing import Any from uuid import uuid4 from fastapi import HTTPException, status from sqlalchemy import or_, select, update from app.application.events.handlers import ( EventHandlerMixin, UnsupportedEventTypeError, ) from app.core.config import get_settings from app.core.constants import ActorValue from app.core.http.pagination import bounded_limit from app.core.utils.time import utc_now from app.modules.audit.constants import ( AuditAction, AuditRiskLevel, AuditSource, AuditTargetType, ) from app.modules.audit.schemas import AuditLogCreate from app.modules.audit.service import AuditService from app.modules.events.constants import ( EventErrorDetail, EventPayloadKey, EventStatus, ) from app.modules.events.models import DomainEvent from app.modules.events.services import EventService from app.modules.events.services.serialization import _serialize_event class EventDispatchService(EventHandlerMixin): """Claim and dispatch outbox events at the application boundary.""" def __init__(self, db: Any): self.db = db self.events = EventService(db) def get_event(self, event_id: str) -> DomainEvent: return self.events.get_event(event_id) def dispatch_event( self, event_id: str, worker_id: str | None = None, preclaimed: bool = False, ) -> DomainEvent: if preclaimed: lock_owner = worker_id or "" else: lock_owner = f"{worker_id or 'api'}:{uuid4().hex}" record = self.get_event(event_id) if preclaimed else self._claim_event(event_id, lock_owner) if record.status == EventStatus.PROCESSED: return record if preclaimed and record.locked_by != lock_owner: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=EventErrorDetail.EVENT_LOCKED, ) if not preclaimed and record.locked_by != lock_owner: return record settings = get_settings() try: self._handle_event(record) except Exception as exc: self.db.rollback() record = self.get_event(event_id) retryable = not isinstance( exc, UnsupportedEventTypeError ) and record.attempts < self._max_attempts(record) return self._finalize_event( event_id, lock_owner, status_value=(EventStatus.PENDING if retryable else EventStatus.FAILED), last_error=f"{type(exc).__name__}: {exc}"[:2000], next_attempt_at=( utc_now() + timedelta(seconds=settings.event_dispatch_retry_delay_seconds) if retryable else None ), processed_at=None, ) return self._finalize_event( event_id, lock_owner, status_value=EventStatus.PROCESSED, last_error=None, next_attempt_at=None, processed_at=utc_now(), ) def dispatch_pending( self, limit: int = 100, worker_id: str | None = None, ) -> list[dict[str, Any]]: now = utc_now() stmt = ( select(DomainEvent.event_id) .where( DomainEvent.status == EventStatus.PENDING, or_( DomainEvent.next_attempt_at.is_(None), DomainEvent.next_attempt_at <= now, ), or_( DomainEvent.locked_until.is_(None), DomainEvent.locked_until <= now, ), or_( DomainEvent.max_attempts.is_(None), DomainEvent.attempts < DomainEvent.max_attempts, ), ) .order_by(DomainEvent.id.asc()) .limit(bounded_limit(limit)) ) event_ids = list(self.db.execute(stmt).scalars()) lock_owner = f"{worker_id or 'worker'}:{uuid4().hex}" dispatched: list[dict[str, Any]] = [] for event_id in event_ids: try: record = self.dispatch_event( event_id, worker_id=lock_owner, ) except HTTPException as exc: if exc.status_code == status.HTTP_409_CONFLICT: continue raise dispatched.append(_serialize_event(record)) return dispatched def retry_event(self, event_id: str, actor: str = ActorValue.API) -> DomainEvent: record = self.db.execute( select(DomainEvent).where(DomainEvent.event_id == event_id).with_for_update() ).scalar_one_or_none() if record is None: self.db.rollback() raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=EventErrorDetail.EVENT_NOT_FOUND, ) now = utc_now() if ( record.locked_by is not None and record.locked_until is not None and record.locked_until > now ): self.db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=EventErrorDetail.EVENT_LOCKED, ) if record.status == EventStatus.PROCESSED: self.db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=EventErrorDetail.EVENT_NOT_RETRYABLE, ) record.status = EventStatus.PENDING record.attempts = 0 record.max_attempts = record.max_attempts or get_settings().event_dispatch_max_attempts record.last_error = None record.locked_by = None record.locked_until = None record.next_attempt_at = utc_now() self.db.commit() self.db.refresh(record) return record def _finalize_event( self, event_id: str, lock_owner: str, *, status_value: str, last_error: str | None, next_attempt_at: Any, processed_at: Any, ) -> DomainEvent: result = self.db.execute( update(DomainEvent) .where( DomainEvent.event_id == event_id, DomainEvent.locked_by == lock_owner, DomainEvent.status == EventStatus.PENDING, ) .values( status=status_value, last_error=last_error, next_attempt_at=next_attempt_at, processed_at=processed_at, locked_by=None, locked_until=None, ) .execution_options(synchronize_session=False) ) if result.rowcount != 1: self.db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=EventErrorDetail.EVENT_LOCKED, ) self.db.expire_all() record = self.get_event(event_id) self._stage_dispatch_audit(record) self.db.commit() self.db.refresh(record) return record def _stage_dispatch_audit(self, record: DomainEvent) -> None: AuditService(self.db).record( AuditLogCreate( actor=record.actor, source=AuditSource.EVENTS, action=AuditAction.EVENT_DISPATCH, target_type=AuditTargetType.DOMAIN_EVENT, target_id=record.event_id, risk_level=AuditRiskLevel.LOW, response_payload={ EventPayloadKey.STATUS: record.status, EventPayloadKey.ATTEMPTS: record.attempts, EventPayloadKey.ERROR_MESSAGE: record.last_error, }, ) ) def _can_attempt(self, record: DomainEvent) -> bool: return record.attempts < self._max_attempts(record) def _claim_event(self, event_id: str, lock_owner: str) -> DomainEvent: now = utc_now() record = self.db.execute( select(DomainEvent) .where(DomainEvent.event_id == event_id) .with_for_update(skip_locked=True) ).scalar_one_or_none() if record is None: self.db.rollback() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=EventErrorDetail.EVENT_LOCKED, ) if record.status == EventStatus.PROCESSED or not self._can_attempt(record): self.db.commit() return record if record.locked_until is not None and record.locked_until > now: self.db.commit() raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=EventErrorDetail.EVENT_LOCKED, ) record.locked_by = lock_owner record.locked_until = now + timedelta(seconds=get_settings().event_dispatch_lock_seconds) record.status = EventStatus.PENDING record.attempts += 1 self.db.commit() self.db.refresh(record) return record @staticmethod def _max_attempts(record: DomainEvent) -> int: return record.max_attempts or get_settings().event_dispatch_max_attempts