From 62a68792a402a4a877d96384384b90fc32d8283b Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Wed, 22 Jul 2026 08:37:58 +0200 Subject: [PATCH] feat(campaign): persist delivery execution mode --- src/govoplan_campaign/backend/db/models.py | 2 + .../c7a2f91e4b60_delivery_mode.py | 30 +++ .../c7a2f91e4b60_v0110_delivery_mode.py | 30 +++ .../backend/reports/campaigns.py | 47 ++++ src/govoplan_campaign/backend/schemas.py | 2 + src/govoplan_campaign/backend/sending/jobs.py | 44 +++- tests/test_queue_controls.py | 200 ++++++++++++++++++ webui/src/api/campaigns.ts | 7 + .../src/features/campaigns/ReviewSendPage.tsx | 15 +- 9 files changed, 373 insertions(+), 4 deletions(-) create mode 100644 src/govoplan_campaign/backend/migrations/dev_versions/c7a2f91e4b60_delivery_mode.py create mode 100644 src/govoplan_campaign/backend/migrations/versions/c7a2f91e4b60_v0110_delivery_mode.py create mode 100644 tests/test_queue_controls.py diff --git a/src/govoplan_campaign/backend/db/models.py b/src/govoplan_campaign/backend/db/models.py index 52a9b6b..d987656 100644 --- a/src/govoplan_campaign/backend/db/models.py +++ b/src/govoplan_campaign/backend/db/models.py @@ -212,6 +212,8 @@ class CampaignVersion(Base, TimestampMixin): execution_snapshot: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) execution_snapshot_hash: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) execution_snapshot_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + delivery_mode: Mapped[str | None] = mapped_column(String(30), nullable=True, index=True) + delivery_mode_selected_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) campaign: Mapped[Campaign] = relationship(back_populates="versions") diff --git a/src/govoplan_campaign/backend/migrations/dev_versions/c7a2f91e4b60_delivery_mode.py b/src/govoplan_campaign/backend/migrations/dev_versions/c7a2f91e4b60_delivery_mode.py new file mode 100644 index 0000000..2f5a067 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/dev_versions/c7a2f91e4b60_delivery_mode.py @@ -0,0 +1,30 @@ +"""persist the selected Campaign delivery mode + +Revision ID: c7a2f91e4b60 +Revises: 4d5e6f7a9203 +Create Date: 2026-07-22 09:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c7a2f91e4b60" +down_revision = "4d5e6f7a9203" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("campaign_versions") as batch: + batch.add_column(sa.Column("delivery_mode", sa.String(length=30), nullable=True)) + batch.add_column(sa.Column("delivery_mode_selected_at", sa.DateTime(timezone=True), nullable=True)) + batch.create_index("ix_campaign_versions_delivery_mode", ["delivery_mode"], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table("campaign_versions") as batch: + batch.drop_index("ix_campaign_versions_delivery_mode") + batch.drop_column("delivery_mode_selected_at") + batch.drop_column("delivery_mode") diff --git a/src/govoplan_campaign/backend/migrations/versions/c7a2f91e4b60_v0110_delivery_mode.py b/src/govoplan_campaign/backend/migrations/versions/c7a2f91e4b60_v0110_delivery_mode.py new file mode 100644 index 0000000..2f5a067 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/versions/c7a2f91e4b60_v0110_delivery_mode.py @@ -0,0 +1,30 @@ +"""persist the selected Campaign delivery mode + +Revision ID: c7a2f91e4b60 +Revises: 4d5e6f7a9203 +Create Date: 2026-07-22 09:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c7a2f91e4b60" +down_revision = "4d5e6f7a9203" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("campaign_versions") as batch: + batch.add_column(sa.Column("delivery_mode", sa.String(length=30), nullable=True)) + batch.add_column(sa.Column("delivery_mode_selected_at", sa.DateTime(timezone=True), nullable=True)) + batch.create_index("ix_campaign_versions_delivery_mode", ["delivery_mode"], unique=False) + + +def downgrade() -> None: + with op.batch_alter_table("campaign_versions") as batch: + batch.drop_index("ix_campaign_versions_delivery_mode") + batch.drop_column("delivery_mode_selected_at") + batch.drop_column("delivery_mode") diff --git a/src/govoplan_campaign/backend/reports/campaigns.py b/src/govoplan_campaign/backend/reports/campaigns.py index 02eeca7..20f26d3 100644 --- a/src/govoplan_campaign/backend/reports/campaigns.py +++ b/src/govoplan_campaign/backend/reports/campaigns.py @@ -87,6 +87,10 @@ def _version_info( ), "execution_snapshot_hash": version.execution_snapshot_hash, "execution_snapshot_at": version.execution_snapshot_at.isoformat() if version.execution_snapshot_at else None, + "delivery_mode": version.delivery_mode, + "delivery_mode_selected_at": ( + version.delivery_mode_selected_at.isoformat() if version.delivery_mode_selected_at else None + ), } @@ -110,6 +114,12 @@ def _load_delivery_info( "queueable_job_count": 0, "estimated_remaining_send_seconds": None, "estimated_remaining_send_human": None, + "delivery_mode": version.delivery_mode if version else None, + "delivery_mode_selected_at": ( + version.delivery_mode_selected_at.isoformat() + if version and version.delivery_mode_selected_at + else None + ), } if include_diagnostics: default.update( @@ -159,6 +169,10 @@ def _load_delivery_info( "queueable_job_count": snapshot.queueable_job_count, "estimated_remaining_send_seconds": estimated_seconds, "estimated_remaining_send_human": _human_duration(estimated_seconds), + "delivery_mode": version.delivery_mode, + "delivery_mode_selected_at": ( + version.delivery_mode_selected_at.isoformat() if version.delivery_mode_selected_at else None + ), } if include_diagnostics: result.update( @@ -544,6 +558,27 @@ def _campaign_report_cards(version: CampaignVersion | None, jobs: list[CampaignJ send_counts = _counter([job.send_status for job in jobs]) imap_counts = _counter([job.imap_status for job in jobs]) queueable = sum(1 for job in jobs if job.validation_status in {"ready", "warning"} and job.build_status == "built") + queueable_unattempted = sum( + 1 + for job in jobs + if job.attempt_count == 0 + and job.send_status in {"not_queued", "cancelled"} + and job.validation_status in {"ready", "warning"} + and job.build_status == "built" + ) + retry_max_attempts = _retry_max_attempts(version) + retryable = sum( + 1 + for job in jobs + if job.send_status == "failed_temporary" + and (retry_max_attempts is None or job.attempt_count < retry_max_attempts) + ) + cancellable = sum( + 1 + for job in jobs + if job.send_status + not in {"smtp_accepted", "sent", "outcome_unknown", "claimed", "sending", "cancelled"} + ) needs_attention = sum( 1 for job in jobs @@ -562,6 +597,9 @@ def _campaign_report_cards(version: CampaignVersion | None, jobs: list[CampaignJ "jobs_total": len(jobs), "inactive": inactive_entries, "queueable": queueable, + "queueable_unattempted": queueable_unattempted, + "retryable": retryable, + "cancellable": cancellable, "needs_attention": needs_attention, "sent": sent, "smtp_accepted": sent, @@ -576,6 +614,15 @@ def _campaign_report_cards(version: CampaignVersion | None, jobs: list[CampaignJ } +def _retry_max_attempts(version: CampaignVersion | None) -> int | None: + if version is None or not isinstance(version.execution_snapshot, dict): + return None + try: + return ExecutionSnapshot.model_validate(version.execution_snapshot).delivery.retry.max_attempts + except Exception: + return None + + def _inactive_entry_count(version: CampaignVersion | None) -> int: build_summary = version.build_summary if version and isinstance(version.build_summary, dict) else {} return int(build_summary.get("inactive_count") or build_summary.get("inactive_entries_count") or 0) diff --git a/src/govoplan_campaign/backend/schemas.py b/src/govoplan_campaign/backend/schemas.py index 1301f8a..a04b700 100644 --- a/src/govoplan_campaign/backend/schemas.py +++ b/src/govoplan_campaign/backend/schemas.py @@ -112,6 +112,8 @@ class CampaignVersionResponse(BaseModel): build_summary: dict[str, Any] | None = None execution_snapshot_hash: str | None = None execution_snapshot_at: datetime | None = None + delivery_mode: Literal["synchronous", "worker_queue", "database_queue"] | None = None + delivery_mode_selected_at: datetime | None = None @field_validator("editor_state", mode="before") @classmethod diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index 71bf4e2..96df615 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -201,6 +201,14 @@ QUEUEABLE_VALIDATION_STATUSES = { JobValidationStatus.WARNING.value, } SMTP_ACCEPTED_STATUSES = {JobSendStatus.SMTP_ACCEPTED.value, JobSendStatus.SENT.value} +DELIVERY_MODE_SYNCHRONOUS = "synchronous" +DELIVERY_MODE_WORKER_QUEUE = "worker_queue" +DELIVERY_MODE_DATABASE_QUEUE = "database_queue" +DELIVERY_MODES = { + DELIVERY_MODE_SYNCHRONOUS, + DELIVERY_MODE_WORKER_QUEUE, + DELIVERY_MODE_DATABASE_QUEUE, +} 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 | { @@ -273,6 +281,13 @@ def _utcnow() -> datetime: return datetime.now(timezone.utc) +def _set_version_delivery_mode(version: CampaignVersion, mode: str) -> None: + if mode not in DELIVERY_MODES: + raise QueueingError(f"Unsupported Campaign delivery mode: {mode}") + version.delivery_mode = mode + version.delivery_mode_selected_at = _utcnow() + + 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: @@ -311,6 +326,10 @@ def _should_enqueue_celery(enqueue_celery: bool) -> bool: return bool(enqueue_celery and _celery_enabled()) +def _asynchronous_delivery_mode(enqueue_celery: bool) -> str: + return DELIVERY_MODE_WORKER_QUEUE if _should_enqueue_celery(enqueue_celery) else DELIVERY_MODE_DATABASE_QUEUE + + def _campaign_notification_body(campaign: Campaign, status: str) -> str: return { CampaignStatus.QUEUED.value: f"{campaign.name} has been queued for delivery.", @@ -601,11 +620,13 @@ def _persist_campaign_queue( campaign: Campaign, version: CampaignVersion, queued: list[CampaignJob], + delivery_mode: str, ) -> None: if queued: previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value + _set_version_delivery_mode(version, delivery_mode) if version.locked_at is None: version.locked_at = _utcnow() session.add(version) @@ -638,9 +659,13 @@ def queue_campaign_jobs( enqueue_celery: bool = True, include_warnings: bool = True, dry_run: bool = False, + delivery_mode: str | None = None, ) -> QueueCampaignResult: """Move queueable DB jobs to QUEUED and optionally enqueue Celery tasks.""" + selected_delivery_mode = delivery_mode or _asynchronous_delivery_mode(enqueue_celery) + if selected_delivery_mode not in DELIVERY_MODES: + raise QueueingError(f"Unsupported Campaign delivery mode: {selected_delivery_mode}") 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) @@ -667,6 +692,7 @@ def queue_campaign_jobs( campaign=campaign, version=version, queued=queued, + delivery_mode=selected_delivery_mode, ) enqueued_count = _enqueue_campaign_jobs( queued, @@ -680,7 +706,7 @@ def queue_campaign_jobs( skipped_count=skipped_count, blocked_count=blocked_count, enqueued_count=enqueued_count, - delivery_mode="worker_queue" if enqueue_celery else "database_queue", + delivery_mode=selected_delivery_mode, worker_queue_available=_celery_enabled(), dry_run=dry_run, ) @@ -738,6 +764,7 @@ def send_campaign_now( include_warnings=include_warnings, enqueue_celery=False, dry_run=dry_run, + delivery_mode=DELIVERY_MODE_SYNCHRONOUS, ) if dry_run: return SendCampaignNowResult( @@ -939,6 +966,12 @@ def resume_campaign_jobs(session: Session, *, tenant_id: str, campaign_id: str, job.send_status = JobSendStatus.QUEUED.value session.add(job) if jobs: + delivery_mode = _asynchronous_delivery_mode(enqueue_celery) + for version_id in {job.campaign_version_id for job in jobs}: + version = session.get(CampaignVersion, version_id) + if version is not None and version.campaign_id == campaign.id: + _set_version_delivery_mode(version, delivery_mode) + session.add(version) previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value session.add(campaign) @@ -1065,6 +1098,10 @@ def queue_failed_jobs_for_retry( if selected: campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value + _set_version_delivery_mode( + version, + _asynchronous_delivery_mode(enqueue_celery), + ) session.add(campaign) session.add(version) session.commit() @@ -1134,6 +1171,10 @@ def queue_unattempted_jobs( if selected: campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value + _set_version_delivery_mode( + version, + _asynchronous_delivery_mode(enqueue_celery), + ) session.add(campaign) session.add(version) session.commit() @@ -1202,6 +1243,7 @@ def send_single_campaign_job( previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value version.workflow_state = CampaignVersionWorkflowState.QUEUED.value + _set_version_delivery_mode(version, DELIVERY_MODE_SYNCHRONOUS) session.add(campaign) session.add(version) session.commit() diff --git a/tests/test_queue_controls.py b/tests/test_queue_controls.py new file mode 100644 index 0000000..dedc1b0 --- /dev/null +++ b/tests/test_queue_controls.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +import unittest +from unittest.mock import patch + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session + +from govoplan_access.backend.db.models import Account, Group, User +from govoplan_campaign.backend.db.models import ( + Campaign, + CampaignJob, + CampaignStatus, + CampaignVersion, + JobBuildStatus, + JobQueueStatus, + JobSendStatus, + JobValidationStatus, +) +from govoplan_campaign.backend.sending.jobs import ( + QueueingError, + cancel_campaign_jobs, + pause_campaign_jobs, + resume_campaign_jobs, +) +from govoplan_campaign.backend.reports.campaigns import _campaign_report_cards, _version_info +from govoplan_core.core.change_sequence import ChangeSequenceEntry, ChangeSequenceRetentionFloor +from govoplan_core.db.base import Base + + +class CampaignQueueControlTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all( + self.engine, + tables=[ + Account.__table__, + User.__table__, + Group.__table__, + ChangeSequenceEntry.__table__, + ChangeSequenceRetentionFloor.__table__, + Campaign.__table__, + CampaignVersion.__table__, + CampaignJob.__table__, + ], + ) + self.session = Session(self.engine) + self._add_campaign("campaign-1", "tenant-1", "version-1") + self._add_campaign("campaign-2", "tenant-2", "version-2") + self._add_job("queued", queue="queued", send="queued") + self._add_job("paused", queue="paused", send="queued") + self._add_job("failed", queue="draft", send="failed_temporary") + self._add_job("unknown", queue="draft", send="outcome_unknown") + self._add_job("accepted", queue="draft", send="smtp_accepted") + self._add_job("claimed", queue="sending", send="claimed") + self._add_job( + "other-tenant", + tenant_id="tenant-2", + campaign_id="campaign-2", + version_id="version-2", + queue="queued", + send="queued", + ) + self.session.commit() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + def test_pause_and_resume_change_only_valid_jobs_in_the_selected_tenant(self) -> None: + with patch("govoplan_campaign.backend.sending.jobs._emit_campaign_status_notification"): + paused = pause_campaign_jobs( + self.session, + tenant_id="tenant-1", + campaign_id="campaign-1", + ) + resumed = resume_campaign_jobs( + self.session, + tenant_id="tenant-1", + campaign_id="campaign-1", + enqueue_celery=False, + ) + + self.assertEqual(paused["paused_count"], 1) + self.assertEqual(resumed, { + "campaign_id": "campaign-1", + "resumed_count": 2, + "enqueued_count": 0, + }) + self.session.expire_all() + self.assertEqual(self.session.get(Campaign, "campaign-1").status, CampaignStatus.QUEUED.value) + version = self.session.get(CampaignVersion, "version-1") + self.assertEqual(version.delivery_mode, "database_queue") + self.assertIsNotNone(version.delivery_mode_selected_at) + self.assertEqual(self.session.get(CampaignJob, "queued").queue_status, JobQueueStatus.QUEUED.value) + self.assertEqual(self.session.get(CampaignJob, "paused").queue_status, JobQueueStatus.QUEUED.value) + self.assertEqual(self.session.get(CampaignJob, "other-tenant").queue_status, JobQueueStatus.QUEUED.value) + + def test_cancel_is_auditable_idempotent_state_change_and_protects_started_effects(self) -> None: + with patch("govoplan_campaign.backend.sending.jobs._emit_campaign_status_notification"): + first = cancel_campaign_jobs( + self.session, + tenant_id="tenant-1", + campaign_id="campaign-1", + ) + second = cancel_campaign_jobs( + self.session, + tenant_id="tenant-1", + campaign_id="campaign-1", + ) + + self.assertEqual(first["cancelled_count"], 3) + self.assertEqual(first["protected_count"], 3) + self.assertEqual(second["cancelled_count"], 0) + self.assertEqual(second["protected_count"], 3) + self.session.expire_all() + for job_id in ("queued", "paused", "failed"): + job = self.session.get(CampaignJob, job_id) + self.assertEqual(job.queue_status, JobQueueStatus.CANCELLED.value) + self.assertEqual(job.send_status, JobSendStatus.CANCELLED.value) + self.assertEqual(self.session.get(CampaignJob, "unknown").send_status, JobSendStatus.OUTCOME_UNKNOWN.value) + self.assertEqual(self.session.get(CampaignJob, "accepted").send_status, JobSendStatus.SMTP_ACCEPTED.value) + self.assertEqual(self.session.get(CampaignJob, "claimed").send_status, JobSendStatus.CLAIMED.value) + self.assertEqual(self.session.get(CampaignJob, "other-tenant").send_status, JobSendStatus.QUEUED.value) + + def test_controls_fail_closed_for_a_campaign_owned_by_another_tenant(self) -> None: + for control in (pause_campaign_jobs, resume_campaign_jobs, cancel_campaign_jobs): + with self.subTest(control=control.__name__), self.assertRaises(QueueingError): + control( + self.session, + tenant_id="tenant-1", + campaign_id="campaign-2", + ) + + def test_queue_projection_exposes_exact_action_counts_and_persisted_mode_only(self) -> None: + version = self.session.get(CampaignVersion, "version-1") + version.delivery_mode = "worker_queue" + version.delivery_mode_selected_at = version.updated_at + jobs = ( + self.session.query(CampaignJob) + .filter(CampaignJob.tenant_id == "tenant-1") + .all() + ) + + cards = _campaign_report_cards(version, jobs) + projected_version = _version_info(version) + + self.assertEqual(cards["retryable"], 1) + self.assertEqual(cards["queueable_unattempted"], 0) + self.assertEqual(cards["cancellable"], 3) + self.assertEqual(projected_version["delivery_mode"], "worker_queue") + self.assertIn("delivery_mode_selected_at", projected_version) + self.assertNotIn("execution_snapshot", projected_version) + + def _add_campaign(self, campaign_id: str, tenant_id: str, version_id: str) -> None: + self.session.add(Campaign( + id=campaign_id, + tenant_id=tenant_id, + external_id=f"external-{campaign_id}", + name=campaign_id, + status=CampaignStatus.QUEUED.value, + current_version_id=version_id, + settings={}, + mail_profile_policy={}, + )) + self.session.add(CampaignVersion( + id=version_id, + campaign_id=campaign_id, + version_number=1, + raw_json={}, + )) + + def _add_job( + self, + job_id: str, + *, + queue: str, + send: str, + tenant_id: str = "tenant-1", + campaign_id: str = "campaign-1", + version_id: str = "version-1", + ) -> None: + self.session.add(CampaignJob( + id=job_id, + tenant_id=tenant_id, + campaign_id=campaign_id, + campaign_version_id=version_id, + entry_index=len(self.session.new), + entry_id=f"entry-{job_id}", + build_status=JobBuildStatus.BUILT.value, + validation_status=JobValidationStatus.READY.value, + queue_status=queue, + send_status=send, + resolved_attachments=[], + issues_snapshot=[], + )) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/campaigns.ts b/webui/src/api/campaigns.ts index e0f21ea..0244837 100644 --- a/webui/src/api/campaigns.ts +++ b/webui/src/api/campaigns.ts @@ -77,6 +77,8 @@ export type CampaignVersionListItem = { build_summary?: Record | null; execution_snapshot_hash?: string | null; execution_snapshot_at?: string | null; + delivery_mode?: "synchronous" | "worker_queue" | "database_queue" | null; + delivery_mode_selected_at?: string | null; }; export type CampaignVersionDetail = CampaignVersionListItem & { @@ -240,11 +242,16 @@ export type CampaignSummary = { created_at?: string | null; validation_summary?: Record | null; build_summary?: Record | null; + delivery_mode?: "synchronous" | "worker_queue" | "database_queue" | null; + delivery_mode_selected_at?: string | null; } | null; cards?: { jobs_total?: number; inactive?: number; queueable?: number; + queueable_unattempted?: number; + retryable?: number; + cancellable?: number; needs_attention?: number; sent?: number; smtp_accepted?: number; diff --git a/webui/src/features/campaigns/ReviewSendPage.tsx b/webui/src/features/campaigns/ReviewSendPage.tsx index 32f5180..df9447f 100644 --- a/webui/src/features/campaigns/ReviewSendPage.tsx +++ b/webui/src/features/campaigns/ReviewSendPage.tsx @@ -305,6 +305,7 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings const jobsTotal = cards?.jobs_total ?? inlineEntries.filter((entry) => entry.active !== false).length; const sentCount = cards?.sent ?? 0; const failedCount = cards?.failed ?? 0; + const retryableCount = cards?.retryable ?? numberFrom(sendStatusCounts, ["failed_temporary"]); const imapAppended = cards?.imap_appended ?? 0; const imapFailed = cards?.imap_failed ?? 0; const imapPending = numberFrom(imapStatusCounts, ["pending"]); @@ -322,6 +323,8 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings const synchronousEligibleCount = numberFrom(synchronousSendOption, ["eligible_recipient_job_count"]); const synchronousSendAllowed = synchronousSendOption.allowed === true; const workerQueueAvailable = deliveryOptions?.worker_queue_available === true; + const persistedDeliveryMode = version?.delivery_mode ?? null; + const persistedDeliveryModeSelectedAt = version?.delivery_mode_selected_at ?? null; const selectedDryRun = dryRun && !directQueuedSendAllowed; const deliveryPartial = cards?.partially_completed === true || ["partially_sent", "failed_partial", "partially_completed"].includes(currentWorkflowState); const deliveryComplete = queuedOrActiveCount === 0 && @@ -913,14 +916,14 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings } async function runRetryFailed() { - if (busy || !canRetryDelivery || failedCount <= 0 || !workerQueueAvailable) return; + if (busy || !canRetryDelivery || retryableCount <= 0 || !workerQueueAvailable) return; setBusy("retry"); setMessage("Queueing retryable failed messages for background workers…"); setError(""); try { const response = await retryCampaignJobs(settings, campaignId, { enqueue_celery: true }); const result = asRecord(response.result ?? response); - setMessage(`Queued ${String(result.queued_count ?? 0)} retryable message(s); ${String(result.enqueued_count ?? 0)} worker task(s) published.`); + setMessage(`Queued ${String(result.selected_count ?? 0)} retryable message(s); ${String(result.enqueued_count ?? 0)} worker task(s) published.`); await reload(); await refreshDeliveryOptions(true); } catch (err) { @@ -1516,6 +1519,7 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings + @@ -1526,6 +1530,11 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings {deliveryStarted ? "i18n:govoplan-campaign.delivery_activity_is_available_in_the_report_and.cb163d1d" : "i18n:govoplan-campaign.no_real_delivery_has_started_for_this_campaign_v.3b9235ff"} + {persistedDeliveryMode && +

+ This version last entered {humanize(persistedDeliveryMode)} mode{persistedDeliveryModeSelectedAt ? ` at ${formatDateTime(persistedDeliveryModeSelectedAt)}` : ""}. This record remains available after leaving and returning. +

+ }