From 7117673ecc2b2b8c15abf47613cf24487d183408 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 03:03:15 +0200 Subject: [PATCH] Add durable search event indexing contract --- docs/SEARCH_EVENT_INDEXING_CONTRACT.md | 26 ++++++++++ src/govoplan_core/celery_app.py | 46 +++++++++++++++++ src/govoplan_core/core/search.py | 40 +++++++++++++++ tests/test_module_system.py | 4 +- tests/test_platform_event_worker.py | 68 ++++++++++++++++++++++++++ 5 files changed, 183 insertions(+), 1 deletion(-) create mode 100644 docs/SEARCH_EVENT_INDEXING_CONTRACT.md diff --git a/docs/SEARCH_EVENT_INDEXING_CONTRACT.md b/docs/SEARCH_EVENT_INDEXING_CONTRACT.md new file mode 100644 index 0000000..716172c --- /dev/null +++ b/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. diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index af291f0..2b036f7 100644 --- a/src/govoplan_core/celery_app.py +++ b/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 diff --git a/src/govoplan_core/core/search.py b/src/govoplan_core/core/search.py index 7f9e9b1..f1b6a0f 100644 --- a/src/govoplan_core/core/search.py +++ b/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", diff --git a/tests/test_module_system.py b/tests/test_module_system.py index f0d039e..66409a6 100644 --- a/tests/test_module_system.py +++ b/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"], ) diff --git a/tests/test_platform_event_worker.py b/tests/test_platform_event_worker.py index d73dd2d..da589b3 100644 --- a/tests/test_platform_event_worker.py +++ b/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()