From 0aae6f05397b13978acaa4bcd9ba08f60b6417a2 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 20 Aug 2026 04:58:35 +0200 Subject: [PATCH] feat(postbox): schedule lifecycle notification reconciliation --- src/govoplan_core/celery_app.py | 35 +++++++++++++++++++++------- src/govoplan_core/core/postbox.py | 12 ++++++++++ tests/test_postbox_routing_worker.py | 24 +++++++++++++++++++ 3 files changed, 63 insertions(+), 8 deletions(-) diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index 0966e75..1c02e65 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -1188,7 +1188,7 @@ def dispatch_postbox_routes( tenant_id: str | None = None, limit: int = 50, ): - """Deliver due Postbox vacancy escalations from durable route rows.""" + """Deliver due routes and reconcile assignment-derived notification facts.""" from govoplan_core.db.session import get_database @@ -1202,21 +1202,40 @@ def dispatch_postbox_routes( "cancelled": 0, "failed": 0, "route_ids": [], + "lifecycle_scanned": 0, + "lifecycle_changed": 0, + "lifecycle_events": 0, + "lifecycle_notifications": 0, + "lifecycle_notification_failures": 0, } if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING): return defaults + provider = _postbox_routing_provider(registry) + + def dispatch_tenant(effective_tenant_id: str) -> Mapping[str, object]: + route_result = provider.dispatch_due_routes( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ) + lifecycle_result = provider.reconcile_notification_lifecycle( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ) + return { + **route_result, + **{ + f"lifecycle_{key}": value for key, value in lifecycle_result.items() + }, + } + result = _run_tenant_worker_batches( registry, session, capability_name=CAPABILITY_POSTBOX_ROUTING, tenant_id=tenant_id, - operation=lambda effective_tenant_id: _postbox_routing_provider( - registry - ).dispatch_due_routes( # type: ignore[union-attr] - session, - tenant_id=effective_tenant_id, - limit=limit, - ), + operation=dispatch_tenant, defaults=defaults, ) session.commit() diff --git a/src/govoplan_core/core/postbox.py b/src/govoplan_core/core/postbox.py index d6416be..3085806 100644 --- a/src/govoplan_core/core/postbox.py +++ b/src/govoplan_core/core/postbox.py @@ -309,6 +309,7 @@ class PostboxDeliveryRequest: body_text: str | None = None sender_label: str | None = None classification: str = "internal" + action_required: bool = False participants: tuple[PostboxParticipantRef, ...] = () attachments: tuple[PostboxAttachmentRef, ...] = () expires_at: datetime | None = None @@ -511,6 +512,17 @@ class PostboxRoutingProvider(Protocol): ) -> Mapping[str, object]: ... + def reconcile_notification_lifecycle( + self, + session: object, + *, + tenant_id: str | None = None, + limit: int = 50, + ) -> Mapping[str, object]: + """Reconcile assignment-derived Postbox notification facts.""" + + ... + @runtime_checkable class PostboxPortalProjectionProvider(Protocol): diff --git a/tests/test_postbox_routing_worker.py b/tests/test_postbox_routing_worker.py index 892ffd6..d764f31 100644 --- a/tests/test_postbox_routing_worker.py +++ b/tests/test_postbox_routing_worker.py @@ -25,6 +25,16 @@ class _RoutingProvider: del session, tenant_id, limit return {"selected": 0} + def reconcile_notification_lifecycle( + self, + session, + *, + tenant_id=None, + limit=50, + ): + del session, tenant_id, limit + return {"scanned": 0} + class PostboxRoutingWorkerTests(unittest.TestCase): def test_contract_is_runtime_checkable_and_resolved(self) -> None: @@ -59,6 +69,13 @@ class PostboxRoutingWorkerTests(unittest.TestCase): "selected": 1, "delivered": 1, } + provider.reconcile_notification_lifecycle.return_value = { + "scanned": 2, + "changed": 1, + "events": 2, + "notifications": 1, + "notification_failures": 0, + } with ( patch( @@ -81,8 +98,15 @@ class PostboxRoutingWorkerTests(unittest.TestCase): tenant_id="tenant-1", limit=25, ) + provider.reconcile_notification_lifecycle.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) session.commit.assert_called_once_with() self.assertEqual(1, result["delivered"]) + self.assertEqual(2, result["lifecycle_scanned"]) + self.assertEqual(1, result["lifecycle_notifications"]) def test_route_and_periodic_recovery_are_registered(self) -> None: self.assertEqual(