diff --git a/pyproject.toml b/pyproject.toml index 12b71d0..87d7c11 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ requires-python = ">=3.12" license = { file = "LICENSE" } authors = [{ name = "GovOPlaN" }] dependencies = [ - "govoplan-core>=0.1.8", + "govoplan-core>=0.1.14", "jsonschema>=4,<5", "pydantic>=2,<3", "SQLAlchemy>=2,<3", diff --git a/src/govoplan_campaign/backend/db/models.py b/src/govoplan_campaign/backend/db/models.py index 50c4553..e3b0771 100644 --- a/src/govoplan_campaign/backend/db/models.py +++ b/src/govoplan_campaign/backend/db/models.py @@ -8,6 +8,7 @@ from typing import Any from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column, relationship +from govoplan_core.core.concurrency import strong_resource_etag from govoplan_core.db.base import Base, TimestampMixin @@ -188,6 +189,11 @@ class CampaignVersion(Base, TimestampMixin): id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) campaign_id: Mapped[str] = mapped_column(ForeignKey("campaigns.id", ondelete="CASCADE"), nullable=False, index=True) version_number: Mapped[int] = mapped_column(Integer, nullable=False) + edit_revision: Mapped[int] = mapped_column( + Integer, + default=1, + nullable=False, + ) raw_json: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) schema_version: Mapped[str] = mapped_column(String(50), default="1.0", nullable=False) source_filename: Mapped[str | None] = mapped_column(String(500)) @@ -234,6 +240,18 @@ class CampaignVersion(Base, TimestampMixin): campaign: Mapped[Campaign] = relationship(back_populates="versions") + __mapper_args__ = { + "version_id_col": edit_revision, + } + + @property + def strong_etag(self) -> str: + return strong_resource_etag( + "campaign_version", + self.id, + self.edit_revision, + ) + @property def mail_profile_migration_required(self) -> bool: from govoplan_campaign.backend.campaign.mail_profile_boundary import campaign_mail_profile_boundary_violations @@ -365,6 +383,139 @@ class SendAttempt(Base, TimestampMixin): finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) +class CampaignMessageAction(Base, TimestampMixin): + __tablename__ = "campaign_message_actions" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_campaign_message_actions_idempotency", + ), + Index( + "ix_campaign_message_actions_job_created", + "job_id", + "created_at", + ), + Index( + "ix_campaign_message_actions_campaign_kind", + "campaign_id", + "kind", + "status", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True) + campaign_id: Mapped[str] = mapped_column( + ForeignKey("campaigns.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + campaign_version_id: Mapped[str] = mapped_column( + ForeignKey("campaign_versions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + job_id: Mapped[str] = mapped_column( + ForeignKey("campaign_jobs.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + kind: Mapped[str] = mapped_column(String(30), nullable=False, index=True) + idempotency_key: Mapped[str] = mapped_column(String(200), nullable=False) + canonical_request_hash: Mapped[str] = mapped_column(String(64), nullable=False) + reason: Mapped[str | None] = mapped_column(Text, nullable=True) + context: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False) + actor_user_id: Mapped[str | None] = mapped_column( + ForeignKey("access_users.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + actor_api_key_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + message_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + message_size_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True) + recipient_manifest_sha256: Mapped[str] = mapped_column( + String(64), + nullable=False, + ) + recipient_count: Mapped[int] = mapped_column(Integer, nullable=False) + prior_send_status: Mapped[str] = mapped_column(String(50), nullable=False) + prior_attempt_count: Mapped[int] = mapped_column(Integer, nullable=False) + final_send_status: Mapped[str | None] = mapped_column(String(50), nullable=True) + status: Mapped[str] = mapped_column( + String(50), + default="initiated", + nullable=False, + index=True, + ) + accepted_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + refused_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + refusal_summary: Mapped[dict[str, Any]] = mapped_column( + JSON, + default=dict, + nullable=False, + ) + error_type: Mapped[str | None] = mapped_column(String(120), nullable=True) + error_message: Mapped[str | None] = mapped_column(String(500), nullable=True) + linked_send_attempt_id: Mapped[str | None] = mapped_column( + ForeignKey("send_attempts.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + effect_started_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + completed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + + +class CampaignMessageActionAttempt(Base, TimestampMixin): + __tablename__ = "campaign_message_action_attempts" + __table_args__ = ( + UniqueConstraint( + "action_id", + "attempt_number", + name="uq_campaign_message_action_attempt_number", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + action_id: Mapped[str] = mapped_column( + ForeignKey("campaign_message_actions.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + attempt_number: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + status: Mapped[str] = mapped_column( + String(50), + default="initiated", + nullable=False, + index=True, + ) + started_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + ) + effect_started_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + completed_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + ) + accepted_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + refused_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) + outcome_code: Mapped[str | None] = mapped_column(String(80), nullable=True) + diagnostic_summary: Mapped[str | None] = mapped_column( + String(500), + nullable=True, + ) + + class ImapAppendAttempt(Base, TimestampMixin): __tablename__ = "imap_append_attempts" __table_args__ = ( diff --git a/src/govoplan_campaign/backend/integrations.py b/src/govoplan_campaign/backend/integrations.py index e6d9156..f379a6f 100644 --- a/src/govoplan_campaign/backend/integrations.py +++ b/src/govoplan_campaign/backend/integrations.py @@ -10,12 +10,15 @@ from typing import Any, Iterator from govoplan_core.core.postbox import ( CAPABILITY_POSTBOX_DIRECTORY, CAPABILITY_POSTBOX_DELIVERY, + CAPABILITY_POSTBOX_EVIDENCE, PostboxDeliveryCatalogRef, PostboxDeliveryProvider, + PostboxDeliveryReceiptSummaryRef, PostboxDeliveryRequest, PostboxDeliveryResult, PostboxDirectoryEntryRef, PostboxDirectoryProvider, + PostboxEvidenceProvider, PostboxTargetRef, ) from govoplan_campaign.backend.runtime import capability @@ -25,6 +28,7 @@ FILES_CAPABILITY = "files.campaign_attachments" MAIL_CAPABILITY = "mail.campaign_delivery" POSTBOX_CAPABILITY = CAPABILITY_POSTBOX_DELIVERY POSTBOX_DIRECTORY_CAPABILITY = CAPABILITY_POSTBOX_DIRECTORY +POSTBOX_EVIDENCE_CAPABILITY = CAPABILITY_POSTBOX_EVIDENCE class OptionalModuleUnavailable(RuntimeError): @@ -63,6 +67,10 @@ class MailProfileError(OptionalModuleUnavailable): pass +class MailDeliveryCommandError(RuntimeError): + pass + + class PostboxDeliveryUnavailable(OptionalModuleUnavailable): pass @@ -158,6 +166,12 @@ class MailCampaignIntegration: def available(self) -> bool: return self._delegate is not None + @property + def durable_delivery_available(self) -> bool: + return self._delegate is not None and callable( + getattr(self._delegate, "submit_delivery_command", None) + ) + def _require(self) -> Any: if self._delegate is None: raise MailProfileError("Mail module is not available") @@ -220,6 +234,42 @@ class MailCampaignIntegration: except getattr(delegate, "MailProfileError", MailProfileError) as exc: raise MailProfileError(str(exc)) from exc + def submit_delivery_command(self, session: Any, **kwargs: Any) -> dict[str, Any]: + delegate = self._require() + method = getattr(delegate, "submit_delivery_command", None) + if not callable(method): + raise MailDeliveryCommandError( + "The installed Mail module does not provide durable delivery commands" + ) + try: + return dict(method(session, **kwargs)) + except Exception as exc: + raise MailDeliveryCommandError(str(exc)) from exc + + def delivery_command_summary( + self, + session: Any, + *, + tenant_id: str, + command_id: str, + ) -> dict[str, Any]: + delegate = self._require() + method = getattr(delegate, "delivery_command_summary", None) + if not callable(method): + raise MailDeliveryCommandError( + "The installed Mail module does not provide durable delivery status" + ) + try: + return dict( + method( + session, + tenant_id=tenant_id, + command_id=command_id, + ) + ) + except Exception as exc: + raise MailDeliveryCommandError(str(exc)) from exc + def mock_mailbox(self) -> Any | None: if self._delegate is None or not hasattr(self._delegate, "mock_mailbox"): return None @@ -231,6 +281,7 @@ class PostboxCampaignIntegration: self, delivery_delegate: object | None = None, directory_delegate: object | None = None, + evidence_delegate: object | None = None, ) -> None: self._delivery_delegate = ( delivery_delegate @@ -242,6 +293,11 @@ class PostboxCampaignIntegration: if isinstance(directory_delegate, PostboxDirectoryProvider) else None ) + self._evidence_delegate = ( + evidence_delegate + if isinstance(evidence_delegate, PostboxEvidenceProvider) + else None + ) @property def available(self) -> bool: @@ -250,6 +306,10 @@ class PostboxCampaignIntegration: and self._directory_delegate is not None ) + @property + def receipt_evidence_available(self) -> bool: + return self._evidence_delegate is not None + def delivery_catalog( self, session: object, @@ -298,6 +358,28 @@ class PostboxCampaignIntegration: ) return self._delivery_delegate.deliver(session, request) + def delivery_receipt_summaries( + self, + session: object, + *, + tenant_id: str, + delivery_ids: list[str] | tuple[str, ...], + ) -> dict[str, PostboxDeliveryReceiptSummaryRef]: + if self._evidence_delegate is None: + return {} + unique_ids = tuple(dict.fromkeys(delivery_ids)) + summaries: dict[str, PostboxDeliveryReceiptSummaryRef] = {} + for offset in range(0, len(unique_ids), 500): + summaries.update( + self._evidence_delegate.delivery_receipt_summaries( + session, + tenant_id=tenant_id, + producer_module="campaigns", + delivery_ids=unique_ids[offset : offset + 500], + ) + ) + return summaries + def files_integration() -> FilesCampaignIntegration: return FilesCampaignIntegration(capability(FILES_CAPABILITY)) @@ -311,4 +393,5 @@ def postbox_integration() -> PostboxCampaignIntegration: return PostboxCampaignIntegration( capability(POSTBOX_CAPABILITY), capability(POSTBOX_DIRECTORY_CAPABILITY), + capability(POSTBOX_EVIDENCE_CAPABILITY), ) diff --git a/src/govoplan_campaign/backend/manifest.py b/src/govoplan_campaign/backend/manifest.py index 483d694..b6ddfd8 100644 --- a/src/govoplan_campaign/backend/manifest.py +++ b/src/govoplan_campaign/backend/manifest.py @@ -31,6 +31,7 @@ from govoplan_core.db.base import Base from govoplan_core.core.postbox import ( CAPABILITY_POSTBOX_DELIVERY, CAPABILITY_POSTBOX_DIRECTORY, + CAPABILITY_POSTBOX_EVIDENCE, ) from govoplan_core.core.references import CAPABILITY_ACCESS_REFERENCE_OPTIONS from govoplan_campaign.backend.change_tracking import register_campaign_change_tracking @@ -235,6 +236,12 @@ manifest = ModuleManifest( version_max_exclusive="0.3.0", optional=True, ), + ModuleInterfaceRequirement( + name="mail.delivery_commands", + version_min="0.1.0", + version_max_exclusive="0.2.0", + optional=True, + ), ModuleInterfaceRequirement( name="addresses.lookup", version_min="0.1.0", @@ -259,6 +266,12 @@ manifest = ModuleManifest( version_max_exclusive="0.2.0", optional=True, ), + ModuleInterfaceRequirement( + name=CAPABILITY_POSTBOX_EVIDENCE, + version_min="0.1.2", + version_max_exclusive="0.2.0", + optional=True, + ), ), permissions=PERMISSIONS, route_factory=_campaigns_router, @@ -330,6 +343,8 @@ manifest = ModuleManifest( campaign_models.AttachmentBlob, campaign_models.AttachmentInstance, campaign_models.SendAttempt, + campaign_models.CampaignMessageAction, + campaign_models.CampaignMessageActionAttempt, campaign_models.ImapAppendAttempt, campaign_models.PostboxDeliveryAttempt, label="Campaigns", @@ -347,6 +362,8 @@ manifest = ModuleManifest( campaign_models.AttachmentBlob, campaign_models.AttachmentInstance, campaign_models.SendAttempt, + campaign_models.CampaignMessageAction, + campaign_models.CampaignMessageActionAttempt, campaign_models.ImapAppendAttempt, campaign_models.PostboxDeliveryAttempt, label="Campaigns", diff --git a/src/govoplan_campaign/backend/migrations/dev_versions/c1a69e4f2b70_campaign_message_actions.py b/src/govoplan_campaign/backend/migrations/dev_versions/c1a69e4f2b70_campaign_message_actions.py new file mode 100644 index 0000000..051febf --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/dev_versions/c1a69e4f2b70_campaign_message_actions.py @@ -0,0 +1,22 @@ +"""add audit-proof single-message action ledger + +Revision ID: c1a69e4f2b70 +Revises: f0a1b2c3d4e5 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from importlib import import_module + + +message_actions = import_module( + "govoplan_campaign.backend.migrations.versions.c1a69e4f2b70_campaign_message_actions" +) + + +revision = message_actions.revision +down_revision = message_actions.down_revision +branch_labels = message_actions.branch_labels +depends_on = message_actions.depends_on +upgrade = message_actions.upgrade +downgrade = message_actions.downgrade diff --git a/src/govoplan_campaign/backend/migrations/dev_versions/d2b7af503c81_campaign_version_edit_revision.py b/src/govoplan_campaign/backend/migrations/dev_versions/d2b7af503c81_campaign_version_edit_revision.py new file mode 100644 index 0000000..38d53bb --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/dev_versions/d2b7af503c81_campaign_version_edit_revision.py @@ -0,0 +1,22 @@ +"""add monotonic campaign version edit revision + +Revision ID: d2b7af503c81 +Revises: c1a69e4f2b70 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from importlib import import_module + + +edit_revision = import_module( + "govoplan_campaign.backend.migrations.versions.d2b7af503c81_campaign_version_edit_revision" +) + + +revision = edit_revision.revision +down_revision = edit_revision.down_revision +branch_labels = edit_revision.branch_labels +depends_on = edit_revision.depends_on +upgrade = edit_revision.upgrade +downgrade = edit_revision.downgrade diff --git a/src/govoplan_campaign/backend/migrations/versions/c1a69e4f2b70_campaign_message_actions.py b/src/govoplan_campaign/backend/migrations/versions/c1a69e4f2b70_campaign_message_actions.py new file mode 100644 index 0000000..0a519fc --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/versions/c1a69e4f2b70_campaign_message_actions.py @@ -0,0 +1,186 @@ +"""add audit-proof single-message action ledger + +Revision ID: c1a69e4f2b70 +Revises: f0a1b2c3d4e5 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c1a69e4f2b70" +down_revision = "f0a1b2c3d4e5" +branch_labels = None +depends_on = "c91f0a72be34" + + +def upgrade() -> None: + tables = set(sa.inspect(op.get_bind()).get_table_names()) + if "campaign_message_actions" not in tables: + op.create_table( + "campaign_message_actions", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("campaign_id", sa.String(length=36), nullable=False), + sa.Column("campaign_version_id", sa.String(length=36), nullable=False), + sa.Column("job_id", sa.String(length=36), nullable=False), + sa.Column("kind", sa.String(length=30), nullable=False), + sa.Column("idempotency_key", sa.String(length=200), nullable=False), + sa.Column("canonical_request_hash", sa.String(length=64), nullable=False), + sa.Column("reason", sa.Text(), nullable=True), + sa.Column("context", sa.JSON(), nullable=False), + sa.Column("actor_user_id", sa.String(length=36), nullable=True), + sa.Column("actor_api_key_id", sa.String(length=36), nullable=True), + sa.Column("message_sha256", sa.String(length=64), nullable=False), + sa.Column("message_size_bytes", sa.Integer(), nullable=True), + sa.Column( + "recipient_manifest_sha256", + sa.String(length=64), + nullable=False, + ), + sa.Column("recipient_count", sa.Integer(), nullable=False), + sa.Column("prior_send_status", sa.String(length=50), nullable=False), + sa.Column("prior_attempt_count", sa.Integer(), nullable=False), + sa.Column("final_send_status", sa.String(length=50), nullable=True), + sa.Column("status", sa.String(length=50), nullable=False), + sa.Column("accepted_count", sa.Integer(), nullable=False), + sa.Column("refused_count", sa.Integer(), nullable=False), + sa.Column("refusal_summary", sa.JSON(), nullable=False), + sa.Column("error_type", sa.String(length=120), nullable=True), + sa.Column("error_message", sa.String(length=500), nullable=True), + sa.Column("linked_send_attempt_id", sa.String(length=36), nullable=True), + sa.Column("effect_started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["campaign_id"], + ["campaigns.id"], + name=op.f( + "fk_campaign_message_actions_campaign_id_campaigns" + ), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["campaign_version_id"], + ["campaign_versions.id"], + name=op.f( + "fk_campaign_message_actions_campaign_version_id_campaign_versions" + ), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["job_id"], + ["campaign_jobs.id"], + name=op.f( + "fk_campaign_message_actions_job_id_campaign_jobs" + ), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["actor_user_id"], + ["access_users.id"], + name=op.f( + "fk_campaign_message_actions_actor_user_id_access_users" + ), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["linked_send_attempt_id"], + ["send_attempts.id"], + name=op.f( + "fk_campaign_message_actions_linked_send_attempt_id_send_attempts" + ), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_campaign_message_actions"), + ), + sa.UniqueConstraint( + "tenant_id", + "idempotency_key", + name="uq_campaign_message_actions_idempotency", + ), + ) + op.create_index( + "ix_campaign_message_actions_job_created", + "campaign_message_actions", + ["job_id", "created_at"], + unique=False, + ) + op.create_index( + "ix_campaign_message_actions_campaign_kind", + "campaign_message_actions", + ["campaign_id", "kind", "status"], + unique=False, + ) + for column in ( + "tenant_id", + "campaign_id", + "campaign_version_id", + "job_id", + "kind", + "actor_user_id", + "status", + "linked_send_attempt_id", + ): + op.create_index( + op.f(f"ix_campaign_message_actions_{column}"), + "campaign_message_actions", + [column], + unique=False, + ) + + tables = set(sa.inspect(op.get_bind()).get_table_names()) + if "campaign_message_action_attempts" not in tables: + op.create_table( + "campaign_message_action_attempts", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("action_id", sa.String(length=36), nullable=False), + sa.Column("attempt_number", sa.Integer(), nullable=False), + sa.Column("status", sa.String(length=50), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("effect_started_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("completed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("accepted_count", sa.Integer(), nullable=False), + sa.Column("refused_count", sa.Integer(), nullable=False), + sa.Column("outcome_code", sa.String(length=80), nullable=True), + sa.Column("diagnostic_summary", sa.String(length=500), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["action_id"], + ["campaign_message_actions.id"], + name=op.f( + "fk_campaign_message_action_attempts_action_id_campaign_message_actions" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_campaign_message_action_attempts"), + ), + sa.UniqueConstraint( + "action_id", + "attempt_number", + name="uq_campaign_message_action_attempt_number", + ), + ) + for column in ("action_id", "status"): + op.create_index( + op.f(f"ix_campaign_message_action_attempts_{column}"), + "campaign_message_action_attempts", + [column], + unique=False, + ) + + +def downgrade() -> None: + tables = set(sa.inspect(op.get_bind()).get_table_names()) + if "campaign_message_action_attempts" in tables: + op.drop_table("campaign_message_action_attempts") + if "campaign_message_actions" in tables: + op.drop_table("campaign_message_actions") diff --git a/src/govoplan_campaign/backend/migrations/versions/d2b7af503c81_campaign_version_edit_revision.py b/src/govoplan_campaign/backend/migrations/versions/d2b7af503c81_campaign_version_edit_revision.py new file mode 100644 index 0000000..3a92fb2 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/versions/d2b7af503c81_campaign_version_edit_revision.py @@ -0,0 +1,45 @@ +"""add monotonic campaign version edit revision + +Revision ID: d2b7af503c81 +Revises: c1a69e4f2b70 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "d2b7af503c81" +down_revision = "c1a69e4f2b70" +branch_labels = None +depends_on = "c91f0a72be34" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + columns = { + column["name"] + for column in inspector.get_columns("campaign_versions") + } + if "edit_revision" not in columns: + with op.batch_alter_table("campaign_versions") as batch_op: + batch_op.add_column( + sa.Column( + "edit_revision", + sa.Integer(), + server_default="1", + nullable=False, + ) + ) + + +def downgrade() -> None: + inspector = sa.inspect(op.get_bind()) + columns = { + column["name"] + for column in inspector.get_columns("campaign_versions") + } + if "edit_revision" in columns: + with op.batch_alter_table("campaign_versions") as batch_op: + batch_op.drop_column("edit_revision") diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index 2e0ca2c..79a8fcb 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -7,7 +7,12 @@ from typing import Any from uuid import uuid4 from sqlalchemy.orm import Session +from sqlalchemy.orm.exc import StaleDataError +from govoplan_core.core.concurrency import ( + RevisionConflictError, + strong_resource_etag, +) from govoplan_campaign.backend.db.models import ( Campaign, CampaignIssue, @@ -720,12 +725,31 @@ def update_campaign_version( source_base_path: str | None = None, autosave: bool = False, migrate_legacy_mail_settings: bool = False, + expected_revision: int | None = None, commit: bool = True, ) -> CampaignVersion: _assert_update_paths_safe(raw_json, source_filename=source_filename, source_base_path=source_base_path) version = get_campaign_version_for_tenant(session, tenant_id=tenant_id, campaign_id=campaign_id, version_id=version_id) campaign = _require_campaign(session, campaign_id) ensure_current_working_version(campaign, version, action="edit") + if ( + expected_revision is not None + and version.edit_revision != int(expected_revision) + ): + raise RevisionConflictError( + resource_type="campaign_version", + resource_id=version.id, + current_revision=version.edit_revision, + submitted_base_revision=int(expected_revision), + refresh_path=( + f"/api/v1/campaigns/{campaign.id}/versions/{version.id}" + ), + current_etag=strong_resource_etag( + "campaign_version", + version.id, + version.edit_revision, + ), + ) if is_version_locked(version): raise LockedCampaignVersionError( @@ -761,7 +785,42 @@ def update_campaign_version( # Changes invalidate previous build and validation summaries. if raw_json is not None: _invalidate_version_content(session, campaign=campaign, version=version) - _persist_updated_version(session, campaign=campaign, version=version, commit=commit) + try: + _persist_updated_version( + session, + campaign=campaign, + version=version, + commit=commit, + ) + except StaleDataError as exc: + session.rollback() + current_revision = ( + session.query(CampaignVersion.edit_revision) + .filter(CampaignVersion.id == version_id) + .scalar() + ) + if current_revision is None: + raise CampaignPersistenceError( + f"Campaign version not found: {version_id}" + ) from exc + raise RevisionConflictError( + resource_type="campaign_version", + resource_id=version_id, + current_revision=int(current_revision), + submitted_base_revision=int( + expected_revision + if expected_revision is not None + else version.edit_revision + ), + refresh_path=( + f"/api/v1/campaigns/{campaign_id}/versions/{version_id}" + ), + current_etag=strong_resource_etag( + "campaign_version", + version_id, + int(current_revision), + ), + ) from exc return version diff --git a/src/govoplan_campaign/backend/reports/campaigns.py b/src/govoplan_campaign/backend/reports/campaigns.py index cdf5ab1..0680c36 100644 --- a/src/govoplan_campaign/backend/reports/campaigns.py +++ b/src/govoplan_campaign/backend/reports/campaigns.py @@ -9,6 +9,7 @@ from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any +from sqlalchemy import func from sqlalchemy.orm import Session from govoplan_core.settings import settings as core_settings @@ -17,11 +18,14 @@ from govoplan_campaign.backend.db.models import ( Campaign, CampaignIssue, CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, CampaignVersion, ImapAppendAttempt, PostboxDeliveryAttempt, SendAttempt, ) +from govoplan_campaign.backend.integrations import postbox_integration from govoplan_campaign.backend.sending.execution import ExecutionSnapshot from govoplan_campaign.backend.response_security import ( public_campaign_payload, @@ -586,6 +590,7 @@ def _job_evidence_row( *, latest_smtp: SendAttempt | None = None, latest_imap: ImapAppendAttempt | None = None, + latest_message_action: CampaignMessageAction | None = None, include_diagnostics: bool = False, ) -> dict[str, Any]: row = _job_row(job, include_diagnostics=include_diagnostics) @@ -598,6 +603,20 @@ def _job_evidence_row( "postbox_targets": _job_postbox_target_summary(job), "attachment_names": _attachment_names(job.resolved_attachments), **_job_attempt_evidence(latest_smtp=latest_smtp, latest_imap=latest_imap), + "latest_message_action_kind": ( + latest_message_action.kind if latest_message_action else None + ), + "latest_message_action_status": ( + latest_message_action.status if latest_message_action else None + ), + "latest_message_action_id": ( + latest_message_action.id if latest_message_action else None + ), + "latest_message_action_at": ( + _iso_timestamp(latest_message_action.completed_at) + if latest_message_action + else None + ), }) return row @@ -762,6 +781,18 @@ def _campaign_report_payload( campaign_id=campaign.id, version=version, ), + "postbox_receipts": _campaign_postbox_receipt_summary( + session, + tenant_id=tenant_id, + campaign_id=campaign.id, + version=version, + ), + "message_actions": _campaign_message_action_summary( + session, + tenant_id=tenant_id, + campaign_id=campaign.id, + version=version, + ), "delivery": _load_delivery_info( version, pending_job_count=aggregate.pending, @@ -775,6 +806,85 @@ def _campaign_report_payload( return report +def _campaign_postbox_receipt_summary( + session: Session, + *, + tenant_id: str, + campaign_id: str, + version: CampaignVersion | None, +) -> dict[str, object]: + if version is None: + return {"status": "not_applicable", "delivery_count": 0} + rows = ( + session.query(PostboxDeliveryAttempt.provider_delivery_id) + .join( + CampaignJob, + PostboxDeliveryAttempt.job_id == CampaignJob.id, + ) + .filter( + CampaignJob.tenant_id == tenant_id, + CampaignJob.campaign_id == campaign_id, + CampaignJob.campaign_version_id == version.id, + PostboxDeliveryAttempt.tenant_id == tenant_id, + PostboxDeliveryAttempt.provider_delivery_id.is_not(None), + ) + .all() + ) + delivery_ids = tuple( + dict.fromkeys( + str(row[0]) + for row in rows + if row[0] + ) + ) + if not delivery_ids: + return {"status": "not_applicable", "delivery_count": 0} + + integration = postbox_integration() + if not integration.receipt_evidence_available: + return { + "status": "unavailable", + "delivery_count": len(delivery_ids), + "missing_delivery_count": len(delivery_ids), + } + summaries = integration.delivery_receipt_summaries( + session, + tenant_id=tenant_id, + delivery_ids=delivery_ids, + ) + values = tuple(summaries.values()) + return { + "status": "available", + "delivery_count": len(delivery_ids), + "reported_delivery_count": len(values), + "missing_delivery_count": len(delivery_ids) - len(values), + "currently_readable_delivery_count": sum( + summary.currently_readable for summary in values + ), + "read_delivery_count": sum( + summary.read_receipt_count > 0 for summary in values + ), + "acknowledged_delivery_count": sum( + summary.acknowledged_receipt_count > 0 for summary in values + ), + "read_receipt_count": sum( + summary.read_receipt_count for summary in values + ), + "acknowledged_receipt_count": sum( + summary.acknowledged_receipt_count for summary in values + ), + "routed_message_count": sum( + summary.routed_message_count for summary in values + ), + "withdrawn_message_count": sum( + summary.withdrawn_message_count for summary in values + ), + "expired_message_count": sum( + summary.expired_message_count for summary in values + ), + } + + def _campaign_report_campaign_payload(campaign: Campaign) -> dict[str, Any]: return { "id": campaign.id, @@ -815,6 +925,99 @@ def _campaign_report_attempt_counts( "send_attempts": count(SendAttempt), "imap_append_attempts": count(ImapAppendAttempt), "postbox_delivery_attempts": count(PostboxDeliveryAttempt), + "message_actions": count(CampaignMessageAction), + "message_action_attempts": int( + session.query(CampaignMessageActionAttempt) + .join( + CampaignMessageAction, + CampaignMessageAction.id + == CampaignMessageActionAttempt.action_id, + ) + .filter( + CampaignMessageAction.tenant_id == tenant_id, + CampaignMessageAction.campaign_id == campaign_id, + *( + [CampaignMessageAction.campaign_version_id == version.id] + if version is not None + else [False] + ), + ) + .count() + ), + } + + +def _campaign_message_action_summary( + session: Session, + *, + tenant_id: str, + campaign_id: str, + version: CampaignVersion | None, +) -> dict[str, Any]: + base = [ + CampaignMessageAction.tenant_id == tenant_id, + CampaignMessageAction.campaign_id == campaign_id, + ] + if version is not None: + base.append(CampaignMessageAction.campaign_version_id == version.id) + else: + base.append(False) + by_kind = { + str(kind): int(count) + for kind, count in ( + session.query( + CampaignMessageAction.kind, + func.count(CampaignMessageAction.id), + ) + .filter(*base) + .group_by(CampaignMessageAction.kind) + .all() + ) + } + by_status = { + str(action_status): int(count) + for action_status, count in ( + session.query( + CampaignMessageAction.status, + func.count(CampaignMessageAction.id), + ) + .filter(*base) + .group_by(CampaignMessageAction.status) + .all() + ) + } + recent = ( + session.query(CampaignMessageAction) + .filter(*base) + .order_by(CampaignMessageAction.created_at.desc()) + .limit(100) + .all() + ) + return { + "total": sum(by_kind.values()), + "by_kind": by_kind, + "by_status": by_status, + "recent": [ + { + "id": action.id, + "job_id": action.job_id, + "kind": action.kind, + "status": action.status, + "reason": action.reason, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": action.recipient_manifest_sha256, + "recipient_count": action.recipient_count, + "prior_send_status": action.prior_send_status, + "final_send_status": action.final_send_status, + "accepted_count": action.accepted_count, + "refused_count": action.refused_count, + "linked_send_attempt_id": action.linked_send_attempt_id, + "actor_user_id": action.actor_user_id, + "created_at": _iso_timestamp(action.created_at), + "completed_at": _iso_timestamp(action.completed_at), + } + for action in recent + ], } @@ -992,11 +1195,16 @@ def generate_jobs_csv( ImapAppendAttempt, job_ids, ) + latest_message_actions = _latest_message_actions_for_jobs( + session, + job_ids, + ) rows = [ _job_evidence_row( job, latest_smtp=latest_smtp.get(job.id), latest_imap=latest_imap.get(job.id), + latest_message_action=latest_message_actions.get(job.id), include_diagnostics=include_diagnostics, ) for job in jobs @@ -1046,6 +1254,10 @@ def generate_jobs_csv( "latest_imap_folder", "latest_imap_created_at", "latest_imap_updated_at", + "latest_message_action_kind", + "latest_message_action_status", + "latest_message_action_id", + "latest_message_action_at", ] if include_diagnostics: sent_at_index = fieldnames.index("outcome_unknown_at") @@ -1082,3 +1294,26 @@ def _latest_attempts_for_jobs( ): latest[attempt.job_id] = attempt return latest + + +def _latest_message_actions_for_jobs( + session: Session, + job_ids: list[str], + *, + batch_size: int = 500, +) -> dict[str, CampaignMessageAction]: + latest: dict[str, CampaignMessageAction] = {} + for offset in range(0, len(job_ids), batch_size): + batch = job_ids[offset : offset + batch_size] + actions = ( + session.query(CampaignMessageAction) + .filter(CampaignMessageAction.job_id.in_(batch)) + .order_by( + CampaignMessageAction.job_id.asc(), + CampaignMessageAction.created_at.asc(), + ) + .yield_per(batch_size) + ) + for action in actions: + latest[action.job_id] = action + return latest diff --git a/src/govoplan_campaign/backend/reports/emailing.py b/src/govoplan_campaign/backend/reports/emailing.py index dd1c281..6a5e498 100644 --- a/src/govoplan_campaign/backend/reports/emailing.py +++ b/src/govoplan_campaign/backend/reports/emailing.py @@ -13,7 +13,11 @@ from govoplan_campaign.backend.campaign.models import CampaignConfig from govoplan_campaign.backend.campaign.mail_profile_boundary import campaign_mail_profile_id from govoplan_campaign.backend.persistence.campaigns import load_version_config from govoplan_campaign.backend.reports.campaigns import CampaignReportError, generate_campaign_report, generate_jobs_csv -from govoplan_campaign.backend.integrations import SmtpConfigurationError +from govoplan_campaign.backend.integrations import ( + MailDeliveryCommandError, + SmtpConfigurationError, + mail_integration, +) from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError, ensure_execution_snapshot @@ -32,6 +36,9 @@ class CampaignReportEmailResult: attached_jobs_csv: bool attached_report_json: bool accepted_count: int | None = None + command_id: str | None = None + delivery_status: str | None = None + duplicate: bool = False def as_dict(self) -> dict[str, Any]: return { @@ -44,6 +51,22 @@ class CampaignReportEmailResult: "attached_jobs_csv": self.attached_jobs_csv, "attached_report_json": self.attached_report_json, "accepted_count": self.accepted_count, + "command_id": self.command_id, + "delivery_status": self.delivery_status, + "duplicate": self.duplicate, + } + + def audit_dict(self) -> dict[str, Any]: + return { + "campaign_id": self.campaign_id, + "version_id": self.version_id, + "recipient_count": len(self.to), + "dry_run": self.dry_run, + "attached_jobs_csv": self.attached_jobs_csv, + "attached_report_json": self.attached_report_json, + "command_id": self.command_id, + "delivery_status": self.delivery_status, + "duplicate": self.duplicate, } @@ -149,6 +172,8 @@ def send_campaign_report_email( attach_jobs_csv: bool = False, attach_report_json: bool = False, dry_run: bool = False, + idempotency_key: str | None = None, + created_by_user_id: str | None = None, ) -> CampaignReportEmailResult: campaign = session.get(Campaign, campaign_id) if not campaign or campaign.tenant_id != tenant_id: @@ -175,11 +200,16 @@ def send_campaign_report_email( ) from exc if not snapshot.smtp_transport_revision: raise CampaignReportEmailError("Campaign build evidence has no SMTP transport revision") + mail = mail_integration() if not dry_run: - raise CampaignReportEmailError( - "Report email delivery is disabled until it uses a durable, idempotent Mail-owned outbox with unknown-outcome reconciliation." - ) - + if not mail.durable_delivery_available: + raise CampaignReportEmailError( + "Report email delivery requires the durable, idempotent Mail-owned outbox." + ) + if not str(idempotency_key or "").strip(): + raise CampaignReportEmailError( + "Live report email delivery requires an idempotency key" + ) report = generate_campaign_report( session, tenant_id=tenant_id, @@ -202,6 +232,48 @@ def send_campaign_report_email( jobs_csv=jobs_csv, report_json=report_json, ) + if not dry_run: + clean_idempotency_key = str(idempotency_key or "").strip() + from_email, _from_name = _effective_from(config) + if not snapshot.mail_profile_id: + raise CampaignReportEmailError( + "Campaign build evidence has no Mail profile reference" + ) + try: + command = mail.submit_delivery_command( + session, + tenant_id=tenant_id, + command_type="campaign_report", + source_module="campaigns", + source_resource_type="campaign", + source_resource_id=campaign.id, + source_version_id=version.id, + idempotency_key=clean_idempotency_key, + profile_id=snapshot.mail_profile_id, + message_bytes=message.as_bytes(), + envelope_from=from_email, + envelope_recipients=to, + from_header=str(message["From"]), + expected_smtp_transport_revision=snapshot.smtp_transport_revision, + smtp_server_id=snapshot.smtp_server_id, + smtp_credential_id=snapshot.smtp_credential_id, + created_by_user_id=created_by_user_id, + ) + except MailDeliveryCommandError as exc: + raise CampaignReportEmailError(str(exc)) from exc + return CampaignReportEmailResult( + campaign_id=campaign.id, + version_id=version.id, + to=to, + subject=str(message["Subject"]), + dry_run=False, + sent=False, + attached_jobs_csv=jobs_csv is not None, + attached_report_json=report_json is not None, + command_id=str(command["id"]), + delivery_status=str(command["status"]), + duplicate=bool(command.get("duplicate")), + ) return CampaignReportEmailResult( campaign_id=campaign.id, version_id=version.id, diff --git a/src/govoplan_campaign/backend/route_support.py b/src/govoplan_campaign/backend/route_support.py index b61048e..cd633ec 100644 --- a/src/govoplan_campaign/backend/route_support.py +++ b/src/govoplan_campaign/backend/route_support.py @@ -43,6 +43,12 @@ from govoplan_campaign.backend.sending.execution import ( from govoplan_core.audit.logging import audit_from_principal from govoplan_core.auth import ApiPrincipal, has_scope from govoplan_core.core.access import CAPABILITY_ACCESS_DIRECTORY, AccessDirectory +from govoplan_core.core.concurrency import ( + ConcurrencyError, + MissingPreconditionError, + RevisionConflictError, + assert_revision_precondition, +) from govoplan_core.core.runtime import get_registry @@ -359,6 +365,9 @@ def _campaign_version_detail_response( raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc) ) from exc + except RevisionConflictError: + session.rollback() + raise except Exception as exc: session.rollback() if validation_error_status is None: @@ -375,44 +384,109 @@ def _update_campaign_version_detail_response( version_id: str, payload: CampaignVersionUpdateRequest, *, + if_match: str | None, autosave: bool, audit_action: str, ) -> CampaignVersionDetailResponse: _get_campaign_for_principal(session, campaign_id, principal, write=True) current_version = _get_version_for_tenant(session, version_id, principal.tenant_id) + if payload.base_revision is None: + error = MissingPreconditionError( + resource_type="campaign_version", + resource_id=version_id, + ) + raise HTTPException( + status_code=status.HTTP_428_PRECONDITION_REQUIRED, + detail=error.as_dict(), + ) + try: + assert_revision_precondition( + if_match, + resource_type="campaign_version", + resource_id=version_id, + submitted_base_revision=payload.base_revision, + ) + except MissingPreconditionError as exc: + raise HTTPException( + status_code=status.HTTP_428_PRECONDITION_REQUIRED, + detail=exc.as_dict(), + ) from exc + except ConcurrencyError as exc: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail={ + "code": "invalid_precondition", + "message": str(exc), + }, + ) from exc if _recipient_sections_changed(current_version.raw_json, payload.campaign_json): _require_permission(principal, "campaigns:recipient:write") _require_mail_profile_use_if_needed(principal, payload.campaign_json) - return _campaign_version_detail_response( - session, - principal, - campaign_id, - lambda: update_campaign_version( + try: + return _campaign_version_detail_response( session, - tenant_id=principal.tenant_id, - campaign_id=campaign_id, - version_id=version_id, - raw_json=payload.campaign_json, - current_flow=payload.current_flow, - current_step=payload.current_step, - workflow_state=payload.workflow_state, - is_complete=payload.is_complete, - editor_state=payload.editor_state, - source_filename=payload.source_filename, - source_base_path=payload.source_base_path, - autosave=autosave, - migrate_legacy_mail_settings=payload.migrate_legacy_mail_settings, - commit=False, - ), - audit_action=audit_action, - details=lambda version: { - "campaign_id": campaign_id, - "current_flow": version.current_flow, - "current_step": version.current_step, - "legacy_mail_settings_migrated": payload.migrate_legacy_mail_settings, - }, - validation_error_status=status.HTTP_422_UNPROCESSABLE_CONTENT, - ) + principal, + campaign_id, + lambda: update_campaign_version( + session, + tenant_id=principal.tenant_id, + campaign_id=campaign_id, + version_id=version_id, + raw_json=payload.campaign_json, + current_flow=payload.current_flow, + current_step=payload.current_step, + workflow_state=payload.workflow_state, + is_complete=payload.is_complete, + editor_state=payload.editor_state, + source_filename=payload.source_filename, + source_base_path=payload.source_base_path, + autosave=autosave, + migrate_legacy_mail_settings=payload.migrate_legacy_mail_settings, + expected_revision=payload.base_revision, + commit=False, + ), + audit_action=audit_action, + details=lambda version: { + "campaign_id": campaign_id, + "current_flow": version.current_flow, + "current_step": version.current_step, + "base_revision": payload.base_revision, + "result_revision": version.edit_revision, + "reconciliation_kind": payload.reconciliation_kind, + "resolved_conflict_path_count": len( + payload.resolved_conflict_paths + ), + "resolved_conflict_sections": sorted( + { + path.strip("/").split("/", 1)[0][:80] + for path in payload.resolved_conflict_paths[:100] + if path.strip("/") + } + ), + "legacy_mail_settings_migrated": payload.migrate_legacy_mail_settings, + }, + validation_error_status=status.HTTP_422_UNPROCESSABLE_CONTENT, + ) + except RevisionConflictError as exc: + session.rollback() + audit_from_principal( + session, + principal, + action="campaign.version_conflict_detected", + object_type="campaign_version", + object_id=version_id, + details={ + "campaign_id": campaign_id, + "submitted_base_revision": exc.submitted_base_revision, + "current_revision": exc.current_revision, + "retryable": True, + }, + commit=True, + ) + raise HTTPException( + status_code=status.HTTP_412_PRECONDITION_FAILED, + detail=exc.as_dict(), + ) from exc def _require_campaign_profile_use_if_needed( diff --git a/src/govoplan_campaign/backend/routes/delivery.py b/src/govoplan_campaign/backend/routes/delivery.py index 0e4e7bf..f1672c3 100644 --- a/src/govoplan_campaign/backend/routes/delivery.py +++ b/src/govoplan_campaign/backend/routes/delivery.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session @@ -74,6 +75,7 @@ from govoplan_campaign.backend.route_support import ( ) router = APIRouter(prefix="/campaigns", tags=["campaigns"]) +logger = logging.getLogger(__name__) @router.get( @@ -236,13 +238,25 @@ def send_unattempted_campaign_jobs( def send_single_campaign_job_endpoint( campaign_id: str, job_id: str, - payload: CampaignSendJobRequest | None = None, + payload: CampaignSendJobRequest, session: Session = Depends(get_session), - principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:send")), + principal: ApiPrincipal = Depends( + require_any_scope( + "campaigns:campaign:send", + "campaigns:campaign:send_test", + ) + ), ): _get_campaign_for_principal(session, campaign_id, principal, write=True) _require_permission(principal, "campaigns:recipient:read") - payload = payload or CampaignSendJobRequest() + _require_permission( + principal, + ( + "campaigns:campaign:send_test" + if payload.kind == "test" + else "campaigns:campaign:send" + ), + ) _require_campaign_profile_use_if_needed(session, principal, campaign_id, None) try: result = send_single_campaign_job( @@ -250,17 +264,20 @@ def send_single_campaign_job_endpoint( tenant_id=principal.tenant_id, campaign_id=campaign_id, job_id=job_id, + kind=payload.kind, + idempotency_key=payload.idempotency_key, + actor_user_id=principal.user.id, + actor_api_key_id=getattr(principal, "api_key_id", None), + reason=payload.reason, + action_context=payload.context, include_warnings=payload.include_warnings, - dry_run=payload.dry_run, use_rate_limit=payload.use_rate_limit, enqueue_imap_task=payload.enqueue_imap_task, ) audit_from_principal( session, principal, - action="campaign.single_message_sent" - if not payload.dry_run - else "campaign.single_message_send_dry_run", + action=f"campaign.message_{payload.kind}", object_type="campaign_job", object_id=job_id, details=result, @@ -272,8 +289,17 @@ def send_single_campaign_job_endpoint( status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) ) from exc except Exception as exc: + logger.exception( + "Unexpected single-message campaign action failure", + extra={ + "campaign_id": campaign_id, + "job_id": job_id, + "action_kind": payload.kind, + }, + ) raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc) + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="The message action failed because of an internal error.", ) from exc diff --git a/src/govoplan_campaign/backend/routes/jobs.py b/src/govoplan_campaign/backend/routes/jobs.py index 5719c28..85f7bee 100644 --- a/src/govoplan_campaign/backend/routes/jobs.py +++ b/src/govoplan_campaign/backend/routes/jobs.py @@ -23,10 +23,13 @@ from govoplan_campaign.backend.change_tracking import ( ) from govoplan_campaign.backend.db.models import ( CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, ImapAppendAttempt, PostboxDeliveryAttempt, SendAttempt, ) +from govoplan_campaign.backend.integrations import postbox_integration from govoplan_core.db.session import get_session @@ -49,6 +52,27 @@ from govoplan_campaign.backend.services.job_queries import ( router = APIRouter(prefix="/campaigns", tags=["campaigns"]) +def _postbox_receipts_for_attempts( + session: Session, + *, + tenant_id: str, + attempts: list[PostboxDeliveryAttempt], +): + integration = postbox_integration() + if not integration.receipt_evidence_available: + return None + delivery_ids = [ + attempt.provider_delivery_id + for attempt in attempts + if attempt.provider_delivery_id + ] + return integration.delivery_receipt_summaries( + session, + tenant_id=tenant_id, + delivery_ids=delivery_ids, + ) + + @router.get("/{campaign_id}/jobs", response_model=CampaignJobsResponse) def list_jobs( campaign_id: str, @@ -373,12 +397,36 @@ def get_job_detail( ), label="Postbox attempts for this campaign job", ) + message_actions = _job_attempt_rows( + session.query(CampaignMessageAction) + .filter(CampaignMessageAction.job_id == job.id) + .order_by(CampaignMessageAction.created_at.asc()), + label="Single-message actions for this campaign job", + ) + action_ids = [action.id for action in message_actions] + message_action_attempts = ( + _job_attempt_rows( + session.query(CampaignMessageActionAttempt) + .filter(CampaignMessageActionAttempt.action_id.in_(action_ids)) + .order_by(CampaignMessageActionAttempt.started_at.asc()), + label="Single-message action attempts for this campaign job", + ) + if action_ids + else [] + ) return CampaignJobDetailResponse( job=_job_detail_payload(job), attempts=_job_attempts_payload( send_attempts, imap_attempts, postbox_attempts, + message_actions, + message_action_attempts, + postbox_receipts=_postbox_receipts_for_attempts( + session, + tenant_id=principal.tenant_id, + attempts=postbox_attempts, + ), ), ) @@ -428,9 +476,33 @@ def get_job_diagnostics( ), label="Postbox diagnostics for this campaign job", ) + message_actions = _job_attempt_rows( + session.query(CampaignMessageAction) + .filter(CampaignMessageAction.job_id == job.id) + .order_by(CampaignMessageAction.created_at.asc()), + label="Single-message action diagnostics for this campaign job", + ) + action_ids = [action.id for action in message_actions] + message_action_attempts = ( + _job_attempt_rows( + session.query(CampaignMessageActionAttempt) + .filter(CampaignMessageActionAttempt.action_id.in_(action_ids)) + .order_by(CampaignMessageActionAttempt.started_at.asc()), + label="Single-message action-attempt diagnostics for this campaign job", + ) + if action_ids + else [] + ) return _job_diagnostics_payload( job, send_attempts, imap_attempts, postbox_attempts, + message_actions, + message_action_attempts, + postbox_receipts=_postbox_receipts_for_attempts( + session, + tenant_id=principal.tenant_id, + attempts=postbox_attempts, + ), ) diff --git a/src/govoplan_campaign/backend/routes/reports.py b/src/govoplan_campaign/backend/routes/reports.py index fef3a3b..6d19baf 100644 --- a/src/govoplan_campaign/backend/routes/reports.py +++ b/src/govoplan_campaign/backend/routes/reports.py @@ -25,9 +25,11 @@ from govoplan_campaign.backend.reports.emailing import ( send_campaign_report_email, ) from govoplan_campaign.backend.integrations import ( + MailDeliveryCommandError, MailProfileError, SmtpConfigurationError, SmtpSendError, + mail_integration, ) @@ -41,6 +43,24 @@ router = APIRouter(prefix="/campaigns", tags=["campaigns"]) logger = logging.getLogger(__name__) +def _enqueue_mail_command() -> None: + try: + from govoplan_core.celery_app import celery + from govoplan_core.settings import settings + + if settings.celery_enabled: + celery.send_task( + "govoplan.mail.dispatch_outbox", + args=[None, 25], + queue="mail", + ) + except Exception: + logger.warning( + "Mail delivery command is durable but immediate worker wake-up failed", + exc_info=True, + ) + + @router.get("/{campaign_id}/summary") def campaign_summary( campaign_id: str, @@ -164,18 +184,22 @@ def email_campaign_report( attach_jobs_csv=payload.attach_jobs_csv, attach_report_json=payload.attach_report_json, dry_run=payload.dry_run, + idempotency_key=payload.idempotency_key, + created_by_user_id=principal.user.id, ) audit_from_principal( session, principal, - action="report.email_sent" + action="report.email_requested" if not payload.dry_run else "report.email_dry_run", object_type="campaign", object_id=campaign_id, - details=result.as_dict(), + details=result.audit_dict(), commit=True, ) + if not payload.dry_run: + _enqueue_mail_command() return ReportEmailResponse(result=result.as_dict()) except CampaignReportError as exc: raise HTTPException( @@ -184,6 +208,7 @@ def email_campaign_report( except ( CampaignReportEmailError, MailProfileError, + MailDeliveryCommandError, SmtpConfigurationError, SmtpSendError, ) as exc: @@ -196,3 +221,37 @@ def email_campaign_report( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Campaign report email could not be completed.", ) from exc + + +@router.get( + "/{campaign_id}/report/email/{command_id}", + response_model=ReportEmailResponse, +) +def campaign_report_email_status( + campaign_id: str, + command_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(require_scope("campaigns:report:read")), +): + _get_campaign_for_principal(session, campaign_id, principal) + try: + result = mail_integration().delivery_command_summary( + session, + tenant_id=principal.tenant_id, + command_id=command_id, + ) + except MailDeliveryCommandError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + if ( + result.get("source_module") != "campaigns" + or result.get("source_resource_type") != "campaign" + or result.get("source_resource_id") != campaign_id + ): + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Campaign report delivery not found", + ) + return ReportEmailResponse(result=result) diff --git a/src/govoplan_campaign/backend/routes/versions.py b/src/govoplan_campaign/backend/routes/versions.py index cd5f36e..6a9c283 100644 --- a/src/govoplan_campaign/backend/routes/versions.py +++ b/src/govoplan_campaign/backend/routes/versions.py @@ -1,7 +1,7 @@ from __future__ import annotations -from fastapi import APIRouter, Depends, HTTPException, Query, status +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status from sqlalchemy.orm import Session from govoplan_campaign.backend.schemas import ( @@ -101,6 +101,7 @@ def list_versions( def get_version_detail( campaign_id: str, version_id: str, + response: Response, session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:read")), ): @@ -113,10 +114,12 @@ def get_version_detail( campaign_id=campaign_id, version_id=version_id, ) - return CampaignVersionDetailResponse.model_validate( + result = CampaignVersionDetailResponse.model_validate( version, context=_campaign_response_context(principal), ) + response.headers["ETag"] = version.strong_etag + return result except CampaignPersistenceError as exc: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=str(exc) @@ -335,18 +338,23 @@ def update_version_detail( campaign_id: str, version_id: str, payload: CampaignVersionUpdateRequest, + response: Response, + if_match: str | None = Header(default=None, alias="If-Match"), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:update")), ): - return _update_campaign_version_detail_response( + result = _update_campaign_version_detail_response( session, principal, campaign_id, version_id, payload, + if_match=if_match, autosave=False, audit_action="campaign.version_updated", ) + response.headers["ETag"] = result.strong_etag + return result @router.post( @@ -357,18 +365,23 @@ def autosave_version( campaign_id: str, version_id: str, payload: CampaignVersionUpdateRequest, + response: Response, + if_match: str | None = Header(default=None, alias="If-Match"), session: Session = Depends(get_session), principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:update")), ): - return _update_campaign_version_detail_response( + result = _update_campaign_version_detail_response( session, principal, campaign_id, version_id, payload, + if_match=if_match, autosave=True, audit_action="campaign.version_autosaved", ) + response.headers["ETag"] = result.strong_etag + return result @router.post( diff --git a/src/govoplan_campaign/backend/schemas.py b/src/govoplan_campaign/backend/schemas.py index 2c0b14a..ecec616 100644 --- a/src/govoplan_campaign/backend/schemas.py +++ b/src/govoplan_campaign/backend/schemas.py @@ -58,6 +58,9 @@ class CampaignVersionUpdateRequest(BaseModel): source_filename: str | None = None source_base_path: str | None = None migrate_legacy_mail_settings: bool = False + base_revision: int | None = Field(default=None, ge=1) + reconciliation_kind: Literal["none", "auto_merge", "manual"] = "none" + resolved_conflict_paths: list[str] = Field(default_factory=list, max_length=100) @field_validator("editor_state") @classmethod @@ -92,6 +95,8 @@ class CampaignVersionResponse(BaseModel): id: str campaign_id: str version_number: int + edit_revision: int = 1 + strong_etag: str = "" schema_version: str source_filename: str | None = None workflow_state: str = "editing" @@ -440,11 +445,21 @@ class CampaignSendUnattemptedRequest(BaseModel): class CampaignSendJobRequest(BaseModel): model_config = ConfigDict(extra="forbid") + kind: Literal["test", "single_send", "single_resend"] + idempotency_key: str = Field(min_length=1, max_length=200) + reason: str | None = Field(default=None, max_length=2000) + context: dict[str, Any] = Field(default_factory=dict) include_warnings: bool = True - dry_run: bool = False use_rate_limit: bool = True enqueue_imap_task: bool = False + @model_validator(mode="after") + def require_resend_reason(self): + self.reason = (self.reason or "").strip() or None + if self.kind == "single_resend" and not self.reason: + raise ValueError("single_resend requires a reason") + return self + class CampaignResolveOutcomeRequest(BaseModel): model_config = ConfigDict(extra="forbid") @@ -626,6 +641,7 @@ class ReportEmailRequest(BaseModel): attach_jobs_csv: bool = False attach_report_json: bool = False dry_run: bool = False + idempotency_key: str | None = Field(default=None, min_length=1, max_length=200) @field_validator("to") @classmethod diff --git a/src/govoplan_campaign/backend/sending/jobs.py b/src/govoplan_campaign/backend/sending/jobs.py index 04f5155..ffc68c6 100644 --- a/src/govoplan_campaign/backend/sending/jobs.py +++ b/src/govoplan_campaign/backend/sending/jobs.py @@ -2,6 +2,7 @@ from __future__ import annotations import hashlib import json +from collections import Counter from dataclasses import asdict, dataclass from datetime import datetime, timezone from email import policy @@ -11,13 +12,18 @@ from typing import Any from uuid import uuid4 from sqlalchemy import func +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from govoplan_core.core.notifications import NotificationDispatchRequest, notification_dispatch_provider +from govoplan_core.audit.logging import audit_event +from govoplan_core.security.redaction import redact_secret_values from govoplan_core.settings import settings as core_settings from govoplan_campaign.backend.db.models import ( Campaign, CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, CampaignStatus, CampaignVersion, CampaignVersionWorkflowState, @@ -1305,49 +1311,139 @@ def send_single_campaign_job( tenant_id: str, campaign_id: str, job_id: str, + kind: str, + idempotency_key: str, + actor_user_id: str | None, + actor_api_key_id: str | None = None, + reason: str | None = None, + action_context: dict[str, Any] | None = None, 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.""" + """Perform one explicit, idempotent action for one immutable built message.""" 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( + normalized_kind = str(kind or "").strip() + if normalized_kind not in {"test", "single_send", "single_resend"}: + raise QueueingError( + "Single-message action kind must be test, single_send, or single_resend" + ) + clean_key = str(idempotency_key or "").strip() + if not clean_key or len(clean_key) > 200: + raise QueueingError("A bounded idempotency key is required") + clean_reason = " ".join(str(reason or "").split()) or None + if normalized_kind == "single_resend" and not clean_reason: + raise QueueingError("Single-message resend requires a reason") + if clean_reason and len(clean_reason) > 2000: + raise QueueingError("Single-message action reason is too long") + + version = _get_version_for_campaign( session, + campaign, + version_id=job.campaign_version_id, + ) + if normalized_kind == "single_send" and campaign.current_version_id != version.id: + raise QueueingError( + "An initial official send is only available for the current campaign version." + ) + recipients = _recipients_from_job(job) + recipient_manifest_sha256 = hashlib.sha256( + json.dumps( + recipients, + ensure_ascii=True, + separators=(",", ":"), + ).encode("utf-8") + ).hexdigest() + canonical_request_hash = hashlib.sha256( + json.dumps( + { + "tenant_id": tenant_id, + "campaign_id": campaign.id, + "version_id": version.id, + "job_id": job.id, + "kind": normalized_kind, + "message_sha256": job.eml_sha256, + "recipient_manifest_sha256": recipient_manifest_sha256, + "reason": clean_reason, + "context": _bounded_single_action_context(action_context), + }, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + ).hexdigest() + action, duplicate = _create_single_message_action( + session, + tenant_id=tenant_id, + campaign=campaign, version=version, job=job, - include_warnings=include_warnings, - dry_run=dry_run, + kind=normalized_kind, + idempotency_key=clean_key, + canonical_request_hash=canonical_request_hash, + reason=clean_reason, + action_context=_bounded_single_action_context(action_context), + actor_user_id=actor_user_id, + actor_api_key_id=actor_api_key_id, + recipient_manifest_sha256=recipient_manifest_sha256, + recipient_count=len(recipients), ) - 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 deliver via {job.delivery_channel_policy}: " - f"{len(context.envelope_recipients)} Mail recipient(s), " - f"{len(job.resolved_postbox_targets or [])} Postbox target(s)" - ), - ).as_dict(), - } + if duplicate: + return _single_message_action_response(action, duplicate=True) + try: + _validate_single_message_action( + version=version, + job=job, + kind=normalized_kind, + include_warnings=include_warnings, + ) + delivery_context = _send_job_delivery_context(session, job) + except Exception as exc: + _finish_single_message_action( + session, + action=action, + status="initiation_failed", + error_type=exc.__class__.__name__, + error_message=str(exc), + final_send_status=job.send_status, + ) + raise + + if normalized_kind in {"test", "single_resend"}: + return _send_single_message_direct( + session, + campaign=campaign, + version=version, + job=job, + action=action, + delivery_context=delivery_context, + use_rate_limit=use_rate_limit, + enqueue_imap_task=enqueue_imap_task, + ) + + try: + queue_action = _prepare_single_job_for_send( + session, + version=version, + job=job, + include_warnings=include_warnings, + dry_run=False, + ) + except Exception as exc: + _finish_single_message_action( + session, + action=action, + status="initiation_failed", + error_type=exc.__class__.__name__, + error_message=str(exc), + final_send_status=job.send_status, + ) + raise if queue_action == "queued": previous_status = campaign.status campaign.status = CampaignStatus.QUEUED.value @@ -1365,24 +1461,565 @@ def send_single_campaign_job( 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, + action_attempt = _start_single_message_action_attempt(session, action) + try: + result = send_campaign_job( + session, + job_id=job.id, + dry_run=False, + use_rate_limit=use_rate_limit, + enqueue_imap_task=enqueue_imap_task, + ) + except Exception as exc: + latest = ( + session.query(SendAttempt) + .filter(SendAttempt.job_id == job.id) + .order_by(SendAttempt.attempt_number.desc()) + .first() + ) + if latest is not None and latest.started_at is not None: + action.linked_send_attempt_id = latest.id + status = _single_action_status_from_job(job) + _finish_single_message_action( + session, + action=action, + attempt=action_attempt, + status=status, + error_type=exc.__class__.__name__, + error_message=str(exc), + final_send_status=job.send_status, + ) + raise + latest = ( + session.query(SendAttempt) + .filter(SendAttempt.job_id == job.id) + .order_by(SendAttempt.attempt_number.desc()) + .first() ) + if latest is not None: + action.linked_send_attempt_id = latest.id + status = _single_action_status_from_result(result.status) + _finish_single_message_action( + session, + action=action, + attempt=action_attempt, + status=status, + final_send_status=job.send_status, + ) + response = _single_message_action_response(action) + response["queue_action"] = queue_action + response["result"] = result.as_dict() + return response + + +def _bounded_single_action_context( + value: dict[str, Any] | None, +) -> dict[str, Any]: + if not isinstance(value, dict): + return {} + result: dict[str, Any] = {} + for raw_key, raw_value in list(value.items())[:20]: + key = str(raw_key).strip()[:80] + if not key: + continue + if isinstance(raw_value, (str, int, float, bool)) or raw_value is None: + result[key] = ( + str(raw_value)[:500] + if isinstance(raw_value, str) + else raw_value + ) + redacted = redact_secret_values(result) + return redacted if isinstance(redacted, dict) else {} + + +def _create_single_message_action( + session: Session, + *, + tenant_id: str, + campaign: Campaign, + version: CampaignVersion, + job: CampaignJob, + kind: str, + idempotency_key: str, + canonical_request_hash: str, + reason: str | None, + action_context: dict[str, Any], + actor_user_id: str | None, + actor_api_key_id: str | None, + recipient_manifest_sha256: str, + recipient_count: int, +) -> tuple[CampaignMessageAction, bool]: + existing = ( + session.query(CampaignMessageAction) + .filter( + CampaignMessageAction.tenant_id == tenant_id, + CampaignMessageAction.idempotency_key == idempotency_key, + ) + .first() + ) + if existing is not None: + if existing.canonical_request_hash != canonical_request_hash: + raise QueueingError( + "The idempotency key is already bound to another single-message action" + ) + _freeze_replayed_in_progress_action(session, existing) + return existing, True + + action = CampaignMessageAction( + tenant_id=tenant_id, + campaign_id=campaign.id, + campaign_version_id=version.id, + job_id=job.id, + kind=kind, + idempotency_key=idempotency_key, + canonical_request_hash=canonical_request_hash, + reason=reason, + context=action_context, + actor_user_id=actor_user_id, + actor_api_key_id=actor_api_key_id, + message_sha256=str(job.eml_sha256 or ""), + message_size_bytes=job.eml_size_bytes, + recipient_manifest_sha256=recipient_manifest_sha256, + recipient_count=recipient_count, + prior_send_status=job.send_status, + prior_attempt_count=job.attempt_count, + status="initiated", + ) + try: + with session.begin_nested(): + session.add(action) + session.flush() + except IntegrityError: + existing = ( + session.query(CampaignMessageAction) + .filter( + CampaignMessageAction.tenant_id == tenant_id, + CampaignMessageAction.idempotency_key == idempotency_key, + ) + .first() + ) + if ( + existing is None + or existing.canonical_request_hash != canonical_request_hash + ): + raise QueueingError( + "The idempotency key is already bound to another single-message action" + ) from None + _freeze_replayed_in_progress_action(session, existing) + return existing, True + session.commit() + audit_event( + session, + tenant_id=tenant_id, + user_id=actor_user_id, + api_key_id=actor_api_key_id, + action="campaign.message_action_initiated", + object_type="campaign_message_action", + object_id=action.id, + details={ + "campaign_id": campaign.id, + "campaign_version_id": version.id, + "job_id": job.id, + "kind": kind, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": recipient_manifest_sha256, + "recipient_count": recipient_count, + "prior_send_status": job.send_status, + "reason": reason, + }, + ) + session.commit() + return action, False + + +def _freeze_replayed_in_progress_action( + session: Session, + action: CampaignMessageAction, +) -> None: + if action.status != "effect_in_progress": + return + now = _utcnow() + action.status = "outcome_unknown" + action.completed_at = now + action.error_type = "OutcomeUnknown" + action.error_message = ( + "A repeated command found an unfinished provider effect. " + "Automatic resend was stopped." + ) + attempt = ( + session.query(CampaignMessageActionAttempt) + .filter(CampaignMessageActionAttempt.action_id == action.id) + .order_by(CampaignMessageActionAttempt.attempt_number.desc()) + .first() + ) + if attempt is not None: + attempt.status = "outcome_unknown" + attempt.completed_at = now + attempt.outcome_code = "replayed_in_progress" + attempt.diagnostic_summary = action.error_message + session.add(attempt) + session.add(action) + session.commit() + + +def _validate_single_message_action( + *, + version: CampaignVersion, + job: CampaignJob, + kind: str, + include_warnings: bool, +) -> None: + _ensure_version_validated_and_locked(version) + 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_sha256 or (not job.eml_local_path and not job.eml_storage_key): + raise QueueingError( + "This message has no immutable generated EML evidence. Rebuild the campaign before sending." + ) + if DeliveryChannelPolicy(job.delivery_channel_policy) != DeliveryChannelPolicy.MAIL: + raise QueueingError( + "Single-message SMTP actions currently require a Mail-only delivery policy." + ) + if kind == "single_send": + 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." + ) + if job.attempt_count > 0 or job.send_status not in { + JobSendStatus.NOT_QUEUED.value, + JobSendStatus.CANCELLED.value, + JobSendStatus.QUEUED.value, + }: + raise QueueingError( + "Single-send is only available for a message without an official delivery attempt." + ) + elif kind == "single_resend" and 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 it before resending." + ) + + +def _start_single_message_action_attempt( + session: Session, + action: CampaignMessageAction, +) -> CampaignMessageActionAttempt: + now = _utcnow() + attempt = CampaignMessageActionAttempt( + action_id=action.id, + attempt_number=1, + status="effect_in_progress", + started_at=now, + effect_started_at=now, + ) + action.status = "effect_in_progress" + action.effect_started_at = now + session.add(action) + session.add(attempt) + session.commit() + return attempt + + +def _finish_single_message_action( + session: Session, + *, + action: CampaignMessageAction, + status: str, + attempt: CampaignMessageActionAttempt | None = None, + accepted_count: int = 0, + refused_recipients: dict[str, dict[str, Any]] | None = None, + error_type: str | None = None, + error_message: str | None = None, + final_send_status: str | None = None, +) -> None: + now = _utcnow() + refusals = refused_recipients or {} + action.status = status + action.accepted_count = accepted_count + action.refused_count = len(refusals) + action.refusal_summary = dict( + Counter( + str(item.get("classification") or "unknown") + for item in refusals.values() + ) + ) + action.error_type = str(error_type)[:120] if error_type else None + action.error_message = ( + " ".join(str(error_message).split())[:500] + if error_message + else None + ) + action.final_send_status = final_send_status + action.completed_at = now + session.add(action) + if attempt is not None: + attempt.status = status + attempt.completed_at = now + attempt.accepted_count = accepted_count + attempt.refused_count = len(refusals) + attempt.outcome_code = action.error_type or status + attempt.diagnostic_summary = action.error_message + session.add(attempt) + session.commit() + audit_event( + session, + tenant_id=action.tenant_id, + user_id=action.actor_user_id, + api_key_id=action.actor_api_key_id, + action="campaign.message_action_completed", + object_type="campaign_message_action", + object_id=action.id, + details={ + "campaign_id": action.campaign_id, + "campaign_version_id": action.campaign_version_id, + "job_id": action.job_id, + "kind": action.kind, + "status": status, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": action.recipient_manifest_sha256, + "recipient_count": action.recipient_count, + "accepted_count": accepted_count, + "refused_count": len(refusals), + "refusal_summary": action.refusal_summary, + "prior_send_status": action.prior_send_status, + "final_send_status": final_send_status, + "linked_send_attempt_id": action.linked_send_attempt_id, + "reason": action.reason, + }, + ) + session.commit() + + +def _single_action_status_from_result(status: str) -> str: + if status in DELIVERY_ACCEPTED_STATUSES or status == "already_accepted": + return "accepted" + if status == JobSendStatus.OUTCOME_UNKNOWN.value: + return "outcome_unknown" + if status == JobSendStatus.FAILED_TEMPORARY.value: + return "failed_temporary" + if status == JobSendStatus.FAILED_PERMANENT.value: + return "failed_permanent" + return status + + +def _single_action_status_from_job(job: CampaignJob) -> str: + return _single_action_status_from_result(job.send_status) + + +def _single_message_action_response( + action: CampaignMessageAction, + *, + duplicate: bool = False, +) -> dict[str, Any]: 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(), + "campaign_id": action.campaign_id, + "version_id": action.campaign_version_id, + "job_id": action.job_id, + "action_id": action.id, + "action": action.kind, + "kind": action.kind, + "status": action.status, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": action.recipient_manifest_sha256, + "recipient_count": action.recipient_count, + "prior_send_status": action.prior_send_status, + "final_send_status": action.final_send_status, + "accepted_count": action.accepted_count, + "refused_count": action.refused_count, + "refusal_summary": dict(action.refusal_summary or {}), + "reason": action.reason, + "linked_send_attempt_id": action.linked_send_attempt_id, + "created_at": action.created_at, + "effect_started_at": action.effect_started_at, + "completed_at": action.completed_at, + "duplicate": duplicate, + "result": SendJobResult( + job_id=action.job_id, + status=action.status, + attempt_number=action.prior_attempt_count, + message=action.error_message, + ).as_dict(), } +def _send_single_message_direct( + session: Session, + *, + campaign: Campaign, + version: CampaignVersion, + job: CampaignJob, + action: CampaignMessageAction, + delivery_context: _SendJobDeliveryContext, + use_rate_limit: bool, + enqueue_imap_task: bool, +) -> dict[str, Any]: + if ( + delivery_context.envelope_from is None + or not delivery_context.envelope_recipients + ): + _finish_single_message_action( + session, + action=action, + status="initiation_failed", + error_type="SmtpConfigurationError", + error_message="Mail delivery has no frozen envelope sender or recipients.", + final_send_status=job.send_status, + ) + raise SmtpConfigurationError( + "Mail delivery has no frozen envelope sender or recipients." + ) + mail_integration().wait_for_rate_limit( + key=f"tenant:{job.tenant_id}:campaign:{job.campaign_id}:single-action", + messages_per_minute=delivery_context.snapshot.delivery.rate_limit.messages_per_minute, + enabled=use_rate_limit, + ) + attempt = _start_single_message_action_attempt(session, action) + try: + result = mail_integration().send_campaign_email_bytes( + session, + tenant_id=job.tenant_id, + campaign_id=job.campaign_id, + profile_id=delivery_context.snapshot.mail_profile_id, + message_bytes=delivery_context.message_bytes, + envelope_from=delivery_context.envelope_from, + envelope_recipients=delivery_context.envelope_recipients, + from_header=_from_header_from_job(job), + expected_smtp_transport_revision=( + delivery_context.snapshot.smtp_transport_revision or "" + ), + smtp_server_id=delivery_context.snapshot.smtp_server_id, + smtp_credential_id=delivery_context.snapshot.smtp_credential_id, + ) + except SmtpSendError as exc: + if exc.outcome_unknown: + status = "outcome_unknown" + elif exc.temporary: + status = "failed_temporary" + else: + status = "failed_permanent" + _finish_single_message_action( + session, + action=action, + attempt=attempt, + status=status, + error_type=exc.__class__.__name__, + error_message=str(exc), + final_send_status=job.send_status, + ) + return _single_message_action_response(action) + except (MailProfileError, SmtpConfigurationError, SendJobError, OSError) as exc: + _finish_single_message_action( + session, + action=action, + attempt=attempt, + status="failed_permanent", + error_type=exc.__class__.__name__, + error_message=str(exc), + final_send_status=job.send_status, + ) + return _single_message_action_response(action) + except Exception: + _finish_single_message_action( + session, + action=action, + attempt=attempt, + status="outcome_unknown", + error_type="OutcomeUnknown", + error_message=( + "The provider effect started, but the delivery outcome could " + "not be established." + ), + final_send_status=job.send_status, + ) + return _single_message_action_response(action) + + refusals = dict(result.refused_recipients) + accepted_count = result.accepted_count + if accepted_count <= 0: + classifications = { + str(item.get("classification") or "unknown") + for item in refusals.values() + } + if classifications and classifications <= {"temporary"}: + status = "failed_temporary" + elif "unknown" in classifications: + status = "outcome_unknown" + else: + status = "failed_permanent" + else: + status = "accepted_with_refusals" if refusals else "accepted" + + if action.kind == "single_resend" and accepted_count > 0: + if job.send_status not in FULLY_ACCEPTED_STATUSES: + job.queue_status = JobQueueStatus.DRAFT.value + job.send_status = JobSendStatus.SMTP_ACCEPTED.value + job.sent_at = _utcnow() + job.outcome_unknown_at = None + job.last_error = ( + "Some SMTP recipients were refused during explicit resend." + if refusals + else None + ) + job.imap_status = ( + JobImapStatus.PENDING.value + if delivery_context.snapshot.delivery.imap_append_sent.enabled + else JobImapStatus.NOT_REQUESTED.value + ) + files_integration().mark_job_attachment_uses_sent(session, job) + session.add(job) + if campaign.current_version_id == version.id: + _update_campaign_after_job( + session, + job.campaign_id, + job.campaign_version_id, + ) + session.commit() + if ( + enqueue_imap_task + and _celery_enabled() + and job.imap_status == JobImapStatus.PENDING.value + ): + try: + _celery_enqueue_append_sent_job(job.id) + except Exception: + pass + _finish_single_message_action( + session, + action=action, + attempt=attempt, + status=status, + accepted_count=accepted_count, + refused_recipients=refusals, + error_type=( + None + if accepted_count > 0 + else f"Smtp{status.title().replace('_', '')}" + ), + error_message=( + None + if accepted_count > 0 + else "SMTP did not accept an envelope recipient." + ), + final_send_status=job.send_status, + ) + return _single_message_action_response(action) + + def _prepare_single_job_for_send( session: Session, *, diff --git a/src/govoplan_campaign/backend/services/job_queries.py b/src/govoplan_campaign/backend/services/job_queries.py index 5bb3d5d..909baae 100644 --- a/src/govoplan_campaign/backend/services/job_queries.py +++ b/src/govoplan_campaign/backend/services/job_queries.py @@ -1,7 +1,7 @@ from __future__ import annotations import json -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from typing import Literal from fastapi import HTTPException, Query, status @@ -13,6 +13,7 @@ from govoplan_campaign.backend.schemas import ( CampaignJobDiagnosticsResponse, ) from govoplan_core.auth import ApiPrincipal +from govoplan_core.core.postbox import PostboxDeliveryReceiptSummaryRef from govoplan_core.core.change_sequence import ( encode_sequence_watermark, max_sequence_id, @@ -30,6 +31,8 @@ from govoplan_campaign.backend.change_tracking import ( from govoplan_campaign.backend.db.models import ( Campaign, CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, CampaignVersion, ImapAppendAttempt, JobImapStatus, @@ -128,7 +131,12 @@ def _job_attempts_payload( send_attempts: list[SendAttempt], imap_attempts: list[ImapAppendAttempt], postbox_attempts: Sequence[PostboxDeliveryAttempt] = (), + message_actions: Sequence[CampaignMessageAction] = (), + message_action_attempts: Sequence[CampaignMessageActionAttempt] = (), *, + postbox_receipts: ( + Mapping[str, PostboxDeliveryReceiptSummaryRef] | None + ) = None, include_diagnostics: bool = False, ) -> dict[str, list[dict[str, object]]]: smtp_payloads: list[dict[str, object]] = [] @@ -186,11 +194,109 @@ def _job_attempts_payload( payload["evidence"] = attempt.evidence or {} payload["error_type"] = attempt.error_type payload["error_message"] = attempt.error_message + if attempt.provider_delivery_id: + receipt_summary = ( + postbox_receipts.get(attempt.provider_delivery_id) + if postbox_receipts is not None + else None + ) + payload["receipt_summary_status"] = ( + "available" + if receipt_summary is not None + else "not_found" + if postbox_receipts is not None + else "unavailable" + ) + if receipt_summary is not None: + payload["receipt_summary"] = _postbox_receipt_summary_payload( + receipt_summary + ) postbox_payloads.append(payload) + action_attempts_by_action = { + attempt.action_id: attempt + for attempt in message_action_attempts + } + message_action_payloads: list[dict[str, object]] = [] + for action in message_actions: + action_attempt = action_attempts_by_action.get(action.id) + payload = { + "id": action.id, + "kind": action.kind, + "status": action.status, + "reason": action.reason, + "actor_user_id": action.actor_user_id, + "actor_api_key_id": action.actor_api_key_id, + "campaign_version_id": action.campaign_version_id, + "message_sha256": action.message_sha256, + "recipient_manifest_sha256": action.recipient_manifest_sha256, + "recipient_count": action.recipient_count, + "prior_send_status": action.prior_send_status, + "final_send_status": action.final_send_status, + "accepted_count": action.accepted_count, + "refused_count": action.refused_count, + "refusal_summary": action.refusal_summary or {}, + "linked_send_attempt_id": action.linked_send_attempt_id, + "created_at": action.created_at, + "effect_started_at": action.effect_started_at, + "completed_at": action.completed_at, + "attempt": ( + { + "id": action_attempt.id, + "status": action_attempt.status, + "started_at": action_attempt.started_at, + "effect_started_at": action_attempt.effect_started_at, + "completed_at": action_attempt.completed_at, + "accepted_count": action_attempt.accepted_count, + "refused_count": action_attempt.refused_count, + } + if action_attempt is not None + else None + ), + } + if include_diagnostics: + payload["idempotency_key"] = action.idempotency_key + payload["canonical_request_hash"] = action.canonical_request_hash + payload["context"] = action.context or {} + payload["error_type"] = action.error_type + payload["error_message"] = action.error_message + if action_attempt is not None: + payload["attempt"] = { + **dict(payload["attempt"] or {}), + "outcome_code": action_attempt.outcome_code, + "diagnostic_summary": action_attempt.diagnostic_summary, + } + message_action_payloads.append(payload) return { "smtp": smtp_payloads, "imap": imap_payloads, "postbox": postbox_payloads, + "message_actions": message_action_payloads, + } + + +def _postbox_receipt_summary_payload( + summary: PostboxDeliveryReceiptSummaryRef, +) -> dict[str, object]: + return { + "delivery_id": summary.delivery_id, + "message_id": summary.message_id, + "postbox_id": summary.postbox_id, + "delivery_status": summary.delivery_status, + "accepted_at": summary.accepted_at, + "current_holder_count": summary.current_holder_count, + "currently_readable": summary.currently_readable, + "message_count": summary.message_count, + "routed_message_count": summary.routed_message_count, + "readable_message_count": summary.readable_message_count, + "read_receipt_count": summary.read_receipt_count, + "acknowledged_receipt_count": summary.acknowledged_receipt_count, + "withdrawn_message_count": summary.withdrawn_message_count, + "expired_message_count": summary.expired_message_count, + "first_read_at": summary.first_read_at, + "last_read_at": summary.last_read_at, + "first_acknowledged_at": summary.first_acknowledged_at, + "last_acknowledged_at": summary.last_acknowledged_at, + "route_status_counts": dict(summary.route_status_counts), } @@ -199,6 +305,12 @@ def _job_diagnostics_payload( send_attempts: list[SendAttempt], imap_attempts: list[ImapAppendAttempt], postbox_attempts: Sequence[PostboxDeliveryAttempt] = (), + message_actions: Sequence[CampaignMessageAction] = (), + message_action_attempts: Sequence[CampaignMessageActionAttempt] = (), + *, + postbox_receipts: ( + Mapping[str, PostboxDeliveryReceiptSummaryRef] | None + ) = None, ) -> CampaignJobDiagnosticsResponse: return CampaignJobDiagnosticsResponse( job_id=job.id, @@ -221,6 +333,9 @@ def _job_diagnostics_payload( send_attempts, imap_attempts, postbox_attempts, + message_actions, + message_action_attempts, + postbox_receipts=postbox_receipts, include_diagnostics=True, ), ) diff --git a/tests/test_campaign_audit_atomicity.py b/tests/test_campaign_audit_atomicity.py index f836557..7bff57b 100644 --- a/tests/test_campaign_audit_atomicity.py +++ b/tests/test_campaign_audit_atomicity.py @@ -10,6 +10,7 @@ from govoplan_campaign.backend import route_support from govoplan_campaign.backend.routes import campaigns as campaign_routes from govoplan_campaign.backend.routes import versions as version_routes from govoplan_campaign.backend.schemas import CampaignUpdateRequest, CampaignVersionUpdateRequest +from govoplan_core.core.concurrency import strong_resource_etag def _principal() -> SimpleNamespace: @@ -28,6 +29,7 @@ def test_version_update_rolls_back_when_its_audit_record_cannot_be_written() -> raw_json={}, current_flow="manual", current_step="recipients", + edit_revision=1, ) def mutate(*_args, **kwargs): @@ -47,7 +49,15 @@ def test_version_update_rolls_back_when_its_audit_record_cannot_be_written() -> principal, # type: ignore[arg-type] "campaign-1", "version-1", - CampaignVersionUpdateRequest(current_step="recipients"), + CampaignVersionUpdateRequest( + current_step="recipients", + base_revision=1, + ), + if_match=strong_resource_etag( + "campaign_version", + "version-1", + 1, + ), autosave=True, audit_action="campaign.version_autosaved", ) diff --git a/tests/test_campaign_optimistic_concurrency.py b/tests/test_campaign_optimistic_concurrency.py new file mode 100644 index 0000000..7353ec1 --- /dev/null +++ b/tests/test_campaign_optimistic_concurrency.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from sqlalchemy import Column, String, Table, create_engine +from sqlalchemy.orm import Session, sessionmaker + +from govoplan_campaign.backend.db.models import ( + Campaign, + CampaignIssue, + CampaignVersion, +) +from govoplan_campaign.backend.persistence.versions import update_campaign_version +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.core.concurrency import RevisionConflictError +from govoplan_core.db.base import Base + + +class CampaignOptimisticConcurrencyTests(unittest.TestCase): + def setUp(self) -> None: + self.temp_dir = tempfile.TemporaryDirectory() + database_path = Path(self.temp_dir.name) / "campaign.db" + self.engine = create_engine(f"sqlite+pysqlite:///{database_path}") + access_users = Base.metadata.tables.get("access_users") + if access_users is None: + access_users = Table( + "access_users", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + access_groups = Base.metadata.tables.get("access_groups") + if access_groups is None: + access_groups = Table( + "access_groups", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + Base.metadata.create_all( + self.engine, + tables=[ + access_users, + access_groups, + ChangeSequenceEntry.__table__, + Campaign.__table__, + CampaignVersion.__table__, + CampaignIssue.__table__, + ], + ) + self.SessionLocal = sessionmaker( + bind=self.engine, + class_=Session, + expire_on_commit=False, + ) + with self.SessionLocal() as session: + campaign = Campaign( + id="campaign-1", + tenant_id="tenant-1", + external_id="C-1", + name="Campaign", + current_version_id="version-1", + ) + version = CampaignVersion( + id="version-1", + campaign_id=campaign.id, + version_number=1, + raw_json={ + "version": "1.0", + "campaign": { + "id": "C-1", + "name": "Campaign", + "description": "Base", + }, + }, + ) + session.add_all((campaign, version)) + session.commit() + self.addCleanup(self.engine.dispose) + self.addCleanup(self.temp_dir.cleanup) + + @staticmethod + def _update( + session: Session, + *, + expected_revision: int, + name: str, + ) -> CampaignVersion: + with ( + patch( + "govoplan_campaign.backend.persistence.versions._updated_runtime_json", + side_effect=lambda _session, **kwargs: kwargs["raw_json"], + ), + patch( + "govoplan_campaign.backend.persistence.versions._write_campaign_snapshot" + ), + ): + return update_campaign_version( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + version_id="version-1", + raw_json={ + "version": "1.0", + "campaign": { + "id": "C-1", + "name": name, + "description": "Base", + }, + }, + expected_revision=expected_revision, + ) + + def test_only_one_of_two_writers_can_commit_the_same_revision(self) -> None: + first = self.SessionLocal() + second = self.SessionLocal() + self.addCleanup(first.close) + self.addCleanup(second.close) + first.get(CampaignVersion, "version-1") + second.get(CampaignVersion, "version-1") + + saved = self._update( + first, + expected_revision=1, + name="First writer", + ) + self.assertEqual(saved.edit_revision, 2) + + with self.assertRaises(RevisionConflictError) as raised: + self._update( + second, + expected_revision=1, + name="Second writer", + ) + self.assertEqual(raised.exception.current_revision, 2) + self.assertEqual(raised.exception.submitted_base_revision, 1) + + with self.SessionLocal() as verification: + current = verification.get(CampaignVersion, "version-1") + assert current is not None + self.assertEqual(current.raw_json["campaign"]["name"], "First writer") + self.assertEqual(current.edit_revision, 2) + + def test_stale_revision_is_rejected_before_mutation(self) -> None: + with self.SessionLocal() as first: + self._update( + first, + expected_revision=1, + name="First writer", + ) + with self.SessionLocal() as stale: + with self.assertRaises(RevisionConflictError): + self._update( + stale, + expected_revision=1, + name="Stale writer", + ) + stale.rollback() + + with self.SessionLocal() as verification: + current = verification.get(CampaignVersion, "version-1") + assert current is not None + self.assertEqual(current.raw_json["campaign"]["name"], "First writer") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_campaign_reports.py b/tests/test_campaign_reports.py index e3d8e25..2956f89 100644 --- a/tests/test_campaign_reports.py +++ b/tests/test_campaign_reports.py @@ -5,7 +5,9 @@ from types import SimpleNamespace import unittest from unittest.mock import patch +from govoplan_core.core.postbox import PostboxDeliveryReceiptSummaryRef from govoplan_campaign.backend.reports.campaigns import ( + _campaign_postbox_receipt_summary, _job_evidence_row, _latest_by_job_id, _load_delivery_info, @@ -161,6 +163,64 @@ def test_aggregate_report_omits_recipient_level_failures_by_default() -> None: assert payload.call_args.kwargs["include_recent_failures"] is False +def test_postbox_receipt_report_aggregates_provider_state_without_identities() -> None: + class _Query: + def join(self, *args, **kwargs): + return self + + def filter(self, *args, **kwargs): + return self + + def all(self): + return [("delivery-1",), ("delivery-2",), ("delivery-1",)] + + integration = SimpleNamespace( + receipt_evidence_available=True, + delivery_receipt_summaries=lambda session, **kwargs: { + "delivery-1": PostboxDeliveryReceiptSummaryRef( + delivery_id="delivery-1", + message_id="message-1", + postbox_id="postbox-1", + delivery_status="accepted", + accepted_at=_dt(), + currently_readable=True, + read_receipt_count=2, + acknowledged_receipt_count=1, + routed_message_count=1, + ), + "delivery-2": PostboxDeliveryReceiptSummaryRef( + delivery_id="delivery-2", + message_id="message-2", + postbox_id="postbox-2", + delivery_status="accepted_vacant", + accepted_at=_dt(), + currently_readable=False, + expired_message_count=1, + ), + }, + ) + session = SimpleNamespace(query=lambda *args: _Query()) + with patch( + "govoplan_campaign.backend.reports.campaigns.postbox_integration", + return_value=integration, + ): + report = _campaign_postbox_receipt_summary( + session, # type: ignore[arg-type] + tenant_id="tenant-1", + campaign_id="campaign-1", + version=SimpleNamespace(id="version-1"), + ) + + assert report["delivery_count"] == 2 + assert report["currently_readable_delivery_count"] == 1 + assert report["read_delivery_count"] == 1 + assert report["acknowledged_delivery_count"] == 1 + assert report["routed_message_count"] == 1 + assert report["expired_message_count"] == 1 + assert "account_id" not in repr(report) + assert "identity_id" not in repr(report) + + class CampaignReportProjectionTests(unittest.TestCase): def test_evidence_projection(self) -> None: test_job_evidence_row_contains_transport_and_message_evidence() diff --git a/tests/test_postbox_integration.py b/tests/test_postbox_integration.py index 4e1d14d..f69f770 100644 --- a/tests/test_postbox_integration.py +++ b/tests/test_postbox_integration.py @@ -1,11 +1,13 @@ from __future__ import annotations import unittest +from datetime import UTC, datetime from govoplan_core.core.postbox import ( PostboxDeliveryCatalogRef, PostboxDirectoryEntryRef, PostboxDeliveryRequest, + PostboxDeliveryReceiptSummaryRef, PostboxDeliveryResult, PostboxTargetRef, ) @@ -55,11 +57,39 @@ class _PostboxDelivery: classification="internal", ) + def link_evidence(self, session, *, tenant_id, message_id, attachment): + del session, tenant_id, message_id, attachment + return None + + def delivery_receipt_summaries( + self, + session, + *, + tenant_id, + producer_module, + delivery_ids, + ): + del session + assert tenant_id == "tenant-1" + assert producer_module == "campaigns" + return { + delivery_id: PostboxDeliveryReceiptSummaryRef( + delivery_id=delivery_id, + message_id="message-1", + postbox_id="postbox-1", + delivery_status="accepted", + accepted_at=datetime(2026, 7, 30, tzinfo=UTC), + currently_readable=True, + read_receipt_count=1, + ) + for delivery_id in delivery_ids + } + class PostboxCampaignIntegrationTests(unittest.TestCase): def test_optional_delivery_boundary_is_typed_and_explicit(self) -> None: delegate = _PostboxDelivery() - integration = PostboxCampaignIntegration(delegate, delegate) + integration = PostboxCampaignIntegration(delegate, delegate, delegate) request = PostboxDeliveryRequest( tenant_id="tenant-1", target=PostboxTargetRef(postbox_id="postbox-1"), @@ -73,8 +103,15 @@ class PostboxCampaignIntegrationTests(unittest.TestCase): result = integration.deliver(object(), request) self.assertTrue(integration.available) + self.assertTrue(integration.receipt_evidence_available) self.assertEqual("delivery-1", result.delivery_id) self.assertEqual(request, delegate.requests[0][1]) + summaries = integration.delivery_receipt_summaries( + object(), + tenant_id="tenant-1", + delivery_ids=["delivery-1"], + ) + self.assertEqual(1, summaries["delivery-1"].read_receipt_count) def test_missing_postbox_is_reported_without_importing_module_code( self, @@ -91,6 +128,15 @@ class PostboxCampaignIntegrationTests(unittest.TestCase): ) self.assertFalse(integration.available) + self.assertFalse(integration.receipt_evidence_available) + self.assertEqual( + {}, + integration.delivery_receipt_summaries( + object(), + tenant_id="tenant-1", + delivery_ids=["delivery-1"], + ), + ) with self.assertRaises(PostboxDeliveryUnavailable): integration.deliver(object(), request) diff --git a/tests/test_report_email_security.py b/tests/test_report_email_security.py index d1b56b2..73b9dc5 100644 --- a/tests/test_report_email_security.py +++ b/tests/test_report_email_security.py @@ -2,6 +2,7 @@ from __future__ import annotations from types import SimpleNamespace from unittest.mock import patch +from email.message import EmailMessage import pytest from fastapi import HTTPException @@ -183,3 +184,91 @@ def test_report_send_fails_closed_until_durable_mail_outbox_exists() -> None: ) generate_report.assert_not_called() + + +def test_live_report_submits_exact_message_to_durable_mail_command() -> None: + campaign = SimpleNamespace( + id="campaign-1", + tenant_id="tenant-1", + name="Campaign", + external_id="campaign-1", + ) + version = SimpleNamespace( + id="version-1", + campaign_id="campaign-1", + raw_json={"server": {"mail_profile_id": "profile-1"}}, + execution_snapshot={"snapshot_version": "7"}, + ) + config = SimpleNamespace( + server=SimpleNamespace( + profile_capabilities=SimpleNamespace(smtp_available=True) + ), + ) + snapshot = SimpleNamespace( + mail_profile_id="profile-1", + smtp_server_id="smtp-1", + smtp_credential_id="credential-1", + smtp_transport_revision="revision-1", + ) + message = EmailMessage() + message["From"] = "Sender " + message["To"] = "recipient@example.test" + message["Subject"] = "Report" + message.set_content("content") + mail = SimpleNamespace( + durable_delivery_available=True, + submit_delivery_command=lambda _session, **_kwargs: { + "id": "command-1", + "status": "pending", + "duplicate": False, + }, + ) + + class Session: + def get(self, _model, _id): + return campaign + + with ( + patch( + "govoplan_campaign.backend.reports.emailing._selected_version", + return_value=version, + ), + patch( + "govoplan_campaign.backend.reports.emailing._load_config", + return_value=config, + ), + patch( + "govoplan_campaign.backend.reports.emailing.ensure_execution_snapshot", + return_value=snapshot, + ), + patch( + "govoplan_campaign.backend.reports.emailing.generate_campaign_report", + return_value={}, + ), + patch( + "govoplan_campaign.backend.reports.emailing.build_report_message", + return_value=message, + ), + patch( + "govoplan_campaign.backend.reports.emailing._effective_from", + return_value=("sender@example.test", "Sender"), + ), + patch( + "govoplan_campaign.backend.reports.emailing.mail_integration", + return_value=mail, + ), + ): + result = send_campaign_report_email( + Session(), # type: ignore[arg-type] + tenant_id="tenant-1", + campaign_id="campaign-1", + to=["recipient@example.test"], + idempotency_key="request-1", + created_by_user_id="user-1", + ) + + assert result.command_id == "command-1" + assert result.delivery_status == "pending" + assert result.sent is False + assert result.audit_dict()["recipient_count"] == 1 + assert "recipient@example.test" not in repr(result.audit_dict()) diff --git a/tests/test_response_security.py b/tests/test_response_security.py index 4c36ade..564b655 100644 --- a/tests/test_response_security.py +++ b/tests/test_response_security.py @@ -5,6 +5,7 @@ import unittest from datetime import UTC, datetime from types import SimpleNamespace +from govoplan_core.core.postbox import PostboxDeliveryReceiptSummaryRef from govoplan_campaign.backend.response_security import public_campaign_payload from govoplan_campaign.backend.services.job_queries import ( _job_attempts_payload, @@ -88,6 +89,30 @@ def _imap_attempt() -> SimpleNamespace: ) +def _postbox_attempt() -> SimpleNamespace: + return SimpleNamespace( + id="postbox-attempt-1", + target_index=0, + attempt_number=1, + status="accepted", + postbox_id="postbox-1", + address="intake@postbox", + holder_count=1, + vacant=False, + duplicate=False, + target_snapshot={"address": "intake@postbox"}, + started_at=_now(), + finished_at=_now(), + error_code=None, + idempotency_key="private-idempotency-key", + provider_delivery_id="delivery-1", + provider_message_id="message-1", + evidence={"target_snapshot": {"address": "intake@postbox"}}, + error_type=None, + error_message=None, + ) + + def test_public_payload_recursively_removes_infrastructure_locators() -> None: source = { "campaign_file": "/tmp/campaign.json", @@ -243,6 +268,33 @@ def test_operator_diagnostics_include_claim_and_storage_details() -> None: assert "provider-secret" in diagnostics["worker_claim"]["last_error"] +def test_postbox_receipt_summary_is_aggregate_and_contains_no_holder_identity() -> None: + summary = PostboxDeliveryReceiptSummaryRef( + delivery_id="delivery-1", + message_id="message-1", + postbox_id="postbox-1", + delivery_status="accepted", + accepted_at=_now(), + current_holder_count=2, + currently_readable=True, + read_receipt_count=1, + acknowledged_receipt_count=1, + ) + attempts = _job_attempts_payload( + [], + [], + [_postbox_attempt()], # type: ignore[list-item] + postbox_receipts={"delivery-1": summary}, + ) + + payload = attempts["postbox"][0] + assert payload["receipt_summary_status"] == "available" + assert payload["receipt_summary"]["currently_readable"] is True + assert payload["receipt_summary"]["read_receipt_count"] == 1 + assert "account_id" not in repr(payload["receipt_summary"]) + assert "identity_id" not in repr(payload["receipt_summary"]) + + def test_diagnostics_permission_is_operator_only_by_default() -> None: from govoplan_campaign.backend.manifest import PERMISSIONS, ROLE_TEMPLATES diff --git a/tests/test_route_registration.py b/tests/test_route_registration.py index c95207d..5b77c8f 100644 --- a/tests/test_route_registration.py +++ b/tests/test_route_registration.py @@ -38,7 +38,7 @@ def test_campaign_router_composes_every_workflow_operation_once() -> None: actual = _operation_keys(router) assert actual == expected - assert len(actual) == 62 + assert len(actual) == 64 assert not [operation for operation, count in Counter(actual).items() if count > 1] diff --git a/tests/test_single_message_actions.py b/tests/test_single_message_actions.py new file mode 100644 index 0000000..465ed01 --- /dev/null +++ b/tests/test_single_message_actions.py @@ -0,0 +1,417 @@ +from __future__ import annotations + +import unittest +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from sqlalchemy import Column, String, Table, create_engine +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool +from pydantic import ValidationError + +from govoplan_core.core.change_sequence import ChangeSequenceEntry +from govoplan_core.db.base import Base +from govoplan_campaign.backend.db.models import ( + Campaign, + CampaignJob, + CampaignMessageAction, + CampaignMessageActionAttempt, + CampaignVersion, + JobBuildStatus, + JobQueueStatus, + JobSendStatus, + JobValidationStatus, + SendAttempt, +) +from govoplan_campaign.backend.schemas import CampaignSendJobRequest +from govoplan_campaign.backend.sending.jobs import ( + QueueingError, + SendJobResult, + send_single_campaign_job, +) +from govoplan_campaign.backend.integrations import SmtpSendError + + +class _Mail: + def __init__(self, result: object | None = None, error: Exception | None = None): + self.result = result or SimpleNamespace( + accepted_count=1, + refused_recipients={}, + ) + self.error = error + self.send = Mock() + + def wait_for_rate_limit(self, **_kwargs) -> None: + return None + + def send_campaign_email_bytes(self, *_args, **_kwargs): + self.send(*_args, **_kwargs) + if self.error is not None: + raise self.error + return self.result + + +class CampaignSingleMessageActionTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine( + "sqlite+pysqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + access_users = Base.metadata.tables.get("access_users") + if access_users is None: + access_users = Table( + "access_users", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + access_groups = Base.metadata.tables.get("access_groups") + if access_groups is None: + access_groups = Table( + "access_groups", + Base.metadata, + Column("id", String(36), primary_key=True), + ) + Base.metadata.create_all( + self.engine, + tables=[ + access_users, + access_groups, + ChangeSequenceEntry.__table__, + Campaign.__table__, + CampaignVersion.__table__, + CampaignJob.__table__, + SendAttempt.__table__, + CampaignMessageAction.__table__, + CampaignMessageActionAttempt.__table__, + ], + ) + self.SessionLocal = sessionmaker( + bind=self.engine, + class_=Session, + expire_on_commit=False, + ) + with self.SessionLocal() as session: + campaign = Campaign( + id="campaign-1", + tenant_id="tenant-1", + external_id="campaign-1", + name="Campaign", + current_version_id="version-1", + ) + version = CampaignVersion( + id="version-1", + campaign_id=campaign.id, + version_number=1, + raw_json={}, + validation_summary={"ok": True}, + build_summary={"build_token": "build-1"}, + ) + job = CampaignJob( + id="job-1", + tenant_id="tenant-1", + campaign_id=campaign.id, + campaign_version_id=version.id, + entry_index=1, + recipient_email="recipient@example.test", + eml_local_path="/tmp/message.eml", + eml_sha256="a" * 64, + eml_size_bytes=42, + build_status=JobBuildStatus.BUILT.value, + validation_status=JobValidationStatus.READY.value, + queue_status=JobQueueStatus.DRAFT.value, + send_status=JobSendStatus.NOT_QUEUED.value, + delivery_channel_policy="mail", + resolved_recipients={ + "from": {"email": "sender@example.test"}, + "to": [{"email": "recipient@example.test"}], + }, + ) + session.add_all((campaign, version, job)) + session.commit() + self.addCleanup(self.engine.dispose) + + @staticmethod + def _delivery_context(): + return SimpleNamespace( + snapshot=SimpleNamespace( + mail_profile_id="profile-1", + smtp_server_id=None, + smtp_credential_id=None, + smtp_transport_revision="revision-1", + delivery=SimpleNamespace( + rate_limit=SimpleNamespace(messages_per_minute=60), + imap_append_sent=SimpleNamespace(enabled=False), + ), + ), + message_bytes=b"Subject: test\r\n\r\ncontent", + envelope_from="sender@example.test", + envelope_recipients=["recipient@example.test"], + ) + + def _run( + self, + session: Session, + *, + kind: str, + key: str, + mail: _Mail, + reason: str | None = None, + ): + with ( + patch( + "govoplan_campaign.backend.sending.jobs._validate_single_message_action" + ), + patch( + "govoplan_campaign.backend.sending.jobs._send_job_delivery_context", + return_value=self._delivery_context(), + ), + patch( + "govoplan_campaign.backend.sending.jobs.mail_integration", + return_value=mail, + ), + patch( + "govoplan_campaign.backend.sending.jobs.audit_event" + ), + ): + return send_single_campaign_job( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + job_id="job-1", + kind=kind, + idempotency_key=key, + actor_user_id=None, + reason=reason, + ) + + def test_test_action_is_idempotent_and_leaves_official_state_unchanged(self) -> None: + mail = _Mail() + with self.SessionLocal() as session: + job = session.get(CampaignJob, "job-1") + assert job is not None + job.send_status = JobSendStatus.FAILED_TEMPORARY.value + job.attempt_count = 2 + session.commit() + + first = self._run( + session, + kind="test", + key="test-1", + mail=mail, + ) + repeated = self._run( + session, + kind="test", + key="test-1", + mail=mail, + ) + session.refresh(job) + + self.assertEqual(first["status"], "accepted") + self.assertTrue(repeated["duplicate"]) + self.assertEqual(job.send_status, JobSendStatus.FAILED_TEMPORARY.value) + self.assertEqual(job.attempt_count, 2) + self.assertEqual(mail.send.call_count, 1) + with self.assertRaisesRegex(QueueingError, "idempotency key"): + self._run( + session, + kind="single_resend", + key="test-1", + reason="Different command", + mail=mail, + ) + + def test_failed_resend_preserves_prior_success(self) -> None: + mail = _Mail( + error=SmtpSendError( + "Temporary delivery failure", + temporary=True, + ) + ) + with self.SessionLocal() as session: + job = session.get(CampaignJob, "job-1") + assert job is not None + job.send_status = JobSendStatus.SMTP_ACCEPTED.value + job.sent_at = job.created_at + session.commit() + + result = self._run( + session, + kind="single_resend", + key="resend-failed", + reason="Recipient requested another copy", + mail=mail, + ) + session.refresh(job) + + self.assertEqual(result["status"], "failed_temporary") + self.assertEqual(job.send_status, JobSendStatus.SMTP_ACCEPTED.value) + self.assertIsNotNone(job.sent_at) + + def test_successful_resend_completes_previously_failed_message(self) -> None: + mail = _Mail() + with self.SessionLocal() as session: + job = session.get(CampaignJob, "job-1") + assert job is not None + job.send_status = JobSendStatus.FAILED_PERMANENT.value + job.last_error = "old failure" + session.commit() + with ( + patch( + "govoplan_campaign.backend.sending.jobs._validate_single_message_action" + ), + patch( + "govoplan_campaign.backend.sending.jobs._send_job_delivery_context", + return_value=self._delivery_context(), + ), + patch( + "govoplan_campaign.backend.sending.jobs.mail_integration", + return_value=mail, + ), + patch( + "govoplan_campaign.backend.sending.jobs.files_integration" + ) as files, + patch( + "govoplan_campaign.backend.sending.jobs._update_campaign_after_job" + ), + patch( + "govoplan_campaign.backend.sending.jobs.audit_event" + ), + ): + result = send_single_campaign_job( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + job_id="job-1", + kind="single_resend", + idempotency_key="resend-success", + actor_user_id=None, + reason="Approved operator resend", + ) + session.refresh(job) + + self.assertEqual(result["status"], "accepted") + self.assertEqual(job.send_status, JobSendStatus.SMTP_ACCEPTED.value) + self.assertIsNotNone(job.sent_at) + files.return_value.mark_job_attachment_uses_sent.assert_called_once() + + def test_unknown_resend_is_frozen_and_replay_does_not_send_again(self) -> None: + mail = _Mail( + error=SmtpSendError( + "Unknown delivery outcome", + outcome_unknown=True, + ) + ) + with self.SessionLocal() as session: + first = self._run( + session, + kind="single_resend", + key="resend-unknown", + reason="Approved operator resend", + mail=mail, + ) + repeated = self._run( + session, + kind="single_resend", + key="resend-unknown", + reason="Approved operator resend", + mail=mail, + ) + + self.assertEqual(first["status"], "outcome_unknown") + self.assertTrue(repeated["duplicate"]) + self.assertEqual(mail.send.call_count, 1) + + def test_initial_send_links_the_canonical_attempt_and_replay_is_inert(self) -> None: + with self.SessionLocal() as session: + job = session.get(CampaignJob, "job-1") + assert job is not None + + def canonical_send(_session: Session, **_kwargs): + attempt = SendAttempt( + job_id=job.id, + attempt_number=1, + status=JobSendStatus.SMTP_ACCEPTED.value, + started_at=job.created_at, + finished_at=job.created_at, + ) + job.send_status = JobSendStatus.SMTP_ACCEPTED.value + job.queue_status = JobQueueStatus.DRAFT.value + job.attempt_count = 1 + _session.add_all((attempt, job)) + _session.commit() + return SendJobResult( + job_id=job.id, + status=JobSendStatus.SMTP_ACCEPTED.value, + attempt_number=1, + ) + + with ( + patch( + "govoplan_campaign.backend.sending.jobs._validate_single_message_action" + ), + patch( + "govoplan_campaign.backend.sending.jobs._send_job_delivery_context", + return_value=self._delivery_context(), + ), + patch( + "govoplan_campaign.backend.sending.jobs._prepare_single_job_for_send", + return_value="queued", + ), + patch( + "govoplan_campaign.backend.sending.jobs.send_campaign_job", + side_effect=canonical_send, + ) as send, + patch( + "govoplan_campaign.backend.sending.jobs._emit_campaign_status_notification" + ), + patch( + "govoplan_campaign.backend.sending.jobs.audit_event" + ), + ): + first = send_single_campaign_job( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + job_id="job-1", + kind="single_send", + idempotency_key="initial-1", + actor_user_id=None, + ) + repeated = send_single_campaign_job( + session, + tenant_id="tenant-1", + campaign_id="campaign-1", + job_id="job-1", + kind="single_send", + idempotency_key="initial-1", + actor_user_id=None, + ) + + self.assertEqual(first["status"], "accepted") + self.assertIsNotNone(first["linked_send_attempt_id"]) + self.assertTrue(repeated["duplicate"]) + self.assertEqual(send.call_count, 1) + + def test_request_requires_explicit_kind_key_and_resend_reason(self) -> None: + with self.assertRaises(ValidationError): + CampaignSendJobRequest.model_validate({}) + with self.assertRaises(ValidationError): + CampaignSendJobRequest.model_validate( + { + "kind": "single_resend", + "idempotency_key": "resend-1", + } + ) + request = CampaignSendJobRequest.model_validate( + { + "kind": "test", + "idempotency_key": "test-1", + } + ) + self.assertEqual(request.kind, "test") + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/package.json b/webui/package.json index 033a756..9192962 100644 --- a/webui/package.json +++ b/webui/package.json @@ -17,7 +17,7 @@ "read-excel-file": "9.2.0" }, "peerDependencies": { - "@govoplan/core-webui": "^0.1.12", + "@govoplan/core-webui": "^0.1.14", "lucide-react": "^1.23.0", "react": "^19.0.0", "react-dom": "^19.0.0", diff --git a/webui/src/api/campaigns.ts b/webui/src/api/campaigns.ts index 2cc8055..ba305df 100644 --- a/webui/src/api/campaigns.ts +++ b/webui/src/api/campaigns.ts @@ -60,6 +60,8 @@ export type CampaignVersionListItem = { id: string; campaign_id: string; version_number: number; + edit_revision: number; + strong_etag: string; schema_version?: string; source_filename?: string | null; source_base_path?: string | null; @@ -274,6 +276,9 @@ export type CampaignVersionUpdatePayload = { source_filename?: string | null; source_base_path?: string | null; migrate_legacy_mail_settings?: boolean; + base_revision?: number; + reconciliation_kind?: "none" | "auto_merge" | "manual"; + resolved_conflict_paths?: string[]; }; export type CampaignPartialValidationPayload = { @@ -330,6 +335,7 @@ export type CampaignSummary = { issues?: Record; attachments?: Record; attempts?: Record; + postbox_receipts?: Record; delivery?: Record; recent_failures?: Record[]; }; @@ -468,8 +474,11 @@ export type CampaignReviewStatePayload = { }; export type CampaignSendJobPayload = { + kind: "test" | "single_send" | "single_resend"; + idempotency_key: string; + reason?: string; + context?: Record; include_warnings?: boolean; - dry_run?: boolean; use_rate_limit?: boolean; enqueue_imap_task?: boolean; }; @@ -520,6 +529,7 @@ export type CampaignReportEmailPayload = { attach_jobs_csv?: boolean; attach_report_json?: boolean; dry_run?: boolean; + idempotency_key?: string; }; export type AggregateReportCount = { @@ -797,10 +807,12 @@ export async function updateCampaignVersion( settings: ApiSettings, campaignId: string, versionId: string, -payload: CampaignVersionUpdatePayload) +payload: CampaignVersionUpdatePayload, +ifMatch: string) : Promise { return apiFetch(settings, `/api/v1/campaigns/${campaignId}/versions/${versionId}`, { method: "PUT", + headers: { "If-Match": ifMatch }, body: JSON.stringify(payload) }); } @@ -821,10 +833,12 @@ export async function autosaveCampaignVersion( settings: ApiSettings, campaignId: string, versionId: string, -payload: CampaignVersionUpdatePayload) +payload: CampaignVersionUpdatePayload, +ifMatch: string) : Promise { return apiFetch(settings, `/api/v1/campaigns/${campaignId}/versions/${versionId}/autosave`, { method: "POST", + headers: { "If-Match": ifMatch }, body: JSON.stringify(payload) }); } @@ -1021,7 +1035,7 @@ export async function sendCampaignJob( settings: ApiSettings, campaignId: string, jobId: string, -payload: CampaignSendJobPayload = {}) +payload: CampaignSendJobPayload) : Promise> { return apiFetch>(settings, `/api/v1/campaigns/${campaignId}/jobs/${jobId}/send`, { method: "POST", diff --git a/webui/src/features/campaigns/CampaignReportPage.tsx b/webui/src/features/campaigns/CampaignReportPage.tsx index 8598e96..fb60536 100644 --- a/webui/src/features/campaigns/CampaignReportPage.tsx +++ b/webui/src/features/campaigns/CampaignReportPage.tsx @@ -102,6 +102,7 @@ export default function CampaignReportPage({ settings, campaignId }: {settings: const version = data.currentVersion; const cards = data.summary?.cards; const delivery = asRecord(data.summary?.delivery); + const postboxReceipts = asRecord(data.summary?.postbox_receipts); const rateLimit = asRecord(delivery.rate_limit); const imapPolicy = asRecord(delivery.imap_append_sent); @@ -262,8 +263,10 @@ export default function CampaignReportPage({ settings, campaignId }: {settings: attempted += 1; try { const sendResponse = await sendCampaignJob(settings, campaignId, jobId, { + kind: "single_resend", + idempotency_key: crypto.randomUUID(), + reason: "Operator requested synchronous resend from the campaign report.", include_warnings: true, - dry_run: false, use_rate_limit: true, enqueue_imap_task: false }); @@ -338,9 +341,10 @@ export default function CampaignReportPage({ settings, campaignId }: {settings: version_id: version?.id, include_jobs: false, attach_jobs_csv: attachCsv, - attach_report_json: attachJson + attach_report_json: attachJson, + idempotency_key: crypto.randomUUID() }); - setActionMessage(`Report sent to ${recipients.join(", ")}.`); + setActionMessage(`Report queued for ${recipients.join(", ")}.`); setEmailOpen(false); } catch (err) { setActionError(err instanceof Error ? err.message : String(err)); @@ -471,6 +475,30 @@ export default function CampaignReportPage({ settings, campaignId }: {settings:
i18n:govoplan-campaign.execution_snapshot.5a67f098
{delivery.execution_snapshot_hash ? i18nMessage("i18n:govoplan-campaign.value.382bcd25", { value0: String(delivery.execution_snapshot_hash).slice(0, 12) }) : "i18n:govoplan-campaign.missing.92185dc5"}
+ +
+
+
Evidence state
+
+ +
+
+
Accepted deliveries
{String(postboxReceipts.delivery_count ?? 0)}
+
Currently readable
{String(postboxReceipts.currently_readable_delivery_count ?? "—")}
+
Read
{String(postboxReceipts.read_delivery_count ?? "—")}
+
Acknowledged
{String(postboxReceipts.acknowledged_delivery_count ?? "—")}
+
Withdrawn / expired copies
{String(Number(postboxReceipts.withdrawn_message_count ?? 0) + Number(postboxReceipts.expired_message_count ?? 0))}
+
+

i18n:govoplan-campaign.these_actions_never_include_smtp_accepted_or_unr.449d0a80

@@ -563,6 +591,13 @@ export default function CampaignReportPage({ settings, campaignId }: {settings:
i18n:govoplan-campaign.attachments.6771ade6
{String(detail.job.matched_file_count ?? detail.job.attachment_count ?? 0)}
Message SHA-256
{String(detail.job.eml_sha256 ?? "—")}
+ @@ -587,6 +622,93 @@ export default function CampaignReportPage({ settings, campaignId }: {settings: } +function PostboxTargetEvidenceSection({ targets }: { targets: unknown[] }) { + const rows = targets.map(asRecord); + if (rows.length === 0) return null; + const columns: DataGridColumn>[] = [ + { + id: "target", + header: "Frozen Postbox target", + width: "minmax(240px, 1fr)", + minWidth: 220, + resizable: true, + filterable: true, + value: (row) => String(row.address ?? row.name ?? row.postbox_id ?? "—"), + render: (row) => ( + + {String(row.name ?? row.address ?? row.postbox_id ?? "—")} + {String(row.address ?? row.address_key ?? "")} + + ) + }, + { + id: "organization", + header: "Organization / function", + width: 260, + resizable: true, + filterable: true, + value: (row) => + `${String(row.organization_unit_name ?? row.organization_unit_id ?? "")} ${String(row.function_name ?? row.function_id ?? "")}`, + render: (row) => ( + + {String(row.organization_unit_name ?? row.organization_unit_id ?? "—")} + {String(row.function_name ?? row.function_id ?? "—")} + + ) + }, + { + id: "resolution", + header: "Frozen resolution", + width: 220, + resizable: true, + value: (row) => + `${String(row.mode ?? "direct")} ${String(row.context_key ?? "")} ${String(row.template_revision_id ?? "")}`, + render: (row) => ( + + {humanize(String(row.mode ?? "direct"))} + + {row.template_revision_id + ? `Template ${shortEvidenceId(String(row.template_revision_id))}` + : "Exact Postbox"} + {row.context_key ? ` · Context ${String(row.context_key)}` : ""} + + + ) + }, + { + id: "holders", + header: "Build-time holders", + width: 150, + align: "right", + sortable: true, + value: (row) => Number(row.holder_count ?? 0), + render: (row) => + row.vacant === true ? ( + + ) : ( + String(row.holder_count ?? 0) + ) + } + ]; + return ( +
+

Frozen Postbox targets

+

+ These addresses, organization/function references, and template revisions + were resolved during build and are the delivery evidence for this job. +

+ + String(row.target_id ?? row.postbox_id ?? `postbox-target-${index}`) + } + /> +
+ ); +} + type AttachmentEvidenceRow = { id: string; rule: string; @@ -718,6 +840,50 @@ function AttemptHistoryTable({ kind, rows }: {kind: "smtp" | "imap" | "postbox"; kind === "postbox" ? { id: "target", header: "Postbox", width: 220, sortable: true, filterable: true, value: (row) => String(row.address ?? asRecord(row.target).address ?? row.postbox_id ?? "—"), render: (row) => String(row.address ?? asRecord(row.target).address ?? row.postbox_id ?? "—") } : { id: "code", header: "i18n:govoplan-campaign.code.adac6937", width: 110, sortable: true, value: (row) => String(row.smtp_status_code ?? "—"), render: (row) => String(row.smtp_status_code ?? "—") }, + ...(kind === "postbox" ? [ + { + id: "readability", + header: "Current access", + width: 155, + value: (row: Record) => { + const summary = asRecord(row.receipt_summary); + return String( + row.receipt_summary_status === "available" + ? summary.currently_readable === true + ? "readable" + : "unavailable" + : row.receipt_summary_status ?? "unavailable" + ); + }, + render: (row: Record) => { + const summary = asRecord(row.receipt_summary); + const available = row.receipt_summary_status === "available"; + const readable = available && summary.currently_readable === true; + return ( + + ); + } + }, + { + id: "receipts", + header: "Read / acknowledged", + width: 170, + align: "right" as const, + value: (row: Record) => { + const summary = asRecord(row.receipt_summary); + return `${String(summary.read_receipt_count ?? 0)} / ${String(summary.acknowledged_receipt_count ?? 0)}`; + } + } + ] : []), { id: "started", header: "i18n:govoplan-campaign.started.faa9e7e7", width: 180, sortable: true, value: (row) => String(row.started_at ?? row.created_at ?? ""), render: (row) => formatDateTime(String(row.started_at ?? row.created_at ?? "")) }, { id: "finished", header: "i18n:govoplan-campaign.finished.355bcc57", width: 180, sortable: true, value: (row) => String(row.finished_at ?? row.updated_at ?? ""), render: (row) => formatDateTime(String(row.finished_at ?? row.updated_at ?? "")) }, { id: "result", header: "i18n:govoplan-campaign.result.5faa59d4", width: "minmax(240px, 1fr)", minWidth: 200, resizable: true, filterable: true, value: (row) => String(row.smtp_response ?? row.error_message ?? row.error_code ?? "—"), render: (row) => {String(row.smtp_response ?? row.error_message ?? row.error_code ?? "—")} } diff --git a/webui/src/features/campaigns/CampaignWorkspace.tsx b/webui/src/features/campaigns/CampaignWorkspace.tsx index 8cc1b29..ed1efb0 100644 --- a/webui/src/features/campaigns/CampaignWorkspace.tsx +++ b/webui/src/features/campaigns/CampaignWorkspace.tsx @@ -1,22 +1,26 @@ -import { useEffect, useState } from "react"; +import { lazy, useEffect, useState } from "react"; import { Navigate, Route, Routes, useLocation, useNavigate, useParams } from "react-router-dom"; -import { useGuardedNavigate } from "@govoplan/core-webui"; +import { + ConcurrencyConflictProvider, + useGuardedNavigate +} from "@govoplan/core-webui"; import type { ApiSettings, AuthInfo, CampaignWorkspaceSection } from "../../types"; import SectionSidebar from "../../layout/SectionSidebar"; -import CampaignOverviewPage from "./CampaignOverviewPage"; -import CampaignFieldsPage from "./CampaignFieldsPage"; -import GlobalSettingsPage from "./GlobalSettingsPage"; -import RecipientDataPage from "./RecipientDataPage"; -import TemplateDataPage from "./TemplateDataPage"; -import AttachmentsDataPage from "./AttachmentsDataPage"; -import MailSettingsPage from "./MailSettingsPage"; -import ReviewSendPage from "./ReviewSendPage"; -import CreateWizard from "./wizard/CreateWizard"; -import ReviewWizard from "./wizard/ReviewWizard"; -import SendWizard from "./wizard/SendWizard"; -import CampaignJsonView from "./CampaignJsonView"; -import CampaignReportPage from "./CampaignReportPage"; -import CampaignAuditPage from "./CampaignAuditPage"; + +const CampaignOverviewPage = lazy(() => import("./CampaignOverviewPage")); +const CampaignFieldsPage = lazy(() => import("./CampaignFieldsPage")); +const GlobalSettingsPage = lazy(() => import("./GlobalSettingsPage")); +const RecipientDataPage = lazy(() => import("./RecipientDataPage")); +const TemplateDataPage = lazy(() => import("./TemplateDataPage")); +const AttachmentsDataPage = lazy(() => import("./AttachmentsDataPage")); +const MailSettingsPage = lazy(() => import("./MailSettingsPage")); +const ReviewSendPage = lazy(() => import("./ReviewSendPage")); +const CreateWizard = lazy(() => import("./wizard/CreateWizard")); +const ReviewWizard = lazy(() => import("./wizard/ReviewWizard")); +const SendWizard = lazy(() => import("./wizard/SendWizard")); +const CampaignJsonView = lazy(() => import("./CampaignJsonView")); +const CampaignReportPage = lazy(() => import("./CampaignReportPage")); +const CampaignAuditPage = lazy(() => import("./CampaignAuditPage")); const sectionPaths: Record = { overview: "", @@ -37,7 +41,11 @@ const sectionPaths: Record = { }; export default function CampaignWorkspace({ settings, auth }: { settings: ApiSettings; auth: AuthInfo }) { - return ; + return ( + + + + ); } function CampaignWorkspaceInner({ settings, auth }: { settings: ApiSettings; auth: AuthInfo }) { diff --git a/webui/src/features/campaigns/ReviewSendPage.tsx b/webui/src/features/campaigns/ReviewSendPage.tsx index e8d2b70..351f2bd 100644 --- a/webui/src/features/campaigns/ReviewSendPage.tsx +++ b/webui/src/features/campaigns/ReviewSendPage.tsx @@ -36,7 +36,7 @@ import { type CampaignSummary } from "../../api/campaigns"; import { getMockMailboxMessage, type MockMailboxMessage } from "../../api/mail"; -import { Button, hasScope, useDeltaWatermarks, useGuardedNavigate, usePlatformUiCapability, type AuthInfo, type MailDevMailboxUiCapability } from "@govoplan/core-webui"; +import { Button, Dialog, SegmentedControl, hasScope, useDeltaWatermarks, useGuardedNavigate, usePlatformUiCapability, type AuthInfo, type MailDevMailboxUiCapability } from "@govoplan/core-webui"; import { DataGrid, type DataGridColumn, type DataGridListOption, type DataGridQueryState } from "@govoplan/core-webui"; import { DismissibleAlert, TableActionGroup } from "@govoplan/core-webui"; import { ConfirmDialog } from "@govoplan/core-webui"; @@ -159,6 +159,8 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings const [newlyReviewedRequiredKeys, setNewlyReviewedRequiredKeys] = useState>(() => new Set()); const [selectedBuiltIndex, setSelectedBuiltIndex] = useState(null); const [singleSendConfirmIndex, setSingleSendConfirmIndex] = useState(null); + const [singleMessageActionKind, setSingleMessageActionKind] = useState<"test" | "single_send" | "single_resend">("single_send"); + const [singleMessageResendReason, setSingleMessageResendReason] = useState(""); const [mockResult, setMockResult] = useState | null>(null); const [mockClearFirst, setMockClearFirst] = useState(true); const [mockAppendSent, setMockAppendSent] = useState(true); @@ -1103,8 +1105,11 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings setError(""); try { const response = await sendCampaignJob(settings, campaignId, jobId, { + kind: singleMessageActionKind, + idempotency_key: crypto.randomUUID(), + reason: singleMessageActionKind === "single_resend" ? singleMessageResendReason.trim() : undefined, + context: { surface: "campaign_review_preview" }, include_warnings: true, - dry_run: false, use_rate_limit: true, enqueue_imap_task: backgroundWorkersEnabled }); @@ -1112,18 +1117,20 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings const sendResult = asRecord(actionResult.result); const status = String(sendResult.status ?? "submitted"); const attemptCount = Number(sendResult.attempt_number ?? row.attempt_count ?? 0); - setBuiltReviewRows((current) => current.map((item) => String(item.id ?? "") === jobId ? + const finalSendStatus = String(actionResult.final_send_status ?? ""); + if (singleMessageActionKind !== "test") setBuiltReviewRows((current) => current.map((item) => String(item.id ?? "") === jobId ? { ...item, - send_status: status === "already_accepted" ? "smtp_accepted" : status, - queue_status: ["smtp_accepted", "already_accepted"].includes(status) ? "draft" : item.queue_status, + send_status: finalSendStatus || (status === "already_accepted" ? "smtp_accepted" : status), + queue_status: ["smtp_accepted", "already_accepted", "accepted", "accepted_with_refusals"].includes(finalSendStatus || status) ? "draft" : item.queue_status, imap_status: sendResult.imap_status ?? item.imap_status, attempt_count: Number.isFinite(attemptCount) ? attemptCount : item.attempt_count, last_error: sendResult.message ?? "" } : item)); - setMessage(`Single message send finished: ${humanize(status)}.`); + setMessage(`${humanize(singleMessageActionKind)} finished: ${humanize(String(actionResult.status ?? status))}.`); setSingleSendConfirmIndex(null); + setSingleMessageResendReason(""); setSelectedBuiltIndex(null); await refreshQueueStatus(true); await reload(); @@ -1146,6 +1153,7 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings setNewlyReviewedRequiredKeys(new Set()); setSelectedBuiltIndex(null); setSingleSendConfirmIndex(null); + setSingleMessageResendReason(""); setMockResult(null); setSelectedMockMessage(null); resetDeltaWatermark(); @@ -1660,7 +1668,18 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings canStartSingleMessageSend={canStartSingleMessageSend} singleMessageSendBusy={busy === "send"} onSelect={openBuiltMessageAtIndex} - onSendSingle={(targetIndex) => setSingleSendConfirmIndex(targetIndex)} + onSendSingle={(targetIndex) => { + const target = filteredBuiltReviewRows[targetIndex]; + const status = String(target?.send_status ?? "not_queued"); + setSingleMessageActionKind( + ["not_queued", "cancelled", "queued"].includes(status) + && Number(target?.attempt_count ?? 0) === 0 + ? "single_send" + : "single_resend" + ); + setSingleMessageResendReason(""); + setSingleSendConfirmIndex(targetIndex); + }} onClose={() => setSelectedBuiltIndex(null)} /> } @@ -1724,16 +1743,103 @@ export default function ReviewSendPage({ settings, auth, campaignId }: {settings onCancel={() => setCancelDeliveryConfirmOpen(false)} onConfirm={() => void runDeliveryControl("cancel")} /> - setSingleSendConfirmIndex(null)} - onConfirm={() => void sendSingleBuiltMessage()} /> + title="One-message delivery" + portal + closeDisabled={busy === "send"} + onClose={() => { + setSingleSendConfirmIndex(null); + setSingleMessageResendReason(""); + }} + footer={ + <> + + + + } + > + {singleSendConfirmRow && ( +
+ +

+ {singleMessageActionKind === "test" + ? "Transports this exact rendered message once without changing its official delivery state." + : singleMessageActionKind === "single_resend" + ? "Transports this exact message again. A successful resend can complete it; a failed resend preserves an earlier success." + : "Makes the first official delivery attempt and changes only this message after SMTP acceptance."} +

+
+
+
Recipient
+
{formatAddressList(asRecord(singleSendConfirmRow.resolved_recipients).to) || String(singleSendConfirmRow.recipient_email ?? "—")}
+
+
+
Version
+
{String(version?.version_number ?? "—")}
+
+
+
Message
+
{String(singleSendConfirmRow.eml_sha256 ?? "—").slice(0, 16)}
+
+
+
State
+
{humanize(String(singleSendConfirmRow.send_status ?? "not_queued"))}
+
+
+ {singleMessageActionKind === "single_resend" && ( +