Add durable platform event delivery contract

src/govoplan_core/celery_app.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from datetime import datetime, timedelta, timezone
+
 from celery import Celery
 
 from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CampaignDeliveryTaskProvider
@@ -10,6 +12,7 @@ from govoplan_core.core.dataflows import (
 )
 from govoplan_core.core.events import (
     CAPABILITY_PLATFORM_EVENT_OUTBOX,
+    DurableEventConsumer,
     PlatformEvent,
     PlatformEventOutbox,
     publish_platform_event,
@@ -41,6 +44,7 @@ celery.conf.update(
         "govoplan.calendar.dispatch_outbox": {"queue": "calendar"},
         "govoplan.dataflow.dispatch_triggers": {"queue": "dataflow"},
         "govoplan.events.dispatch_outbox": {"queue": "events"},
+        "govoplan.events.purge_outbox": {"queue": "events"},
     },
     worker_prefetch_multiplier=1,
     task_acks_late=True,
@@ -61,6 +65,11 @@ celery.conf.update(
             "schedule": 10.0,
             "args": (100,),
         },
+        "platform-event-retention-daily": {
+            "task": "govoplan.events.purge_outbox",
+            "schedule": 24 * 60 * 60.0,
+            "args": (500,),
+        },
     },
 )
 
@@ -231,7 +240,7 @@ def dispatch_dataflow_triggers(self, limit: int = 100):
     max_retries=0,
 )
 def dispatch_platform_events(self, limit: int = 100):
-    """Deliver committed platform events once across all worker processes."""
+    """Deliver committed platform events through persistent consumer ledgers."""
 
     from govoplan_core.db.session import get_database
 
@@ -239,21 +248,68 @@ def dispatch_platform_events(self, limit: int = 100):
         registry = _platform_registry()
         outbox = _platform_event_outbox(registry)
         if outbox is None:
-            return {"selected": 0, "dispatched": 0, "failed": 0}
+            return {
+                "selected": 0,
+                "delivered": 0,
+                "retrying": 0,
+                "quarantined": 0,
+                "dispatched": 0,
+                "observer_failed": 0,
+            }
         dataflow_dispatcher = _dataflow_trigger_dispatcher(registry)
+        consumers = ()
+        if dataflow_dispatcher is not None:
+            def deliver_to_dataflow(
+                event: PlatformEvent,
+                _delivery_key: str,
+            ) -> None:
+                dataflow_dispatcher.ingest_event(
+                    session,
+                    event=event,
+                )
 
-        def dispatch(event: PlatformEvent) -> None:
-            if (
-                dataflow_dispatcher is not None
-                and event.classification in {"public", "internal"}
-            ):
-                dataflow_dispatcher.ingest_event(session, event=event)
-            publish_platform_event(event)
+            consumers = (
+                DurableEventConsumer(
+                    consumer_id="dataflow.event-triggers.v1",
+                    event_types=frozenset({"*"}),
+                    classifications=frozenset({"public", "internal"}),
+                    handler=deliver_to_dataflow,
+                ),
+            )
 
         result = dict(
             outbox.dispatch_pending(
                 session,
-                dispatcher=dispatch,
+                consumers=consumers,
+                observer=publish_platform_event,
+                limit=limit,
+            )
+        )
+        session.commit()
+        return result
+
+
+@celery.task(
+    name="govoplan.events.purge_outbox",
+    bind=True,
+    max_retries=0,
+)
+def purge_platform_events(self, limit: int = 500):
+    """Remove old terminal event envelopes while retaining quarantine evidence."""
+
+    from govoplan_core.db.session import get_database
+
+    with get_database().SessionLocal() as session:
+        outbox = _platform_event_outbox()
+        if outbox is None:
+            return {"deleted": 0}
+        before = datetime.now(timezone.utc) - timedelta(
+            days=settings.platform_event_outbox_terminal_retention_days
+        )
+        result = dict(
+            outbox.purge_terminal(
+                session,
+                before=before,
                 limit=limit,
             )
         )