Refactor campaign job queue selection

src/govoplan_campaign/backend/sending/jobs.py

@@ -161,6 +161,19 @@ QUEUEABLE_VALIDATION_STATUSES = {
 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),
@@ -324,6 +337,147 @@ def _celery_enqueue_append_sent_job(job_id: str) -> None:
     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,
     *,
@@ -339,92 +493,34 @@ def queue_campaign_jobs(
     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)
-    try:
+    _ensure_campaign_execution_snapshot(session, version)
-        ensure_execution_snapshot(session, version)
+    allowed_validation = _queue_validation_statuses(
-    except ExecutionSnapshotError as exc:
+        include_warnings=include_warnings
-        raise QueueingError(str(exc)) from exc
+    )
-
+    jobs = _campaign_jobs_for_queue(
-    allowed_validation = {JobValidationStatus.READY.value}
+        session,
-    if include_warnings:
+        tenant_id=tenant_id,
-        allowed_validation.add(JobValidationStatus.WARNING.value)
+        version_id=version.id,
-
-    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] = []
     reviewed_needs_review_keys = _reviewed_needs_review_keys(version)
-    skipped_count = 0
+    queued, skipped_count, blocked_count = _select_campaign_jobs_for_queue(
-    blocked_count = 0
+        session,
-    for job in jobs:
+        jobs=jobs,
-        if job.send_status in SMTP_ACCEPTED_STATUSES | {
+        allowed_validation=allowed_validation,
-            JobSendStatus.CLAIMED.value,
+        reviewed_needs_review_keys=reviewed_needs_review_keys,
-            JobSendStatus.SENDING.value,
+        dry_run=dry_run,
-            JobSendStatus.OUTCOME_UNKNOWN.value,
+    )
-            JobSendStatus.FAILED_TEMPORARY.value,
-            JobSendStatus.FAILED_PERMANENT.value,
-            JobSendStatus.CANCELLED.value,
-        }:
-            # Initial queueing never doubles as retry/reconciliation. Those
-            # states require the dedicated explicit actions below.
-            skipped_count += 1
-            continue
-        if job.queue_status in {JobQueueStatus.CANCELLED.value, JobQueueStatus.SENDING.value, JobQueueStatus.PAUSED.value}:
-            skipped_count += 1
-            continue
-        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:
-            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.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 queued:
+        _persist_campaign_queue(
-            previous_status = campaign.status
+            session,
-            campaign.status = CampaignStatus.QUEUED.value
+            campaign=campaign,
-            version.workflow_state = CampaignVersionWorkflowState.QUEUED.value
+            version=version,
-            if version.locked_at is None:
+            queued=queued,
-                version.locked_at = _utcnow()
+        )
-            session.add(version)
+    enqueued_count = _enqueue_campaign_jobs(
-            if previous_status != campaign.status:
+        queued,
-                _emit_campaign_status_notification(
+        enabled=_should_enqueue_celery(enqueue_celery) and not dry_run,
-                    session,
+    )
-                    campaign=campaign,
-                    status=campaign.status,
-                    previous_status=previous_status,
-                    version_id=version.id,
-                )
-        session.add(campaign)
-        session.commit()
-
-    enqueued_count = 0
-    if _should_enqueue_celery(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,

tests/test_sending_jobs.py

@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+
+from govoplan_campaign.backend.db.models import (
+    JobBuildStatus,
+    JobQueueStatus,
+    JobSendStatus,
+    JobValidationStatus,
+)
+from govoplan_campaign.backend.sending.jobs import (
+    _queue_validation_statuses,
+    _select_campaign_jobs_for_queue,
+)
+
+
+class FakeSession:
+    def __init__(self) -> None:
+        self.added: list[object] = []
+
+    def add(self, value: object) -> None:
+        self.added.append(value)
+
+
+def _job(entry_id: str, **overrides):
+    values = {
+        "entry_id": entry_id,
+        "entry_index": int(entry_id),
+        "send_status": JobSendStatus.NOT_QUEUED.value,
+        "queue_status": JobQueueStatus.DRAFT.value,
+        "validation_status": JobValidationStatus.READY.value,
+        "build_status": JobBuildStatus.BUILT.value,
+        "eml_local_path": f"{entry_id}.eml",
+        "eml_storage_key": None,
+        "last_error": "old error",
+        "queued_at": None,
+        "claimed_at": "claimed",
+        "claim_token": "token",
+        "smtp_started_at": "started",
+        "outcome_unknown_at": "unknown",
+    }
+    values.update(overrides)
+    return SimpleNamespace(**values)
+
+
+class CampaignQueueSelectionTests(unittest.TestCase):
+    def test_selects_queueable_jobs_without_reclassifying_retry_states(self):
+        skipped_send = _job("1", send_status=JobSendStatus.FAILED_TEMPORARY.value)
+        skipped_queue = _job("2", queue_status=JobQueueStatus.PAUSED.value)
+        blocked_validation = _job(
+            "3",
+            validation_status=JobValidationStatus.WARNING.value,
+        )
+        blocked_missing_eml = _job(
+            "4",
+            eml_local_path=None,
+            eml_storage_key=None,
+        )
+        ready = _job("5")
+        reviewed = _job(
+            "6",
+            validation_status=JobValidationStatus.NEEDS_REVIEW.value,
+        )
+        session = FakeSession()
+
+        queued, skipped_count, blocked_count = _select_campaign_jobs_for_queue(
+            session,
+            jobs=[
+                skipped_send,
+                skipped_queue,
+                blocked_validation,
+                blocked_missing_eml,
+                ready,
+                reviewed,
+            ],
+            allowed_validation=_queue_validation_statuses(include_warnings=False),
+            reviewed_needs_review_keys={"6"},
+            dry_run=False,
+        )
+
+        self.assertEqual(queued, [ready, reviewed])
+        self.assertEqual(skipped_count, 2)
+        self.assertEqual(blocked_count, 2)
+        self.assertIn("generated EML", blocked_missing_eml.last_error)
+        self.assertEqual(session.added, [ready, reviewed])
+        for job in queued:
+            self.assertEqual(job.queue_status, JobQueueStatus.QUEUED.value)
+            self.assertEqual(job.send_status, JobSendStatus.QUEUED.value)
+            self.assertIsNotNone(job.queued_at)
+            self.assertIsNone(job.claimed_at)
+            self.assertIsNone(job.claim_token)
+            self.assertIsNone(job.smtp_started_at)
+            self.assertIsNone(job.outcome_unknown_at)
+            self.assertIsNone(job.last_error)
+
+    def test_dry_run_does_not_mutate_queueable_job(self):
+        warning = _job(
+            "1",
+            validation_status=JobValidationStatus.WARNING.value,
+        )
+        session = FakeSession()
+
+        queued, skipped_count, blocked_count = _select_campaign_jobs_for_queue(
+            session,
+            jobs=[warning],
+            allowed_validation=_queue_validation_statuses(include_warnings=True),
+            reviewed_needs_review_keys=set(),
+            dry_run=True,
+        )
+
+        self.assertEqual(queued, [warning])
+        self.assertEqual((skipped_count, blocked_count), (0, 0))
+        self.assertEqual(warning.queue_status, JobQueueStatus.DRAFT.value)
+        self.assertEqual(warning.send_status, JobSendStatus.NOT_QUEUED.value)
+        self.assertEqual(session.added, [])
+
+
+if __name__ == "__main__":
+    unittest.main()