diff --git a/docs/MODULE_ARCHITECTURE.md b/docs/MODULE_ARCHITECTURE.md index 294aebe..0e96d52 100644 --- a/docs/MODULE_ARCHITECTURE.md +++ b/docs/MODULE_ARCHITECTURE.md @@ -1278,6 +1278,27 @@ runtime activation with the active tenant's effective entitlement. Entitlement does not grant a permission. Access authorization must still allow every API operation and resource. +The same boundary applies outside authenticated request handling: + +- capability factories retain their owning module, and tenant-scoped capability + lookup treats a provider that is unavailable to the tenant as absent; +- workers partition scheduled scans by tenant before claiming rows; +- new work is rejected while a module is unavailable, while already accepted + durable work remains in provider-owned storage and is reported as + `operator_action_required` instead of being dropped or executed; +- Workflow, Dataflow, event consumers, reconciliation jobs, and external-effect + outboxes run inside a tenant execution context, so their optional capability + calls inherit the same provider checks; +- public signed-link modules declare a `public_tenant_resolver`; valid token + context is resolved before the route runs and the module entitlement is then + enforced without requiring an authenticated principal. + +Entitlement resolution uses a bounded process-local cache. A local policy +mutation invalidates its tenant entry immediately; changes made by another node +become authoritative after `TENANT_MODULE_ENTITLEMENT_CACHE_TTL_SECONDS` +(five seconds by default). This is a bounded staleness optimization, not an +authorization grant: a cache miss or resolution failure fails closed. + Users and groups do not own another module-runtime state. Every WebUI module already contributes a root `.module` View surface, so personal and group module visibility is expressed through Views. View policy controls who diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index 2b036f7..4f7d6d3 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Callable, Mapping from datetime import datetime, timedelta, timezone from importlib.metadata import PackageNotFoundError, version import logging @@ -42,7 +43,12 @@ from govoplan_core.core.idm import ( ) from govoplan_core.core.module_management import ( load_startup_enabled_modules, - startup_candidate_module_ids, +) +from govoplan_core.core.module_entitlements import ( + TenantModuleAdmission, + TenantModuleOperatorActionRequired, + TenantWorkState, + tenant_execution_scope, ) from govoplan_core.core.mail import ( CAPABILITY_MAIL_BOUNCE_PROCESSING, @@ -50,7 +56,6 @@ from govoplan_core.core.mail import ( MailBounceProcessingProvider, MailDeliveryOutboxProvider, ) -from govoplan_core.core.modules import ModuleContext from govoplan_core.core.notifications import ( CAPABILITY_NOTIFICATIONS_DISPATCH, NotificationDispatchProvider, @@ -66,7 +71,7 @@ from govoplan_core.core.workflows import ( WorkflowTriggerDispatcher, ) from govoplan_core.core.registry import PlatformRegistry -from govoplan_core.core.runtime import configure_runtime +from govoplan_core.core.worker_runtime import build_worker_platform_registry from govoplan_core.core.runtime_coordination import ( RuntimeIdentity, bind_process_runtime_identity, @@ -81,10 +86,6 @@ from govoplan_core.core.search import ( ) from govoplan_core.settings import settings from govoplan_core.db.session import configure_database, get_database -from govoplan_core.server.registry import ( - available_module_manifests, - build_platform_registry, -) configure_database(settings.database_url) @@ -195,6 +196,150 @@ _worker_consumer: object | None = None logger = logging.getLogger("govoplan.worker.runtime") +def _worker_admissions( + registry: PlatformRegistry, + session: object, + *, + capability_name: str, + tenant_id: str | None, + work_state: TenantWorkState = "accepted", +) -> tuple[TenantModuleAdmission, ...]: + owner = registry.capability_owner(capability_name) + if owner is None: + raise RuntimeError( + f"Worker capability has no owning module: {capability_name}" + ) + resolver = registry.tenant_entitlement_resolver() + if tenant_id is not None: + return ( + resolver.admission( + session, + tenant_id=tenant_id, + module_id=owner, + work_state=work_state, + ), + ) + return resolver.active_tenant_admissions( + session, + module_id=owner, + work_state=work_state, + ) + + +def _worker_operator_actions( + admissions: tuple[TenantModuleAdmission, ...], +) -> list[dict[str, object]]: + return [ + admission.payload() + for admission in admissions + if not admission.allowed + ] + + +def _merge_tenant_worker_results( + results: list[tuple[str, Mapping[str, object]]], + *, + defaults: Mapping[str, object], + operator_actions: list[dict[str, object]], +) -> dict[str, object]: + merged = dict(defaults) + for _tenant_id, result in results: + for key, value in result.items(): + current = merged.get(key) + if isinstance(value, bool): + merged[key] = value + elif isinstance(value, int | float): + merged[key] = ( + (current if isinstance(current, int | float) else 0) + + value + ) + elif isinstance(value, list): + merged[key] = [ + *(current if isinstance(current, list) else []), + *value, + ] + elif key not in merged: + merged[key] = value + merged["tenant_results"] = { + tenant_id: dict(result) for tenant_id, result in results + } + merged["operator_action_required"] = len(operator_actions) + merged["operator_actions"] = operator_actions + return merged + + +def _run_tenant_worker_batches( + registry: PlatformRegistry, + session: object, + *, + capability_name: str, + tenant_id: str | None, + operation: Callable[[str], Mapping[str, object]], + defaults: Mapping[str, object], + work_state: TenantWorkState = "accepted", +) -> dict[str, object]: + admissions = _worker_admissions( + registry, + session, + capability_name=capability_name, + tenant_id=tenant_id, + work_state=work_state, + ) + operator_actions = _worker_operator_actions(admissions) + results: list[tuple[str, Mapping[str, object]]] = [] + resolver = registry.tenant_entitlement_resolver() + for admission in admissions: + if not admission.allowed: + continue + with tenant_execution_scope( + resolver, + session, + tenant_id=admission.tenant_id, + work_state=work_state, + ): + results.append( + (admission.tenant_id, operation(admission.tenant_id)) + ) + if tenant_id is not None and len(results) == 1 and not operator_actions: + return dict(results[0][1]) + return _merge_tenant_worker_results( + results, + defaults=defaults, + operator_actions=operator_actions, + ) + + +def _run_tenant_worker_item( + registry: PlatformRegistry, + session: object, + *, + capability_name: str, + tenant_id: str, + operation: Callable[[], Mapping[str, object]], +) -> dict[str, object]: + admissions = _worker_admissions( + registry, + session, + capability_name=capability_name, + tenant_id=tenant_id, + work_state="accepted", + ) + admission = admissions[0] + if not admission.allowed: + return { + "status": "operator_action_required", + "operator_action_required": 1, + "operator_actions": [admission.payload()], + } + with tenant_execution_scope( + registry.tenant_entitlement_resolver(), + session, + tenant_id=tenant_id, + work_state="accepted", + ): + return dict(operation()) + + def _core_version() -> str: try: return version("govoplan-core") @@ -397,41 +542,33 @@ def _worker_acceptance_delivery_count(task, probe_id: str) -> int: def _platform_registry() -> PlatformRegistry: - raw_enabled_modules = load_startup_enabled_modules(settings.enabled_modules) - candidate_modules = startup_candidate_module_ids( - settings.enabled_modules, raw_enabled_modules - ) - available_modules = available_module_manifests( - enabled_modules=candidate_modules, ignore_load_errors=True - ) - enabled_modules = load_startup_enabled_modules( - settings.enabled_modules, available=available_modules - ) - registry = build_platform_registry(enabled_modules) - context = ModuleContext(registry=registry, settings=settings) - configure_runtime(context) - registry.configure_capability_context(context) - return registry + return build_worker_platform_registry(settings) -def _campaign_delivery_tasks() -> CampaignDeliveryTaskProvider: - registry = _platform_registry() +def _campaign_delivery_tasks( + registry: PlatformRegistry | None = None, +) -> CampaignDeliveryTaskProvider: + registry = registry or _platform_registry() capability = registry.require_capability(CAPABILITY_CAMPAIGNS_DELIVERY_TASKS) if not isinstance(capability, CampaignDeliveryTaskProvider): raise RuntimeError("Campaign delivery task capability is invalid") return capability -def _notification_dispatch() -> NotificationDispatchProvider: - registry = _platform_registry() +def _notification_dispatch( + registry: PlatformRegistry | None = None, +) -> NotificationDispatchProvider: + registry = registry or _platform_registry() capability = registry.require_capability(CAPABILITY_NOTIFICATIONS_DISPATCH) if not isinstance(capability, NotificationDispatchProvider): raise RuntimeError("Notification dispatch capability is invalid") return capability -def _calendar_outbox() -> CalendarOutboxProvider | None: - registry = _platform_registry() +def _calendar_outbox( + registry: PlatformRegistry | None = None, +) -> CalendarOutboxProvider | None: + registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_CALENDAR_OUTBOX): return None capability = registry.require_capability(CAPABILITY_CALENDAR_OUTBOX) @@ -440,8 +577,10 @@ def _calendar_outbox() -> CalendarOutboxProvider | None: return capability -def _mail_delivery_outbox() -> MailDeliveryOutboxProvider | None: - registry = _platform_registry() +def _mail_delivery_outbox( + registry: PlatformRegistry | None = None, +) -> MailDeliveryOutboxProvider | None: + registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX): return None capability = registry.require_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX) @@ -450,8 +589,10 @@ def _mail_delivery_outbox() -> MailDeliveryOutboxProvider | None: return capability -def _mail_bounce_processing() -> MailBounceProcessingProvider | None: - registry = _platform_registry() +def _mail_bounce_processing( + registry: PlatformRegistry | None = None, +) -> MailBounceProcessingProvider | None: + registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_MAIL_BOUNCE_PROCESSING): return None capability = registry.require_capability(CAPABILITY_MAIL_BOUNCE_PROCESSING) @@ -570,10 +711,27 @@ def send_email(self, job_id: str): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - return dict( - _campaign_delivery_tasks().send_campaign_job( - session, job_id=job_id, enqueue_imap_task=True + registry = _platform_registry() + provider = _campaign_delivery_tasks(registry) + tenant_id = provider.tenant_id_for_job(session, job_id=job_id) + if tenant_id is None: + return dict( + provider.send_campaign_job( + session, + job_id=job_id, + enqueue_imap_task=True, + ) ) + return _run_tenant_worker_item( + registry, + session, + capability_name=CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, + tenant_id=tenant_id, + operation=lambda: provider.send_campaign_job( + session, + job_id=job_id, + enqueue_imap_task=True, + ), ) @@ -584,9 +742,21 @@ def append_sent(self, job_id: str): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: + registry = _platform_registry() + provider = _campaign_delivery_tasks(registry) + tenant_id = provider.tenant_id_for_job(session, job_id=job_id) try: - return dict( - _campaign_delivery_tasks().append_sent_for_job(session, job_id=job_id) + if tenant_id is None: + return dict(provider.append_sent_for_job(session, job_id=job_id)) + return _run_tenant_worker_item( + registry, + session, + capability_name=CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, + tenant_id=tenant_id, + operation=lambda: provider.append_sent_for_job( + session, + job_id=job_id, + ), ) except Exception as exc: if getattr(exc, "temporary", None) is True: @@ -599,11 +769,30 @@ def deliver_notification(self, notification_id: str): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - result = dict( - _notification_dispatch().deliver_notification( - session, notification_id=notification_id - ) + registry = _platform_registry() + provider = _notification_dispatch(registry) + tenant_id = provider.tenant_id_for_notification( + session, + notification_id=notification_id, ) + if tenant_id is None: + result = dict( + provider.deliver_notification( + session, + notification_id=notification_id, + ) + ) + else: + result = _run_tenant_worker_item( + registry, + session, + capability_name=CAPABILITY_NOTIFICATIONS_DISPATCH, + tenant_id=tenant_id, + operation=lambda: provider.deliver_notification( + session, + notification_id=notification_id, + ), + ) session.commit() return result @@ -613,10 +802,20 @@ def deliver_pending_notifications(self, tenant_id: str | None = None, limit: int from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - result = dict( - _notification_dispatch().deliver_pending( - session, tenant_id=tenant_id, limit=limit - ) + registry = _platform_registry() + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_NOTIFICATIONS_DISPATCH, + tenant_id=tenant_id, + operation=lambda effective_tenant_id: _notification_dispatch( + registry + ).deliver_pending( + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults={"selected": 0, "delivered": 0, "failed": 0}, ) session.commit() return result @@ -633,24 +832,32 @@ def dispatch_mail_outbox( from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _mail_delivery_outbox() - if provider is None: - return { - "selected": 0, - "accepted": 0, - "partially_refused": 0, - "retrying": 0, - "failed": 0, - "outcome_unknown": 0, - "command_ids": [], - } - return dict( - provider.dispatch_due( + registry = _platform_registry() + defaults = { + "selected": 0, + "accepted": 0, + "partially_refused": 0, + "retrying": 0, + "failed": 0, + "outcome_unknown": 0, + "command_ids": [], + } + if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX): + return defaults + return _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_MAIL_DELIVERY_OUTBOX, + tenant_id=tenant_id, + operation=lambda effective_tenant_id: _mail_delivery_outbox( + registry + ).dispatch_due( # type: ignore[union-attr] session, - tenant_id=tenant_id, + tenant_id=effective_tenant_id, limit=limit, worker_id=getattr(self.request, "hostname", None), - ) + ), + defaults=defaults, ) @@ -661,10 +868,24 @@ def purge_mail_outbox(self, limit: int = 250): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _mail_delivery_outbox() - if provider is None: + registry = _platform_registry() + if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX): return {"purged": 0} - return dict(provider.purge_expired(session, limit=limit)) + return _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_MAIL_DELIVERY_OUTBOX, + tenant_id=None, + operation=lambda effective_tenant_id: _mail_delivery_outbox( + registry + ).purge_expired( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults={"purged": 0}, + work_state="new", + ) @celery.task(name="govoplan.mail.scan_bounces", bind=True, max_retries=0) @@ -678,15 +899,29 @@ def scan_mail_bounces( from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _mail_bounce_processing() - if provider is None: - return { - "sources": 0, - "processed_messages": 0, - "observations": 0, - "failures": [], - } - result = dict(provider.scan_due(session, tenant_id=tenant_id, limit=limit)) + registry = _platform_registry() + defaults = { + "sources": 0, + "processed_messages": 0, + "observations": 0, + "failures": [], + } + if not registry.has_capability(CAPABILITY_MAIL_BOUNCE_PROCESSING): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_MAIL_BOUNCE_PROCESSING, + tenant_id=tenant_id, + operation=lambda effective_tenant_id: _mail_bounce_processing( + registry + ).scan_due( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults=defaults, + ) session.commit() return result @@ -698,16 +933,30 @@ def dispatch_calendar_outbox(self, tenant_id: str | None = None, limit: int = 50 from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _calendar_outbox() - if provider is None: - return { - "processed": 0, - "succeeded": 0, - "retrying": 0, - "failed": 0, - "operations": [], - } - result = dict(provider.dispatch_due(session, tenant_id=tenant_id, limit=limit)) + registry = _platform_registry() + defaults = { + "processed": 0, + "succeeded": 0, + "retrying": 0, + "failed": 0, + "operations": [], + } + if not registry.has_capability(CAPABILITY_CALENDAR_OUTBOX): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_CALENDAR_OUTBOX, + tenant_id=tenant_id, + operation=lambda effective_tenant_id: _calendar_outbox( + registry + ).dispatch_due( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults=defaults, + ) session.commit() return result @@ -723,17 +972,31 @@ def dispatch_dataflow_triggers(self, limit: int = 100): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _dataflow_trigger_dispatcher() - if provider is None: - return { - "queued": 0, - "processed": 0, - "succeeded": 0, - "failed": 0, - "blocked": 0, - "skipped": 0, - } - result = dict(provider.dispatch_due(session, limit=limit)) + registry = _platform_registry() + defaults = { + "queued": 0, + "processed": 0, + "succeeded": 0, + "failed": 0, + "blocked": 0, + "skipped": 0, + } + if not registry.has_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, + tenant_id=None, + operation=lambda effective_tenant_id: _dataflow_trigger_dispatcher( + registry + ).dispatch_due( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults=defaults, + ) session.commit() return result @@ -749,21 +1012,30 @@ def dispatch_dataflow_runs(self, limit: int = 10): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _dataflow_run_worker() - if provider is None: - return { - "claimed": 0, - "succeeded": 0, - "retrying": 0, - "failed": 0, - "cancelled": 0, - } - result = dict( - provider.dispatch_pending( + registry = _platform_registry() + defaults = { + "claimed": 0, + "succeeded": 0, + "retrying": 0, + "failed": 0, + "cancelled": 0, + } + if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_DATAFLOW_RUN_WORKER, + tenant_id=None, + operation=lambda effective_tenant_id: _dataflow_run_worker( + registry + ).dispatch_pending( # type: ignore[union-attr] session, + tenant_id=effective_tenant_id, limit=limit, worker_id=getattr(self.request, "hostname", None), - ) + ), + defaults=defaults, ) session.commit() return result @@ -780,10 +1052,24 @@ def purge_dataflow_runs(self, limit: int = 500): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _dataflow_run_worker() - if provider is None: + registry = _platform_registry() + if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER): return {"purged": 0} - result = dict(provider.purge_expired(session, limit=limit)) + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_DATAFLOW_RUN_WORKER, + tenant_id=None, + operation=lambda effective_tenant_id: _dataflow_run_worker( + registry + ).purge_expired( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults={"purged": 0}, + work_state="new", + ) session.commit() return result @@ -799,15 +1085,29 @@ def reconcile_workflow_instances(self, limit: int = 50): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _workflow_runtime_worker() - if provider is None: - return { - "inspected": 0, - "advanced": 0, - "waiting": 0, - "failed": 0, - } - result = dict(provider.reconcile_pending(session, limit=limit)) + registry = _platform_registry() + defaults = { + "inspected": 0, + "advanced": 0, + "waiting": 0, + "failed": 0, + } + if not registry.has_capability(CAPABILITY_WORKFLOW_RUNTIME_WORKER): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_WORKFLOW_RUNTIME_WORKER, + tenant_id=None, + operation=lambda effective_tenant_id: _workflow_runtime_worker( + registry + ).reconcile_pending( # type: ignore[union-attr] + session, + tenant_id=effective_tenant_id, + limit=limit, + ), + defaults=defaults, + ) session.commit() return result @@ -827,23 +1127,31 @@ def dispatch_postbox_routes( from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _postbox_routing_provider() - if provider is None: - return { - "selected": 0, - "delivered": 0, - "vacant": 0, - "rescheduled": 0, - "cancelled": 0, - "failed": 0, - "route_ids": [], - } - result = dict( - provider.dispatch_due_routes( + registry = _platform_registry() + defaults = { + "selected": 0, + "delivered": 0, + "vacant": 0, + "rescheduled": 0, + "cancelled": 0, + "failed": 0, + "route_ids": [], + } + if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING): + return defaults + 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=tenant_id, + tenant_id=effective_tenant_id, limit=limit, - ) + ), + defaults=defaults, ) session.commit() return result @@ -864,15 +1172,23 @@ def expire_idm_assignments( from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - provider = _idm_assignment_lifecycle() - if provider is None: - return {"selected": 0, "expired": 0, "assignment_ids": []} - result = dict( - provider.process_expired( + registry = _platform_registry() + defaults = {"selected": 0, "expired": 0, "assignment_ids": []} + if not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE): + return defaults + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE, + tenant_id=tenant_id, + operation=lambda effective_tenant_id: _idm_assignment_lifecycle( + registry + ).process_expired( # type: ignore[union-attr] session, - tenant_id=tenant_id, + tenant_id=effective_tenant_id, limit=limit, - ) + ), + defaults=defaults, ) session.commit() return result @@ -890,94 +1206,178 @@ def dispatch_platform_events(self, limit: int = 100): with get_database().SessionLocal() as session: registry = _platform_registry() - outbox = _platform_event_outbox(registry) - if outbox is None: - return { - "selected": 0, - "delivered": 0, - "retrying": 0, - "quarantined": 0, - "dispatched": 0, - "observer_failed": 0, - } - 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: + defaults = { + "selected": 0, + "delivered": 0, + "retrying": 0, + "quarantined": 0, + "dispatched": 0, + "observer_failed": 0, + } + if not registry.has_capability(CAPABILITY_PLATFORM_EVENT_OUTBOX): + return defaults - def deliver_to_dataflow( - event: PlatformEvent, - _delivery_key: str, - ) -> None: - dataflow_dispatcher.ingest_event( + def dispatch_for_scope( + tenant_id: str | None, + *, + tenantless_only: bool = False, + ) -> Mapping[str, object]: + consumers: list[DurableEventConsumer] = [] + search_coordinator: SearchIndexCoordinator | None = None + + def consumer_admission( + capability_name: str, + ) -> TenantModuleAdmission | None: + if tenant_id is None: + return None + return _worker_admissions( + registry, session, - event=event, + capability_name=capability_name, + tenant_id=tenant_id, + work_state="accepted", + )[0] + + def blocked_handler(admission: TenantModuleAdmission): + def preserve_for_operator( + _event: PlatformEvent, + _delivery_key: str, + ) -> None: + raise TenantModuleOperatorActionRequired(admission) + + return preserve_for_operator + + if registry.has_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER): + admission = consumer_admission( + CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER + ) + dataflow_dispatcher = ( + _dataflow_trigger_dispatcher(registry) + if admission is None or admission.allowed + else None ) - consumers.append( - DurableEventConsumer( - consumer_id="dataflow.event-triggers.v1", - event_types=frozenset({"*"}), - classifications=frozenset({"public", "internal"}), - handler=deliver_to_dataflow, - ) - ) - if workflow_dispatcher is not None: + def deliver_to_dataflow( + event: PlatformEvent, + _delivery_key: str, + ) -> None: + assert dataflow_dispatcher is not None + dataflow_dispatcher.ingest_event(session, event=event) - def deliver_to_workflow( - event: PlatformEvent, - _delivery_key: str, - ) -> None: - workflow_dispatcher.ingest_event( - session, - event=event, + consumers.append( + DurableEventConsumer( + consumer_id="dataflow.event-triggers.v1", + event_types=frozenset({"*"}), + classifications=frozenset({"public", "internal"}), + handler=( + deliver_to_dataflow + if admission is None or admission.allowed + else blocked_handler(admission) + ), + ) ) - - consumers.append( - DurableEventConsumer( - consumer_id="workflow.event-triggers.v1", - event_types=frozenset({"*"}), - classifications=frozenset({"public", "internal"}), - handler=deliver_to_workflow, + if registry.has_capability(CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER): + admission = consumer_admission( + CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER + ) + workflow_dispatcher = ( + _workflow_trigger_dispatcher(registry) + if admission is None or admission.allowed + else None ) - ) - 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, + def deliver_to_workflow( + event: PlatformEvent, + _delivery_key: str, + ) -> None: + assert workflow_dispatcher is not None + workflow_dispatcher.ingest_event(session, event=event) + + consumers.append( + DurableEventConsumer( + consumer_id="workflow.event-triggers.v1", + event_types=frozenset({"*"}), + classifications=frozenset({"public", "internal"}), + handler=( + deliver_to_workflow + if admission is None or admission.allowed + else blocked_handler(admission) + ), + ) ) + if registry.has_capability(CAPABILITY_SEARCH_INDEX_WRITER): + admission = consumer_admission(CAPABILITY_SEARCH_INDEX_WRITER) + search_coordinator = ( + _search_index_coordinator(registry) + if admission is None or admission.allowed + else None + ) + + def deliver_to_search( + event: PlatformEvent, + delivery_key: str, + ) -> None: + assert search_coordinator is not 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, + consumers.append( + DurableEventConsumer( + consumer_id="search.indexing.v1", + event_types=frozenset({"*"}), + classifications=frozenset({"public", "internal"}), + handler=( + deliver_to_search + if admission is None or admission.allowed + else blocked_handler(admission) + ), + ) ) - ) - result = dict( - outbox.dispatch_pending( - session, - consumers=tuple(consumers), - observer=publish_platform_event, - limit=limit, - ) - ) - if search_coordinator is not None: - result["search_changes"] = dict( - search_coordinator.process_changes( + outbox = _platform_event_outbox(registry) + assert outbox is not None + result = dict( + outbox.dispatch_pending( session, + tenant_id=tenant_id, + tenantless_only=tenantless_only, + consumers=tuple(consumers), + observer=publish_platform_event, limit=limit, ) ) + if search_coordinator is not None and tenant_id is not None: + result["search_changes"] = dict( + search_coordinator.process_changes( + session, + tenant_id=tenant_id, + limit=limit, + ) + ) + return result + + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_PLATFORM_EVENT_OUTBOX, + tenant_id=None, + operation=dispatch_for_scope, + defaults=defaults, + ) + system_result = dict( + dispatch_for_scope(None, tenantless_only=True) + ) + for key in defaults: + value = system_result.get(key) + current = result.get(key) + if isinstance(value, int | float): + result[key] = ( + current if isinstance(current, int | float) else 0 + ) + value + result["system_result"] = system_result session.commit() return result @@ -993,18 +1393,39 @@ def purge_platform_events(self, limit: int = 500): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: - outbox = _platform_event_outbox() - if outbox is None: + registry = _platform_registry() + if not registry.has_capability(CAPABILITY_PLATFORM_EVENT_OUTBOX): return {"deleted": 0} before = datetime.now(timezone.utc) - timedelta( days=settings.platform_event_outbox_terminal_retention_days ) - result = dict( - outbox.purge_terminal( + result = _run_tenant_worker_batches( + registry, + session, + capability_name=CAPABILITY_PLATFORM_EVENT_OUTBOX, + tenant_id=None, + operation=lambda effective_tenant_id: _platform_event_outbox( + registry + ).purge_terminal( # type: ignore[union-attr] session, + tenant_id=effective_tenant_id, + before=before, + limit=limit, + ), + defaults={"deleted": 0}, + work_state="new", + ) + system_result = dict( + _platform_event_outbox(registry).purge_terminal( # type: ignore[union-attr] + session, + tenantless_only=True, before=before, limit=limit, ) ) + result["deleted"] = int(result.get("deleted", 0)) + int( + system_result.get("deleted", 0) + ) + result["system_result"] = system_result session.commit() return result diff --git a/src/govoplan_core/core/campaigns.py b/src/govoplan_core/core/campaigns.py index de2927d..1aed444 100644 --- a/src/govoplan_core/core/campaigns.py +++ b/src/govoplan_core/core/campaigns.py @@ -95,6 +95,9 @@ class CampaignPolicyContextProvider(Protocol): @runtime_checkable class CampaignDeliveryTaskProvider(Protocol): + def tenant_id_for_job(self, session: object, *, job_id: str) -> str | None: + ... + def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True) -> Mapping[str, object]: ... diff --git a/src/govoplan_core/core/dataflows.py b/src/govoplan_core/core/dataflows.py index 3a126ab..e16e5e8 100644 --- a/src/govoplan_core/core/dataflows.py +++ b/src/govoplan_core/core/dataflows.py @@ -178,6 +178,7 @@ class DataflowTriggerDispatcher(Protocol): self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> Mapping[str, object]: @@ -204,6 +205,7 @@ class DataflowRunWorker(Protocol): self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 10, worker_id: str | None = None, @@ -214,6 +216,7 @@ class DataflowRunWorker(Protocol): self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 500, ) -> Mapping[str, object]: diff --git a/src/govoplan_core/core/events.py b/src/govoplan_core/core/events.py index a576848..1ede34f 100644 --- a/src/govoplan_core/core/events.py +++ b/src/govoplan_core/core/events.py @@ -182,6 +182,8 @@ class PlatformEventOutbox(Protocol): self, session: object, *, + tenant_id: str | None = None, + tenantless_only: bool = False, consumers: Sequence[DurableEventConsumer] = (), observer: EventHandler | None = None, limit: int = 100, @@ -203,6 +205,8 @@ class PlatformEventOutbox(Protocol): self, session: object, *, + tenant_id: str | None = None, + tenantless_only: bool = False, before: datetime, limit: int = 500, ) -> Mapping[str, int]: diff --git a/src/govoplan_core/core/lifecycle.py b/src/govoplan_core/core/lifecycle.py index 6506561..f65315f 100644 --- a/src/govoplan_core/core/lifecycle.py +++ b/src/govoplan_core/core/lifecycle.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence +from collections.abc import AsyncIterator, Mapping, Sequence from dataclasses import dataclass from threading import RLock @@ -10,7 +10,11 @@ from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal, get_api_principal from govoplan_core.core.module_management import ModuleManagementError, REQUIRED_PLATFORM_MODULES, plan_desired_enabled_modules -from govoplan_core.core.module_entitlements import tenant_module_entitlement_state +from govoplan_core.core.module_entitlements import ( + ModuleEntitlementResolutionError, + TenantModuleUnavailable, + tenant_execution_scope, +) from govoplan_core.core.module_lifecycle_recovery import ( ModuleLifecycleRecovery, begin_runtime_graph_recovery, @@ -24,7 +28,6 @@ from govoplan_core.core.workflows import ( ) from govoplan_core.db.session import get_session from govoplan_core.server.route_validation import validate_router_can_mount -from govoplan_core.tenancy.scope import Tenant @dataclass(frozen=True, slots=True) @@ -37,60 +40,74 @@ class ModuleLifecycleResult: def require_module_active(module_id: str): - def dependency( + async def dependency( request: Request, session: Session = Depends(get_session), authorization: str | None = Header(default=None), x_api_key: str | None = Header(default=None, alias="X-API-Key"), - ) -> None: + ) -> AsyncIterator[None]: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry) or not registry.has_module(module_id): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Module is disabled: {module_id}") - # Public module routes must remain reachable without Access. When an - # authenticated request is present, cache its principal and apply the - # active tenant's module entitlement before the owning route executes. + tenant_id: str | None = None if not authorization and not x_api_key and not request.cookies: - return - try: - principal = get_api_principal( - request, - session, - authorization=authorization, - x_api_key=x_api_key, - ) - except HTTPException as exc: - if exc.status_code in { - status.HTTP_401_UNAUTHORIZED, - status.HTTP_403_FORBIDDEN, - }: + public_resolver = registry.public_tenant_resolver(module_id) + if public_resolver is not None: + tenant_id = public_resolver(request, session) + if tenant_id is None: + yield return - raise - if not isinstance(principal, ApiPrincipal) or principal.principal.tenant_id is None: - return + else: + try: + principal = get_api_principal( + request, + session, + authorization=authorization, + x_api_key=x_api_key, + ) + except HTTPException as exc: + if exc.status_code in { + status.HTTP_401_UNAUTHORIZED, + status.HTTP_403_FORBIDDEN, + }: + yield + return + raise + if ( + not isinstance(principal, ApiPrincipal) + or principal.principal.tenant_id is None + ): + yield + return + tenant_id = principal.principal.tenant_id + + resolver = registry.tenant_entitlement_resolver() try: - tenant = session.get(Tenant, principal.principal.tenant_id) - except (RuntimeError, SQLAlchemyError) as exc: + admission = resolver.require( + session, + tenant_id=tenant_id, + module_id=module_id, + work_state="interactive", + ) + except TenantModuleUnavailable as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Module is unavailable in the active tenant: {module_id}", + ) from exc + except (ModuleEntitlementResolutionError, RuntimeError, SQLAlchemyError) as exc: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Tenant module entitlement could not be resolved.", ) from exc - if tenant is None: - raise HTTPException( - status_code=status.HTTP_403_FORBIDDEN, - detail="The active tenant is unavailable.", - ) - manifests = {manifest.id: manifest for manifest in registry.manifests()} - entitlement = tenant_module_entitlement_state( - tenant.settings or {}, - manifests, - runtime_active_modules=manifests, - ) - if module_id not in entitlement.effective_modules: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"Module is unavailable in the active tenant: {module_id}", - ) + request.state.govoplan_module_admission = admission + with tenant_execution_scope( + resolver, + session, + tenant_id=tenant_id, + work_state="interactive", + ): + yield return dependency diff --git a/src/govoplan_core/core/mail.py b/src/govoplan_core/core/mail.py index 75215e9..0571f90 100644 --- a/src/govoplan_core/core/mail.py +++ b/src/govoplan_core/core/mail.py @@ -56,6 +56,7 @@ class MailDeliveryOutboxProvider(Protocol): self, session: object, *, + tenant_id: str | None = None, limit: int = 250, ) -> Mapping[str, object]: ... diff --git a/src/govoplan_core/core/module_entitlements.py b/src/govoplan_core/core/module_entitlements.py index e8cf06f..0cec560 100644 --- a/src/govoplan_core/core/module_entitlements.py +++ b/src/govoplan_core/core/module_entitlements.py @@ -1,8 +1,13 @@ from __future__ import annotations -from collections.abc import Iterable, Mapping +from collections import OrderedDict +from collections.abc import Iterable, Iterator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass -from typing import Any +from threading import RLock +from time import monotonic +from typing import Any, Literal from govoplan_core.core.modules import ModuleManifest @@ -20,6 +25,30 @@ class ModuleEntitlementConflict(ModuleEntitlementError): pass +class ModuleEntitlementResolutionError(ModuleEntitlementError): + pass + + +class TenantModuleUnavailable(ModuleEntitlementError): + def __init__(self, admission: "TenantModuleAdmission") -> None: + self.admission = admission + super().__init__(admission.reason) + + +class TenantModuleOperatorActionRequired(ModuleEntitlementError): + def __init__(self, admission: "TenantModuleAdmission") -> None: + self.admission = admission + super().__init__(admission.reason) + + +TenantWorkState = Literal["interactive", "new", "accepted"] +TenantAdmissionDisposition = Literal[ + "allowed", + "rejected", + "operator_action_required", +] + + @dataclass(frozen=True, slots=True) class TenantModuleItem: id: str @@ -48,6 +77,331 @@ class TenantModuleEntitlementState: diagnostics: tuple[dict[str, str], ...] = () +@dataclass(frozen=True, slots=True) +class TenantModuleAdmission: + tenant_id: str + module_id: str + revision: int + work_state: TenantWorkState + allowed: bool + disposition: TenantAdmissionDisposition + reason: str + + def payload(self) -> dict[str, object]: + return { + "tenant_id": self.tenant_id, + "module_id": self.module_id, + "entitlement_revision": self.revision, + "work_state": self.work_state, + "allowed": self.allowed, + "disposition": self.disposition, + "reason": self.reason, + } + + +@dataclass(frozen=True, slots=True) +class _CachedTenantEntitlement: + expires_at: float + tenant_active: bool + state: TenantModuleEntitlementState + + +class TenantModuleEntitlementResolver: + """Resolve tenant-effective modules with bounded process-local caching. + + Cache entries are explicitly invalidated by local mutations and expire + quickly so changes made on another application node become authoritative + without requiring a database lookup for every capability call. + """ + + def __init__( + self, + registry: object, + *, + ttl_seconds: float = 5.0, + max_entries: int = 2048, + ) -> None: + self._registry = registry + self._ttl_seconds = max(0.0, min(float(ttl_seconds), 300.0)) + self._max_entries = max(1, int(max_entries)) + self._cache: OrderedDict[str, _CachedTenantEntitlement] = OrderedDict() + self._lock = RLock() + + def resolve( + self, + session: object, + tenant_id: str, + ) -> TenantModuleEntitlementState: + normalized_tenant_id = str(tenant_id or "").strip() + if not normalized_tenant_id: + raise ModuleEntitlementResolutionError("Tenant id is required") + + cached = self._cached(normalized_tenant_id) + if cached is not None: + if not cached.tenant_active: + raise ModuleEntitlementResolutionError( + f"Tenant is inactive: {normalized_tenant_id}" + ) + return cached.state + + from govoplan_core.tenancy.scope import Tenant + + getter = getattr(session, "get", None) + if not callable(getter): + raise ModuleEntitlementResolutionError( + "Tenant module entitlement resolution requires a database session" + ) + tenant = getter(Tenant, normalized_tenant_id) + if tenant is None: + raise ModuleEntitlementResolutionError( + f"Tenant is unavailable: {normalized_tenant_id}" + ) + state = self._state_from_settings(getattr(tenant, "settings", None)) + tenant_active = bool(getattr(tenant, "is_active", False)) + self._store(normalized_tenant_id, tenant_active=tenant_active, state=state) + if not tenant_active: + raise ModuleEntitlementResolutionError( + f"Tenant is inactive: {normalized_tenant_id}" + ) + return state + + def admission( + self, + session: object, + *, + tenant_id: str, + module_id: str, + work_state: TenantWorkState = "interactive", + ) -> TenantModuleAdmission: + if work_state not in {"interactive", "new", "accepted"}: + raise ModuleEntitlementError(f"Unsupported tenant work state: {work_state}") + normalized_module_id = str(module_id or "").strip() + if not normalized_module_id: + raise ModuleEntitlementError("Module id is required") + state = self.resolve(session, tenant_id) + allowed = normalized_module_id in state.effective_modules + if allowed: + return TenantModuleAdmission( + tenant_id=str(tenant_id), + module_id=normalized_module_id, + revision=state.revision, + work_state=work_state, + allowed=True, + disposition="allowed", + reason="The module is effective for this tenant.", + ) + accepted = work_state == "accepted" + return TenantModuleAdmission( + tenant_id=str(tenant_id), + module_id=normalized_module_id, + revision=state.revision, + work_state=work_state, + allowed=False, + disposition=( + "operator_action_required" if accepted else "rejected" + ), + reason=( + "Accepted durable work was preserved because the owning module " + "is no longer effective for this tenant; an operator must resume " + "the module or resolve the work explicitly." + if accepted + else "The module is not effective for this tenant." + ), + ) + + def require( + self, + session: object, + *, + tenant_id: str, + module_id: str, + work_state: TenantWorkState = "interactive", + ) -> TenantModuleAdmission: + admission = self.admission( + session, + tenant_id=tenant_id, + module_id=module_id, + work_state=work_state, + ) + if admission.allowed: + return admission + if admission.disposition == "operator_action_required": + raise TenantModuleOperatorActionRequired(admission) + raise TenantModuleUnavailable(admission) + + def effective_tenant_ids( + self, + session: object, + *, + module_id: str, + ) -> tuple[str, ...]: + """Return active tenants that may admit new work for one module.""" + + return tuple( + admission.tenant_id + for admission in self.active_tenant_admissions( + session, + module_id=module_id, + work_state="new", + ) + if admission.allowed + ) + + def active_tenant_admissions( + self, + session: object, + *, + module_id: str, + work_state: TenantWorkState = "new", + ) -> tuple[TenantModuleAdmission, ...]: + """Resolve one admission per active tenant with a single DB query.""" + + from govoplan_core.tenancy.scope import Tenant + + query = getattr(session, "query", None) + if not callable(query): + raise ModuleEntitlementResolutionError( + "Tenant module entitlement resolution requires a database session" + ) + tenants = ( + query(Tenant) + .filter(Tenant.is_active.is_(True)) + .order_by(Tenant.id.asc()) + .all() + ) + admissions: list[TenantModuleAdmission] = [] + for tenant in tenants: + state = self._state_from_settings(getattr(tenant, "settings", None)) + self._store(tenant.id, tenant_active=True, state=state) + allowed = module_id in state.effective_modules + accepted = work_state == "accepted" + admissions.append( + TenantModuleAdmission( + tenant_id=tenant.id, + module_id=module_id, + revision=state.revision, + work_state=work_state, + allowed=allowed, + disposition=( + "allowed" + if allowed + else "operator_action_required" + if accepted + else "rejected" + ), + reason=( + "The module is effective for this tenant." + if allowed + else "Accepted durable work was preserved because the owning module is no longer effective for this tenant; an operator must resume the module or resolve the work explicitly." + if accepted + else "The module is not effective for this tenant." + ), + ) + ) + return tuple(admissions) + + def invalidate(self, tenant_id: str | None = None) -> None: + with self._lock: + if tenant_id is None: + self._cache.clear() + else: + self._cache.pop(str(tenant_id), None) + + def _state_from_settings( + self, + settings: Mapping[str, object] | None, + ) -> TenantModuleEntitlementState: + manifests_method = getattr(self._registry, "manifests", None) + if not callable(manifests_method): + raise ModuleEntitlementResolutionError( + "Tenant module entitlement resolver has no platform registry" + ) + manifests = {manifest.id: manifest for manifest in manifests_method()} + return tenant_module_entitlement_state( + settings, + manifests, + runtime_active_modules=manifests, + ) + + def _cached(self, tenant_id: str) -> _CachedTenantEntitlement | None: + now = monotonic() + with self._lock: + cached = self._cache.get(tenant_id) + if cached is None: + return None + if cached.expires_at <= now: + self._cache.pop(tenant_id, None) + return None + self._cache.move_to_end(tenant_id) + return cached + + def _store( + self, + tenant_id: str, + *, + tenant_active: bool, + state: TenantModuleEntitlementState, + ) -> None: + if self._ttl_seconds <= 0: + return + with self._lock: + self._cache[str(tenant_id)] = _CachedTenantEntitlement( + expires_at=monotonic() + self._ttl_seconds, + tenant_active=tenant_active, + state=state, + ) + self._cache.move_to_end(str(tenant_id)) + while len(self._cache) > self._max_entries: + self._cache.popitem(last=False) + + +@dataclass(frozen=True, slots=True) +class TenantExecutionContext: + resolver: TenantModuleEntitlementResolver + session: object + tenant_id: str + work_state: TenantWorkState + + def require_module(self, module_id: str) -> TenantModuleAdmission: + return self.resolver.require( + self.session, + tenant_id=self.tenant_id, + module_id=module_id, + work_state=self.work_state, + ) + + +_TENANT_EXECUTION_CONTEXT: ContextVar[TenantExecutionContext | None] = ContextVar( + "govoplan_tenant_execution_context", + default=None, +) + + +def current_tenant_execution_context() -> TenantExecutionContext | None: + return _TENANT_EXECUTION_CONTEXT.get() + + +@contextmanager +def tenant_execution_scope( + resolver: TenantModuleEntitlementResolver, + session: object, + *, + tenant_id: str, + work_state: TenantWorkState = "interactive", +) -> Iterator[TenantExecutionContext]: + context = TenantExecutionContext( + resolver=resolver, + session=session, + tenant_id=str(tenant_id), + work_state=work_state, + ) + token = _TENANT_EXECUTION_CONTEXT.set(context) + try: + yield context + finally: + _TENANT_EXECUTION_CONTEXT.reset(token) + + def tenant_module_entitlement_state( settings: Mapping[str, object] | None, manifests: Mapping[str, ModuleManifest], @@ -474,9 +828,18 @@ __all__ = [ "TENANT_PROTECTED_MODULES", "ModuleEntitlementConflict", "ModuleEntitlementError", + "ModuleEntitlementResolutionError", + "TenantExecutionContext", + "TenantModuleAdmission", + "TenantModuleEntitlementResolver", "TenantModuleEntitlementState", "TenantModuleItem", + "TenantModuleOperatorActionRequired", + "TenantModuleUnavailable", + "TenantWorkState", + "current_tenant_execution_context", "module_entitlement_payload", + "tenant_execution_scope", "tenant_module_entitlement_state", "update_system_tenant_module_policy", "update_tenant_module_selection", diff --git a/src/govoplan_core/core/modules.py b/src/govoplan_core/core/modules.py index 899ab98..94128d7 100644 --- a/src/govoplan_core/core/modules.py +++ b/src/govoplan_core/core/modules.py @@ -408,6 +408,7 @@ class DeleteVetoProviderRegistration: RouteFactory = Callable[[ModuleContext], "APIRouter"] CapabilityFactory = Callable[[ModuleContext], object] +PublicTenantResolver = Callable[[object, object], str | None] DocumentationProvider = Callable[[DocumentationContext], Iterable[DocumentationTopic]] LifecycleHook = Callable[[ModuleContext], None] @@ -426,6 +427,7 @@ class ModuleManifest: permissions: tuple[PermissionDefinition, ...] = () role_templates: tuple[RoleTemplate, ...] = () route_factory: RouteFactory | None = None + public_tenant_resolver: PublicTenantResolver | None = None migration_spec: MigrationSpec | None = None nav_items: tuple[NavItem, ...] = () frontend: FrontendModule | None = None diff --git a/src/govoplan_core/core/notifications.py b/src/govoplan_core/core/notifications.py index e768663..dd00009 100644 --- a/src/govoplan_core/core/notifications.py +++ b/src/govoplan_core/core/notifications.py @@ -33,6 +33,14 @@ class NotificationDispatchRequest: @runtime_checkable class NotificationDispatchProvider(Protocol): + def tenant_id_for_notification( + self, + session: object, + *, + notification_id: str, + ) -> str | None: + ... + def enqueue_notification( self, session: object, diff --git a/src/govoplan_core/core/poll_participation.py b/src/govoplan_core/core/poll_participation.py index 08cf25e..9db43bc 100644 --- a/src/govoplan_core/core/poll_participation.py +++ b/src/govoplan_core/core/poll_participation.py @@ -129,6 +129,16 @@ class PollParticipationContextRef: response: PollGovernedResponseRef | None = None +@dataclass(frozen=True, slots=True) +class PollPublicInvitationRef: + """Non-sensitive routing identity for one valid governed invitation.""" + + invitation_id: str + tenant_id: str + poll_id: str + gateway: PollResponseGatewayRef + + @runtime_checkable class PollParticipationGatewayProvider(Protocol): def create_governed_invitation( @@ -156,6 +166,17 @@ class PollParticipationGatewayProvider(Protocol): ... + def resolve_public_invitation( + self, + session: object, + *, + token: str, + gateway: PollResponseGatewayRef, + ) -> PollPublicInvitationRef: + """Resolve tenant routing without disclosing participant details.""" + + ... + def submit_governed_response( self, session: object, @@ -266,6 +287,7 @@ __all__ = [ "PollParticipationContextRef", "PollParticipationGatewayProvider", "PollParticipationPolicy", + "PollPublicInvitationRef", "PollResponseGatewayRef", "participation_token_fingerprint", "poll_participation_gateway_provider", diff --git a/src/govoplan_core/core/registry.py b/src/govoplan_core/core/registry.py index bf85af0..b0ce653 100644 --- a/src/govoplan_core/core/registry.py +++ b/src/govoplan_core/core/registry.py @@ -25,6 +25,13 @@ from govoplan_core.core.modules import ( TenantSummaryProvider, user_workflow_scope_condition_issues, ) +from govoplan_core.core.module_entitlements import ( + TenantModuleEntitlementResolver, + TenantModuleUnavailable, + TenantWorkState, + current_tenant_execution_context, + tenant_execution_scope, +) from govoplan_core.core.ownership import ( OwnershipProviderRegistration, ResourceOwnershipProvider, @@ -83,8 +90,10 @@ class PlatformRegistry: self._delete_veto_providers: dict[str, list[DeleteVetoProviderRegistration]] = defaultdict(list) self._ownership_providers: dict[str, OwnershipProviderRegistration] = {} self._capability_factories: dict[str, CapabilityFactory] = {} + self._capability_factory_owners: dict[str, str] = {} self._capabilities: dict[str, object] = {} self._capability_context: ModuleContext | None = None + self._tenant_entitlement_resolver = TenantModuleEntitlementResolver(self) self._search_provider_registrations: list[RegisteredSearchProvider] = [] self._search_providers: dict[str, SearchProvider] = {} self._search_source_registrations: list[ @@ -142,6 +151,9 @@ class PlatformRegistry: }) self._ownership_providers = dict(replacement._ownership_providers) self._capability_factories = dict(replacement._capability_factories) + self._capability_factory_owners = dict( + replacement._capability_factory_owners + ) self._search_provider_registrations = list( replacement._search_provider_registrations ) @@ -151,6 +163,7 @@ class PlatformRegistry: self._capabilities.clear() self._search_providers.clear() self._search_sources.clear() + self._tenant_entitlement_resolver.invalidate() return snapshot def get(self, module_id: str) -> ModuleManifest | None: @@ -258,6 +271,23 @@ class PlatformRegistry: def configure_capability_context(self, context: ModuleContext) -> None: self._capability_context = context + self._tenant_entitlement_resolver = TenantModuleEntitlementResolver( + self, + ttl_seconds=float( + getattr( + context.settings, + "tenant_module_entitlement_cache_ttl_seconds", + 5.0, + ) + ), + max_entries=int( + getattr( + context.settings, + "tenant_module_entitlement_cache_max_entries", + 2048, + ) + ), + ) self._capabilities.clear() self._search_providers.clear() self._search_sources.clear() @@ -266,6 +296,7 @@ class PlatformRegistry: if name in self._capability_factories: raise RegistryError(f"Duplicate capability: {name}") self._capability_factories[name] = factory + self._capability_factory_owners[name] = module_id def has_capability(self, name: str) -> bool: return name in self._capability_factories @@ -273,7 +304,69 @@ class PlatformRegistry: def capability_names(self) -> tuple[str, ...]: return tuple(sorted(self._capability_factories)) + def capability_owner(self, name: str) -> str | None: + return self._capability_factory_owners.get(name) + + def public_tenant_resolver(self, module_id: str): + manifest = self.get(module_id) + return manifest.public_tenant_resolver if manifest is not None else None + + def tenant_entitlement_resolver(self) -> TenantModuleEntitlementResolver: + return self._tenant_entitlement_resolver + + def invalidate_tenant_entitlement(self, tenant_id: str | None = None) -> None: + self._tenant_entitlement_resolver.invalidate(tenant_id) + + def tenant_capability( + self, + name: str, + session: object, + *, + tenant_id: str, + work_state: TenantWorkState = "interactive", + ) -> object | None: + with tenant_execution_scope( + self._tenant_entitlement_resolver, + session, + tenant_id=tenant_id, + work_state=work_state, + ): + return self.capability(name) + + def require_tenant_capability( + self, + name: str, + session: object, + *, + tenant_id: str, + work_state: TenantWorkState = "interactive", + ) -> object: + owner = self.capability_owner(name) + if owner is not None: + self._tenant_entitlement_resolver.require( + session, + tenant_id=tenant_id, + module_id=owner, + work_state=work_state, + ) + capability = self.tenant_capability( + name, + session, + tenant_id=tenant_id, + work_state=work_state, + ) + if capability is None: + raise RegistryError(f"Required capability is not available: {name}") + return capability + def capability(self, name: str) -> object | None: + execution = current_tenant_execution_context() + owner = self._capability_factory_owners.get(name) + if execution is not None and owner is not None: + try: + execution.require_module(owner) + except TenantModuleUnavailable: + return None if name in self._capabilities: return self._capabilities[name] factory = self._capability_factories.get(name) @@ -314,6 +407,12 @@ class PlatformRegistry: return () providers: list[tuple[RegisteredSearchProvider, SearchProvider]] = [] for registered in self.search_provider_registrations(): + execution = current_tenant_execution_context() + if execution is not None: + try: + execution.require_module(registered.module_id) + except TenantModuleUnavailable: + continue key = f"{registered.module_id}:{registered.registration.id}" provider = self._search_providers.get(key) if provider is None: @@ -352,6 +451,12 @@ class PlatformRegistry: tuple[RegisteredSearchSourceProvider, SearchSourceProvider] ] = [] for registered in self.search_source_registrations(): + execution = current_tenant_execution_context() + if execution is not None: + try: + execution.require_module(registered.module_id) + except TenantModuleUnavailable: + continue key = f"{registered.module_id}:{registered.registration.id}" provider = self._search_sources.get(key) if provider is None: @@ -1020,6 +1125,11 @@ def _validate_manifest_frontend(manifest: ModuleManifest) -> None: ) if frontend.package_name is not None and not _NPM_PACKAGE_RE.match(frontend.package_name): raise RegistryError(f"Module {manifest.id!r} has invalid frontend package name {frontend.package_name!r}") + if frontend.public_routes and manifest.public_tenant_resolver is None: + raise RegistryError( + f"Module {manifest.id!r} exposes public frontend routes without a " + "public tenant resolver" + ) for route in (*frontend.routes, *frontend.settings_routes, *frontend.public_routes): _validate_frontend_route(manifest.id, route.path, route.component) for route in (*frontend.routes, *frontend.settings_routes): diff --git a/src/govoplan_core/core/worker_runtime.py b/src/govoplan_core/core/worker_runtime.py new file mode 100644 index 0000000..f7ee8dd --- /dev/null +++ b/src/govoplan_core/core/worker_runtime.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from govoplan_core.core.module_management import ( + load_startup_enabled_modules, + startup_candidate_module_ids, +) +from govoplan_core.core.modules import ModuleContext +from govoplan_core.core.registry import PlatformRegistry +from govoplan_core.core.runtime import configure_runtime +from govoplan_core.server.registry import ( + available_module_manifests, + build_platform_registry, +) + + +def build_worker_platform_registry(settings: object) -> PlatformRegistry: + """Build the active capability graph used by an out-of-process worker.""" + + configured_modules = getattr(settings, "enabled_modules", "") + raw_enabled_modules = load_startup_enabled_modules(configured_modules) + candidate_modules = startup_candidate_module_ids( + configured_modules, + raw_enabled_modules, + ) + available_modules = available_module_manifests( + enabled_modules=candidate_modules, + ignore_load_errors=True, + ) + enabled_modules = load_startup_enabled_modules( + configured_modules, + available=available_modules, + ) + registry = build_platform_registry(enabled_modules) + context = ModuleContext(registry=registry, settings=settings) + configure_runtime(context) + registry.configure_capability_context(context) + return registry + + +__all__ = ["build_worker_platform_registry"] diff --git a/src/govoplan_core/core/workflows.py b/src/govoplan_core/core/workflows.py index 5635f5b..8c407d3 100644 --- a/src/govoplan_core/core/workflows.py +++ b/src/govoplan_core/core/workflows.py @@ -127,6 +127,7 @@ class WorkflowRuntimeWorker(Protocol): self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> Mapping[str, object]: ... @@ -140,6 +141,7 @@ class WorkflowTriggerDispatcher(Protocol): self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> Mapping[str, object]: ... diff --git a/src/govoplan_core/settings.py b/src/govoplan_core/settings.py index 20730d8..6cb0ec6 100644 --- a/src/govoplan_core/settings.py +++ b/src/govoplan_core/settings.py @@ -198,6 +198,18 @@ class Settings(BaseSettings): le=100_000, alias="AUTH_PRINCIPAL_CACHE_MAX_ENTRIES", ) + tenant_module_entitlement_cache_ttl_seconds: int = Field( + default=5, + ge=0, + le=300, + alias="TENANT_MODULE_ENTITLEMENT_CACHE_TTL_SECONDS", + ) + tenant_module_entitlement_cache_max_entries: int = Field( + default=2048, + ge=1, + le=100_000, + alias="TENANT_MODULE_ENTITLEMENT_CACHE_MAX_ENTRIES", + ) auth_login_throttle_enabled: bool = Field(default=True, alias="AUTH_LOGIN_THROTTLE_ENABLED") auth_login_throttle_identity_limit: int = Field( default=10, diff --git a/tests/test_access_contracts.py b/tests/test_access_contracts.py index 1b4e97e..8d91a94 100644 --- a/tests/test_access_contracts.py +++ b/tests/test_access_contracts.py @@ -426,6 +426,10 @@ class _FakeCampaignPolicyContextProvider: class _FakeCampaignDeliveryTaskProvider: + def tenant_id_for_job(self, session: object, *, job_id: str): + del session, job_id + return "tenant-1" + def send_campaign_job(self, session: object, *, job_id: str, enqueue_imap_task: bool = True): del session return {"job_id": job_id, "enqueue_imap_task": enqueue_imap_task} diff --git a/tests/test_calendar_outbox_worker.py b/tests/test_calendar_outbox_worker.py index bd1d810..a043d58 100644 --- a/tests/test_calendar_outbox_worker.py +++ b/tests/test_calendar_outbox_worker.py @@ -4,6 +4,7 @@ import unittest from unittest.mock import MagicMock, patch from govoplan_core.celery_app import celery, dispatch_calendar_outbox +from tests.worker_test_support import allowed_worker_admissions class CalendarOutboxWorkerTests(unittest.TestCase): @@ -22,6 +23,10 @@ class CalendarOutboxWorkerTests(unittest.TestCase): with ( patch("govoplan_core.celery_app._calendar_outbox", return_value=provider), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), patch("govoplan_core.db.session.get_database", return_value=database), ): result = dispatch_calendar_outbox.run("tenant-1", 25) diff --git a/tests/test_dataflow_run_worker.py b/tests/test_dataflow_run_worker.py index 328c4a6..37e3f13 100644 --- a/tests/test_dataflow_run_worker.py +++ b/tests/test_dataflow_run_worker.py @@ -8,6 +8,7 @@ from govoplan_core.celery_app import ( dispatch_dataflow_runs, purge_dataflow_runs, ) +from tests.worker_test_support import allowed_worker_admissions class DataflowRunWorkerTests(unittest.TestCase): @@ -30,11 +31,16 @@ class DataflowRunWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = dispatch_dataflow_runs.run(7) provider.dispatch_pending.assert_called_once_with( session, + tenant_id="tenant-1", limit=7, worker_id=ANY, ) @@ -57,10 +63,18 @@ class DataflowRunWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = purge_dataflow_runs.run(25) - provider.purge_expired.assert_called_once_with(session, limit=25) + provider.purge_expired.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) session.commit.assert_called_once_with() self.assertEqual(2, result["purged"]) diff --git a/tests/test_dataflow_trigger_worker.py b/tests/test_dataflow_trigger_worker.py index fb1e69f..a2f0f3f 100644 --- a/tests/test_dataflow_trigger_worker.py +++ b/tests/test_dataflow_trigger_worker.py @@ -4,6 +4,7 @@ import unittest from unittest.mock import MagicMock, patch from govoplan_core.celery_app import celery, dispatch_dataflow_triggers +from tests.worker_test_support import allowed_worker_admissions class DataflowTriggerWorkerTests(unittest.TestCase): @@ -30,10 +31,18 @@ class DataflowTriggerWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = dispatch_dataflow_triggers.run(25) - provider.dispatch_due.assert_called_once_with(session, limit=25) + provider.dispatch_due.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) session.commit.assert_called_once_with() self.assertEqual(result["succeeded"], 1) diff --git a/tests/test_idm_assignment_lifecycle_worker.py b/tests/test_idm_assignment_lifecycle_worker.py index 9e49f43..6b46ba3 100644 --- a/tests/test_idm_assignment_lifecycle_worker.py +++ b/tests/test_idm_assignment_lifecycle_worker.py @@ -11,6 +11,7 @@ from govoplan_core.core.idm import ( ) from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry +from tests.worker_test_support import allowed_worker_admissions class _Lifecycle: @@ -70,6 +71,10 @@ class IdmAssignmentLifecycleWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = expire_idm_assignments.run("tenant-1", 25) diff --git a/tests/test_mail_delivery_worker.py b/tests/test_mail_delivery_worker.py index 399fd16..ab94a9e 100644 --- a/tests/test_mail_delivery_worker.py +++ b/tests/test_mail_delivery_worker.py @@ -10,6 +10,7 @@ from govoplan_core.celery_app import ( dispatch_mail_outbox, purge_mail_outbox, ) +from tests.worker_test_support import allowed_worker_admissions class _Provider: @@ -35,6 +36,10 @@ class MailDeliveryWorkerTests(unittest.TestCase): "govoplan_core.celery_app._mail_delivery_outbox", return_value=_Provider(), ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = dispatch_mail_outbox.run("tenant-1", 7) @@ -56,10 +61,15 @@ class MailDeliveryWorkerTests(unittest.TestCase): "govoplan_core.celery_app._mail_delivery_outbox", return_value=_Provider(), ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = purge_mail_outbox.run(19) self.assertIs(result["session"], session) + self.assertEqual(result["tenant_id"], "tenant-1") self.assertEqual(result["limit"], 19) def test_worker_routes_and_schedules_are_declared(self) -> None: diff --git a/tests/test_module_entitlements.py b/tests/test_module_entitlements.py index 61e2bf2..f72f32d 100644 --- a/tests/test_module_entitlements.py +++ b/tests/test_module_entitlements.py @@ -3,22 +3,27 @@ from __future__ import annotations import unittest from pathlib import Path import tempfile +from types import SimpleNamespace from unittest.mock import patch from fastapi import APIRouter, Depends, FastAPI from fastapi.testclient import TestClient from govoplan_core.auth import ApiPrincipal, get_api_principal +from govoplan_core.celery_app import _run_tenant_worker_batches from govoplan_core.core.access import PrincipalRef from govoplan_core.core.lifecycle import require_module_active from govoplan_core.core.module_entitlements import ( ModuleEntitlementConflict, ModuleEntitlementError, + TenantModuleEntitlementResolver, + TenantModuleOperatorActionRequired, + TenantModuleUnavailable, tenant_module_entitlement_state, update_system_tenant_module_policy, update_tenant_module_selection, ) -from govoplan_core.core.modules import ModuleManifest +from govoplan_core.core.modules import ModuleContext, ModuleManifest from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.session import configure_database, get_database from govoplan_core.server.platform import create_platform_router @@ -177,6 +182,75 @@ class TenantModuleEntitlementTests(unittest.TestCase): self.assertEqual({"access", "admin"}, set(state.effective_modules)) self.assertTrue(state.diagnostics) + def test_resolver_caches_and_invalidates_tenant_state(self) -> None: + registry = PlatformRegistry() + for manifest in self.manifests.values(): + registry.register(manifest) + tenant = SimpleNamespace(id="tenant-1", is_active=True, settings={}) + + class CountingSession: + calls = 0 + + def get(self, _model, _tenant_id): + self.calls += 1 + return tenant + + session = CountingSession() + resolver = TenantModuleEntitlementResolver( + registry, + ttl_seconds=60, + max_entries=2, + ) + + resolver.resolve(session, "tenant-1") + resolver.resolve(session, "tenant-1") + self.assertEqual(1, session.calls) + + resolver.invalidate("tenant-1") + resolver.resolve(session, "tenant-1") + self.assertEqual(2, session.calls) + + def test_new_and_accepted_work_have_distinct_disable_semantics(self) -> None: + registry = PlatformRegistry() + for manifest in self.manifests.values(): + registry.register(manifest) + settings, _state = update_system_tenant_module_policy( + {}, + self.manifests, + available_modules=(), + forced_modules=(), + enabled_modules=(), + expected_revision=0, + ) + tenant = SimpleNamespace( + id="tenant-1", + is_active=True, + settings=settings, + ) + session = SimpleNamespace(get=lambda _model, _tenant_id: tenant) + resolver = TenantModuleEntitlementResolver(registry, ttl_seconds=0) + + with self.assertRaises(TenantModuleUnavailable) as rejected: + resolver.require( + session, + tenant_id="tenant-1", + module_id="files", + work_state="new", + ) + self.assertEqual("rejected", rejected.exception.admission.disposition) + + with self.assertRaises(TenantModuleOperatorActionRequired) as preserved: + resolver.require( + session, + tenant_id="tenant-1", + module_id="files", + work_state="accepted", + ) + self.assertEqual( + "operator_action_required", + preserved.exception.admission.disposition, + ) + class TenantModuleEntitlementRouteTests(unittest.TestCase): def setUp(self) -> None: @@ -288,6 +362,144 @@ class TenantModuleEntitlementRouteTests(unittest.TestCase): self.assertEqual(200, response.status_code, response.text) + def test_public_tenant_route_enforces_module_entitlement(self) -> None: + registry = PlatformRegistry() + for manifest in self.manifests: + registry.register( + ModuleManifest( + id=manifest.id, + name=manifest.name, + version=manifest.version, + dependencies=manifest.dependencies, + public_tenant_resolver=( + (lambda _request, _session: "tenant-1") + if manifest.id == "files" + else None + ), + ) + ) + app = FastAPI() + app.state.govoplan_registry = registry + guarded = APIRouter(dependencies=[Depends(require_module_active("files"))]) + + @guarded.get("/public-files/{token}") + def public_files_route(token: str): + return {"token": token} + + app.include_router(guarded) + with TestClient(app) as client: + response = client.get("/public-files/example") + + self.assertEqual(404, response.status_code, response.text) + self.assertEqual( + "Module is unavailable in the active tenant: files", + response.json()["detail"], + ) + + def test_tenant_capability_rejects_unavailable_provider(self) -> None: + registry = PlatformRegistry() + for manifest in self.manifests: + registry.register( + ModuleManifest( + id=manifest.id, + name=manifest.name, + version=manifest.version, + dependencies=manifest.dependencies, + capability_factories=( + {"files.example": lambda _context: object()} + if manifest.id == "files" + else {} + ), + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=SimpleNamespace()) + ) + with get_database().session() as session: + with self.assertRaises(TenantModuleUnavailable): + registry.require_tenant_capability( + "files.example", + session, + tenant_id="tenant-1", + ) + + def test_request_context_treats_unavailable_optional_capability_as_absent(self) -> None: + registry = PlatformRegistry() + for manifest in self.manifests: + registry.register( + ModuleManifest( + id=manifest.id, + name=manifest.name, + version=manifest.version, + dependencies=manifest.dependencies, + capability_factories=( + {"files.example": lambda _context: object()} + if manifest.id == "files" + else {} + ), + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=SimpleNamespace()) + ) + app = FastAPI() + app.state.govoplan_registry = registry + guarded = APIRouter(dependencies=[Depends(require_module_active("admin"))]) + + @guarded.get("/admin-capability") + def admin_capability_route(): + return {"files_available": registry.capability("files.example") is not None} + + app.include_router(guarded) + with patch( + "govoplan_core.core.lifecycle.get_api_principal", + return_value=self.principal, + ), TestClient(app) as client: + response = client.get( + "/admin-capability", + headers={"Authorization": "Bearer test"}, + ) + + self.assertEqual(200, response.status_code, response.text) + self.assertFalse(response.json()["files_available"]) + + def test_worker_preserves_accepted_work_for_operator_when_disabled(self) -> None: + registry = PlatformRegistry() + for manifest in self.manifests: + registry.register( + ModuleManifest( + id=manifest.id, + name=manifest.name, + version=manifest.version, + dependencies=manifest.dependencies, + capability_factories=( + {"files.worker": lambda _context: object()} + if manifest.id == "files" + else {} + ), + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=SimpleNamespace()) + ) + invoked: list[str] = [] + with get_database().session() as session: + result = _run_tenant_worker_batches( + registry, + session, + capability_name="files.worker", + tenant_id="tenant-1", + operation=lambda tenant_id: invoked.append(tenant_id) or {}, + defaults={"processed": 0}, + ) + + self.assertEqual([], invoked) + self.assertEqual(1, result["operator_action_required"]) + self.assertEqual( + "operator_action_required", + result["operator_actions"][0]["disposition"], + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_module_system.py b/tests/test_module_system.py index 66409a6..b79bfaa 100644 --- a/tests/test_module_system.py +++ b/tests/test_module_system.py @@ -520,6 +520,7 @@ class ModuleSystemTests(unittest.TestCase): id="example", name="Example", version="test", + public_tenant_resolver=lambda _request, _session: "tenant-1", frontend=FrontendModule( module_id="example", package_name="@govoplan/example-webui", @@ -536,6 +537,7 @@ class ModuleSystemTests(unittest.TestCase): id="example", name="Example", version="test", + public_tenant_resolver=lambda _request, _session: "tenant-1", frontend=FrontendModule( module_id="example", package_name="@govoplan/example-webui", @@ -584,6 +586,7 @@ class ModuleSystemTests(unittest.TestCase): id=module_id, name=module_id.title(), version="test", + public_tenant_resolver=lambda _request, _session: "tenant-1", frontend=FrontendModule( module_id=module_id, public_routes=( diff --git a/tests/test_platform_event_worker.py b/tests/test_platform_event_worker.py index da589b3..337d4ec 100644 --- a/tests/test_platform_event_worker.py +++ b/tests/test_platform_event_worker.py @@ -10,6 +10,10 @@ from govoplan_core.celery_app import ( purge_platform_events, ) from govoplan_core.core.events import PlatformEvent +from govoplan_core.core.dataflows import CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER +from govoplan_core.core.events import CAPABILITY_PLATFORM_EVENT_OUTBOX +from govoplan_core.core.search import CAPABILITY_SEARCH_INDEX_WRITER +from tests.worker_test_support import allowed_worker_admissions class PlatformEventWorkerTests(unittest.TestCase): @@ -18,16 +22,30 @@ class PlatformEventWorkerTests(unittest.TestCase): 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, - } + outbox.dispatch_pending.side_effect = ( + { + "selected": 1, + "delivered": 1, + "retrying": 0, + "quarantined": 0, + "dispatched": 1, + "observer_failed": 0, + }, + { + "selected": 0, + "delivered": 0, + "retrying": 0, + "quarantined": 0, + "dispatched": 0, + "observer_failed": 0, + }, + ) dataflow = MagicMock() registry = MagicMock() + registry.has_capability.side_effect = lambda name: name in { + CAPABILITY_PLATFORM_EVENT_OUTBOX, + CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, + } with ( patch( @@ -54,12 +72,21 @@ class PlatformEventWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = dispatch_platform_events.run(25) - call = outbox.dispatch_pending.call_args + self.assertEqual(2, outbox.dispatch_pending.call_count) + call = outbox.dispatch_pending.call_args_list[0] self.assertEqual(session, call.args[0]) self.assertEqual(25, call.kwargs["limit"]) + self.assertEqual("tenant-1", call.kwargs["tenant_id"]) + system_call = outbox.dispatch_pending.call_args_list[1] + self.assertTrue(system_call.kwargs["tenantless_only"]) + self.assertIsNone(system_call.kwargs["tenant_id"]) consumer = call.kwargs["consumers"][0] self.assertEqual( "dataflow.event-triggers.v1", @@ -86,14 +113,24 @@ class PlatformEventWorkerTests(unittest.TestCase): 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, - } + outbox.dispatch_pending.side_effect = ( + { + "selected": 1, + "delivered": 1, + "retrying": 0, + "quarantined": 0, + "dispatched": 1, + "observer_failed": 0, + }, + { + "selected": 0, + "delivered": 0, + "retrying": 0, + "quarantined": 0, + "dispatched": 0, + "observer_failed": 0, + }, + ) search = MagicMock() search.process_changes.return_value = { "selected": 1, @@ -101,11 +138,16 @@ class PlatformEventWorkerTests(unittest.TestCase): "retrying": 0, "quarantined": 0, } + registry = MagicMock() + registry.has_capability.side_effect = lambda name: name in { + CAPABILITY_PLATFORM_EVENT_OUTBOX, + CAPABILITY_SEARCH_INDEX_WRITER, + } with ( patch( "govoplan_core.celery_app._platform_registry", - return_value=MagicMock(), + return_value=registry, ), patch( "govoplan_core.celery_app._platform_event_outbox", @@ -127,10 +169,16 @@ class PlatformEventWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = dispatch_platform_events.run(25) - consumer = outbox.dispatch_pending.call_args.kwargs["consumers"][0] + consumer = outbox.dispatch_pending.call_args_list[0].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") @@ -141,7 +189,11 @@ class PlatformEventWorkerTests(unittest.TestCase): event=event, delivery_key=delivery_key, ) - search.process_changes.assert_called_once_with(session, limit=25) + search.process_changes.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) self.assertEqual(1, result["search_changes"]["applied"]) session.commit.assert_called_once_with() @@ -150,9 +202,20 @@ class PlatformEventWorkerTests(unittest.TestCase): database = MagicMock() database.SessionLocal.return_value.__enter__.return_value = session outbox = MagicMock() - outbox.purge_terminal.return_value = {"deleted": 2} + outbox.purge_terminal.side_effect = ( + {"deleted": 2}, + {"deleted": 1}, + ) + registry = MagicMock() + registry.has_capability.side_effect = lambda name: ( + name == CAPABILITY_PLATFORM_EVENT_OUTBOX + ) with ( + patch( + "govoplan_core.celery_app._platform_registry", + return_value=registry, + ), patch( "govoplan_core.celery_app._platform_event_outbox", return_value=outbox, @@ -166,17 +229,25 @@ class PlatformEventWorkerTests(unittest.TestCase): "platform_event_outbox_terminal_retention_days", 30, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = purge_platform_events.run(75) - call = outbox.purge_terminal.call_args + self.assertEqual(2, outbox.purge_terminal.call_count) + call = outbox.purge_terminal.call_args_list[0] self.assertEqual(session, call.args[0]) self.assertEqual(75, call.kwargs["limit"]) + self.assertEqual("tenant-1", call.kwargs["tenant_id"]) + system_call = outbox.purge_terminal.call_args_list[1] + self.assertTrue(system_call.kwargs["tenantless_only"]) before = call.kwargs["before"] self.assertIsInstance(before, datetime) self.assertEqual(timezone.utc, before.tzinfo) session.commit.assert_called_once_with() - self.assertEqual({"deleted": 2}, result) + self.assertEqual(3, result["deleted"]) def test_worker_routes_and_periodic_tasks_are_registered(self) -> None: self.assertEqual( diff --git a/tests/test_poll_participation_contract.py b/tests/test_poll_participation_contract.py index e79ac2f..8c0f390 100644 --- a/tests/test_poll_participation_contract.py +++ b/tests/test_poll_participation_contract.py @@ -18,6 +18,9 @@ class _CompleteGateway: def resolve_participation(self, *args, **kwargs): raise NotImplementedError + def resolve_public_invitation(self, *args, **kwargs): + raise NotImplementedError + def submit_governed_response(self, *args, **kwargs): raise NotImplementedError diff --git a/tests/test_postbox_routing_worker.py b/tests/test_postbox_routing_worker.py index 00dc447..892ffd6 100644 --- a/tests/test_postbox_routing_worker.py +++ b/tests/test_postbox_routing_worker.py @@ -11,6 +11,7 @@ from govoplan_core.core.postbox import ( postbox_routing_provider, ) from govoplan_core.core.registry import PlatformRegistry +from tests.worker_test_support import allowed_worker_admissions class _RoutingProvider: @@ -68,6 +69,10 @@ class PostboxRoutingWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = dispatch_postbox_routes.run("tenant-1", 25) diff --git a/tests/test_workflow_runtime_worker.py b/tests/test_workflow_runtime_worker.py index 70c3ab0..d9447ea 100644 --- a/tests/test_workflow_runtime_worker.py +++ b/tests/test_workflow_runtime_worker.py @@ -22,6 +22,7 @@ from govoplan_core.core.workflows import ( workflow_runtime_worker, workflow_trigger_dispatcher, ) +from tests.worker_test_support import allowed_worker_admissions class _Worker: @@ -141,10 +142,18 @@ class WorkflowRuntimeWorkerTests(unittest.TestCase): "govoplan_core.db.session.get_database", return_value=database, ), + patch( + "govoplan_core.celery_app._worker_admissions", + side_effect=allowed_worker_admissions, + ), ): result = reconcile_workflow_instances.run(25) - worker.reconcile_pending.assert_called_once_with(session, limit=25) + worker.reconcile_pending.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) session.commit.assert_called_once_with() self.assertEqual(1, result["advanced"]) diff --git a/tests/worker_test_support.py b/tests/worker_test_support.py new file mode 100644 index 0000000..ece6432 --- /dev/null +++ b/tests/worker_test_support.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from govoplan_core.core.module_entitlements import ( + TenantModuleAdmission, + TenantWorkState, +) + + +def allowed_worker_admissions( + _registry, + _session, + *, + capability_name: str, + tenant_id: str | None, + work_state: TenantWorkState = "accepted", +) -> tuple[TenantModuleAdmission, ...]: + return ( + TenantModuleAdmission( + tenant_id=tenant_id or "tenant-1", + module_id=capability_name.split(".", 1)[0], + revision=1, + work_state=work_state, + allowed=True, + disposition="allowed", + reason="Test tenant permits the worker capability.", + ), + )