feat(campaign): support reviewed single-message delivery

This commit is contained in:
2026-07-20 20:07:18 +02:00
parent ad34365f6c
commit 12036b1f36

View File

@@ -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.") 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: def _utcnow() -> datetime:
return datetime.now(timezone.utc) return datetime.now(timezone.utc)
@@ -342,6 +358,7 @@ def queue_campaign_jobs(
raise QueueingError("Campaign version has no jobs. Build messages before queueing.") raise QueueingError("Campaign version has no jobs. Build messages before queueing.")
queued: list[CampaignJob] = [] queued: list[CampaignJob] = []
reviewed_needs_review_keys = _reviewed_needs_review_keys(version)
skipped_count = 0 skipped_count = 0
blocked_count = 0 blocked_count = 0
for job in jobs: 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}: if job.queue_status in {JobQueueStatus.CANCELLED.value, JobQueueStatus.SENDING.value, JobQueueStatus.PAUSED.value}:
skipped_count += 1 skipped_count += 1
continue 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 blocked_count += 1
continue continue
if not job.eml_local_path and not job.eml_storage_key: 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( def reconcile_job_outcome(
session: Session, session: Session,
*, *,
@@ -1168,7 +1326,7 @@ def _preflight_send_campaign_job(
) )
if job.send_status == JobSendStatus.CLAIMED.value: if job.send_status == JobSendStatus.CLAIMED.value:
return SendJobResult( return SendJobResult(
job_id=job_id, job_id=job.id,
status="already_claimed", status="already_claimed",
attempt_number=job.attempt_count, attempt_number=job.attempt_count,
dry_run=dry_run, dry_run=dry_run,