from __future__ import annotations 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 sqlalchemy.orm import Session from app.db.models import ( Campaign, CampaignJob, CampaignStatus, CampaignVersion, JobBuildStatus, JobImapStatus, JobQueueStatus, JobSendStatus, JobValidationStatus, ImapAppendAttempt, SendAttempt, ) from app.mailer.campaign.loader import load_campaign_config from app.mailer.campaign.models import CampaignConfig from app.mailer.persistence.campaigns import _write_campaign_snapshot from app.mailer.sending.rate_limit import wait_for_rate_limit from app.mailer.sending.smtp import SmtpConfigurationError, SmtpSendError, send_email_message from app.mailer.sending.imap import ImapAppendError, ImapConfigurationError, append_message_to_sent 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 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, } QUEUEABLE_VALIDATION_STATUSES = { JobValidationStatus.READY.value, JobValidationStatus.WARNING.value, } 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_current_version(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 current 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 _load_version_campaign_config(version: CampaignVersion) -> tuple[Path, CampaignConfig]: snapshot_path = _write_campaign_snapshot(version) return snapshot_path, load_campaign_config(snapshot_path) def _celery_enqueue_send_job(job_id: str) -> None: from app.celery_app import celery celery.send_task("multimailer.send_email", args=[job_id], queue="send_email") def _celery_enqueue_append_sent_job(job_id: str) -> None: from app.celery_app import celery celery.send_task("multimailer.append_sent", args=[job_id], queue="append_sent") 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) allowed_validation = {JobValidationStatus.READY.value} if include_warnings: allowed_validation.add(JobValidationStatus.WARNING.value) 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.") queued: list[CampaignJob] = [] skipped_count = 0 blocked_count = 0 for job in jobs: if job.queue_status in {JobQueueStatus.CANCELLED.value, JobQueueStatus.SENDING.value} or job.send_status == JobSendStatus.SENT.value: skipped_count += 1 continue if job.build_status != JobBuildStatus.BUILT.value or job.validation_status not in allowed_validation: blocked_count += 1 continue 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." blocked_count += 1 continue queued.append(job) if not dry_run: job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.QUEUED.value job.queued_at = _utcnow() job.last_error = None session.add(job) if not dry_run: if queued: campaign.status = CampaignStatus.QUEUED.value session.add(campaign) session.commit() enqueued_count = 0 if enqueue_celery and not dry_run: for job in queued: _celery_enqueue_send_job(job.id) enqueued_count += 1 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 enqueue_existing_queued_jobs(session: Session, *, tenant_id: str, campaign_id: str) -> int: 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: campaign.status = CampaignStatus.QUEUED.value session.add(campaign) session.commit() enqueued_count = 0 if 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, CampaignJob.send_status.notin_([JobSendStatus.SENT.value]), ) .all() ) for job in jobs: if job.queue_status != JobQueueStatus.SENDING.value: job.queue_status = JobQueueStatus.CANCELLED.value job.send_status = JobSendStatus.CANCELLED.value session.add(job) campaign.status = CampaignStatus.CANCELLED.value session.add(campaign) session.commit() return {"campaign_id": campaign.id, "cancelled_count": len(jobs)} 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}") return path.read_bytes() 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, config: CampaignConfig) -> 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 config.server.smtp and config.server.smtp.username: return config.server.smtp.username raise SmtpConfigurationError("No envelope sender could be determined") 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 _record_attempt_start(session: Session, job: CampaignJob) -> SendAttempt: attempt = SendAttempt( job_id=job.id, attempt_number=job.attempt_count + 1, started_at=_utcnow(), ) job.attempt_count += 1 job.queue_status = JobQueueStatus.SENDING.value job.send_status = JobSendStatus.SENDING.value job.last_error = None session.add(attempt) session.add(job) session.commit() return attempt def _update_campaign_after_job(session: Session, campaign_id: str) -> None: session.flush() campaign = session.get(Campaign, campaign_id) if not campaign: return remaining = ( session.query(CampaignJob) .filter( CampaignJob.campaign_id == campaign_id, CampaignJob.queue_status.in_([JobQueueStatus.QUEUED.value, JobQueueStatus.SENDING.value, JobQueueStatus.PAUSED.value]), ) .count() ) failed = ( session.query(CampaignJob) .filter( CampaignJob.campaign_id == campaign_id, CampaignJob.send_status.in_([JobSendStatus.FAILED_TEMPORARY.value, JobSendStatus.FAILED_PERMANENT.value]), ) .count() ) sent = session.query(CampaignJob).filter(CampaignJob.campaign_id == campaign_id, CampaignJob.send_status == JobSendStatus.SENT.value).count() if remaining: campaign.status = CampaignStatus.QUEUED.value elif failed: campaign.status = CampaignStatus.FAILED.value if not sent else CampaignStatus.NEEDS_REVIEW.value elif sent: campaign.status = CampaignStatus.SENT.value session.add(campaign) 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}") 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 == JobSendStatus.SENT.value: return SendJobResult(job_id=job_id, status="already_sent", attempt_number=job.attempt_count, dry_run=dry_run) if job.queue_status not in {JobQueueStatus.QUEUED.value, JobQueueStatus.SENDING.value}: raise SendJobError(f"Job is not queued: {job.queue_status}") version = session.get(CampaignVersion, job.campaign_version_id) if not version: raise SendJobError("Campaign version not found") _, config = _load_version_campaign_config(version) if not config.server.smtp: raise SmtpConfigurationError("Campaign has no SMTP configuration") message = _load_eml_for_job(job) envelope_from = _sender_from_job(job, config) envelope_recipients = _recipients_from_job(job) if not envelope_recipients: raise SmtpConfigurationError("No envelope recipients could be determined") 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(envelope_recipients)} recipient(s) from {envelope_from}", ) attempt = _record_attempt_start(session, job) try: wait_for_rate_limit( key=f"tenant:{job.tenant_id}:campaign:{job.campaign_id}", messages_per_minute=config.delivery.rate_limit.messages_per_minute, enabled=use_rate_limit, ) result = send_email_message( message, smtp_config=config.server.smtp, envelope_from=envelope_from, envelope_recipients=envelope_recipients, ) attempt.finished_at = _utcnow() attempt.smtp_response = json.dumps(asdict(result), default=str) job.queue_status = JobQueueStatus.DRAFT.value job.send_status = JobSendStatus.SENT.value job.sent_at = _utcnow() if config.delivery.imap_append_sent.enabled: job.imap_status = JobImapStatus.PENDING.value else: job.imap_status = JobImapStatus.NOT_REQUESTED.value job.last_error = None session.add(attempt) session.add(job) _update_campaign_after_job(session, job.campaign_id) session.commit() if enqueue_imap_task and job.imap_status == JobImapStatus.PENDING.value: _celery_enqueue_append_sent_job(job.id) return SendJobResult(job_id=job.id, status="sent", attempt_number=attempt.attempt_number) except (SmtpConfigurationError, SmtpSendError, SendJobError, OSError) as exc: attempt.finished_at = _utcnow() attempt.error_type = exc.__class__.__name__ attempt.error_message = str(exc) job.last_error = str(exc) retry = getattr(exc, "temporary", None) is True max_attempts = config.delivery.retry.max_attempts if retry and job.attempt_count < max_attempts: job.queue_status = JobQueueStatus.QUEUED.value job.send_status = JobSendStatus.FAILED_TEMPORARY.value else: job.queue_status = JobQueueStatus.DRAFT.value job.send_status = JobSendStatus.FAILED_PERMANENT.value session.add(attempt) session.add(job) _update_campaign_after_job(session, job.campaign_id) session.commit() raise 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(config: CampaignConfig) -> str: delivery_folder = config.delivery.imap_append_sent.folder if delivery_folder and delivery_folder != "auto": return delivery_folder if config.server.imap and config.server.imap.sent_folder and config.server.imap.sent_folder != "auto": return config.server.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 != JobSendStatus.SENT.value: return AppendSentResult(job_id=job_id, status="not_sent", attempt_number=0, dry_run=dry_run, message="Job has not been sent") 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") _, config = _load_version_campaign_config(version) if not config.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) if not config.server.imap or not config.server.imap.enabled: job.imap_status = JobImapStatus.SKIPPED.value job.last_error = "IMAP append requested, but server.imap is missing or disabled" 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(config) 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: result = append_message_to_sent( message_bytes, imap_config=config.server.imap, folder=None if folder == "auto" else folder, ) attempt.status = "appended" attempt.folder = result.folder job.imap_status = JobImapStatus.APPENDED.value job.last_error = None 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 (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, 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 == JobSendStatus.SENT.value, CampaignJob.imap_status.in_([JobImapStatus.PENDING.value, JobImapStatus.FAILED.value]), ) .order_by(CampaignJob.entry_index.asc()) .all() ) if not dry_run and enqueue_celery: for job in jobs: _celery_enqueue_append_sent_job(job.id) return { "campaign_id": campaign.id, "pending_count": len(jobs), "enqueued_count": 0 if dry_run or not enqueue_celery else len(jobs), "dry_run": dry_run, } def next_retry_delay(config: CampaignConfig, attempt_count: int) -> int: delays = config.delivery.retry.backoff_seconds or [60] index = max(0, min(attempt_count - 1, len(delays) - 1)) return int(delays[index])