from __future__ import annotations import hashlib import json from dataclasses import asdict, dataclass from datetime import datetime, timezone from email import policy from email.parser import BytesParser from pathlib import Path from typing import Any from uuid import uuid4 from sqlalchemy.orm import Session from govoplan_core.core.notifications import NotificationDispatchRequest, notification_dispatch_provider from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import ( Campaign, CampaignJob, CampaignStatus, CampaignVersion, CampaignVersionWorkflowState, JobBuildStatus, JobImapStatus, JobQueueStatus, JobSendStatus, JobValidationStatus, ImapAppendAttempt, SendAttempt, ) from govoplan_campaign.backend.sending.execution import ExecutionSnapshot, ExecutionSnapshotError, ensure_execution_snapshot, runtime_imap_config, runtime_smtp_config from govoplan_campaign.backend.runtime import get_registry from govoplan_campaign.backend.integrations import ( ImapAppendError, ImapConfigurationError, MailProfileError, SmtpConfigurationError, SmtpSendError, files_integration, mail_integration, ) class QueueingError(RuntimeError): pass class SendJobError(RuntimeError): pass @dataclass(frozen=True, slots=True) class QueueCampaignResult: campaign_id: str version_id: str queued_count: int skipped_count: int blocked_count: int enqueued_count: int dry_run: bool = False def as_dict(self) -> dict[str, Any]: return { "campaign_id": self.campaign_id, "version_id": self.version_id, "queued_count": self.queued_count, "skipped_count": self.skipped_count, "blocked_count": self.blocked_count, "enqueued_count": self.enqueued_count, "dry_run": self.dry_run, } @dataclass(frozen=True, slots=True) class SendCampaignNowResult: campaign_id: str version_id: str attempted_count: int sent_count: int failed_count: int outcome_unknown_count: int skipped_count: int dry_run: bool = False results: list[dict[str, Any]] | None = None def as_dict(self) -> dict[str, Any]: return { "campaign_id": self.campaign_id, "version_id": self.version_id, "attempted_count": self.attempted_count, "sent_count": self.sent_count, "failed_count": self.failed_count, "outcome_unknown_count": self.outcome_unknown_count, "skipped_count": self.skipped_count, "dry_run": self.dry_run, "results": self.results or [], } @dataclass(frozen=True, slots=True) class SendJobResult: job_id: str status: str attempt_number: int dry_run: bool = False message: str | None = None def as_dict(self) -> dict[str, Any]: return { "job_id": self.job_id, "status": self.status, "attempt_number": self.attempt_number, "dry_run": self.dry_run, "message": self.message, } @dataclass(frozen=True, slots=True) class AppendSentResult: job_id: str status: str attempt_number: int dry_run: bool = False folder: str | None = None message: str | None = None def as_dict(self) -> dict[str, Any]: return { "job_id": self.job_id, "status": self.status, "attempt_number": self.attempt_number, "dry_run": self.dry_run, "folder": self.folder, "message": self.message, } @dataclass(frozen=True, slots=True) class _CampaignDeliveryCounts: accepted: int unknown: int failed: int active: int cancelled: int not_started: int @dataclass(frozen=True, slots=True) class _SendJobDeliveryContext: version: CampaignVersion snapshot: ExecutionSnapshot message_bytes: bytes envelope_from: str envelope_recipients: list[str] QUEUEABLE_VALIDATION_STATUSES = { JobValidationStatus.READY.value, JobValidationStatus.WARNING.value, } SMTP_ACCEPTED_STATUSES = {JobSendStatus.SMTP_ACCEPTED.value, JobSendStatus.SENT.value} AUTOMATICALLY_SENDABLE_STATUSES = {JobSendStatus.QUEUED.value} EXPLICIT_RETRY_STATUSES = {JobSendStatus.FAILED_TEMPORARY.value, JobSendStatus.FAILED_PERMANENT.value} INITIAL_QUEUE_SKIPPED_SEND_STATUSES = SMTP_ACCEPTED_STATUSES | { JobSendStatus.CLAIMED.value, JobSendStatus.SENDING.value, JobSendStatus.OUTCOME_UNKNOWN.value, JobSendStatus.FAILED_TEMPORARY.value, JobSendStatus.FAILED_PERMANENT.value, JobSendStatus.CANCELLED.value, } INITIAL_QUEUE_SKIPPED_QUEUE_STATUSES = { JobQueueStatus.CANCELLED.value, JobQueueStatus.SENDING.value, JobQueueStatus.PAUSED.value, } CAMPAIGN_STATUS_NOTIFICATION_EVENTS = { CampaignStatus.QUEUED.value: ("campaign.queued", "Campaign queued", 1), CampaignStatus.SENDING.value: ("campaign.sending", "Campaign sending started", 1), CampaignStatus.SENT.value: ("campaign.sent", "Campaign sent", 1), CampaignStatus.PARTIALLY_COMPLETED.value: ("campaign.partially_completed", "Campaign partially completed", 4), CampaignStatus.OUTCOME_UNKNOWN.value: ("campaign.outcome_unknown", "Campaign requires review", 5), CampaignStatus.FAILED.value: ("campaign.failed", "Campaign failed", 5), CampaignStatus.CANCELLED.value: ("campaign.cancelled", "Campaign cancelled", 2), } def _version_user_lock_state(version: CampaignVersion) -> str | None: state = getattr(version, "user_lock_state", None) if state in {"temporary", "permanent"}: return state return "permanent" if version.published_at else None def _version_is_user_locked(version: CampaignVersion) -> bool: return _version_user_lock_state(version) is not None def _version_is_validated_and_locked(version: CampaignVersion) -> bool: validation = version.validation_summary if isinstance(version.validation_summary, dict) else {} return bool(version.locked_at and validation.get("ok") is True and not _version_is_user_locked(version)) def _ensure_version_validated_and_locked(version: CampaignVersion) -> None: state = _version_user_lock_state(version) if state == "temporary": raise QueueingError("This version has a temporary user lock. Unlock it before queueing, dry-run or sending.") if state == "permanent": raise QueueingError("This version is permanently user-locked. Create an editable copy instead.") if not _version_is_validated_and_locked(version): raise QueueingError("Campaign version must be validated and locked before building, queueing, dry-run or sending.") def _reviewed_needs_review_keys(version: CampaignVersion) -> set[str]: build_summary = version.build_summary if isinstance(version.build_summary, dict) else {} build_token = str(build_summary.get("build_token") or build_summary.get("built_at") or "") editor_state = version.editor_state if isinstance(version.editor_state, dict) else {} review_state = editor_state.get("review_send") if isinstance(editor_state.get("review_send"), dict) else {} if not build_token or str(review_state.get("build_token") or "") != build_token: return set() if review_state.get("inspection_complete") is not True: return set() return {str(value) for value in (review_state.get("reviewed_message_keys") or []) if str(value).strip()} def _job_review_key(job: CampaignJob) -> str: return str(job.entry_id or job.entry_index) def _utcnow() -> datetime: return datetime.now(timezone.utc) def _get_campaign_for_tenant(session: Session, *, campaign_id: str, tenant_id: str) -> Campaign: campaign = session.query(Campaign).filter(Campaign.id == campaign_id, Campaign.tenant_id == tenant_id).one_or_none() if not campaign: raise QueueingError(f"Campaign not found or not accessible: {campaign_id}") return campaign def _get_version_for_campaign( session: Session, campaign: Campaign, version_id: str | None = None, ) -> CampaignVersion: wanted = version_id or campaign.current_version_id if not wanted: raise QueueingError("Campaign has no selected version") version = session.get(CampaignVersion, wanted) if not version or version.campaign_id != campaign.id: raise QueueingError(f"Campaign version not found or not part of campaign: {wanted}") return version def _get_current_version(session: Session, campaign: Campaign, version_id: str | None = None) -> CampaignVersion: version = _get_version_for_campaign(session, campaign, version_id) if campaign.current_version_id != version.id: raise QueueingError( "Historical campaign versions cannot start a new execution. Use the explicit retry or reconciliation actions for an existing execution." ) return version def _celery_enabled() -> bool: return bool(core_settings.celery_enabled) def _should_enqueue_celery(enqueue_celery: bool) -> bool: return bool(enqueue_celery and _celery_enabled()) def _campaign_notification_body(campaign: Campaign, status: str) -> str: return { CampaignStatus.QUEUED.value: f"{campaign.name} has been queued for delivery.", CampaignStatus.SENDING.value: f"{campaign.name} started sending.", CampaignStatus.SENT.value: f"{campaign.name} finished sending.", CampaignStatus.PARTIALLY_COMPLETED.value: f"{campaign.name} finished with partial delivery results. Review the campaign report.", CampaignStatus.OUTCOME_UNKNOWN.value: f"{campaign.name} has unresolved delivery outcomes and needs operator review.", CampaignStatus.FAILED.value: f"{campaign.name} failed during delivery. Review the failed jobs.", CampaignStatus.CANCELLED.value: f"{campaign.name} was cancelled.", }.get(status, f"{campaign.name} changed status to {status}.") def _emit_campaign_status_notification( session: Session, *, campaign: Campaign, status: str, previous_status: str | None = None, version_id: str | None = None, message: str | None = None, ) -> None: provider = notification_dispatch_provider(get_registry()) if provider is None: return event = CAMPAIGN_STATUS_NOTIFICATION_EVENTS.get(status) if event is None: return event_kind, title, priority = event try: provider.enqueue_notification( session, NotificationDispatchRequest( tenant_id=campaign.tenant_id, source_module="campaigns", source_resource_type="campaign", source_resource_id=campaign.id, event_kind=event_kind, channel="inbox", subject=f"{title}: {campaign.name}", body_text=message or _campaign_notification_body(campaign, status), action_url=f"/campaigns/{campaign.id}", priority=priority, payload={ "campaign_id": campaign.id, "version_id": version_id or campaign.current_version_id, "status": status, "previous_status": previous_status, }, metadata={"campaign_id": campaign.id, "version_id": version_id or campaign.current_version_id}, ), enqueue_delivery=False, ) except Exception: return def _celery_enqueue_send_job(job_id: str) -> None: from govoplan_core.celery_app import celery celery.send_task("govoplan.campaigns.send_email", args=[job_id], queue="send_email") def _celery_enqueue_append_sent_job(job_id: str) -> None: from govoplan_core.celery_app import celery celery.send_task("govoplan.campaigns.append_sent", args=[job_id], queue="append_sent") def _ensure_campaign_execution_snapshot( session: Session, version: CampaignVersion, ) -> None: try: ensure_execution_snapshot(session, version) except ExecutionSnapshotError as exc: raise QueueingError(str(exc)) from exc def _queue_validation_statuses(*, include_warnings: bool) -> set[str]: allowed_validation = {JobValidationStatus.READY.value} if include_warnings: allowed_validation.add(JobValidationStatus.WARNING.value) return allowed_validation def _campaign_jobs_for_queue( session: Session, *, tenant_id: str, version_id: str, ) -> list[CampaignJob]: jobs = ( session.query(CampaignJob) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_version_id == version_id, ) .order_by(CampaignJob.entry_index.asc()) .all() ) if not jobs: raise QueueingError( "Campaign version has no jobs. Build messages before queueing." ) return jobs def _initial_queue_disposition( job: CampaignJob, *, allowed_validation: set[str], reviewed_needs_review_keys: set[str], ) -> str: if job.send_status in INITIAL_QUEUE_SKIPPED_SEND_STATUSES: # Initial queueing never doubles as retry/reconciliation. Those states # require the dedicated explicit actions below. return "skipped" if job.queue_status in INITIAL_QUEUE_SKIPPED_QUEUE_STATUSES: return "skipped" validation_allowed = job.validation_status in allowed_validation or ( job.validation_status == JobValidationStatus.NEEDS_REVIEW.value and _job_review_key(job) in reviewed_needs_review_keys ) if job.build_status != JobBuildStatus.BUILT.value or not validation_allowed: return "blocked" if not job.eml_local_path and not job.eml_storage_key: job.last_error = ( "Job has no generated EML path/storage key. Rebuild with write_eml " "enabled before queueing." ) return "blocked" return "queued" def _mark_campaign_job_queued(session: Session, job: CampaignJob) -> None: job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.QUEUED.value job.queued_at = _utcnow() job.claimed_at = None job.claim_token = None job.smtp_started_at = None job.outcome_unknown_at = None job.last_error = None session.add(job) def _select_campaign_jobs_for_queue( session: Session, *, jobs: list[CampaignJob], allowed_validation: set[str], reviewed_needs_review_keys: set[str], dry_run: bool, ) -> tuple[list[CampaignJob], int, int]: queued: list[CampaignJob] = [] skipped_count = 0 blocked_count = 0 for job in jobs: disposition = _initial_queue_disposition( job, allowed_validation=allowed_validation, reviewed_needs_review_keys=reviewed_needs_review_keys, ) if disposition == "skipped": skipped_count += 1 continue if disposition == "blocked": blocked_count += 1 continue queued.append(job) if not dry_run: _mark_campaign_job_queued(session, job) return queued, skipped_count, blocked_count def _persist_campaign_queue( session: Session, *, campaign: Campaign, version: CampaignVersion, queued: list[CampaignJob], ) -> None: if queued: previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value if version.locked_at is None: version.locked_at = _utcnow() session.add(version) if previous_status != campaign.status: _emit_campaign_status_notification( session, campaign=campaign, status=campaign.status, previous_status=previous_status, version_id=version.id, ) session.add(campaign) session.commit() def _enqueue_campaign_jobs(queued: list[CampaignJob], *, enabled: bool) -> int: if not enabled: return 0 for job in queued: _celery_enqueue_send_job(job.id) return len(queued) def queue_campaign_jobs( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, enqueue_celery: bool = True, include_warnings: bool = True, dry_run: bool = False, ) -> QueueCampaignResult: """Move queueable DB jobs to QUEUED and optionally enqueue Celery tasks.""" campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) version = _get_current_version(session, campaign, version_id=version_id) _ensure_version_validated_and_locked(version) _ensure_campaign_execution_snapshot(session, version) allowed_validation = _queue_validation_statuses( include_warnings=include_warnings ) jobs = _campaign_jobs_for_queue( session, tenant_id=tenant_id, version_id=version.id, ) reviewed_needs_review_keys = _reviewed_needs_review_keys(version) queued, skipped_count, blocked_count = _select_campaign_jobs_for_queue( session, jobs=jobs, allowed_validation=allowed_validation, reviewed_needs_review_keys=reviewed_needs_review_keys, dry_run=dry_run, ) if not dry_run: _persist_campaign_queue( session, campaign=campaign, version=version, queued=queued, ) enqueued_count = _enqueue_campaign_jobs( queued, enabled=_should_enqueue_celery(enqueue_celery) and not dry_run, ) return QueueCampaignResult( campaign_id=campaign.id, version_id=version.id, queued_count=len(queued), skipped_count=skipped_count, blocked_count=blocked_count, enqueued_count=enqueued_count, dry_run=dry_run, ) def send_campaign_now( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, include_warnings: bool = True, dry_run: bool = False, use_rate_limit: bool = True, enqueue_imap_task: bool = False, ) -> SendCampaignNowResult: """Queue and send a small campaign synchronously. This is intended for WebUI test/small-campaign flows. Large campaigns can still use queue_campaign_jobs with Celery workers. """ campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) version = _get_current_version(session, campaign, version_id=version_id) queue_result = queue_campaign_jobs( session, tenant_id=tenant_id, campaign_id=campaign.id, version_id=version.id, include_warnings=include_warnings, enqueue_celery=False, dry_run=dry_run, ) if dry_run: return SendCampaignNowResult( campaign_id=campaign.id, version_id=version.id, attempted_count=0, sent_count=0, failed_count=0, outcome_unknown_count=0, skipped_count=queue_result.skipped_count + queue_result.blocked_count, dry_run=True, results=[queue_result.as_dict()], ) jobs = ( session.query(CampaignJob) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_version_id == version.id, CampaignJob.queue_status == JobQueueStatus.QUEUED.value, CampaignJob.send_status.in_([JobSendStatus.QUEUED.value, JobSendStatus.FAILED_TEMPORARY.value]), ) .order_by(CampaignJob.entry_index.asc()) .all() ) results: list[dict[str, Any]] = [] sent_count = 0 failed_count = 0 outcome_unknown_count = 0 skipped_after_queue = 0 for job in jobs: try: result = send_campaign_job( session, job_id=job.id, dry_run=False, use_rate_limit=use_rate_limit, enqueue_imap_task=enqueue_imap_task, ) result_dict = result.as_dict() results.append(result_dict) if result.status in {JobSendStatus.SMTP_ACCEPTED.value, "already_accepted"}: sent_count += 1 elif result.status == JobSendStatus.OUTCOME_UNKNOWN.value: outcome_unknown_count += 1 else: skipped_after_queue += 1 except Exception as exc: # keep sending other jobs and return per-job details failed_count += 1 results.append({"job_id": job.id, "status": "failed", "message": str(exc)}) return SendCampaignNowResult( campaign_id=campaign.id, version_id=version.id, attempted_count=len(jobs), sent_count=sent_count, failed_count=failed_count, outcome_unknown_count=outcome_unknown_count, skipped_count=queue_result.skipped_count + queue_result.blocked_count + skipped_after_queue, dry_run=False, results=results, ) def enqueue_existing_queued_jobs(session: Session, *, tenant_id: str, campaign_id: str) -> int: if not _celery_enabled(): return 0 campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) jobs = ( session.query(CampaignJob) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id, CampaignJob.queue_status == JobQueueStatus.QUEUED.value, CampaignJob.send_status.in_([JobSendStatus.QUEUED.value, JobSendStatus.FAILED_TEMPORARY.value]), ) .order_by(CampaignJob.entry_index.asc()) .all() ) for job in jobs: _celery_enqueue_send_job(job.id) return len(jobs) def pause_campaign_jobs(session: Session, *, tenant_id: str, campaign_id: str) -> dict[str, Any]: campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) changed = ( session.query(CampaignJob) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id, CampaignJob.queue_status == JobQueueStatus.QUEUED.value, ) .update({CampaignJob.queue_status: JobQueueStatus.PAUSED.value}, synchronize_session=False) ) if changed: campaign.status = CampaignStatus.READY_TO_QUEUE.value session.add(campaign) session.commit() return {"campaign_id": campaign.id, "paused_count": int(changed)} def resume_campaign_jobs(session: Session, *, tenant_id: str, campaign_id: str, enqueue_celery: bool = True) -> dict[str, Any]: campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) jobs = ( session.query(CampaignJob) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id, CampaignJob.queue_status == JobQueueStatus.PAUSED.value, ) .order_by(CampaignJob.entry_index.asc()) .all() ) for job in jobs: job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.QUEUED.value session.add(job) if jobs: previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value session.add(campaign) if previous_status != campaign.status: _emit_campaign_status_notification( session, campaign=campaign, status=campaign.status, previous_status=previous_status, ) session.commit() enqueued_count = 0 if _should_enqueue_celery(enqueue_celery): for job in jobs: _celery_enqueue_send_job(job.id) enqueued_count += 1 return {"campaign_id": campaign.id, "resumed_count": len(jobs), "enqueued_count": enqueued_count} def cancel_campaign_jobs(session: Session, *, tenant_id: str, campaign_id: str) -> dict[str, Any]: campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) jobs = ( session.query(CampaignJob) .filter(CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id) .all() ) cancelled_count = 0 protected_count = 0 version_ids: set[str] = set() for job in jobs: version_ids.add(job.campaign_version_id) if job.send_status in SMTP_ACCEPTED_STATUSES | { JobSendStatus.OUTCOME_UNKNOWN.value, JobSendStatus.CLAIMED.value, JobSendStatus.SENDING.value, }: protected_count += 1 continue if job.send_status != JobSendStatus.CANCELLED.value: job.queue_status = JobQueueStatus.CANCELLED.value job.send_status = JobSendStatus.CANCELLED.value job.claim_token = None session.add(job) cancelled_count += 1 for version_id in version_ids: _update_campaign_after_job(session, campaign.id, version_id) session.commit() return { "campaign_id": campaign.id, "cancelled_count": cancelled_count, "protected_count": protected_count, "campaign_status": campaign.status, } def _eligible_jobs_for_explicit_action( session: Session, *, tenant_id: str, campaign: Campaign, version: CampaignVersion, job_ids: list[str] | None, ) -> list[CampaignJob]: query = session.query(CampaignJob).filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id, CampaignJob.campaign_version_id == version.id, ) if job_ids: query = query.filter(CampaignJob.id.in_(job_ids)) return query.order_by(CampaignJob.entry_index.asc()).all() def queue_failed_jobs_for_retry( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, job_ids: list[str] | None = None, include_permanent: bool = False, force_max_attempts: bool = False, enqueue_celery: bool = True, dry_run: bool = False, ) -> dict[str, Any]: """Explicitly queue failed jobs; never includes uncertain or accepted jobs.""" campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) version = _get_version_for_campaign(session, campaign, version_id=version_id) _ensure_version_validated_and_locked(version) snapshot = ensure_execution_snapshot(session, version) allowed = {JobSendStatus.FAILED_TEMPORARY.value} if include_permanent: allowed.add(JobSendStatus.FAILED_PERMANENT.value) selected: list[CampaignJob] = [] skipped: list[dict[str, str]] = [] for job in _eligible_jobs_for_explicit_action( session, tenant_id=tenant_id, campaign=campaign, version=version, job_ids=job_ids, ): if job.send_status not in allowed: skipped.append({"job_id": job.id, "reason": f"status {job.send_status} is not an explicit retry candidate"}) continue if not force_max_attempts and job.attempt_count >= snapshot.delivery.retry.max_attempts: skipped.append({"job_id": job.id, "reason": "configured maximum attempts reached"}) continue selected.append(job) if not dry_run: job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.QUEUED.value job.queued_at = _utcnow() job.claimed_at = None job.claim_token = None job.smtp_started_at = None job.outcome_unknown_at = None session.add(job) if not dry_run: if selected: campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value session.add(campaign) session.add(version) session.commit() enqueued = 0 if _should_enqueue_celery(enqueue_celery) and not dry_run: for job in selected: _celery_enqueue_send_job(job.id) enqueued += 1 return { "campaign_id": campaign.id, "version_id": version.id, "action": "retry_failed", "selected_count": len(selected), "enqueued_count": enqueued, "skipped": skipped, "dry_run": dry_run, } def queue_unattempted_jobs( session: Session, *, tenant_id: str, campaign_id: str, version_id: str | None = None, job_ids: list[str] | None = None, enqueue_celery: bool = True, dry_run: bool = False, ) -> dict[str, Any]: """Explicitly queue built jobs that have never started an SMTP attempt.""" campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) version = _get_version_for_campaign(session, campaign, version_id=version_id) _ensure_version_validated_and_locked(version) ensure_execution_snapshot(session, version) selected: list[CampaignJob] = [] skipped: list[dict[str, str]] = [] for job in _eligible_jobs_for_explicit_action( session, tenant_id=tenant_id, campaign=campaign, version=version, job_ids=job_ids, ): eligible = ( job.attempt_count == 0 and job.send_status in {JobSendStatus.NOT_QUEUED.value, JobSendStatus.CANCELLED.value} and job.build_status == JobBuildStatus.BUILT.value and job.validation_status in QUEUEABLE_VALIDATION_STATUSES ) if not eligible: skipped.append({"job_id": job.id, "reason": "job is not an unattempted queueable message"}) continue selected.append(job) if not dry_run: job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.QUEUED.value job.queued_at = _utcnow() job.claimed_at = None job.claim_token = None job.smtp_started_at = None job.outcome_unknown_at = None job.last_error = None session.add(job) if not dry_run: if selected: campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value session.add(campaign) session.add(version) session.commit() enqueued = 0 if _should_enqueue_celery(enqueue_celery) and not dry_run: for job in selected: _celery_enqueue_send_job(job.id) enqueued += 1 return { "campaign_id": campaign.id, "version_id": version.id, "action": "send_unattempted", "selected_count": len(selected), "enqueued_count": enqueued, "skipped": skipped, "dry_run": dry_run, } def send_single_campaign_job( session: Session, *, tenant_id: str, campaign_id: str, job_id: str, include_warnings: bool = True, dry_run: bool = False, use_rate_limit: bool = True, enqueue_imap_task: bool = False, ) -> dict[str, Any]: """Explicitly queue and send one built recipient message through the audit send path.""" campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) job = session.get(CampaignJob, job_id) if not job or job.tenant_id != tenant_id or job.campaign_id != campaign.id: raise QueueingError("Campaign job not found or not accessible") version = _get_current_version(session, campaign, version_id=job.campaign_version_id) _ensure_version_validated_and_locked(version) ensure_execution_snapshot(session, version) queue_action = _prepare_single_job_for_send( session, version=version, job=job, include_warnings=include_warnings, dry_run=dry_run, ) if dry_run: context = _send_job_delivery_context(session, job) return { "campaign_id": campaign.id, "version_id": version.id, "job_id": job.id, "action": "single_send", "queue_action": queue_action, "dry_run": True, "result": SendJobResult( job_id=job.id, status="dry_run", attempt_number=job.attempt_count, dry_run=True, message=f"Would send to {len(context.envelope_recipients)} recipient(s) from {context.envelope_from}", ).as_dict(), } if queue_action == "queued": previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value session.add(campaign) session.add(version) session.commit() if previous_status != campaign.status: _emit_campaign_status_notification( session, campaign=campaign, status=campaign.status, previous_status=previous_status, version_id=version.id, ) result = send_campaign_job( session, job_id=job.id, dry_run=False, use_rate_limit=use_rate_limit, enqueue_imap_task=enqueue_imap_task, ) return { "campaign_id": campaign.id, "version_id": version.id, "job_id": job.id, "action": "single_send", "queue_action": queue_action, "dry_run": False, "result": result.as_dict(), } def _prepare_single_job_for_send( session: Session, *, version: CampaignVersion, job: CampaignJob, include_warnings: bool, dry_run: bool, ) -> str: if job.queue_status == JobQueueStatus.QUEUED.value and job.send_status == JobSendStatus.QUEUED.value: return "already_queued" if job.send_status in SMTP_ACCEPTED_STATUSES: raise QueueingError("This message has already been accepted by SMTP and cannot be sent again from preview.") if job.send_status in {JobSendStatus.CLAIMED.value, JobSendStatus.SENDING.value, JobSendStatus.OUTCOME_UNKNOWN.value}: raise QueueingError(f"This message is in delivery state {job.send_status}; reconcile or wait before sending it again.") if job.send_status in {JobSendStatus.FAILED_TEMPORARY.value, JobSendStatus.FAILED_PERMANENT.value}: raise QueueingError("This message has failed before. Use the explicit retry action so the retry is visible in the delivery protocol.") if job.attempt_count > 0: raise QueueingError("This message already has SMTP attempts. Use retry or reconciliation instead of preview send.") if job.build_status != JobBuildStatus.BUILT.value: raise QueueingError("This message has not been built yet.") if not _single_job_validation_allowed(version, job, include_warnings=include_warnings): raise QueueingError(f"This message cannot be sent while validation status is {job.validation_status}.") if not job.eml_local_path and not job.eml_storage_key: raise QueueingError("This message has no generated EML evidence. Rebuild the campaign before sending.") if job.queue_status not in {JobQueueStatus.DRAFT.value, JobQueueStatus.CANCELLED.value}: raise QueueingError(f"This message cannot be sent from queue state {job.queue_status}.") if dry_run: return "would_queue" job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.QUEUED.value job.queued_at = _utcnow() job.claimed_at = None job.claim_token = None job.smtp_started_at = None job.outcome_unknown_at = None job.last_error = None session.add(job) return "queued" def _single_job_validation_allowed( version: CampaignVersion, job: CampaignJob, *, include_warnings: bool, ) -> bool: allowed = {JobValidationStatus.READY.value} if include_warnings: allowed.add(JobValidationStatus.WARNING.value) if job.validation_status in allowed: return True return ( job.validation_status == JobValidationStatus.NEEDS_REVIEW.value and _job_review_key(job) in _reviewed_needs_review_keys(version) ) def reconcile_job_outcome( session: Session, *, tenant_id: str, campaign_id: str, job_id: str, decision: str, note: str | None = None, ) -> dict[str, Any]: """Record an operator decision for an uncertain/incomplete worker state. ``smtp_accepted`` protects the message from retry and enables IMAP append. ``not_sent`` converts it into a failed-temporary candidate; retry remains a separate explicit action. """ campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) job = session.get(CampaignJob, job_id) if not job or job.tenant_id != tenant_id or job.campaign_id != campaign.id: raise QueueingError("Campaign job not found or not accessible") if job.send_status not in { JobSendStatus.OUTCOME_UNKNOWN.value, JobSendStatus.SENDING.value, JobSendStatus.CLAIMED.value, }: raise QueueingError(f"Job status {job.send_status} does not require reconciliation") version = _get_version_for_campaign(session, campaign, version_id=job.campaign_version_id) snapshot = ensure_execution_snapshot(session, version) now = _utcnow() attempt = _unfinished_attempt(session, job) if decision == "smtp_accepted": job.send_status = JobSendStatus.SMTP_ACCEPTED.value job.queue_status = JobQueueStatus.DRAFT.value job.sent_at = job.sent_at or now job.outcome_unknown_at = None job.claim_token = None job.last_error = note or "Operator confirmed SMTP acceptance after an uncertain outcome." job.imap_status = JobImapStatus.PENDING.value if snapshot.delivery.imap_append_sent.enabled else JobImapStatus.NOT_REQUESTED.value files_integration().mark_job_attachment_uses_sent(session, job) attempt_status = "reconciled_smtp_accepted" elif decision == "not_sent": job.send_status = JobSendStatus.FAILED_TEMPORARY.value job.queue_status = JobQueueStatus.DRAFT.value job.outcome_unknown_at = None job.claim_token = None job.last_error = note or "Operator confirmed that SMTP did not accept this message." attempt_status = "reconciled_not_sent" else: raise QueueingError("decision must be 'smtp_accepted' or 'not_sent'") if attempt: attempt.status = attempt_status attempt.finished_at = now attempt.error_type = "OperatorReconciliation" attempt.error_message = job.last_error session.add(attempt) session.add(job) _update_campaign_after_job(session, campaign.id, version.id) session.commit() return { "campaign_id": campaign.id, "version_id": version.id, "job_id": job.id, "decision": decision, "send_status": job.send_status, "imap_status": job.imap_status, "note": job.last_error, } def _verify_eml_evidence(job: CampaignJob, payload: bytes) -> None: if job.eml_size_bytes is not None and len(payload) != job.eml_size_bytes: raise SendJobError( f"Generated EML size mismatch for job {job.id}: expected {job.eml_size_bytes}, got {len(payload)}" ) if job.eml_sha256: actual = hashlib.sha256(payload).hexdigest() if actual != job.eml_sha256: raise SendJobError( f"Generated EML SHA-256 mismatch for job {job.id}; rebuild the campaign version before delivery" ) if job.message_id_header: parsed = BytesParser(policy=policy.default).parsebytes(payload) message_id = parsed.get("Message-ID") parsed_message_id = str(message_id) if message_id else None if parsed_message_id != job.message_id_header: raise SendJobError( f"Generated EML Message-ID mismatch for job {job.id}; rebuild the campaign version before delivery" ) def _load_eml_bytes_for_job(job: CampaignJob) -> bytes: if job.eml_local_path: path = Path(job.eml_local_path) if not path.exists(): raise SendJobError(f"Generated EML file does not exist: {path}") payload = path.read_bytes() _verify_eml_evidence(job, payload) return payload raise SendJobError("Only local EML paths are supported for sending in this implementation step") def _load_eml_for_job(job: CampaignJob): return BytesParser(policy=policy.default).parsebytes(_load_eml_bytes_for_job(job)) def _addresses_from_job(job: CampaignJob, field: str) -> list[str]: data = job.resolved_recipients or {} values = data.get(field) or [] return [item.get("email") for item in values if isinstance(item, dict) and item.get("email")] def _sender_from_job(job: CampaignJob, snapshot: ExecutionSnapshot) -> str: data = job.resolved_recipients or {} bounce_to = _addresses_from_job(job, "bounce_to") if bounce_to: return bounce_to[0] from_data = data.get("from") if isinstance(data, dict) else None if isinstance(from_data, dict) and from_data.get("email"): return from_data["email"] if snapshot.smtp.username: return snapshot.smtp.username raise SmtpConfigurationError("No envelope sender could be determined") def _from_header_from_job(job: CampaignJob, snapshot: ExecutionSnapshot) -> str | None: data = job.resolved_recipients or {} from_data = data.get("from") if isinstance(data, dict) else None if isinstance(from_data, dict) and from_data.get("email"): return str(from_data["email"]) return snapshot.smtp.username def _recipients_from_job(job: CampaignJob) -> list[str]: recipients: list[str] = [] for field in ["to", "cc", "bcc"]: recipients.extend(_addresses_from_job(job, field)) # Preserve order while de-duplicating. return list(dict.fromkeys(recipients)) def _claim_job_for_sending(session: Session, job: CampaignJob) -> str | None: """Atomically claim a queued job and return the claim token. A duplicate task can observe CLAIMED/SENDING but cannot acquire a second claim. CLAIMED is deliberately not timed out automatically: an operator can safely release it after checking worker state, while an automatic timeout could race a slow worker. """ claim_token = str(uuid4()) changed = ( session.query(CampaignJob) .filter( CampaignJob.id == job.id, CampaignJob.queue_status == JobQueueStatus.QUEUED.value, CampaignJob.send_status == JobSendStatus.QUEUED.value, ) .update( { CampaignJob.queue_status: JobQueueStatus.SENDING.value, CampaignJob.send_status: JobSendStatus.CLAIMED.value, CampaignJob.claimed_at: _utcnow(), CampaignJob.claim_token: claim_token, CampaignJob.last_error: None, }, synchronize_session=False, ) ) session.commit() session.expire_all() return claim_token if changed == 1 else None def _record_attempt_start(session: Session, job: CampaignJob, claim_token: str) -> SendAttempt: now = _utcnow() attempt = SendAttempt( job_id=job.id, attempt_number=job.attempt_count + 1, status="smtp_in_progress", claim_token=claim_token, started_at=now, ) job.attempt_count += 1 job.queue_status = JobQueueStatus.SENDING.value job.send_status = JobSendStatus.SENDING.value job.smtp_started_at = now job.outcome_unknown_at = None job.last_error = None session.add(attempt) session.add(job) session.commit() return attempt def _unfinished_attempt(session: Session, job: CampaignJob) -> SendAttempt | None: return ( session.query(SendAttempt) .filter(SendAttempt.job_id == job.id, SendAttempt.finished_at.is_(None)) .order_by(SendAttempt.attempt_number.desc()) .first() ) def mark_job_outcome_unknown(session: Session, job: CampaignJob, *, reason: str) -> SendJobResult: """Freeze an uncertain SMTP result and prohibit automatic redelivery.""" now = _utcnow() attempt = _unfinished_attempt(session, job) if attempt: attempt.status = "outcome_unknown" attempt.finished_at = now attempt.error_type = "OutcomeUnknown" attempt.error_message = reason session.add(attempt) job.queue_status = JobQueueStatus.DRAFT.value job.send_status = JobSendStatus.OUTCOME_UNKNOWN.value job.outcome_unknown_at = now job.last_error = reason session.add(job) _update_campaign_after_job(session, job.campaign_id, job.campaign_version_id) session.commit() return SendJobResult( job_id=job.id, status=JobSendStatus.OUTCOME_UNKNOWN.value, attempt_number=job.attempt_count, message=reason, ) def _update_campaign_after_job(session: Session, campaign_id: str, version_id: str | None = None) -> None: session.flush() campaign = session.get(Campaign, campaign_id) if not campaign: return previous_status = campaign.status version = session.get(CampaignVersion, version_id) if version_id else None counts = _campaign_delivery_counts(session, campaign_id=campaign_id, version_id=version_id) _apply_campaign_delivery_status(campaign, version, counts) if version and (counts.accepted or counts.unknown): if version.locked_at is None: version.locked_at = _utcnow() session.add(version) session.add(campaign) if campaign.status != previous_status: _emit_campaign_status_notification( session, campaign=campaign, status=campaign.status, previous_status=previous_status, version_id=version_id, ) def _campaign_delivery_counts(session: Session, *, campaign_id: str, version_id: str | None) -> _CampaignDeliveryCounts: base_filters = _campaign_delivery_base_filters(campaign_id=campaign_id, version_id=version_id) counts = { status: session.query(CampaignJob).filter(*base_filters, CampaignJob.send_status == status).count() for status in {item.value for item in JobSendStatus} } return _CampaignDeliveryCounts( accepted=sum(counts.get(status, 0) for status in SMTP_ACCEPTED_STATUSES), unknown=counts.get(JobSendStatus.OUTCOME_UNKNOWN.value, 0), failed=counts.get(JobSendStatus.FAILED_TEMPORARY.value, 0) + counts.get(JobSendStatus.FAILED_PERMANENT.value, 0), active=( counts.get(JobSendStatus.QUEUED.value, 0) + counts.get(JobSendStatus.CLAIMED.value, 0) + counts.get(JobSendStatus.SENDING.value, 0) ), cancelled=counts.get(JobSendStatus.CANCELLED.value, 0), not_started=_queueable_not_started_job_count(session, base_filters), ) def _campaign_delivery_base_filters(*, campaign_id: str, version_id: str | None) -> list[object]: base_filters: list[object] = [CampaignJob.campaign_id == campaign_id] if version_id: base_filters.append(CampaignJob.campaign_version_id == version_id) return base_filters def _queueable_not_started_job_count(session: Session, base_filters: list[object]) -> int: return ( session.query(CampaignJob) .filter( *base_filters, CampaignJob.send_status == JobSendStatus.NOT_QUEUED.value, CampaignJob.build_status == JobBuildStatus.BUILT.value, CampaignJob.validation_status.in_(list(QUEUEABLE_VALIDATION_STATUSES)), ) .count() ) def _apply_campaign_delivery_status( campaign: Campaign, version: CampaignVersion | None, counts: _CampaignDeliveryCounts, ) -> None: status_pair = _campaign_delivery_status_pair(counts) if status_pair is None: return campaign_status, workflow_state = status_pair campaign.status = campaign_status if version: version.workflow_state = workflow_state def _campaign_delivery_status_pair(counts: _CampaignDeliveryCounts) -> tuple[str, str] | None: if counts.active: return CampaignStatus.SENDING.value, CampaignVersionWorkflowState.SENDING.value if counts.accepted and (counts.failed or counts.unknown or counts.cancelled or counts.not_started): return CampaignStatus.PARTIALLY_COMPLETED.value, CampaignVersionWorkflowState.PARTIALLY_COMPLETED.value if counts.unknown: return CampaignStatus.OUTCOME_UNKNOWN.value, CampaignVersionWorkflowState.OUTCOME_UNKNOWN.value if counts.failed: return CampaignStatus.FAILED.value, CampaignVersionWorkflowState.FAILED.value if counts.accepted: return CampaignStatus.SENT.value, CampaignVersionWorkflowState.COMPLETED.value if counts.cancelled: return CampaignStatus.CANCELLED.value, CampaignVersionWorkflowState.CANCELLED.value return None def send_campaign_job( session: Session, *, job_id: str, dry_run: bool = False, use_rate_limit: bool = True, enqueue_imap_task: bool = False, ) -> SendJobResult: job = session.get(CampaignJob, job_id) if not job: raise SendJobError(f"Job not found: {job_id}") preflight_result = _preflight_send_campaign_job(session, job, dry_run=dry_run) if preflight_result is not None: return preflight_result context = _send_job_delivery_context(session, job) if dry_run: return SendJobResult( job_id=job.id, status="dry_run", attempt_number=job.attempt_count, dry_run=True, message=f"Would send to {len(context.envelope_recipients)} recipient(s) from {context.envelope_from}", ) claimed = _claimed_campaign_job_for_delivery(session, job) if isinstance(claimed, SendJobResult): return claimed claimed_job, claim_token = claimed return _send_claimed_campaign_job( session, job=claimed_job, claim_token=claim_token, context=context, use_rate_limit=use_rate_limit, enqueue_imap_task=enqueue_imap_task, ) def _preflight_send_campaign_job( session: Session, job: CampaignJob, *, dry_run: bool, ) -> SendJobResult | None: if job.queue_status == JobQueueStatus.CANCELLED.value or job.send_status == JobSendStatus.CANCELLED.value: return SendJobResult(job_id=job.id, status="cancelled", attempt_number=job.attempt_count, dry_run=dry_run) if job.queue_status == JobQueueStatus.PAUSED.value: return SendJobResult(job_id=job.id, status="paused", attempt_number=job.attempt_count, dry_run=dry_run) if job.send_status in SMTP_ACCEPTED_STATUSES: return SendJobResult(job_id=job.id, status="already_accepted", attempt_number=job.attempt_count, dry_run=dry_run) if job.send_status == JobSendStatus.OUTCOME_UNKNOWN.value: return SendJobResult( job_id=job.id, status=JobSendStatus.OUTCOME_UNKNOWN.value, attempt_number=job.attempt_count, dry_run=dry_run, message="SMTP outcome is unresolved; reconcile it before any retry.", ) if job.send_status == JobSendStatus.SENDING.value: return mark_job_outcome_unknown( session, job, reason="A delivery task resumed while the previous SMTP attempt was still marked in progress. Automatic resend was stopped.", ) if job.send_status == JobSendStatus.CLAIMED.value: return SendJobResult( job_id=job.id, status="already_claimed", attempt_number=job.attempt_count, dry_run=dry_run, message="Another worker has claimed this job, or a stale pre-SMTP claim requires operator review.", ) if job.queue_status != JobQueueStatus.QUEUED.value or job.send_status != JobSendStatus.QUEUED.value: raise SendJobError(f"Job is not explicitly queued for delivery: queue={job.queue_status}, send={job.send_status}") return None def _send_job_delivery_context(session: Session, job: CampaignJob) -> _SendJobDeliveryContext: version = session.get(CampaignVersion, job.campaign_version_id) if not version: raise SendJobError("Campaign version not found") try: snapshot = ensure_execution_snapshot(session, version) except ExecutionSnapshotError as exc: raise SendJobError(str(exc)) from exc message_bytes = _load_eml_bytes_for_job(job) envelope_from = _sender_from_job(job, snapshot) envelope_recipients = _recipients_from_job(job) if not envelope_recipients: raise SmtpConfigurationError("No envelope recipients could be determined") return _SendJobDeliveryContext( version=version, snapshot=snapshot, message_bytes=message_bytes, envelope_from=envelope_from, envelope_recipients=envelope_recipients, ) def _claimed_campaign_job_for_delivery( session: Session, job: CampaignJob, ) -> tuple[CampaignJob, str] | SendJobResult: claim_token = _claim_job_for_sending(session, job) if claim_token is None: return _not_claimed_send_job_result(session, job) job = session.get(CampaignJob, job.id) if job is None: raise SendJobError("Claimed campaign job disappeared before send.") return job, claim_token def _not_claimed_send_job_result(session: Session, job: CampaignJob) -> SendJobResult: current = session.get(CampaignJob, job.id) if not current: raise SendJobError(f"Job disappeared while claiming: {job.id}") if current.send_status == JobSendStatus.SENDING.value: return mark_job_outcome_unknown( session, current, reason="A duplicate/redelivered task found an unfinished SMTP attempt. Automatic resend was stopped.", ) return SendJobResult( job_id=current.id, status="not_claimed", attempt_number=current.attempt_count, message=f"Job is no longer queueable: {current.send_status}", ) def _send_claimed_campaign_job( session: Session, *, job: CampaignJob, claim_token: str, context: _SendJobDeliveryContext, use_rate_limit: bool, enqueue_imap_task: bool, ) -> SendJobResult: mail_integration().wait_for_rate_limit( key=f"tenant:{job.tenant_id}:campaign:{job.campaign_id}", messages_per_minute=context.snapshot.delivery.rate_limit.messages_per_minute, enabled=use_rate_limit, ) attempt = _record_attempt_start(session, job, claim_token) try: smtp_config = runtime_smtp_config(session, context.version, context.snapshot) mail_integration().assert_mail_policy_allows_send( session, tenant_id=job.tenant_id, campaign_id=job.campaign_id, smtp=smtp_config, envelope_sender=context.envelope_from, from_header=_from_header_from_job(job, context.snapshot), recipients=context.envelope_recipients, ) result = mail_integration().send_email_bytes( context.message_bytes, smtp_config=smtp_config, envelope_from=context.envelope_from, envelope_recipients=context.envelope_recipients, ) return _record_smtp_send_success( session, job=job, attempt=attempt, snapshot=context.snapshot, result=result, enqueue_imap_task=enqueue_imap_task, ) except SmtpSendError as exc: outcome_unknown = _record_smtp_send_error(session, job=job, attempt=attempt, exc=exc) if outcome_unknown is not None: return outcome_unknown raise except (MailProfileError, SmtpConfigurationError, SendJobError, OSError) as exc: _record_permanent_send_error(session, job=job, attempt=attempt, exc=exc) raise def _record_smtp_send_success( session: Session, *, job: CampaignJob, attempt: SendAttempt, snapshot: ExecutionSnapshot, result: object, enqueue_imap_task: bool, ) -> SendJobResult: if result.accepted_count <= 0: raise SmtpSendError("SMTP did not accept any envelope recipients", temporary=False) refused_warning = _refused_recipient_warning(result) attempt.finished_at = _utcnow() attempt.status = "smtp_accepted_with_refusals" if refused_warning else JobSendStatus.SMTP_ACCEPTED.value attempt.smtp_response = json.dumps(asdict(result), default=str) job.queue_status = JobQueueStatus.DRAFT.value job.send_status = JobSendStatus.SMTP_ACCEPTED.value job.sent_at = _utcnow() job.claim_token = None job.outcome_unknown_at = None job.imap_status = JobImapStatus.PENDING.value if snapshot.delivery.imap_append_sent.enabled else JobImapStatus.NOT_REQUESTED.value job.last_error = refused_warning files_integration().mark_job_attachment_uses_sent(session, job) session.add(attempt) session.add(job) _update_campaign_after_job(session, job.campaign_id, job.campaign_version_id) session.commit() if enqueue_imap_task and _celery_enabled() and job.imap_status == JobImapStatus.PENDING.value: _celery_enqueue_append_sent_job(job.id) return SendJobResult( job_id=job.id, status=JobSendStatus.SMTP_ACCEPTED.value, attempt_number=attempt.attempt_number, message=refused_warning, ) def _refused_recipient_warning(result: object) -> str | None: if not result.refused_recipients: return None return ( f"SMTP accepted {result.accepted_count}/{len(result.envelope_recipients)} envelope recipient(s); " f"refused recipients: {json.dumps(result.refused_recipients, default=str, sort_keys=True)}" ) def _record_smtp_send_error( session: Session, *, job: CampaignJob, attempt: SendAttempt, exc: SmtpSendError, ) -> SendJobResult | None: if getattr(exc, "outcome_unknown", False): attempt.status = JobSendStatus.OUTCOME_UNKNOWN.value attempt.finished_at = _utcnow() attempt.error_type = exc.__class__.__name__ attempt.error_message = str(exc) session.add(attempt) job.claim_token = None session.add(job) session.flush() return mark_job_outcome_unknown(session, job, reason=str(exc)) attempt.finished_at = _utcnow() attempt.error_type = exc.__class__.__name__ attempt.error_message = str(exc) retryable = bool(getattr(exc, "temporary", False)) job.last_error = str(exc) job.queue_status = JobQueueStatus.DRAFT.value job.send_status = JobSendStatus.FAILED_TEMPORARY.value if retryable else JobSendStatus.FAILED_PERMANENT.value job.claim_token = None attempt.status = job.send_status session.add(attempt) session.add(job) _update_campaign_after_job(session, job.campaign_id, job.campaign_version_id) session.commit() return None def _record_permanent_send_error( session: Session, *, job: CampaignJob, attempt: SendAttempt, exc: Exception, ) -> None: attempt.finished_at = _utcnow() attempt.error_type = exc.__class__.__name__ attempt.error_message = str(exc) attempt.status = JobSendStatus.FAILED_PERMANENT.value job.last_error = str(exc) job.queue_status = JobQueueStatus.DRAFT.value job.send_status = JobSendStatus.FAILED_PERMANENT.value job.claim_token = None session.add(attempt) session.add(job) _update_campaign_after_job(session, job.campaign_id, job.campaign_version_id) session.commit() def _record_imap_attempt_start(session: Session, job: CampaignJob) -> ImapAppendAttempt: existing_count = session.query(ImapAppendAttempt).filter(ImapAppendAttempt.job_id == job.id).count() attempt = ImapAppendAttempt( job_id=job.id, attempt_number=existing_count + 1, status="running", ) job.imap_status = JobImapStatus.PENDING.value job.last_error = None session.add(attempt) session.add(job) session.commit() return attempt def _effective_imap_folder(snapshot: ExecutionSnapshot) -> str: delivery_folder = snapshot.delivery.imap_append_sent.folder if delivery_folder and delivery_folder != "auto": return delivery_folder if snapshot.imap and snapshot.imap.sent_folder and snapshot.imap.sent_folder != "auto": return snapshot.imap.sent_folder return "auto" def append_sent_for_job(session: Session, *, job_id: str, dry_run: bool = False) -> AppendSentResult: """Append one successfully sent job's exact EML to the configured IMAP Sent folder.""" job = session.get(CampaignJob, job_id) if not job: raise SendJobError(f"Job not found: {job_id}") if job.send_status not in SMTP_ACCEPTED_STATUSES: return AppendSentResult(job_id=job_id, status="not_sent", attempt_number=0, dry_run=dry_run, message="SMTP has not accepted this job") if job.imap_status == JobImapStatus.NOT_REQUESTED.value: return AppendSentResult(job_id=job_id, status="not_requested", attempt_number=0, dry_run=dry_run) if job.imap_status == JobImapStatus.APPENDED.value: attempts = session.query(ImapAppendAttempt).filter(ImapAppendAttempt.job_id == job.id).count() return AppendSentResult(job_id=job_id, status="already_appended", attempt_number=attempts, dry_run=dry_run) if job.imap_status == JobImapStatus.SKIPPED.value: attempts = session.query(ImapAppendAttempt).filter(ImapAppendAttempt.job_id == job.id).count() return AppendSentResult(job_id=job_id, status="skipped", attempt_number=attempts, dry_run=dry_run) version = session.get(CampaignVersion, job.campaign_version_id) if not version: raise SendJobError("Campaign version not found") try: snapshot = ensure_execution_snapshot(session, version) except ExecutionSnapshotError as exc: raise SendJobError(str(exc)) from exc if not snapshot.delivery.imap_append_sent.enabled: job.imap_status = JobImapStatus.NOT_REQUESTED.value session.add(job) session.commit() return AppendSentResult(job_id=job.id, status="not_requested", attempt_number=0, dry_run=dry_run) imap_config = runtime_imap_config(session, version, snapshot) if not imap_config: job.imap_status = JobImapStatus.SKIPPED.value job.last_error = "IMAP append requested, but the execution snapshot has no IMAP configuration" session.add(job) session.commit() return AppendSentResult(job_id=job.id, status="skipped", attempt_number=0, dry_run=dry_run, message=job.last_error) message_bytes = _load_eml_bytes_for_job(job) folder = _effective_imap_folder(snapshot) if dry_run: attempts = session.query(ImapAppendAttempt).filter(ImapAppendAttempt.job_id == job.id).count() return AppendSentResult( job_id=job.id, status="dry_run", attempt_number=attempts, dry_run=True, folder=folder, message=f"Would append {len(message_bytes)} bytes to IMAP folder {folder!r}", ) attempt = _record_imap_attempt_start(session, job) try: mail_integration().assert_mail_policy_allows_send( session, tenant_id=job.tenant_id, campaign_id=job.campaign_id, smtp=snapshot.smtp, imap=imap_config, ) result = mail_integration().append_message_to_sent( message_bytes, imap_config=imap_config, folder=None if folder == "auto" else folder, ) attempt.status = "appended" attempt.folder = result.folder job.imap_status = JobImapStatus.APPENDED.value job.last_error = None files_integration().mark_job_attachment_uses_sent(session, job) session.add(attempt) session.add(job) session.commit() return AppendSentResult(job_id=job.id, status="appended", attempt_number=attempt.attempt_number, folder=result.folder) except (MailProfileError, ImapConfigurationError, ImapAppendError, SendJobError, OSError) as exc: attempt.status = "failed" attempt.folder = None if folder == "auto" else folder attempt.error_message = str(exc) job.imap_status = JobImapStatus.FAILED.value job.last_error = str(exc) session.add(attempt) session.add(job) session.commit() raise def enqueue_pending_imap_appends( session: Session, *, tenant_id: str, campaign_id: str, enqueue_celery: bool = True, run_inline: bool = False, dry_run: bool = False, ) -> dict[str, Any]: campaign = _get_campaign_for_tenant(session, campaign_id=campaign_id, tenant_id=tenant_id) jobs = ( session.query(CampaignJob) .filter( CampaignJob.tenant_id == tenant_id, CampaignJob.campaign_id == campaign.id, CampaignJob.send_status.in_(list(SMTP_ACCEPTED_STATUSES)), CampaignJob.imap_status.in_([JobImapStatus.PENDING.value, JobImapStatus.FAILED.value]), ) .order_by(CampaignJob.entry_index.asc()) .all() ) should_enqueue = _should_enqueue_celery(enqueue_celery) and not dry_run and not run_inline results: list[dict[str, Any]] = [] appended_count = 0 failed_count = 0 skipped_count = 0 if run_inline or dry_run: for job in jobs: try: result = append_sent_for_job(session, job_id=job.id, dry_run=dry_run) payload = result.as_dict() results.append(payload) if result.status == JobImapStatus.APPENDED.value: appended_count += 1 elif result.status in {"skipped", "not_requested", "not_sent", "already_appended", "dry_run"}: skipped_count += 1 except Exception as exc: # keep processing later jobs and expose per-job details failed_count += 1 results.append({"job_id": job.id, "status": "failed", "message": str(exc)}) elif should_enqueue: for job in jobs: _celery_enqueue_append_sent_job(job.id) return { "campaign_id": campaign.id, "pending_count": len(jobs), "enqueued_count": len(jobs) if should_enqueue else 0, "processed_count": len(results) if run_inline and not dry_run else 0, "appended_count": appended_count, "failed_count": failed_count, "skipped_count": skipped_count, "dry_run": dry_run, "run_inline": run_inline, "results": results, } def next_retry_delay(snapshot: ExecutionSnapshot, attempt_count: int) -> int: delays = snapshot.delivery.retry.backoff_seconds or [60] index = max(0, min(attempt_count - 1, len(delays) - 1)) return int(delays[index])