diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index 313512a..0893289 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -198,6 +198,22 @@ def _ensure_version_validated_and_locked(version: CampaignVersion) -> None: 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) @@ -342,6 +358,7 @@ def queue_campaign_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: @@ -360,7 +377,11 @@ def queue_campaign_jobs( if job.queue_status in {JobQueueStatus.CANCELLED.value, JobQueueStatus.SENDING.value, JobQueueStatus.PAUSED.value}: skipped_count += 1 continue - if job.build_status != JobBuildStatus.BUILT.value or job.validation_status not in allowed_validation: + 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: @@ -778,6 +799,143 @@ def queue_unattempted_jobs( } +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, *, @@ -1168,7 +1326,7 @@ def _preflight_send_campaign_job( ) if job.send_status == JobSendStatus.CLAIMED.value: return SendJobResult( - job_id=job_id, + job_id=job.id, status="already_claimed", attempt_number=job.attempt_count, dry_run=dry_run,