diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index 0893289..73824e9 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/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_execution_snapshot(session, version) - except ExecutionSnapshotError as exc: - raise QueueingError(str(exc)) from exc - - 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() + _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, ) - 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 - blocked_count = 0 - for job in jobs: - if job.send_status in 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 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) - + 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: - 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() - - 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 + _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, diff --git a/tests/test_sending_jobs.py b/tests/test_sending_jobs.py new file mode 100644 index 0000000..bc91d26 --- /dev/null +++ b/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()