govoplan-audit/src/govoplan_audit/backend/outbox.py

from __future__ import annotations

from collections.abc import Callable, Mapping
from datetime import datetime, timedelta, timezone
from typing import Any, cast

from sqlalchemy import or_
from sqlalchemy.orm import Session

from govoplan_audit.backend.db.models import AuditOutboxEvent
from govoplan_core.core.events import (
    EventActorRef,
    EventClassification,
    EventObjectRef,
    EventTenantRef,
    PlatformEvent,
    ensure_event_trace,
    publish_platform_event,
)

EventDispatcher = Callable[[PlatformEvent], None]


class SqlAuditOutbox:
    def enqueue(self, session: object, event: PlatformEvent) -> AuditOutboxEvent:
        db = _session(session)
        traced = ensure_event_trace(event)
        item = AuditOutboxEvent(
            event_id=traced.event_id,
            event_type=traced.type,
            module_id=traced.module_id,
            correlation_id=traced.correlation_id,
            causation_id=traced.causation_id,
            classification=traced.classification,
            payload=traced.to_dict(),
            status="pending",
        )
        db.add(item)
        db.flush()
        return item

    def dispatch_pending(
        self,
        session: object,
        *,
        dispatcher: EventDispatcher = publish_platform_event,
        limit: int = 100,
    ) -> dict[str, int]:
        db = _session(session)
        now = datetime.now(timezone.utc)
        rows = (
            db.query(AuditOutboxEvent)
            .filter(
                AuditOutboxEvent.status.in_(("pending", "failed")),
                or_(AuditOutboxEvent.next_attempt_at.is_(None), AuditOutboxEvent.next_attempt_at <= now),
            )
            .order_by(AuditOutboxEvent.created_at.asc(), AuditOutboxEvent.id.asc())
            .limit(max(1, min(int(limit), 500)))
            .all()
        )
        counts = {"selected": len(rows), "dispatched": 0, "failed": 0}
        for row in rows:
            try:
                dispatcher(_event_from_payload(row.payload))
            except Exception as exc:  # noqa: BLE001 - dispatcher errors must be retained for retry/diagnostics.
                row.status = "failed"
                row.attempts += 1
                row.last_error = str(exc)
                row.next_attempt_at = now + _retry_delay(row.attempts)
                counts["failed"] += 1
                continue
            row.status = "dispatched"
            row.attempts += 1
            row.dispatched_at = now
            row.next_attempt_at = None
            row.last_error = None
            counts["dispatched"] += 1
        db.flush()
        return counts


def enqueue_platform_event(session: object, event: PlatformEvent) -> AuditOutboxEvent:
    return SqlAuditOutbox().enqueue(session, event)


def dispatch_pending_platform_events(
    session: object,
    *,
    dispatcher: EventDispatcher = publish_platform_event,
    limit: int = 100,
) -> dict[str, int]:
    return SqlAuditOutbox().dispatch_pending(session, dispatcher=dispatcher, limit=limit)


def _retry_delay(attempts: int) -> timedelta:
    seconds = min(300, max(1, 2 ** max(0, attempts - 1)))
    return timedelta(seconds=seconds)


def _event_from_payload(payload: Mapping[str, Any]) -> PlatformEvent:
    return PlatformEvent(
        type=str(payload["type"]),
        module_id=str(payload["module_id"]),
        payload=_mapping(payload.get("payload")),
        occurred_at=_datetime(payload.get("occurred_at")),
        event_id=str(payload["event_id"]),
        correlation_id=_optional_str(payload.get("correlation_id")),
        causation_id=_optional_str(payload.get("causation_id")),
        actor=_actor_ref(payload.get("actor")),
        tenant=_tenant_ref(payload.get("tenant")),
        subject=_object_ref(payload.get("subject")),
        resource=_object_ref(payload.get("resource")),
        classification=cast(EventClassification, str(payload.get("classification") or "internal")),
    )


def _session(session: object) -> Session:
    if not isinstance(session, Session):
        raise TypeError("Audit outbox requires a SQLAlchemy Session")
    return session


def _mapping(value: object) -> dict[str, Any]:
    return dict(value) if isinstance(value, Mapping) else {}


def _optional_str(value: object) -> str | None:
    return str(value) if value is not None else None


def _datetime(value: object) -> datetime:
    if isinstance(value, datetime):
        return value
    if isinstance(value, str):
        return datetime.fromisoformat(value)
    return datetime.now(timezone.utc)


def _actor_ref(value: object) -> EventActorRef | None:
    data = _mapping(value)
    if not data:
        return None
    return EventActorRef(type=str(data["type"]), id=_optional_str(data.get("id")), label=_optional_str(data.get("label")))


def _tenant_ref(value: object) -> EventTenantRef | None:
    data = _mapping(value)
    if not data:
        return None
    return EventTenantRef(id=str(data["id"]), slug=_optional_str(data.get("slug")), label=_optional_str(data.get("label")))


def _object_ref(value: object) -> EventObjectRef | None:
    data = _mapping(value)
    if not data:
        return None
    return EventObjectRef(type=str(data["type"]), id=_optional_str(data.get("id")), label=_optional_str(data.get("label")))