from __future__ import annotations

from datetime import datetime, timedelta, timezone
from importlib.metadata import PackageNotFoundError, version
import logging

from celery import Celery
from celery.signals import heartbeat_sent, worker_ready, worker_shutdown

from govoplan_core.core.campaigns import (
    CAPABILITY_CAMPAIGNS_DELIVERY_TASKS,
    CampaignDeliveryTaskProvider,
)
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,
    startup_candidate_module_ids,
)
from govoplan_core.core.mail import (
    CAPABILITY_MAIL_BOUNCE_PROCESSING,
    CAPABILITY_MAIL_DELIVERY_OUTBOX,
    MailBounceProcessingProvider,
    MailDeliveryOutboxProvider,
)
from govoplan_core.core.modules import ModuleContext
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.runtime import configure_runtime
from govoplan_core.core.runtime_coordination import (
    RuntimeIdentity,
    heartbeat_runtime_node,
    register_runtime_node,
    runtime_identity,
    stop_runtime_node,
)
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)

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.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,
    beat_schedule={
        "calendar-outbox-every-minute": {
            "task": "govoplan.calendar.dispatch_outbox",
            "schedule": 60.0,
            "args": (None, 100),
        },
        "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 _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,
        )
    return _worker_identity


@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={"process": "celery-worker"},
            )
            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={"process": "celery-worker"},
            )
            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"


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


def _campaign_delivery_tasks() -> CampaignDeliveryTaskProvider:
    registry = _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()
    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()
    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() -> MailDeliveryOutboxProvider | None:
    registry = _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() -> MailBounceProcessingProvider | None:
    registry = _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 _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.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:
        return dict(
            _campaign_delivery_tasks().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:
        try:
            return dict(
                _campaign_delivery_tasks().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:
        result = dict(
            _notification_dispatch().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:
        result = dict(
            _notification_dispatch().deliver_pending(
                session, tenant_id=tenant_id, limit=limit
            )
        )
        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:
        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(
                session,
                tenant_id=tenant_id,
                limit=limit,
                worker_id=getattr(self.request, "hostname", None),
            )
        )


@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:
        provider = _mail_delivery_outbox()
        if provider is None:
            return {"purged": 0}
        return dict(provider.purge_expired(session, limit=limit))


@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:
        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))
        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:
        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))
        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:
        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))
        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:
        provider = _dataflow_run_worker()
        if provider is None:
            return {
                "claimed": 0,
                "succeeded": 0,
                "retrying": 0,
                "failed": 0,
                "cancelled": 0,
            }
        result = dict(
            provider.dispatch_pending(
                session,
                limit=limit,
                worker_id=getattr(self.request, "hostname", None),
            )
        )
        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:
        provider = _dataflow_run_worker()
        if provider is None:
            return {"purged": 0}
        result = dict(provider.purge_expired(session, limit=limit))
        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:
        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))
        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 Postbox vacancy escalations from durable route rows."""

    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(
                session,
                tenant_id=tenant_id,
                limit=limit,
            )
        )
        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:
        provider = _idm_assignment_lifecycle()
        if provider is None:
            return {"selected": 0, "expired": 0, "assignment_ids": []}
        result = dict(
            provider.process_expired(
                session,
                tenant_id=tenant_id,
                limit=limit,
            )
        )
        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()
        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)
        consumers: list[DurableEventConsumer] = []
        if dataflow_dispatcher is not None:

            def deliver_to_dataflow(
                event: PlatformEvent,
                _delivery_key: str,
            ) -> 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 workflow_dispatcher is not None:

            def deliver_to_workflow(
                event: PlatformEvent,
                _delivery_key: str,
            ) -> 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,
                )
            )

        result = dict(
            outbox.dispatch_pending(
                session,
                consumers=tuple(consumers),
                observer=publish_platform_event,
                limit=limit,
            )
        )
        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:
        outbox = _platform_event_outbox()
        if outbox is None:
            return {"deleted": 0}
        before = datetime.now(timezone.utc) - timedelta(
            days=settings.platform_event_outbox_terminal_retention_days
        )
        result = dict(
            outbox.purge_terminal(
                session,
                before=before,
                limit=limit,
            )
        )
        session.commit()
        return result
