Add durable search event indexing contract

docs/SEARCH_EVENT_INDEXING_CONTRACT.md

@@ -0,0 +1,26 @@
+# Search event indexing contract
+
+Core defines, but does not implement, the optional Search indexing boundary.
+Feature modules register `SearchSourceProvider` implementations for bounded
+backfills and live authorization checks. A provider may additionally implement
+`SearchEventSourceProvider` to translate a committed `PlatformEvent` into one
+or more authoritative `SearchIndexChange` values.
+
+When the Search index-writer capability is active, the platform event worker
+uses the durable consumer identity `search.indexing.v1`. It accepts only public
+and internal events, passes the outbox delivery key to each event-capable
+source, and then advances a bounded batch of queued index changes in the same
+worker transaction. Stable change IDs make delivery replay idempotent.
+
+The boundary has three non-negotiable rules:
+
+- a source may emit changes only for its registered module, provider, resource
+  type, and event tenant;
+- Search validates every upsert document before queueing it and rejects secret
+  metadata keys;
+- an index ACL is only a candidate filter. Resources marked for authorization
+  recheck are returned only after the owning source explicitly allows the
+  current principal at query time.
+
+Search and its worker remain optional. Core-only startup and feature-module
+operation do not require the Search package.

src/govoplan_core/celery_app.py

@@ -75,6 +75,10 @@ from govoplan_core.core.runtime_coordination import (
     runtime_identity,
     stop_runtime_node,
 )
+from govoplan_core.core.search import (
+    CAPABILITY_SEARCH_INDEX_WRITER,
+    SearchIndexCoordinator,
+)
 from govoplan_core.settings import settings
 from govoplan_core.db.session import configure_database, get_database
 from govoplan_core.server.registry import (
@@ -528,6 +532,20 @@ def _platform_event_outbox(
     return capability
 
 
+def _search_index_coordinator(
+    registry: PlatformRegistry | None = None,
+) -> SearchIndexCoordinator | None:
+    registry = registry or _platform_registry()
+    if not registry.has_capability(CAPABILITY_SEARCH_INDEX_WRITER):
+        return None
+    capability = registry.require_capability(
+        CAPABILITY_SEARCH_INDEX_WRITER
+    )
+    if not isinstance(capability, SearchIndexCoordinator):
+        raise RuntimeError("Search index coordinator capability is invalid")
+    return capability
+
+
 def _idm_assignment_lifecycle(
     registry: PlatformRegistry | None = None,
 ) -> IdmAssignmentLifecycle | None:
@@ -884,6 +902,7 @@ def dispatch_platform_events(self, limit: int = 100):
             }
         dataflow_dispatcher = _dataflow_trigger_dispatcher(registry)
         workflow_dispatcher = _workflow_trigger_dispatcher(registry)
+        search_coordinator = _search_index_coordinator(registry)
         consumers: list[DurableEventConsumer] = []
         if dataflow_dispatcher is not None:
 
@@ -923,6 +942,26 @@ def dispatch_platform_events(self, limit: int = 100):
                     handler=deliver_to_workflow,
                 )
             )
+        if search_coordinator is not None:
+
+            def deliver_to_search(
+                event: PlatformEvent,
+                delivery_key: str,
+            ) -> None:
+                search_coordinator.ingest_event(
+                    session,
+                    event=event,
+                    delivery_key=delivery_key,
+                )
+
+            consumers.append(
+                DurableEventConsumer(
+                    consumer_id="search.indexing.v1",
+                    event_types=frozenset({"*"}),
+                    classifications=frozenset({"public", "internal"}),
+                    handler=deliver_to_search,
+                )
+            )
 
         result = dict(
             outbox.dispatch_pending(
@@ -932,6 +971,13 @@ def dispatch_platform_events(self, limit: int = 100):
                 limit=limit,
             )
         )
+        if search_coordinator is not None:
+            result["search_changes"] = dict(
+                search_coordinator.process_changes(
+                    session,
+                    limit=limit,
+                )
+            )
         session.commit()
         return result
 

src/govoplan_core/core/search.py

@@ -5,6 +5,7 @@ from dataclasses import dataclass, field
 from datetime import datetime
 from typing import Literal, Protocol, runtime_checkable
 
+from govoplan_core.core.events import PlatformEvent
 from govoplan_core.core.external_references import ExternalObjectReference
 from govoplan_core.core.modules import ModuleContext
 
@@ -379,6 +380,43 @@ class SearchSourceProvider(Protocol):
         """Return an explicit decision for every requested reference key."""
 
 
