from __future__ import annotations from celery import Celery from govoplan_core.core.campaigns import CAPABILITY_CAMPAIGNS_DELIVERY_TASKS, CampaignDeliveryTaskProvider 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.runtime import configure_runtime from govoplan_core.settings import settings from govoplan_core.db.session import configure_database from govoplan_core.server.registry import available_module_manifests, build_platform_registry configure_database(settings.database_url) celery = Celery( "multimailer", broker=settings.redis_url, backend=settings.redis_url, ) celery.conf.update( task_default_queue="default", task_routes={ "multimailer.send_email": {"queue": "send_email"}, "multimailer.append_sent": {"queue": "append_sent"}, }, worker_prefetch_multiplier=1, task_acks_late=True, task_reject_on_worker_lost=True, ) @celery.task(name="multimailer.ping") def ping(): return "pong" def _campaign_delivery_tasks() -> CampaignDeliveryTaskProvider: 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) capability = registry.require_capability(CAPABILITY_CAMPAIGNS_DELIVERY_TASKS) if not isinstance(capability, CampaignDeliveryTaskProvider): raise RuntimeError("Campaign delivery task capability is invalid") return capability @celery.task(name="multimailer.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="multimailer.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