From 057e660b1707539b6429211c0b42b686020b5a62 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 21 Jul 2026 17:13:49 +0200 Subject: [PATCH] feat: add Campaign delivery ownership and IMAP claim primitives --- .../backend/campaign/mail_profile_boundary.py | 66 ++++++++++++++++++ src/govoplan_campaign/backend/db/models.py | 14 ++++ .../3c4d5e6f8192_imap_append_claim.py | 67 +++++++++++++++++++ .../3c4d5e6f8192_v019_imap_append_claim.py | 67 +++++++++++++++++++ 4 files changed, 214 insertions(+) create mode 100644 src/govoplan_campaign/backend/campaign/mail_profile_boundary.py create mode 100644 src/govoplan_campaign/backend/migrations/dev_versions/3c4d5e6f8192_imap_append_claim.py create mode 100644 src/govoplan_campaign/backend/migrations/versions/3c4d5e6f8192_v019_imap_append_claim.py diff --git a/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py b/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py new file mode 100644 index 0000000..2da7dff --- /dev/null +++ b/src/govoplan_campaign/backend/campaign/mail_profile_boundary.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from typing import Any + + +CAMPAIGN_MAIL_SERVER_KEYS = frozenset({"mail_profile_id"}) + + +class CampaignMailProfileBoundaryError(ValueError): + """Raised when campaign JSON owns mail transport configuration. + + SMTP/IMAP endpoints and credentials are Mail-module data. Campaign JSON + may select one Mail-owned profile, but it must never copy or override that + profile's transport configuration. + """ + + +def campaign_mail_profile_id(raw_json: dict[str, Any] | None) -> str | None: + server = raw_json.get("server") if isinstance(raw_json, dict) else None + if not isinstance(server, dict): + return None + value = server.get("mail_profile_id") + if not isinstance(value, str): + return None + normalized = value.strip() + return normalized or None + + +def campaign_mail_profile_boundary_violations(raw_json: dict[str, Any] | None) -> tuple[str, ...]: + server = raw_json.get("server") if isinstance(raw_json, dict) else None + if not isinstance(server, dict): + return () + + violations = [f"/server/{key}" for key in sorted(server) if key not in CAMPAIGN_MAIL_SERVER_KEYS] + if "mail_profile_id" in server: + profile_id = server["mail_profile_id"] + if not isinstance(profile_id, str) or not profile_id.strip(): + violations.append("/server/mail_profile_id") + return tuple(violations) + + +def assert_campaign_uses_mail_profile_reference( + raw_json: dict[str, Any] | None, + *, + require_profile: bool = False, +) -> None: + violations = campaign_mail_profile_boundary_violations(raw_json) + if violations: + fields = ", ".join(violations) + raise CampaignMailProfileBoundaryError( + "Campaign JSON may only reference a Mail-module profile through " + f"server.mail_profile_id; remove campaign-local SMTP/IMAP settings ({fields}), " + "select an authorized Mail profile, and save a new campaign version." + ) + if require_profile and campaign_mail_profile_id(raw_json) is None: + raise CampaignMailProfileBoundaryError( + "Campaign delivery requires server.mail_profile_id. Select an authorized, active " + "profile from the Mail module and validate the campaign again." + ) + + +def public_campaign_mail_server(raw_json: dict[str, Any] | None) -> dict[str, str]: + """Return the complete public/persisted Campaign-to-Mail contract.""" + + profile_id = campaign_mail_profile_id(raw_json) + return {"mail_profile_id": profile_id} if profile_id else {} diff --git a/src/govoplan_campaign/backend/db/models.py b/src/govoplan_campaign/backend/db/models.py index bfb0376..4cc23e6 100644 --- a/src/govoplan_campaign/backend/db/models.py +++ b/src/govoplan_campaign/backend/db/models.py @@ -92,7 +92,9 @@ class JobSendStatus(StrEnum): class JobImapStatus(StrEnum): NOT_REQUESTED = "not_requested" PENDING = "pending" + APPENDING = "appending" APPENDED = "appended" + OUTCOME_UNKNOWN = "outcome_unknown" FAILED = "failed" SKIPPED = "skipped" @@ -213,6 +215,12 @@ class CampaignVersion(Base, TimestampMixin): campaign: Mapped[Campaign] = relationship(back_populates="versions") + @property + def mail_profile_migration_required(self) -> bool: + from govoplan_campaign.backend.campaign.mail_profile_boundary import campaign_mail_profile_boundary_violations + + return bool(campaign_mail_profile_boundary_violations(self.raw_json)) + class CampaignJob(Base, TimestampMixin): __tablename__ = "campaign_jobs" @@ -246,6 +254,8 @@ class CampaignJob(Base, TimestampMixin): claim_token: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) smtp_started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) outcome_unknown_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + imap_claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + imap_claim_token: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) resolved_recipients: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) @@ -315,10 +325,14 @@ class SendAttempt(Base, TimestampMixin): class ImapAppendAttempt(Base, TimestampMixin): __tablename__ = "imap_append_attempts" + __table_args__ = ( + UniqueConstraint("job_id", "attempt_number", name="uq_imap_append_attempts_job_attempt"), + ) id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) job_id: Mapped[str] = mapped_column(ForeignKey("campaign_jobs.id", ondelete="CASCADE"), nullable=False, index=True) attempt_number: Mapped[int] = mapped_column(Integer, nullable=False) + claim_token: Mapped[str | None] = mapped_column(String(36), nullable=True) folder: Mapped[str | None] = mapped_column(String(500)) status: Mapped[str] = mapped_column(String(50), nullable=False) error_message: Mapped[str | None] = mapped_column(Text) diff --git a/src/govoplan_campaign/backend/migrations/dev_versions/3c4d5e6f8192_imap_append_claim.py b/src/govoplan_campaign/backend/migrations/dev_versions/3c4d5e6f8192_imap_append_claim.py new file mode 100644 index 0000000..3959ac6 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/dev_versions/3c4d5e6f8192_imap_append_claim.py @@ -0,0 +1,67 @@ +"""add durable IMAP append claim and attempt idempotency + +Revision ID: 3c4d5e6f8192 +Revises: 2c3d4e5f7081 +Create Date: 2026-07-21 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "3c4d5e6f8192" +down_revision = "2c3d4e5f7081" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("campaign_jobs") as batch: + batch.add_column(sa.Column("imap_claimed_at", sa.DateTime(timezone=True), nullable=True)) + batch.add_column(sa.Column("imap_claim_token", sa.String(length=36), nullable=True)) + batch.create_index("ix_campaign_jobs_imap_claim_token", ["imap_claim_token"], unique=False) + with op.batch_alter_table("imap_append_attempts") as batch: + batch.add_column(sa.Column("claim_token", sa.String(length=36), nullable=True)) + _renumber_attempts() + with op.batch_alter_table("imap_append_attempts") as batch: + batch.create_unique_constraint( + "uq_imap_append_attempts_job_attempt", + ["job_id", "attempt_number"], + ) + + +def downgrade() -> None: + with op.batch_alter_table("imap_append_attempts") as batch: + batch.drop_constraint("uq_imap_append_attempts_job_attempt", type_="unique") + batch.drop_column("claim_token") + with op.batch_alter_table("campaign_jobs") as batch: + batch.drop_index("ix_campaign_jobs_imap_claim_token") + batch.drop_column("imap_claim_token") + batch.drop_column("imap_claimed_at") + + +def _renumber_attempts() -> None: + """Make historical attempt numbers unique per job before constraining them.""" + + bind = op.get_bind() + rows = list( + bind.execute( + sa.text( + "SELECT id, job_id FROM imap_append_attempts " + "ORDER BY job_id, created_at, id" + ) + ).mappings() + ) + per_job: dict[str, int] = {} + for row in rows: + job_id = str(row["job_id"]) + attempt_number = per_job.get(job_id, 0) + 1 + per_job[job_id] = attempt_number + bind.execute( + sa.text( + "UPDATE imap_append_attempts SET attempt_number = :attempt_number " + "WHERE id = :attempt_id" + ), + {"attempt_number": attempt_number, "attempt_id": row["id"]}, + ) diff --git a/src/govoplan_campaign/backend/migrations/versions/3c4d5e6f8192_v019_imap_append_claim.py b/src/govoplan_campaign/backend/migrations/versions/3c4d5e6f8192_v019_imap_append_claim.py new file mode 100644 index 0000000..3959ac6 --- /dev/null +++ b/src/govoplan_campaign/backend/migrations/versions/3c4d5e6f8192_v019_imap_append_claim.py @@ -0,0 +1,67 @@ +"""add durable IMAP append claim and attempt idempotency + +Revision ID: 3c4d5e6f8192 +Revises: 2c3d4e5f7081 +Create Date: 2026-07-21 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "3c4d5e6f8192" +down_revision = "2c3d4e5f7081" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.batch_alter_table("campaign_jobs") as batch: + batch.add_column(sa.Column("imap_claimed_at", sa.DateTime(timezone=True), nullable=True)) + batch.add_column(sa.Column("imap_claim_token", sa.String(length=36), nullable=True)) + batch.create_index("ix_campaign_jobs_imap_claim_token", ["imap_claim_token"], unique=False) + with op.batch_alter_table("imap_append_attempts") as batch: + batch.add_column(sa.Column("claim_token", sa.String(length=36), nullable=True)) + _renumber_attempts() + with op.batch_alter_table("imap_append_attempts") as batch: + batch.create_unique_constraint( + "uq_imap_append_attempts_job_attempt", + ["job_id", "attempt_number"], + ) + + +def downgrade() -> None: + with op.batch_alter_table("imap_append_attempts") as batch: + batch.drop_constraint("uq_imap_append_attempts_job_attempt", type_="unique") + batch.drop_column("claim_token") + with op.batch_alter_table("campaign_jobs") as batch: + batch.drop_index("ix_campaign_jobs_imap_claim_token") + batch.drop_column("imap_claim_token") + batch.drop_column("imap_claimed_at") + + +def _renumber_attempts() -> None: + """Make historical attempt numbers unique per job before constraining them.""" + + bind = op.get_bind() + rows = list( + bind.execute( + sa.text( + "SELECT id, job_id FROM imap_append_attempts " + "ORDER BY job_id, created_at, id" + ) + ).mappings() + ) + per_job: dict[str, int] = {} + for row in rows: + job_id = str(row["job_id"]) + attempt_number = per_job.get(job_id, 0) + 1 + per_job[job_id] = attempt_number + bind.execute( + sa.text( + "UPDATE imap_append_attempts SET attempt_number = :attempt_number " + "WHERE id = :attempt_id" + ), + {"attempt_number": attempt_number, "attempt_id": row["id"]}, + )