+@runtime_checkable
+class SearchEventSourceProvider(Protocol):
+    """Optional source extension for committed, idempotent index deltas."""
+
+    def index_changes_for_event(
+        self,
+        session: object,
+        *,
+        event: PlatformEvent,
+        delivery_key: str,
+    ) -> Sequence[SearchIndexChange]:
+        """Translate one committed event into authoritative index changes."""
+
+
+@runtime_checkable
+class SearchIndexCoordinator(Protocol):
+    """Worker-facing orchestration surface exposed by the Search module."""
+
+    def ingest_event(
+        self,
+        session: object,
+        *,
+        event: PlatformEvent,
+        delivery_key: str,
+    ) -> Mapping[str, int]:
+        ...
+
+    def process_changes(
+        self,
+        session: object,
+        *,
+        limit: int = 100,
+        tenant_id: str | None = None,
+    ) -> Mapping[str, int]:
+        ...
+
+
 SearchProviderFactory = Callable[[ModuleContext], SearchProvider]
 SearchSourceProviderFactory = Callable[
     [ModuleContext],
@@ -455,8 +493,10 @@ __all__ = [
     "SearchBackfillRequest",
     "SearchContextKind",
     "SearchDocument",
+    "SearchEventSourceProvider",
     "SearchIndexChange",
     "SearchIndexChangeKind",
+    "SearchIndexCoordinator",
     "SearchIndexWriter",
     "SearchProvider",
     "SearchProviderFactory",

tests/test_module_system.py

@@ -250,6 +250,7 @@ class ModuleSystemTests(unittest.TestCase):
                 "postbox",
                 "approvals",
                 "reporting",
+                "search",
             ),
         )
         self.assertEqual(manifests["dashboard"].dependencies, ())
@@ -3517,7 +3518,7 @@ finally:
             "version_max_exclusive": "0.2.0",
         }, modules["files"]["requires_interfaces"])
         self.assertEqual(
-            ["campaigns", "encryption"],
+            ["campaigns", "encryption", "search"],
             modules["files"]["optional_dependencies"],
         )
         self.assertIn({"name": "mail.campaign_delivery", "version": "0.2.0"}, modules["mail"]["provides_interfaces"])
@@ -3581,6 +3582,7 @@ finally:
                 "postbox",
                 "approvals",
                 "reporting",
+                "search",
             ],
             modules["campaigns"]["optional_dependencies"],
         )

tests/test_platform_event_worker.py

@@ -46,6 +46,10 @@ class PlatformEventWorkerTests(unittest.TestCase):
                 "govoplan_core.celery_app._workflow_trigger_dispatcher",
                 return_value=None,
             ),
+            patch(
+                "govoplan_core.celery_app._search_index_coordinator",
+                return_value=None,
+            ),
             patch(
                 "govoplan_core.db.session.get_database",
                 return_value=database,
@@ -77,6 +81,70 @@ class PlatformEventWorkerTests(unittest.TestCase):
         session.commit.assert_called_once_with()
         self.assertEqual(1, result["delivered"])
 
+    def test_dispatch_uses_a_durable_search_consumer_and_processes_changes(self) -> None:
+        session = MagicMock()
+        database = MagicMock()
+        database.SessionLocal.return_value.__enter__.return_value = session
+        outbox = MagicMock()
+        outbox.dispatch_pending.return_value = {
+            "selected": 1,
+            "delivered": 1,
+            "retrying": 0,
+            "quarantined": 0,
+            "dispatched": 1,
+            "observer_failed": 0,
+        }
+        search = MagicMock()
+        search.process_changes.return_value = {
+            "selected": 1,
+            "applied": 1,
+            "retrying": 0,
+            "quarantined": 0,
+        }
+
+        with (
+            patch(
+                "govoplan_core.celery_app._platform_registry",
+                return_value=MagicMock(),
+            ),
+            patch(
+                "govoplan_core.celery_app._platform_event_outbox",
+                return_value=outbox,
+            ),
+            patch(
+                "govoplan_core.celery_app._dataflow_trigger_dispatcher",
+                return_value=None,
+            ),
+            patch(
+                "govoplan_core.celery_app._workflow_trigger_dispatcher",
+                return_value=None,
+            ),
+            patch(
+                "govoplan_core.celery_app._search_index_coordinator",
+                return_value=search,
+            ),
+            patch(
+                "govoplan_core.db.session.get_database",
+                return_value=database,
+            ),
+        ):
+            result = dispatch_platform_events.run(25)
+
+        consumer = outbox.dispatch_pending.call_args.kwargs["consumers"][0]
+        self.assertEqual("search.indexing.v1", consumer.consumer_id)
+        self.assertEqual(frozenset({"*"}), consumer.event_types)
+        event = PlatformEvent(type="files.file.updated", module_id="files")
+        delivery_key = consumer.delivery_key(event)
+        consumer.handler(event, delivery_key)
+        search.ingest_event.assert_called_once_with(
+            session,
+            event=event,
+            delivery_key=delivery_key,
+        )
+        search.process_changes.assert_called_once_with(session, limit=25)
+        self.assertEqual(1, result["search_changes"]["applied"])
+        session.commit.assert_called_once_with()
+
     def test_retention_task_uses_the_configured_terminal_window(self) -> None:
         session = MagicMock()
         database = MagicMock()