from __future__ import annotations from collections.abc import Callable, Mapping from datetime import datetime, timedelta, timezone from importlib.metadata import PackageNotFoundError, version import logging import os import time from celery import Celery from celery.signals import ( heartbeat_sent, task_prerun, worker_process_init, worker_ready, worker_shutdown, ) from govoplan_core.core.campaigns import ( CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CAPABILITY_CAMPAIGNS_SCHEDULES, CampaignDeliveryTaskProvider, CampaignScheduleProvider, ) from govoplan_core.core.calendar import ( CAPABILITY_CALENDAR_OUTBOX, CalendarOutboxProvider, ) from govoplan_core.core.dataflows import ( CAPABILITY_DATAFLOW_RUN_WORKER, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER, DataflowRunWorker, DataflowTriggerDispatcher, ) from govoplan_core.core.events import ( CAPABILITY_PLATFORM_EVENT_OUTBOX, DurableEventConsumer, PlatformEvent, PlatformEventOutbox, publish_platform_event, ) from govoplan_core.core.idm import ( CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE, IdmAssignmentLifecycle, ) from govoplan_core.core.module_management import ( load_startup_enabled_modules, ) from govoplan_core.core.module_entitlements import ( TenantModuleAdmission, TenantModuleOperatorActionRequired, TenantWorkState, tenant_execution_scope, ) from govoplan_core.core.mail import ( CAPABILITY_MAIL_BOUNCE_PROCESSING, CAPABILITY_MAIL_DELIVERY_OUTBOX, MailBounceProcessingProvider, MailDeliveryOutboxProvider, ) from govoplan_core.core.notifications import ( CAPABILITY_NOTIFICATIONS_DISPATCH, NotificationDispatchProvider, ) from govoplan_core.core.postbox import ( CAPABILITY_POSTBOX_ROUTING, PostboxRoutingProvider, ) from govoplan_core.core.workflows import ( CAPABILITY_WORKFLOW_RUNTIME_WORKER, CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER, WorkflowRuntimeWorker, WorkflowTriggerDispatcher, ) from govoplan_core.core.registry import PlatformRegistry from govoplan_core.core.worker_runtime import build_worker_platform_registry from govoplan_core.core.runtime_coordination import ( RuntimeIdentity, bind_process_runtime_identity, heartbeat_runtime_node, register_runtime_node, 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 configure_database(settings.database_url) celery = Celery( "govoplan", broker=settings.redis_url, backend=settings.redis_url, ) celery.conf.update( task_default_queue="default", task_routes={ "govoplan.campaigns.send_email": {"queue": "send_email"}, "govoplan.campaigns.append_sent": {"queue": "append_sent"}, "govoplan.campaigns.dispatch_schedules": {"queue": "default"}, "govoplan.notifications.deliver": {"queue": "notifications"}, "govoplan.notifications.deliver_pending": {"queue": "notifications"}, "govoplan.mail.dispatch_outbox": {"queue": "mail"}, "govoplan.mail.purge_outbox": {"queue": "mail"}, "govoplan.mail.scan_bounces": {"queue": "mail"}, "govoplan.calendar.dispatch_outbox": {"queue": "calendar"}, "govoplan.dataflow.dispatch_runs": {"queue": "dataflow"}, "govoplan.dataflow.purge_runs": {"queue": "dataflow"}, "govoplan.dataflow.dispatch_triggers": {"queue": "dataflow"}, "govoplan.workflow.reconcile": {"queue": "workflow"}, "govoplan.postbox.dispatch_routes": {"queue": "postbox"}, "govoplan.events.dispatch_outbox": {"queue": "events"}, "govoplan.events.purge_outbox": {"queue": "events"}, "govoplan.idm.expire_assignments": {"queue": "idm"}, }, worker_prefetch_multiplier=1, task_acks_late=True, task_reject_on_worker_lost=True, task_track_started=True, broker_transport_options={ "visibility_timeout": settings.celery_visibility_timeout_seconds, }, result_backend_transport_options={ "visibility_timeout": settings.celery_visibility_timeout_seconds, }, visibility_timeout=settings.celery_visibility_timeout_seconds, beat_schedule={ "calendar-outbox-every-minute": { "task": "govoplan.calendar.dispatch_outbox", "schedule": 60.0, "args": (None, 100), }, "campaign-schedules-every-minute": { "task": "govoplan.campaigns.dispatch_schedules", "schedule": 60.0, "args": (None, 50), }, "mail-outbox-every-five-seconds": { "task": "govoplan.mail.dispatch_outbox", "schedule": 5.0, "args": (None, 25), }, "mail-outbox-retention-daily": { "task": "govoplan.mail.purge_outbox", "schedule": 24 * 60 * 60.0, "args": (250,), }, "mail-bounces-every-five-minutes": { "task": "govoplan.mail.scan_bounces", "schedule": 5 * 60.0, "args": (None, 250), }, "dataflow-triggers-every-minute": { "task": "govoplan.dataflow.dispatch_triggers", "schedule": 60.0, "args": (100,), }, "dataflow-runs-every-five-seconds": { "task": "govoplan.dataflow.dispatch_runs", "schedule": 5.0, "args": (10,), }, "dataflow-run-retention-daily": { "task": "govoplan.dataflow.purge_runs", "schedule": 24 * 60 * 60.0, "args": (500,), }, "workflow-reconcile-every-five-seconds": { "task": "govoplan.workflow.reconcile", "schedule": 5.0, "args": (50,), }, "postbox-routes-every-minute": { "task": "govoplan.postbox.dispatch_routes", "schedule": 60.0, "args": (None, 50), }, "platform-events-every-ten-seconds": { "task": "govoplan.events.dispatch_outbox", "schedule": 10.0, "args": (100,), }, "platform-event-retention-daily": { "task": "govoplan.events.purge_outbox", "schedule": 24 * 60 * 60.0, "args": (500,), }, "idm-assignment-expiry-every-minute": { "task": "govoplan.idm.expire_assignments", "schedule": 60.0, "args": (None, 100), }, }, ) _worker_identity: RuntimeIdentity | None = None _worker_draining = False _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") except PackageNotFoundError: return "development" def _worker_runtime_identity(sender: object | None = None) -> RuntimeIdentity: global _worker_identity if _worker_identity is None: hostname = str(getattr(sender, "hostname", "") or "").strip() or None module_ids = tuple(load_startup_enabled_modules(settings.enabled_modules)) _worker_identity = runtime_identity( settings, software_version=_core_version(), module_ids=module_ids, role="worker", node_id=hostname, ) bind_process_runtime_identity(_worker_identity) return _worker_identity def _worker_metadata() -> dict[str, object]: raw_concurrency = str(os.getenv("CELERY_WORKER_CONCURRENCY") or "").strip() return { "process": "celery-worker", "worker_pool": str(os.getenv("GOVOPLAN_WORKER_POOL") or "default"), "concurrency": int(raw_concurrency) if raw_concurrency.isdigit() else None, } @worker_process_init.connect def _reset_worker_process_database(**_kwargs) -> None: global _worker_identity # SQLAlchemy pools must not be shared across prefork child processes. _worker_identity = None bind_process_runtime_identity(None) configure_database(settings.database_url, dispose_previous=True) @task_prerun.connect def _bind_worker_effect_identity(task=None, **_kwargs) -> None: _worker_runtime_identity(task) @worker_ready.connect def _register_worker_runtime(sender=None, **_kwargs) -> None: global _worker_consumer _worker_consumer = sender identity = _worker_runtime_identity(sender) try: with get_database().SessionLocal() as session: node = register_runtime_node( session, identity, metadata=_worker_metadata(), ) session.commit() except Exception: _set_worker_consumers(identity, draining=True) logger.exception( "worker runtime registration failed; queues were disabled node_id=%s", identity.node_id, ) raise _set_worker_consumers(identity, draining=node.state == "draining") @heartbeat_sent.connect def _heartbeat_worker_runtime(sender=None, **_kwargs) -> None: identity = _worker_runtime_identity(sender) try: with get_database().SessionLocal() as session: node = heartbeat_runtime_node( session, identity, metadata=_worker_metadata(), ) session.commit() except Exception: # noqa: BLE001 - uncertain authority must fail closed _set_worker_consumers(identity, draining=True) logger.exception( "worker runtime heartbeat failed; queues were disabled node_id=%s", identity.node_id, ) return _set_worker_consumers(identity, draining=node.state == "draining") def _set_worker_consumers( identity: RuntimeIdentity, *, draining: bool, ) -> None: global _worker_draining if draining == _worker_draining: return consumer = _worker_consumer if consumer is not None: method_name = "cancel_task_queue" if draining else "add_task_queue" method = getattr(consumer, method_name) destination = None else: method = ( celery.control.cancel_consumer if draining else celery.control.add_consumer ) destination = [identity.node_id] for queue in identity.queues: if destination is None: method(queue) else: method(queue, destination=destination) _worker_draining = draining @worker_shutdown.connect def _stop_worker_runtime(sender=None, **_kwargs) -> None: identity = _worker_identity if identity is None: return try: with get_database().SessionLocal() as session: stop_runtime_node(session, identity) session.commit() except Exception: # noqa: BLE001 - shutdown must continue logger.exception( "worker runtime stop marker failed node_id=%s", identity.node_id, ) @celery.task(name="govoplan.ping") def ping(): return "pong" @celery.task( bind=True, name="govoplan.worker.acceptance", max_retries=1, acks_late=True, reject_on_worker_lost=True, track_started=True, ) def worker_acceptance_probe( task, probe_id: str, *, mode: str = "complete", delay_seconds: float = 0.0, track_delivery: bool = False, ): """Exercise broker delivery semantics without touching business data.""" if mode not in {"complete", "retry_once"}: raise ValueError(f"Unsupported worker acceptance mode: {mode}") normalized_probe_id = str(probe_id).strip() if not normalized_probe_id or len(normalized_probe_id) > 120: raise ValueError("Worker acceptance probe_id must contain 1-120 characters") bounded_delay = max(0.0, min(float(delay_seconds), 120.0)) delivery_count = _worker_acceptance_delivery_count( task, normalized_probe_id, ) if track_delivery else None task.update_state( state="PROGRESS", meta={ "phase": "started", "probe_id": normalized_probe_id, "retries": int(task.request.retries or 0), "delivery_count": delivery_count, }, ) if mode == "retry_once" and int(task.request.retries or 0) == 0: raise task.retry(countdown=0.2) if bounded_delay: time.sleep(bounded_delay) delivery_info = task.request.delivery_info or {} return { "probe_id": normalized_probe_id, "mode": mode, "retries": int(task.request.retries or 0), "redelivered": bool(delivery_info.get("redelivered")), "delivery_count": delivery_count, } def _worker_acceptance_delivery_count(task, probe_id: str) -> int: client = getattr(task.backend, "client", None) if client is None: raise RuntimeError( "Worker acceptance delivery tracking requires a Redis result backend" ) key = f"govoplan:worker-acceptance:{probe_id}" count = int(client.incr(key)) client.expire(key, 60 * 60) return count def _platform_registry() -> PlatformRegistry: return build_worker_platform_registry(settings) 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 _campaign_schedules( registry: PlatformRegistry | None = None, ) -> CampaignScheduleProvider | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_CAMPAIGNS_SCHEDULES): return None capability = registry.require_capability(CAPABILITY_CAMPAIGNS_SCHEDULES) if not isinstance(capability, CampaignScheduleProvider): raise RuntimeError("Campaign schedule capability is invalid") return capability 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( 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) if not isinstance(capability, CalendarOutboxProvider): raise RuntimeError("Calendar outbox capability is invalid") return capability 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) if not isinstance(capability, MailDeliveryOutboxProvider): raise RuntimeError("Mail delivery outbox capability is invalid") return capability 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) if not isinstance(capability, MailBounceProcessingProvider): raise RuntimeError("Mail bounce-processing capability is invalid") return capability def _dataflow_trigger_dispatcher( registry: PlatformRegistry | None = None, ) -> DataflowTriggerDispatcher | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER): return None capability = registry.require_capability(CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER) if not isinstance(capability, DataflowTriggerDispatcher): raise RuntimeError("Dataflow trigger dispatcher capability is invalid") return capability def _dataflow_run_worker( registry: PlatformRegistry | None = None, ) -> DataflowRunWorker | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER): return None capability = registry.require_capability(CAPABILITY_DATAFLOW_RUN_WORKER) if not isinstance(capability, DataflowRunWorker): raise RuntimeError("Dataflow run worker capability is invalid") return capability def _workflow_runtime_worker( registry: PlatformRegistry | None = None, ) -> WorkflowRuntimeWorker | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_WORKFLOW_RUNTIME_WORKER): return None capability = registry.require_capability(CAPABILITY_WORKFLOW_RUNTIME_WORKER) if not isinstance(capability, WorkflowRuntimeWorker): raise RuntimeError("Workflow runtime worker capability is invalid") return capability def _workflow_trigger_dispatcher( registry: PlatformRegistry | None = None, ) -> WorkflowTriggerDispatcher | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER): return None capability = registry.require_capability(CAPABILITY_WORKFLOW_TRIGGER_DISPATCHER) if not isinstance(capability, WorkflowTriggerDispatcher): raise RuntimeError("Workflow trigger dispatcher capability is invalid") return capability def _postbox_routing_provider( registry: PlatformRegistry | None = None, ) -> PostboxRoutingProvider | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING): return None capability = registry.require_capability(CAPABILITY_POSTBOX_ROUTING) if not isinstance(capability, PostboxRoutingProvider): raise RuntimeError("Postbox routing capability is invalid") return capability def _platform_event_outbox( registry: PlatformRegistry | None = None, ) -> PlatformEventOutbox | None: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_PLATFORM_EVENT_OUTBOX): return None capability = registry.require_capability(CAPABILITY_PLATFORM_EVENT_OUTBOX) if not isinstance(capability, PlatformEventOutbox): raise RuntimeError("Platform event outbox capability is invalid") 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: registry = registry or _platform_registry() if not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE): return None capability = registry.require_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE) if not isinstance(capability, IdmAssignmentLifecycle): raise RuntimeError("IDM assignment lifecycle capability is invalid") return capability @celery.task( name="govoplan.campaigns.dispatch_schedules", bind=True, max_retries=0, ) def dispatch_campaign_schedules( self, tenant_id: str | None = None, limit: int = 50, ): """Prepare manual drafts or governed autonomous Mail commands for due schedules.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: registry = _platform_registry() defaults = { "selected": 0, "prepared": 0, "autonomous_prepared": 0, "failed": 0, "completed": 0, "coalesced": 0, "duplicates": 0, "deferred": 0, "campaign_ids": [], "operator_actions": [], "refreshed": { "checked": 0, "accepted": 0, "uncertain": 0, "failed": 0, "skipped": 0, }, } if not registry.has_capability(CAPABILITY_CAMPAIGNS_SCHEDULES): return defaults result = _run_tenant_worker_batches( registry, session, capability_name=CAPABILITY_CAMPAIGNS_SCHEDULES, tenant_id=tenant_id, operation=lambda effective_tenant_id: _campaign_schedules( registry ).dispatch_due( # type: ignore[union-attr] session, tenant_id=effective_tenant_id, limit=limit, ), defaults=defaults, work_state="new", ) session.commit() return result @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0) def send_email(self, job_id: str): """Send one explicitly queued campaign job. SMTP failures are persisted but are not retried implicitly. A worker-loss redelivery is safe because the delivery service converts an unfinished SMTP attempt into ``outcome_unknown`` instead of transmitting again. """ 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) 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, ), ) @celery.task(name="govoplan.campaigns.append_sent", bind=True, max_retries=None) def append_sent(self, job_id: str): """Append the exact sent MIME to the configured IMAP Sent folder.""" 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: 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: raise self.retry(exc=exc, countdown=300) raise @celery.task(name="govoplan.notifications.deliver", bind=True, max_retries=0) def deliver_notification(self, notification_id: str): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task(name="govoplan.notifications.deliver_pending", bind=True, max_retries=0) def deliver_pending_notifications(self, tenant_id: str | None = None, limit: int = 50): from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task(name="govoplan.mail.dispatch_outbox", bind=True, max_retries=0) def dispatch_mail_outbox( self, tenant_id: str | None = None, limit: int = 25, ): """Drain durable Mail commands; retry and reconciliation live in Mail.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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=effective_tenant_id, limit=limit, worker_id=getattr(self.request, "hostname", None), ), defaults=defaults, ) @celery.task(name="govoplan.mail.purge_outbox", bind=True, max_retries=0) def purge_mail_outbox(self, limit: int = 250): """Minimize expired Mail payloads while retaining delivery evidence.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: registry = _platform_registry() if not registry.has_capability(CAPABILITY_MAIL_DELIVERY_OUTBOX): return {"purged": 0} 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) def scan_mail_bounces( self, tenant_id: str | None = None, limit: int = 250, ): """Read configured DSN folders without mutating provider mailbox flags.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task(name="govoplan.calendar.dispatch_outbox", bind=True, max_retries=0) def dispatch_calendar_outbox(self, tenant_id: str | None = None, limit: int = 50): """Drain durable Calendar operations; retry timing lives in the database.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task( name="govoplan.dataflow.dispatch_triggers", bind=True, max_retries=0, ) def dispatch_dataflow_triggers(self, limit: int = 100): """Drain durable Dataflow trigger deliveries and due schedules.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task( name="govoplan.dataflow.dispatch_runs", bind=True, max_retries=0, ) def dispatch_dataflow_runs(self, limit: int = 10): """Claim and execute durable Dataflow runs outside the API process.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task( name="govoplan.dataflow.purge_runs", bind=True, max_retries=0, ) def purge_dataflow_runs(self, limit: int = 500): """Apply Dataflow evidence-retention policy without deleting run records.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: registry = _platform_registry() if not registry.has_capability(CAPABILITY_DATAFLOW_RUN_WORKER): return {"purged": 0} 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 @celery.task( name="govoplan.workflow.reconcile", bind=True, max_retries=0, ) def reconcile_workflow_instances(self, limit: int = 50): """Resume asynchronous Workflow steps from durable provider state.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 @celery.task( name="govoplan.postbox.dispatch_routes", bind=True, max_retries=0, ) def dispatch_postbox_routes( self, tenant_id: str | None = None, limit: int = 50, ): """Deliver due routes and reconcile assignment-derived notification facts.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: registry = _platform_registry() defaults = { "selected": 0, "delivered": 0, "vacant": 0, "rescheduled": 0, "cancelled": 0, "failed": 0, "route_ids": [], "lifecycle_scanned": 0, "lifecycle_changed": 0, "lifecycle_events": 0, "lifecycle_notifications": 0, "lifecycle_notification_failures": 0, } if not registry.has_capability(CAPABILITY_POSTBOX_ROUTING): return defaults provider = _postbox_routing_provider(registry) def dispatch_tenant(effective_tenant_id: str) -> Mapping[str, object]: route_result = provider.dispatch_due_routes( # type: ignore[union-attr] session, tenant_id=effective_tenant_id, limit=limit, ) lifecycle_result = provider.reconcile_notification_lifecycle( # type: ignore[union-attr] session, tenant_id=effective_tenant_id, limit=limit, ) return { **route_result, **{ f"lifecycle_{key}": value for key, value in lifecycle_result.items() }, } result = _run_tenant_worker_batches( registry, session, capability_name=CAPABILITY_POSTBOX_ROUTING, tenant_id=tenant_id, operation=dispatch_tenant, defaults=defaults, ) session.commit() return result @celery.task( name="govoplan.idm.expire_assignments", bind=True, max_retries=0, ) def expire_idm_assignments( self, tenant_id: str | None = None, limit: int = 100, ): """Emit idempotent lifecycle events for elapsed IDM assignments.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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=effective_tenant_id, limit=limit, ), defaults=defaults, ) session.commit() return result @celery.task( name="govoplan.events.dispatch_outbox", bind=True, max_retries=0, ) def dispatch_platform_events(self, limit: int = 100): """Deliver committed platform events through persistent consumer ledgers.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: registry = _platform_registry() 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 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, 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 ) def deliver_to_dataflow( event: PlatformEvent, _delivery_key: str, ) -> None: assert dataflow_dispatcher is not None dataflow_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) ), ) ) 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 ) 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 if admission is None or admission.allowed else blocked_handler(admission) ), ) ) 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 @celery.task( name="govoplan.events.purge_outbox", bind=True, max_retries=0, ) def purge_platform_events(self, limit: int = 500): """Remove old terminal event envelopes while retaining quarantine evidence.""" from govoplan_core.db.session import get_database with get_database().SessionLocal() as session: 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 = _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