Add audit command and outbox foundations

src/govoplan_audit/backend/outbox.py

@@ -0,0 +1,157 @@
+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")))