Persist durable event consumer delivery state

src/govoplan_audit/backend/outbox.py

@@ -1,14 +1,18 @@
 from __future__ import annotations
 
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from datetime import datetime, timedelta, timezone
 from typing import Any, cast
 
-from sqlalchemy import or_
+from sqlalchemy import delete, func, or_, select
 from sqlalchemy.orm import Session
 
-from govoplan_audit.backend.db.models import AuditOutboxEvent
+from govoplan_audit.backend.db.models import (
+    AuditOutboxDelivery,
+    AuditOutboxEvent,
+)
 from govoplan_core.core.events import (
+    DurableEventConsumer,
     EventActorRef,
     EventClassification,
     EventObjectRef,
@@ -22,9 +26,23 @@ EventDispatcher = Callable[[PlatformEvent], None]
 
 
 class SqlAuditOutbox:
+    def __init__(self, *, max_attempts: int = 8) -> None:
+        self._max_attempts = max(1, min(int(max_attempts), 100))
+
     def enqueue(self, session: object, event: PlatformEvent) -> AuditOutboxEvent:
         db = _session(session)
         traced = ensure_event_trace(event)
+        existing = db.scalar(
+            select(AuditOutboxEvent).where(
+                AuditOutboxEvent.event_id == traced.event_id
+            )
+        )
+        if existing is not None:
+            if existing.payload != traced.to_dict():
+                raise ValueError(
+                    "A different platform event already uses this event id"
+                )
+            return existing
         item = AuditOutboxEvent(
             event_id=traced.event_id,
             event_type=traced.type,
@@ -42,15 +60,20 @@ class SqlAuditOutbox:
         self,
         session: object,
         *,
-        dispatcher: EventDispatcher = publish_platform_event,
+        consumers: Sequence[DurableEventConsumer] = (),
+        observer: EventDispatcher | None = publish_platform_event,
+        dispatcher: EventDispatcher | None = None,
         limit: int = 100,
     ) -> dict[str, int]:
         db = _session(session)
         now = datetime.now(timezone.utc)
+        consumers_by_id = _consumer_map(consumers)
         rows = (
             db.query(AuditOutboxEvent)
             .filter(
-                AuditOutboxEvent.status.in_(("pending", "failed")),
+                AuditOutboxEvent.status.in_(
+                    ("pending", "failed", "retrying")
+                ),
                 or_(AuditOutboxEvent.next_attempt_at.is_(None), AuditOutboxEvent.next_attempt_at <= now),
             )
             .order_by(AuditOutboxEvent.created_at.asc(), AuditOutboxEvent.id.asc())
@@ -58,26 +81,443 @@ class SqlAuditOutbox:
             .limit(max(1, min(int(limit), 500)))
             .all()
         )
-        counts = {"selected": len(rows), "dispatched": 0, "failed": 0}
+        counts = {
+            "selected": len(rows),
+            "delivered": 0,
+            "retrying": 0,
+            "quarantined": 0,
+            "dispatched": 0,
+            "observer_failed": 0,
+        }
+        effective_observer = dispatcher or observer
         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
+            event = _event_from_payload(row.payload)
+            deliveries = _event_deliveries(
+                db,
+                row=row,
+                event=event,
+                consumers=consumers_by_id.values(),
+            )
+            _dispatch_event_deliveries(
+                event,
+                deliveries=deliveries,
+                consumers_by_id=consumers_by_id,
+                now=now,
+                max_attempts=self._max_attempts,
+                counts=counts,
+            )
+            _finish_event_dispatch(
+                row,
+                deliveries=deliveries,
+                observer=effective_observer,
+                event=event,
+                now=now,
+                counts=counts,
+            )
         db.flush()
         return counts
 
+    def replay_delivery(
+        self,
+        session: object,
+        *,
+        event_id: str,
+        consumer_id: str,
+        operator_id: str,
+        reason: str,
+    ) -> dict[str, object]:
+        db = _session(session)
+        clean_reason = reason.strip()
+        clean_operator_id = operator_id.strip()
+        if not clean_reason or len(clean_reason) > 2000:
+            raise ValueError(
+                "Replay reason must contain between 1 and 2000 characters"
+            )
+        if not clean_operator_id or len(clean_operator_id) > 128:
+            raise ValueError("Replay operator id is invalid")
+        row = db.scalar(
+            select(AuditOutboxDelivery)
+            .join(
+                AuditOutboxEvent,
+                AuditOutboxEvent.id
+                == AuditOutboxDelivery.outbox_event_id,
+            )
+            .where(
+                AuditOutboxEvent.event_id == event_id,
+                AuditOutboxDelivery.consumer_id == consumer_id,
+            )
+            .with_for_update()
+        )
+        if row is None:
+            raise LookupError("Platform event delivery was not found")
+        if row.status not in {"retrying", "quarantined"}:
+            raise ValueError(
+                "Only retrying or quarantined deliveries can be replayed"
+            )
+        now = datetime.now(timezone.utc)
+        row.status = "pending"
+        row.attempts = 0
+        row.next_attempt_at = now
+        row.quarantined_at = None
+        row.last_error = None
+        row.replay_count += 1
+        row.last_replayed_at = now
+        row.last_replayed_by = clean_operator_id
+        row.last_replay_reason = clean_reason
+        event_row = db.get(AuditOutboxEvent, row.outbox_event_id)
+        if event_row is None:
+            raise LookupError("Platform event envelope was not found")
+        event_row.status = "pending"
+        event_row.next_attempt_at = now
+        event_row.last_error = None
+        event_row.dispatched_at = None
+        db.flush()
+        return _delivery_state(row, event_id=event_row.event_id)
+
+    def purge_terminal(
+        self,
+        session: object,
+        *,
+        before: datetime,
+        limit: int = 500,
+    ) -> dict[str, int]:
+        db = _session(session)
+        ids = tuple(
+            db.scalars(
+                select(AuditOutboxEvent.id)
+                .where(
+                    AuditOutboxEvent.status == "dispatched",
+                    AuditOutboxEvent.dispatched_at.is_not(None),
+                    AuditOutboxEvent.dispatched_at < before,
+                )
+                .order_by(
+                    AuditOutboxEvent.dispatched_at,
+                    AuditOutboxEvent.id,
+                )
+                .limit(max(1, min(int(limit), 5000)))
+            )
+        )
+        if ids:
+            db.execute(
+                delete(AuditOutboxEvent).where(
+                    AuditOutboxEvent.id.in_(ids)
+                )
+            )
+        db.flush()
+        return {"deleted": len(ids)}
+
+    def delivery_metrics(
+        self,
+        session: object,
+    ) -> dict[str, object]:
+        db = _session(session)
+        event_counts = {
+            str(status): int(count)
+            for status, count in db.execute(
+                select(
+                    AuditOutboxEvent.status,
+                    func.count(AuditOutboxEvent.id),
+                ).group_by(AuditOutboxEvent.status)
+            )
+        }
+        delivery_counts = {
+            str(status): int(count)
+            for status, count in db.execute(
+                select(
+                    AuditOutboxDelivery.status,
+                    func.count(AuditOutboxDelivery.id),
+                ).group_by(AuditOutboxDelivery.status)
+            )
+        }
+        consumer_counts = {
+            str(consumer_id): {
+                str(status): int(count)
+                for status, count in values
+            }
+            for consumer_id, values in _consumer_delivery_counts(db).items()
+        }
+        oldest_due = db.scalar(
+            select(func.min(AuditOutboxDelivery.created_at)).where(
+                AuditOutboxDelivery.status.in_(
+                    ("pending", "retrying")
+                )
+            )
+        )
+        return {
+            "events": event_counts,
+            "deliveries": delivery_counts,
+            "consumers": consumer_counts,
+            "oldest_due_at": (
+                oldest_due.isoformat()
+                if isinstance(oldest_due, datetime)
+                else None
+            ),
+        }
+
+
+def _consumer_map(
+    consumers: Sequence[DurableEventConsumer],
+) -> dict[str, DurableEventConsumer]:
+    result: dict[str, DurableEventConsumer] = {}
+    for consumer in consumers:
+        if consumer.consumer_id in result:
+            raise ValueError(
+                f"Duplicate durable event consumer: {consumer.consumer_id}"
+            )
+        result[consumer.consumer_id] = consumer
+    return result
+
+
+def _event_deliveries(
+    session: Session,
+    *,
+    row: AuditOutboxEvent,
+    event: PlatformEvent,
+    consumers: Sequence[DurableEventConsumer],
+) -> list[AuditOutboxDelivery]:
+    existing = {
+        delivery.consumer_id: delivery
+        for delivery in session.scalars(
+            select(AuditOutboxDelivery).where(
+                AuditOutboxDelivery.outbox_event_id == row.id
+            )
+        )
+    }
+    for consumer in consumers:
+        if (
+            consumer.consumer_id in existing
+            or not consumer.accepts(event)
+        ):
+            continue
+        delivery = AuditOutboxDelivery(
+            outbox_event_id=row.id,
+            consumer_id=consumer.consumer_id,
+            delivery_key=consumer.delivery_key(event),
+            policy_decision_ref=consumer.policy_decision_ref,
+            status="pending",
+        )
+        session.add(delivery)
+        existing[consumer.consumer_id] = delivery
+    session.flush()
+    return sorted(
+        existing.values(),
+        key=lambda item: (item.created_at, item.id),
+    )
+
+
+def _dispatch_event_deliveries(
+    event: PlatformEvent,
+    *,
+    deliveries: Sequence[AuditOutboxDelivery],
+    consumers_by_id: Mapping[str, DurableEventConsumer],
+    now: datetime,
+    max_attempts: int,
+    counts: dict[str, int],
+) -> None:
+    for delivery in deliveries:
+        if not _delivery_is_due(delivery, now=now):
+            continue
+        consumer = consumers_by_id.get(delivery.consumer_id)
+        if consumer is None:
+            _record_delivery_failure(
+                delivery,
+                error="Durable event consumer is not registered",
+                now=now,
+                max_attempts=max_attempts,
+                counts=counts,
+            )
+            continue
+        if not consumer.accepts(event):
+            _quarantine_delivery(
+                delivery,
+                error=(
+                    "The current durable subscription no longer permits "
+                    "this event"
+                ),
+                now=now,
+                counts=counts,
+            )
+            continue
+        if (
+            event.classification in {"confidential", "restricted"}
+            and delivery.policy_decision_ref
+            != consumer.policy_decision_ref
+        ):
+            _quarantine_delivery(
+                delivery,
+                error=(
+                    "The policy decision for this classified event "
+                    "subscription changed"
+                ),
+                now=now,
+                counts=counts,
+            )
+            continue
+        try:
+            consumer.handler(event, delivery.delivery_key)
+        except Exception as exc:  # noqa: BLE001 - failures must be persisted.
+            _record_delivery_failure(
+                delivery,
+                error=str(exc),
+                now=now,
+                max_attempts=max_attempts,
+                counts=counts,
+            )
+            continue
+        delivery.status = "delivered"
+        delivery.attempts += 1
+        delivery.delivered_at = now
+        delivery.next_attempt_at = None
+        delivery.quarantined_at = None
+        delivery.last_error = None
+        counts["delivered"] += 1
+
+
+def _delivery_is_due(
+    delivery: AuditOutboxDelivery,
+    *,
+    now: datetime,
+) -> bool:
+    if delivery.status not in {"pending", "retrying"}:
+        return False
+    if delivery.next_attempt_at is None:
+        return True
+    return _as_utc(delivery.next_attempt_at) <= now
+
+
+def _record_delivery_failure(
+    delivery: AuditOutboxDelivery,
+    *,
+    error: str,
+    now: datetime,
+    max_attempts: int,
+    counts: dict[str, int],
+) -> None:
+    delivery.attempts += 1
+    delivery.last_error = _bounded_error(error)
+    if delivery.attempts >= max_attempts:
+        _quarantine_delivery(
+            delivery,
+            error=delivery.last_error,
+            now=now,
+            counts=counts,
+        )
+        return
+    delivery.status = "retrying"
+    delivery.next_attempt_at = now + _retry_delay(delivery.attempts)
+    counts["retrying"] += 1
+
+
+def _quarantine_delivery(
+    delivery: AuditOutboxDelivery,
+    *,
+    error: str,
+    now: datetime,
+    counts: dict[str, int],
+) -> None:
+    delivery.status = "quarantined"
+    delivery.quarantined_at = now
+    delivery.next_attempt_at = None
+    delivery.last_error = _bounded_error(error)
+    counts["quarantined"] += 1
+
+
+def _finish_event_dispatch(
+    row: AuditOutboxEvent,
+    *,
+    deliveries: Sequence[AuditOutboxDelivery],
+    observer: EventDispatcher | None,
+    event: PlatformEvent,
+    now: datetime,
+    counts: dict[str, int],
+) -> None:
+    row.attempts += 1
+    quarantined = [
+        item for item in deliveries
+        if item.status == "quarantined"
+    ]
+    outstanding = [
+        item for item in deliveries
+        if item.status in {"pending", "retrying"}
+    ]
+    if quarantined:
+        row.status = "quarantined"
+        row.next_attempt_at = None
+        row.last_error = quarantined[0].last_error
+        return
+    if outstanding:
+        row.status = "retrying"
+        due_times = [
+            item.next_attempt_at
+            for item in outstanding
+            if item.next_attempt_at is not None
+        ]
+        row.next_attempt_at = min(due_times) if due_times else now
+        row.last_error = next(
+            (
+                item.last_error
+                for item in outstanding
+                if item.last_error
+            ),
+            None,
+        )
+        return
+    if observer is not None:
+        try:
+            observer(event)
+        except Exception as exc:  # noqa: BLE001 - observers are non-durable.
+            counts["observer_failed"] += 1
+            row.last_error = _bounded_error(
+                f"Non-durable observer failed: {exc}"
+            )
+        else:
+            row.last_error = None
+    else:
+        row.last_error = None
+    row.status = "dispatched"
+    row.dispatched_at = now
+    row.next_attempt_at = None
+    counts["dispatched"] += 1
+
+
+def _delivery_state(
+    delivery: AuditOutboxDelivery,
+    *,
+    event_id: str,
+) -> dict[str, object]:
+    return {
+        "event_id": event_id,
+        "consumer_id": delivery.consumer_id,
+        "delivery_key": delivery.delivery_key,
+        "status": delivery.status,
+        "attempts": delivery.attempts,
+        "replay_count": delivery.replay_count,
+        "last_replayed_at": delivery.last_replayed_at,
+        "last_replayed_by": delivery.last_replayed_by,
+        "last_replay_reason": delivery.last_replay_reason,
+        "last_error": delivery.last_error,
+    }
+
+
+def _consumer_delivery_counts(
+    session: Session,
+) -> dict[str, list[tuple[str, int]]]:
+    result: dict[str, list[tuple[str, int]]] = {}
+    for consumer_id, status, count in session.execute(
+        select(
+            AuditOutboxDelivery.consumer_id,
+            AuditOutboxDelivery.status,
+            func.count(AuditOutboxDelivery.id),
+        ).group_by(
+            AuditOutboxDelivery.consumer_id,
+            AuditOutboxDelivery.status,
+        )
+    ):
+        result.setdefault(str(consumer_id), []).append(
+            (str(status), int(count))
+        )
+    return result
+
 
 def enqueue_platform_event(session: object, event: PlatformEvent) -> AuditOutboxEvent:
     return SqlAuditOutbox().enqueue(session, event)
@@ -97,6 +537,17 @@ def _retry_delay(attempts: int) -> timedelta:
     return timedelta(seconds=seconds)
 
 
+def _as_utc(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def _bounded_error(value: str) -> str:
+    clean = value.strip() or "Unknown durable event delivery failure"
+    return clean[:4000]
+
+
 def _event_from_payload(payload: Mapping[str, Any]) -> PlatformEvent:
     return PlatformEvent(
         type=str(payload["type"]),