diff --git a/README.md b/README.md index 8bbcac8..3476268 100644 --- a/README.md +++ b/README.md @@ -33,10 +33,13 @@ revision comparison, credential resolution, policy checks, and SMTP/IMAP effects inside Mail. Consumer-visible outcomes are sanitized: provider banners, raw response bytes, hosts, account identities, and credentials are not returned. -A remaining observability slice, tracked in [govoplan-mail#17](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17), is a Mail-owned durable outbox and transport-attempt store with -restricted diagnostic access, retention controls, and correlation identifiers. -Until that exists, Campaign retains only sanitized delivery evidence; raw -provider diagnostics must not be copied into consumer records. +Mail also owns a durable delivery-command outbox for effects that do not +already have a consumer-owned job ledger, including Campaign report messages. +It persists the command and attempt before SMTP, binds idempotency keys to +canonical request hashes, distinguishes partial refusal and unknown outcome, +and requires explicit evidence-backed reconciliation before any deliberate +resend. Business readers receive only counts and sanitized state; recipient +refusal details require `mail:delivery:diagnostic`. SMTP effects decrypt only SMTP credentials; Sent-folder effects decrypt only IMAP credentials. A connection loss after an effect starts is surfaced as an diff --git a/docs/MAIL_HANDBOOK.md b/docs/MAIL_HANDBOOK.md index b500337..208aa89 100644 --- a/docs/MAIL_HANDBOOK.md +++ b/docs/MAIL_HANDBOOK.md @@ -82,10 +82,20 @@ effect. ### Provider outcomes -SMTP acceptance, recipient refusal, temporary/permanent failure, connection -loss, and unknown outcome are distinct. IMAP append is a separate operation and -outcome. Mail returns a sanitized transport result; the consuming domain owns -the durable business job and its retry/reconciliation semantics. +SMTP acceptance, partial or complete recipient refusal, temporary/permanent +failure, connection loss, and unknown outcome are distinct. IMAP append is a +separate operation and outcome. Ordinary Campaign recipient delivery retains +its Campaign-owned job ledger. Mail owns an encrypted durable command and +attempt ledger for report delivery and other effects that do not have such a +consumer ledger. + +The Mail worker commits an attempt before it starts SMTP and then commits an +effect-start marker before opening the provider effect. Worker redelivery may +recover a stale pre-effect claim, but an accepted, partially accepted, +in-progress-after-effect, or unknown command is never sent automatically. +Unknown outcomes require a separately authorized reconciliation with an +external evidence reference. A deliberate resend creates a new command with a +new idempotency key and links it to the prior command. ## User tasks @@ -354,11 +364,11 @@ Security invariants: Credential replacement/deletion emits canonical non-secret Mail audit events. Profile/policy create/update/deactivate currently emits change-feed evidence, but connection tests and the complete administrative lifecycle do not yet have -equivalent canonical audit events. Consumer send/retry/reconciliation evidence -belongs primarily to the consuming module today. A Mail-owned restricted -provider-attempt ledger with correlation, retention, and unknown-outcome -reconciliation remains part of the durable outbox work in -[`govoplan-mail#17`](https://git.add-ideas.de/GovOPlaN/govoplan-mail/issues/17). +equivalent canonical audit events. Mail delivery commands emit request, +terminal-outcome, reconciliation, and deliberate-resend events without +addresses or raw provider responses. MIME, envelope addresses, detailed +refusals, and reconciliation notes are encrypted and minimized after their +retention deadline while hashes, counts, attempts, and decisions remain. ## Acceptance checklist diff --git a/src/govoplan_mail/backend/capabilities.py b/src/govoplan_mail/backend/capabilities.py index e1ad0ec..c07db41 100644 --- a/src/govoplan_mail/backend/capabilities.py +++ b/src/govoplan_mail/backend/capabilities.py @@ -511,6 +511,32 @@ class MailCampaignCapability: append_campaign_message_to_sent = staticmethod(append_campaign_message_to_sent) wait_for_rate_limit = staticmethod(wait_for_rate_limit) + @staticmethod + def submit_delivery_command(session: Session, **kwargs: Any) -> dict[str, object]: + from govoplan_mail.backend.delivery_outbox import submit_delivery_command + + return submit_delivery_command(session, **kwargs) + + @staticmethod + def delivery_command_summary( + session: Session, + *, + tenant_id: str, + command_id: str, + ) -> dict[str, object]: + from govoplan_mail.backend.delivery_outbox import ( + delivery_command_summary, + get_delivery_command, + ) + + return delivery_command_summary( + get_delivery_command( + session, + tenant_id=tenant_id, + command_id=command_id, + ) + ) + @staticmethod def mock_mailbox(): from govoplan_mail.backend.dev import mock_mailbox @@ -521,3 +547,10 @@ class MailCampaignCapability: def campaign_capability(context: ModuleContext) -> MailCampaignCapability: configure_runtime(settings=context.settings) return MailCampaignCapability() + + +def delivery_outbox_capability(context: ModuleContext): + from govoplan_mail.backend.delivery_outbox import MailDeliveryOutboxCapability + + configure_runtime(settings=context.settings) + return MailDeliveryOutboxCapability() diff --git a/src/govoplan_mail/backend/db/models.py b/src/govoplan_mail/backend/db/models.py index 5461955..3b66faa 100644 --- a/src/govoplan_mail/backend/db/models.py +++ b/src/govoplan_mail/backend/db/models.py @@ -152,3 +152,137 @@ class MailMailboxMessageIndex(Base, TimestampMixin): body_preview: Mapped[str | None] = mapped_column(Text, nullable=True) attachment_count: Mapped[int] = mapped_column(Integer, default=0, nullable=False) indexed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) + + +class MailDeliveryCommand(Base, TimestampMixin): + __tablename__ = "mail_delivery_commands" + __table_args__ = ( + UniqueConstraint( + "tenant_id", + "command_type", + "idempotency_key", + name="uq_mail_delivery_commands_idempotency", + ), + Index( + "ix_mail_delivery_commands_dispatch", + "status", + "next_attempt_at", + "created_at", + ), + Index( + "ix_mail_delivery_commands_source", + "tenant_id", + "source_module", + "source_resource_type", + "source_resource_id", + ), + ) + + 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) + command_type: Mapped[str] = mapped_column(String(60), nullable=False, index=True) + source_module: Mapped[str] = mapped_column(String(80), nullable=False, index=True) + source_resource_type: Mapped[str] = mapped_column(String(80), nullable=False) + source_resource_id: Mapped[str | None] = mapped_column(String(120), nullable=True) + source_version_id: Mapped[str | None] = mapped_column(String(120), nullable=True) + idempotency_key: Mapped[str] = mapped_column(String(200), nullable=False) + canonical_request_hash: Mapped[str] = mapped_column(String(64), nullable=False) + profile_id: Mapped[str] = mapped_column( + ForeignKey("mail_server_profiles.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ) + smtp_server_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + smtp_credential_id: Mapped[str | None] = mapped_column(String(36), nullable=True) + expected_smtp_transport_revision: Mapped[str] = mapped_column(String(120), nullable=False) + envelope_from_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) + envelope_recipients_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) + from_header_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) + message_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) + message_sha256: Mapped[str] = mapped_column(String(64), nullable=False) + message_size_bytes: Mapped[int] = mapped_column(BigInteger, nullable=False) + recipient_count: Mapped[int] = mapped_column(Integer, nullable=False) + status: Mapped[str] = mapped_column(String(40), nullable=False, default="pending", index=True) + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + next_attempt_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), + nullable=True, + index=True, + ) + claimed_by: Mapped[str | None] = mapped_column(String(255), nullable=True) + claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=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) + accepted_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + refused_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + refusal_summary: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict) + refusal_details_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) + failure_code: Mapped[str | None] = mapped_column(String(80), nullable=True) + failure_summary: Mapped[str | None] = mapped_column(String(500), nullable=True) + created_by_user_id: Mapped[str | None] = mapped_column( + ForeignKey("access_users.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + supersedes_command_id: Mapped[str | None] = mapped_column( + ForeignKey("mail_delivery_commands.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) + expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, index=True) + payload_purged_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class MailDeliveryAttempt(Base, TimestampMixin): + __tablename__ = "mail_delivery_attempts" + __table_args__ = ( + UniqueConstraint( + "command_id", + "attempt_number", + name="uq_mail_delivery_attempts_number", + ), + Index("ix_mail_delivery_attempts_command_started", "command_id", "started_at"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + command_id: Mapped[str] = mapped_column( + ForeignKey("mail_delivery_commands.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + attempt_number: Mapped[int] = mapped_column(Integer, nullable=False) + worker_id: Mapped[str | None] = mapped_column(String(255), nullable=True) + status: Mapped[str] = mapped_column(String(40), 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, nullable=False, default=0) + refused_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + outcome_code: Mapped[str | None] = mapped_column(String(80), nullable=True) + diagnostic_summary: Mapped[str | None] = mapped_column(String(500), nullable=True) + + +class MailDeliveryReconciliation(Base, TimestampMixin): + __tablename__ = "mail_delivery_reconciliations" + __table_args__ = ( + Index( + "ix_mail_delivery_reconciliations_command_created", + "command_id", + "created_at", + ), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + command_id: Mapped[str] = mapped_column( + ForeignKey("mail_delivery_commands.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + decision: Mapped[str] = mapped_column(String(40), nullable=False) + evidence_reference: Mapped[str] = mapped_column(String(500), nullable=False) + note_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True) + created_by_user_id: Mapped[str | None] = mapped_column( + ForeignKey("access_users.id", ondelete="SET NULL"), + nullable=True, + index=True, + ) diff --git a/src/govoplan_mail/backend/delivery_outbox.py b/src/govoplan_mail/backend/delivery_outbox.py new file mode 100644 index 0000000..742fd27 --- /dev/null +++ b/src/govoplan_mail/backend/delivery_outbox.py @@ -0,0 +1,880 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +from collections import Counter +from datetime import datetime, timedelta, timezone +from typing import Any + +from sqlalchemy import and_, or_, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from govoplan_core.audit.logging import audit_event +from govoplan_core.security.secrets import decrypt_secret, encrypt_secret +from govoplan_mail.backend.capabilities import send_campaign_email_bytes +from govoplan_mail.backend.db.models import ( + MailDeliveryAttempt, + MailDeliveryCommand, + MailDeliveryReconciliation, +) +from govoplan_mail.backend.mail_profiles import MailProfileError +from govoplan_mail.backend.sending.smtp import SmtpConfigurationError, SmtpSendError + + +DISPATCHABLE_STATUSES = frozenset({"pending", "temporary_failure", "reconciled_not_accepted"}) +TERMINAL_STATUSES = frozenset( + { + "accepted", + "partially_refused", + "permanent_failure", + "outcome_unknown", + "reconciled_accepted", + "cancelled", + } +) +NON_RETRYABLE_STATUSES = TERMINAL_STATUSES | frozenset({"claimed", "in_progress"}) +DEFAULT_PAYLOAD_RETENTION_DAYS = 30 +STALE_CLAIM_AFTER = timedelta(minutes=10) + + +class MailDeliveryError(RuntimeError): + pass + + +class MailDeliveryIdempotencyConflict(MailDeliveryError): + pass + + +class MailDeliveryNotFound(MailDeliveryError): + pass + + +class MailDeliveryStateError(MailDeliveryError): + pass + + +def utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def _bounded_text(value: object | None, *, limit: int = 500) -> str | None: + if value is None: + return None + candidate = " ".join(str(value).split()) + return candidate[:limit] or None + + +def _canonical_hash(payload: dict[str, object]) -> str: + encoded = json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _encrypt_json(value: object) -> str: + encrypted = encrypt_secret( + json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + ) + if not encrypted: + raise MailDeliveryError("Mail delivery evidence could not be encrypted") + return encrypted + + +def _decrypt_json(value: str | None) -> Any: + plaintext = decrypt_secret(value) + if plaintext is None: + return None + return json.loads(plaintext) + + +def _message_bytes(command: MailDeliveryCommand) -> bytes: + encoded = decrypt_secret(command.message_encrypted) + if encoded is None: + raise MailDeliveryStateError("Mail delivery payload is no longer available") + try: + message = base64.b64decode(encoded.encode("ascii"), validate=True) + except (ValueError, TypeError) as exc: + raise MailDeliveryStateError("Mail delivery payload is invalid") from exc + if hashlib.sha256(message).hexdigest() != command.message_sha256: + raise MailDeliveryStateError("Mail delivery payload integrity check failed") + return message + + +def _delivery_payload( + *, + command_type: str, + source_module: str, + source_resource_type: str, + source_resource_id: str | None, + source_version_id: str | None, + profile_id: str, + smtp_server_id: str | None, + smtp_credential_id: str | None, + expected_smtp_transport_revision: str, + envelope_from: str, + envelope_recipients: list[str], + from_header: str | None, + message_sha256: str, +) -> dict[str, object]: + return { + "command_type": command_type, + "source_module": source_module, + "source_resource_type": source_resource_type, + "source_resource_id": source_resource_id, + "source_version_id": source_version_id, + "profile_id": profile_id, + "smtp_server_id": smtp_server_id, + "smtp_credential_id": smtp_credential_id, + "expected_smtp_transport_revision": expected_smtp_transport_revision, + "envelope_from": envelope_from, + "envelope_recipients": envelope_recipients, + "from_header": from_header, + "message_sha256": message_sha256, + } + + +def submit_delivery_command( + session: Session, + *, + tenant_id: str, + command_type: str, + source_module: str, + source_resource_type: str, + source_resource_id: str | None, + source_version_id: str | None, + idempotency_key: str, + profile_id: str, + message_bytes: bytes, + envelope_from: str, + envelope_recipients: list[str], + from_header: str | None, + expected_smtp_transport_revision: str, + smtp_server_id: str | None = None, + smtp_credential_id: str | None = None, + created_by_user_id: str | None = None, + retention_days: int = DEFAULT_PAYLOAD_RETENTION_DAYS, + supersedes_command_id: str | None = None, +) -> dict[str, object]: + clean_key = idempotency_key.strip() + clean_recipients = [str(value).strip() for value in envelope_recipients if str(value).strip()] + if not clean_key or len(clean_key) > 200: + raise MailDeliveryError("A bounded idempotency key is required") + if not clean_recipients: + raise MailDeliveryError("At least one envelope recipient is required") + if retention_days < 1: + raise MailDeliveryError("Mail payload retention must be at least one day") + digest = hashlib.sha256(message_bytes).hexdigest() + request_hash = _canonical_hash( + _delivery_payload( + command_type=command_type, + source_module=source_module, + source_resource_type=source_resource_type, + source_resource_id=source_resource_id, + source_version_id=source_version_id, + profile_id=profile_id, + smtp_server_id=smtp_server_id, + smtp_credential_id=smtp_credential_id, + expected_smtp_transport_revision=expected_smtp_transport_revision, + envelope_from=envelope_from, + envelope_recipients=clean_recipients, + from_header=from_header, + message_sha256=digest, + ) + ) + existing = session.scalar( + select(MailDeliveryCommand).where( + MailDeliveryCommand.tenant_id == tenant_id, + MailDeliveryCommand.command_type == command_type, + MailDeliveryCommand.idempotency_key == clean_key, + ) + ) + if existing is not None: + if existing.canonical_request_hash != request_hash: + raise MailDeliveryIdempotencyConflict( + "The idempotency key is already bound to a different mail command" + ) + return delivery_command_summary(existing, duplicate=True) + + now = utcnow() + command = MailDeliveryCommand( + tenant_id=tenant_id, + command_type=command_type, + source_module=source_module, + source_resource_type=source_resource_type, + source_resource_id=source_resource_id, + source_version_id=source_version_id, + idempotency_key=clean_key, + canonical_request_hash=request_hash, + profile_id=profile_id, + smtp_server_id=smtp_server_id, + smtp_credential_id=smtp_credential_id, + expected_smtp_transport_revision=expected_smtp_transport_revision, + envelope_from_encrypted=encrypt_secret(envelope_from), + envelope_recipients_encrypted=_encrypt_json(clean_recipients), + from_header_encrypted=encrypt_secret(from_header), + message_encrypted=encrypt_secret(base64.b64encode(message_bytes).decode("ascii")), + message_sha256=digest, + message_size_bytes=len(message_bytes), + recipient_count=len(clean_recipients), + status="pending", + next_attempt_at=now, + created_by_user_id=created_by_user_id, + supersedes_command_id=supersedes_command_id, + expires_at=now + timedelta(days=retention_days), + ) + try: + with session.begin_nested(): + session.add(command) + session.flush() + except IntegrityError: + existing = session.scalar( + select(MailDeliveryCommand).where( + MailDeliveryCommand.tenant_id == tenant_id, + MailDeliveryCommand.command_type == command_type, + MailDeliveryCommand.idempotency_key == clean_key, + ) + ) + if existing is None or existing.canonical_request_hash != request_hash: + raise MailDeliveryIdempotencyConflict( + "The idempotency key is already bound to a different mail command" + ) from None + return delivery_command_summary(existing, duplicate=True) + + audit_event( + session, + tenant_id=tenant_id, + user_id=created_by_user_id, + action="mail.delivery_requested", + object_type="mail_delivery_command", + object_id=command.id, + details={ + "command_type": command_type, + "source_module": source_module, + "source_resource_type": source_resource_type, + "source_resource_id": source_resource_id, + "recipient_count": len(clean_recipients), + "message_sha256": digest, + "supersedes_command_id": supersedes_command_id, + }, + ) + return delivery_command_summary(command) + + +def get_delivery_command( + session: Session, + *, + tenant_id: str, + command_id: str, +) -> MailDeliveryCommand: + command = session.get(MailDeliveryCommand, command_id) + if command is None or command.tenant_id != tenant_id: + raise MailDeliveryNotFound("Mail delivery command not found") + return command + + +def delivery_command_summary( + command: MailDeliveryCommand, + *, + duplicate: bool = False, +) -> dict[str, object]: + return { + "id": command.id, + "tenant_id": command.tenant_id, + "command_type": command.command_type, + "source_module": command.source_module, + "source_resource_type": command.source_resource_type, + "source_resource_id": command.source_resource_id, + "source_version_id": command.source_version_id, + "status": command.status, + "recipient_count": command.recipient_count, + "accepted_count": command.accepted_count, + "refused_count": command.refused_count, + "refusal_summary": dict(command.refusal_summary or {}), + "attempt_count": command.attempt_count, + "failure_code": command.failure_code, + "failure_summary": command.failure_summary, + "message_sha256": command.message_sha256, + "message_size_bytes": command.message_size_bytes, + "created_at": command.created_at, + "completed_at": command.completed_at, + "expires_at": command.expires_at, + "payload_purged_at": command.payload_purged_at, + "safe_to_retry": command.status in DISPATCHABLE_STATUSES, + "outcome_known": command.status + not in {"claimed", "in_progress", "outcome_unknown"}, + "duplicate": duplicate, + } + + +def delivery_command_diagnostics( + session: Session, + *, + tenant_id: str, + command_id: str, +) -> dict[str, object]: + command = get_delivery_command(session, tenant_id=tenant_id, command_id=command_id) + attempts = session.scalars( + select(MailDeliveryAttempt) + .where(MailDeliveryAttempt.command_id == command.id) + .order_by(MailDeliveryAttempt.attempt_number) + ).all() + reconciliations = session.scalars( + select(MailDeliveryReconciliation) + .where(MailDeliveryReconciliation.command_id == command.id) + .order_by(MailDeliveryReconciliation.created_at) + ).all() + refusals = _decrypt_json(command.refusal_details_encrypted) or {} + return { + **delivery_command_summary(command), + "refused_recipients": refusals, + "attempts": [ + { + "id": attempt.id, + "attempt_number": attempt.attempt_number, + "worker_id": attempt.worker_id, + "status": attempt.status, + "started_at": attempt.started_at, + "effect_started_at": attempt.effect_started_at, + "completed_at": attempt.completed_at, + "accepted_count": attempt.accepted_count, + "refused_count": attempt.refused_count, + "outcome_code": attempt.outcome_code, + "diagnostic_summary": attempt.diagnostic_summary, + } + for attempt in attempts + ], + "reconciliations": [ + { + "id": item.id, + "decision": item.decision, + "evidence_reference": item.evidence_reference, + "note": decrypt_secret(item.note_encrypted), + "created_by_user_id": item.created_by_user_id, + "created_at": item.created_at, + } + for item in reconciliations + ], + } + + +def _current_attempt( + session: Session, + command: MailDeliveryCommand, +) -> MailDeliveryAttempt | None: + return session.scalar( + select(MailDeliveryAttempt) + .where( + MailDeliveryAttempt.command_id == command.id, + MailDeliveryAttempt.attempt_number == command.attempt_count, + ) + .limit(1) + ) + + +def _recover_stale_commands( + session: Session, + *, + now: datetime, + tenant_id: str | None, +) -> tuple[int, int]: + cutoff = now - STALE_CLAIM_AFTER + clauses = [ + MailDeliveryCommand.status.in_(("claimed", "in_progress")), + MailDeliveryCommand.claimed_at <= cutoff, + ] + if tenant_id: + clauses.append(MailDeliveryCommand.tenant_id == tenant_id) + commands = session.scalars( + select(MailDeliveryCommand).where(*clauses).order_by(MailDeliveryCommand.claimed_at) + ).all() + recovered = 0 + unknown = 0 + for command in commands: + attempt = _current_attempt(session, command) + effect_started = command.effect_started_at or ( + attempt.effect_started_at if attempt else None + ) + if effect_started is None: + command.status = "pending" + command.next_attempt_at = now + command.claimed_at = None + command.claimed_by = None + if attempt is not None: + attempt.status = "claim_abandoned" + attempt.completed_at = now + attempt.outcome_code = "worker_lost_before_effect" + recovered += 1 + continue + command.status = "outcome_unknown" + command.completed_at = now + command.next_attempt_at = None + command.failure_code = "worker_lost_after_effect_start" + command.failure_summary = ( + "Delivery outcome is unknown because the worker stopped after transmission began." + ) + if attempt is not None: + attempt.status = "outcome_unknown" + attempt.completed_at = now + attempt.outcome_code = command.failure_code + attempt.diagnostic_summary = command.failure_summary + unknown += 1 + if commands: + session.commit() + return recovered, unknown + + +def _claim_command( + session: Session, + *, + command_id: str, + now: datetime, + worker_id: str | None, +) -> tuple[MailDeliveryCommand, MailDeliveryAttempt] | None: + command = session.scalar( + select(MailDeliveryCommand) + .where( + MailDeliveryCommand.id == command_id, + MailDeliveryCommand.status.in_(DISPATCHABLE_STATUSES), + or_( + MailDeliveryCommand.next_attempt_at.is_(None), + MailDeliveryCommand.next_attempt_at <= now, + ), + MailDeliveryCommand.payload_purged_at.is_(None), + ) + .with_for_update(skip_locked=True) + ) + if command is None: + session.rollback() + return None + command.attempt_count += 1 + command.status = "claimed" + command.claimed_by = _bounded_text(worker_id, limit=255) + command.claimed_at = now + command.effect_started_at = None + command.next_attempt_at = None + command.failure_code = None + command.failure_summary = None + attempt = MailDeliveryAttempt( + command_id=command.id, + attempt_number=command.attempt_count, + worker_id=command.claimed_by, + status="claimed", + started_at=now, + ) + session.add(attempt) + session.commit() + return command, attempt + + +def _mark_effect_started( + session: Session, + command: MailDeliveryCommand, + attempt: MailDeliveryAttempt, +) -> None: + now = utcnow() + command.status = "in_progress" + command.effect_started_at = now + attempt.status = "in_progress" + attempt.effect_started_at = now + session.commit() + + +def _refusal_summary(refusals: dict[str, dict[str, int | str]]) -> dict[str, int]: + classifications = Counter( + str(item.get("classification") or "unknown") + for item in refusals.values() + ) + return dict(sorted(classifications.items())) + + +def _record_outcome( + session: Session, + *, + command: MailDeliveryCommand, + attempt: MailDeliveryAttempt, + status: str, + accepted_count: int = 0, + refusals: dict[str, dict[str, int | str]] | None = None, + failure_code: str | None = None, + failure_summary: str | None = None, +) -> None: + now = utcnow() + refusal_map = refusals or {} + command.status = status + command.accepted_count = accepted_count + command.refused_count = len(refusal_map) + command.refusal_summary = _refusal_summary(refusal_map) + command.refusal_details_encrypted = ( + _encrypt_json(refusal_map) if refusal_map else None + ) + command.failure_code = failure_code + command.failure_summary = _bounded_text(failure_summary) + command.claimed_by = None + command.claimed_at = None + command.next_attempt_at = ( + now + timedelta(minutes=min(60, 2 ** min(command.attempt_count, 5))) + if status == "temporary_failure" + else None + ) + if status != "temporary_failure": + command.completed_at = now + attempt.status = status + attempt.completed_at = now + attempt.accepted_count = accepted_count + attempt.refused_count = len(refusal_map) + attempt.outcome_code = failure_code or status + attempt.diagnostic_summary = _bounded_text(failure_summary) + session.commit() + + audit_event( + session, + tenant_id=command.tenant_id, + user_id=command.created_by_user_id, + action="mail.delivery_completed", + object_type="mail_delivery_command", + object_id=command.id, + details={ + "status": status, + "attempt_number": attempt.attempt_number, + "accepted_count": accepted_count, + "refused_count": len(refusal_map), + "refusal_summary": command.refusal_summary, + "failure_code": failure_code, + "source_module": command.source_module, + "source_resource_id": command.source_resource_id, + }, + ) + session.commit() + + +def _process_claimed( + session: Session, + command: MailDeliveryCommand, + attempt: MailDeliveryAttempt, +) -> str: + try: + message = _message_bytes(command) + envelope_from = decrypt_secret(command.envelope_from_encrypted) + recipients = _decrypt_json(command.envelope_recipients_encrypted) + from_header = decrypt_secret(command.from_header_encrypted) + if not envelope_from or not isinstance(recipients, list) or not recipients: + raise MailDeliveryStateError("Mail delivery envelope is unavailable") + except MailDeliveryStateError as exc: + _record_outcome( + session, + command=command, + attempt=attempt, + status="permanent_failure", + failure_code="payload_unavailable", + failure_summary=str(exc), + ) + return "permanent_failure" + + _mark_effect_started(session, command, attempt) + try: + result = send_campaign_email_bytes( + session, + tenant_id=command.tenant_id, + campaign_id=str(command.source_resource_id or ""), + profile_id=command.profile_id, + message_bytes=message, + envelope_from=envelope_from, + envelope_recipients=[str(item) for item in recipients], + from_header=from_header, + expected_smtp_transport_revision=command.expected_smtp_transport_revision, + smtp_server_id=command.smtp_server_id, + smtp_credential_id=command.smtp_credential_id, + ) + except SmtpSendError as exc: + if exc.outcome_unknown: + status = "outcome_unknown" + code = "smtp_outcome_unknown" + elif exc.temporary: + status = "temporary_failure" + code = "smtp_temporary_failure" + else: + status = "permanent_failure" + code = "smtp_permanent_failure" + _record_outcome( + session, + command=command, + attempt=attempt, + status=status, + failure_code=code, + failure_summary=str(exc), + ) + return status + except (MailProfileError, SmtpConfigurationError) as exc: + _record_outcome( + session, + command=command, + attempt=attempt, + status="permanent_failure", + failure_code="authorization_or_configuration_changed", + failure_summary=str(exc), + ) + return "permanent_failure" + except Exception: + _record_outcome( + session, + command=command, + attempt=attempt, + status="outcome_unknown", + failure_code="unexpected_error_after_effect_start", + failure_summary="Mail delivery outcome is unknown after transmission began.", + ) + return "outcome_unknown" + + refusals = dict(result.refused_recipients) + accepted_count = result.accepted_count + if not refusals: + status = "accepted" + elif accepted_count > 0: + status = "partially_refused" + elif all( + item.get("classification") == "temporary" for item in refusals.values() + ): + status = "temporary_failure" + elif any( + item.get("classification") == "unknown" for item in refusals.values() + ): + status = "outcome_unknown" + else: + status = "permanent_failure" + _record_outcome( + session, + command=command, + attempt=attempt, + status=status, + accepted_count=accepted_count, + refusals=refusals, + failure_code=None if status == "accepted" else f"smtp_{status}", + failure_summary=None if status == "accepted" else "One or more recipients were refused.", + ) + return status + + +def dispatch_due( + session: Session, + *, + tenant_id: str | None = None, + limit: int = 25, + worker_id: str | None = None, +) -> dict[str, object]: + bounded_limit = max(1, min(int(limit), 100)) + now = utcnow() + recovered, recovered_unknown = _recover_stale_commands( + session, + now=now, + tenant_id=tenant_id, + ) + clauses = [ + MailDeliveryCommand.status.in_(DISPATCHABLE_STATUSES), + MailDeliveryCommand.payload_purged_at.is_(None), + or_( + MailDeliveryCommand.next_attempt_at.is_(None), + MailDeliveryCommand.next_attempt_at <= now, + ), + ] + if tenant_id: + clauses.append(MailDeliveryCommand.tenant_id == tenant_id) + command_ids = list( + session.scalars( + select(MailDeliveryCommand.id) + .where(*clauses) + .order_by(MailDeliveryCommand.next_attempt_at, MailDeliveryCommand.created_at) + .limit(bounded_limit) + ).all() + ) + counters: Counter[str] = Counter() + processed_ids: list[str] = [] + for command_id in command_ids: + claimed = _claim_command( + session, + command_id=command_id, + now=utcnow(), + worker_id=worker_id, + ) + if claimed is None: + continue + command, attempt = claimed + outcome = _process_claimed(session, command, attempt) + counters[outcome] += 1 + processed_ids.append(command.id) + return { + "selected": len(processed_ids), + "accepted": counters["accepted"], + "partially_refused": counters["partially_refused"], + "retrying": counters["temporary_failure"], + "failed": counters["permanent_failure"], + "outcome_unknown": counters["outcome_unknown"] + recovered_unknown, + "recovered_before_effect": recovered, + "command_ids": processed_ids, + } + + +def reconcile_delivery_command( + session: Session, + *, + tenant_id: str, + command_id: str, + decision: str, + evidence_reference: str, + note: str | None, + user_id: str, +) -> dict[str, object]: + command = get_delivery_command(session, tenant_id=tenant_id, command_id=command_id) + if command.status not in {"outcome_unknown", "in_progress"}: + raise MailDeliveryStateError( + "Only a delivery with an unknown outcome can be reconciled" + ) + clean_decision = decision.strip().casefold() + if clean_decision not in {"accepted", "not_accepted"}: + raise MailDeliveryStateError( + "Reconciliation decision must be accepted or not_accepted" + ) + clean_evidence = _bounded_text(evidence_reference) + if not clean_evidence: + raise MailDeliveryStateError("An evidence reference is required") + item = MailDeliveryReconciliation( + command_id=command.id, + decision=clean_decision, + evidence_reference=clean_evidence, + note_encrypted=encrypt_secret(note), + created_by_user_id=user_id, + ) + session.add(item) + command.status = ( + "reconciled_accepted" + if clean_decision == "accepted" + else "reconciled_not_accepted" + ) + command.completed_at = utcnow() if clean_decision == "accepted" else None + command.next_attempt_at = None + command.failure_code = f"reconciled_{clean_decision}" + command.failure_summary = "Delivery outcome was reconciled from external evidence." + audit_event( + session, + tenant_id=tenant_id, + user_id=user_id, + action="mail.delivery_reconciled", + object_type="mail_delivery_command", + object_id=command.id, + details={ + "decision": clean_decision, + "evidence_reference": clean_evidence, + }, + ) + session.commit() + return delivery_command_summary(command) + + +def resend_delivery_command( + session: Session, + *, + tenant_id: str, + command_id: str, + idempotency_key: str, + user_id: str, +) -> dict[str, object]: + command = get_delivery_command(session, tenant_id=tenant_id, command_id=command_id) + if command.status not in { + "outcome_unknown", + "reconciled_not_accepted", + "permanent_failure", + "partially_refused", + }: + raise MailDeliveryStateError( + "A deliberate resend is only available after a terminal or reconciled failure" + ) + if command.payload_purged_at is not None: + raise MailDeliveryStateError("The retained delivery payload is no longer available") + message = _message_bytes(command) + recipients = _decrypt_json(command.envelope_recipients_encrypted) + envelope_from = decrypt_secret(command.envelope_from_encrypted) + if not envelope_from or not isinstance(recipients, list): + raise MailDeliveryStateError("The retained delivery envelope is unavailable") + result = submit_delivery_command( + session, + tenant_id=tenant_id, + command_type=command.command_type, + source_module=command.source_module, + source_resource_type=command.source_resource_type, + source_resource_id=command.source_resource_id, + source_version_id=command.source_version_id, + idempotency_key=idempotency_key, + profile_id=command.profile_id, + message_bytes=message, + envelope_from=envelope_from, + envelope_recipients=[str(item) for item in recipients], + from_header=decrypt_secret(command.from_header_encrypted), + expected_smtp_transport_revision=command.expected_smtp_transport_revision, + smtp_server_id=command.smtp_server_id, + smtp_credential_id=command.smtp_credential_id, + created_by_user_id=user_id, + retention_days=max( + 1, + (command.expires_at - utcnow()).days, + ), + supersedes_command_id=command.id, + ) + audit_event( + session, + tenant_id=tenant_id, + user_id=user_id, + action="mail.delivery_resend_requested", + object_type="mail_delivery_command", + object_id=str(result["id"]), + details={"supersedes_command_id": command.id}, + ) + session.commit() + return result + + +def purge_expired( + session: Session, + *, + limit: int = 250, +) -> dict[str, object]: + now = utcnow() + commands = session.scalars( + select(MailDeliveryCommand) + .where( + MailDeliveryCommand.expires_at <= now, + MailDeliveryCommand.payload_purged_at.is_(None), + ) + .order_by(MailDeliveryCommand.expires_at) + .limit(max(1, min(int(limit), 1000))) + ).all() + for command in commands: + command.envelope_from_encrypted = None + command.envelope_recipients_encrypted = None + command.from_header_encrypted = None + command.message_encrypted = None + command.refusal_details_encrypted = None + command.payload_purged_at = now + reconciliations = session.scalars( + select(MailDeliveryReconciliation) + .join( + MailDeliveryCommand, + MailDeliveryCommand.id == MailDeliveryReconciliation.command_id, + ) + .where( + MailDeliveryCommand.payload_purged_at == now, + MailDeliveryReconciliation.note_encrypted.is_not(None), + ) + ).all() + for reconciliation in reconciliations: + reconciliation.note_encrypted = None + session.commit() + return {"purged": len(commands)} + + +class MailDeliveryOutboxCapability: + dispatch_due = staticmethod(dispatch_due) + purge_expired = staticmethod(purge_expired) diff --git a/src/govoplan_mail/backend/documentation.py b/src/govoplan_mail/backend/documentation.py index efd4215..2c3fb1a 100644 --- a/src/govoplan_mail/backend/documentation.py +++ b/src/govoplan_mail/backend/documentation.py @@ -4,7 +4,13 @@ from typing import Any from sqlalchemy.orm import Session -from govoplan_core.core.modules import DocumentationCondition, DocumentationContext, DocumentationLink, DocumentationTopic +from govoplan_core.core.modules import ( + DocumentationCondition, + DocumentationConfigurationDecision, + DocumentationContext, + DocumentationLink, + DocumentationTopic, +) from govoplan_mail.backend.mail_profiles import ( EffectiveMailProfilePolicy, MailProfileError, @@ -35,6 +41,60 @@ def documentation_topics(context: DocumentationContext) -> tuple[DocumentationTo return tuple(topics) +def documentation_configuration_states( + context: DocumentationContext, + keys: tuple[str, ...], +) -> dict[str, DocumentationConfigurationDecision]: + if "mail_profile_policy" not in keys: + return {} + principal = context.principal + tenant_id = str(getattr(principal, "tenant_id", "") or "") + session = context.session + if not tenant_id or not isinstance(session, Session): + return { + "mail_profile_policy": DocumentationConfigurationDecision( + key="mail_profile_policy", + state="unavailable", + reason="The effective tenant policy cannot be evaluated in this context.", + ) + } + try: + policy = effective_mail_profile_policy_for_scope( + session, + tenant_id=tenant_id, + scope_type="tenant", + ) + except MailProfileError: + return { + "mail_profile_policy": DocumentationConfigurationDecision( + key="mail_profile_policy", + state="unavailable", + reason="The effective tenant policy could not be evaluated.", + ) + } + tenant_sources = [ + source + for source in policy.source_policies + if source.get("scope_type") == "tenant" + ] + explicitly_configured = any( + bool(source.get("applied_fields")) + for source in tenant_sources + ) + return { + "mail_profile_policy": DocumentationConfigurationDecision( + key="mail_profile_policy", + state="enabled" if explicitly_configured else "inherited", + source="tenant" if explicitly_configured else "system default", + reason=( + "A tenant policy is configured." + if explicitly_configured + else "The tenant uses the inherited system policy." + ), + ) + } + + def _tenant_mail_policy_topic(context: DocumentationContext) -> DocumentationTopic | None: principal = context.principal tenant_id = str(getattr(principal, "tenant_id", "") or "") diff --git a/src/govoplan_mail/backend/manifest.py b/src/govoplan_mail/backend/manifest.py index 46220c0..5fd4e96 100644 --- a/src/govoplan_mail/backend/manifest.py +++ b/src/govoplan_mail/backend/manifest.py @@ -6,9 +6,11 @@ from pathlib import Path from sqlalchemy import inspect from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER +from govoplan_core.core.mail import CAPABILITY_MAIL_DELIVERY_OUTBOX from govoplan_core.core.module_guards import drop_table_retirement_provider, persistent_table_uninstall_guard from govoplan_core.core.modules import ( DocumentationCondition, + DocumentationConfigurationProviderRegistration, DocumentationLink, DocumentationTopic, FrontendModule, @@ -24,7 +26,10 @@ from govoplan_core.core.modules import ( ) from govoplan_core.core.views import ViewSurface from govoplan_core.db.base import Base -from govoplan_mail.backend.documentation import documentation_topics +from govoplan_mail.backend.documentation import ( + documentation_configuration_states, + documentation_topics, +) from govoplan_mail.backend.db import models as mail_models # noqa: F401 - populate Mail ORM metadata @@ -35,6 +40,9 @@ _mail_table_retirement_provider = drop_table_retirement_provider( mail_models.MailProfilePolicy, mail_models.MailMailboxFolderIndex, mail_models.MailMailboxMessageIndex, + mail_models.MailDeliveryReconciliation, + mail_models.MailDeliveryAttempt, + mail_models.MailDeliveryCommand, label="Mail", ) @@ -90,6 +98,16 @@ PERMISSIONS = ( "Create, edit, and deactivate only the current account's user-scoped mail profiles within effective policy.", ), _permission("mail:secret:manage", "Manage mail secrets", "Create or replace stored SMTP/IMAP credentials."), + _permission( + "mail:delivery:diagnostic", + "Inspect mail delivery diagnostics", + "Inspect bounded recipient-level refusal and attempt evidence for durable Mail commands.", + ), + _permission( + "mail:delivery:reconcile", + "Reconcile mail delivery outcomes", + "Reconcile unknown delivery outcomes and explicitly authorize deliberate resend commands.", + ), _permission( "mail:secret:manage_own", "Manage own mail secrets", @@ -109,6 +127,8 @@ ROLE_TEMPLATES = ( "mail:mailbox:read", "mail:profile:write", "mail:secret:manage", + "mail:delivery:diagnostic", + "mail:delivery:reconcile", ), ), RoleTemplate( @@ -158,6 +178,8 @@ manifest = ModuleManifest( optional_dependencies=("campaigns", "addresses"), provides_interfaces=( ModuleInterfaceProvider(name="mail.campaign_delivery", version="0.2.0"), + ModuleInterfaceProvider(name="mail.delivery_commands", version="0.1.0"), + ModuleInterfaceProvider(name="mail.delivery_outbox", version="0.1.0"), ), requires_interfaces=( ModuleInterfaceRequirement( @@ -219,11 +241,18 @@ manifest = ModuleManifest( mail_models.MailProfilePolicy, mail_models.MailMailboxFolderIndex, mail_models.MailMailboxMessageIndex, + mail_models.MailDeliveryReconciliation, + mail_models.MailDeliveryAttempt, + mail_models.MailDeliveryCommand, label="Mail", ), ), capability_factories={ "mail.campaign_delivery": lambda context: __import__("govoplan_mail.backend.capabilities", fromlist=["campaign_capability"]).campaign_capability(context), + CAPABILITY_MAIL_DELIVERY_OUTBOX: lambda context: __import__( + "govoplan_mail.backend.capabilities", + fromlist=["delivery_outbox_capability"], + ).delivery_outbox_capability(context), }, documentation=( DocumentationTopic( @@ -427,7 +456,7 @@ manifest = ModuleManifest( id="mail.reference.campaign-delivery-contract", title="Integrate Campaign through the Mail delivery contract", summary="Campaign freezes a Mail profile reference and opaque revision; Mail re-authorizes, revision-checks, resolves credentials, and performs the effect in one call.", - body="The mail.campaign_delivery 0.2 contract never returns decrypted credentials or resolved SMTP/IMAP configuration. Mail compares the expected random transport revision before decrypting protocol-specific credentials and returns only bounded sanitized outcomes. Campaign owns durable recipient jobs, retry, and reconciliation. Live report emailing remains disabled until the planned Mail-owned idempotent outbox and attempt ledger are implemented.", + body="The mail.campaign_delivery 0.2 contract never returns decrypted credentials or resolved SMTP/IMAP configuration. Mail compares the expected random transport revision before decrypting protocol-specific credentials and returns only bounded sanitized outcomes. Campaign owns ordinary recipient jobs; report messages use Mail's encrypted idempotent delivery-command and attempt ledger. Effect-start evidence prevents blind redelivery, unknown outcomes require explicit reconciliation, and raw recipient refusals require Mail diagnostic authority.", layer="available", documentation_types=("admin", "user"), audience=("integrator", "campaign_manager", "campaign_sender", "release_reviewer"), @@ -460,6 +489,12 @@ manifest = ModuleManifest( ), ), documentation_providers=(documentation_topics,), + documentation_configuration_providers=( + DocumentationConfigurationProviderRegistration( + keys=("mail_profile_policy",), + resolve=documentation_configuration_states, + ), + ), ) diff --git a/src/govoplan_mail/backend/migrations/dev_versions/82a3b4c5d6e7_mail_delivery_outbox.py b/src/govoplan_mail/backend/migrations/dev_versions/82a3b4c5d6e7_mail_delivery_outbox.py new file mode 100644 index 0000000..c36e24d --- /dev/null +++ b/src/govoplan_mail/backend/migrations/dev_versions/82a3b4c5d6e7_mail_delivery_outbox.py @@ -0,0 +1,22 @@ +"""add durable mail delivery outbox + +Revision ID: 82a3b4c5d6e7 +Revises: 7192a3bcdef0 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from importlib import import_module + + +delivery_outbox = import_module( + "govoplan_mail.backend.migrations.versions.82a3b4c5d6e7_mail_delivery_outbox" +) + + +revision = delivery_outbox.revision +down_revision = delivery_outbox.down_revision +branch_labels = delivery_outbox.branch_labels +depends_on = delivery_outbox.depends_on +upgrade = delivery_outbox.upgrade +downgrade = delivery_outbox.downgrade diff --git a/src/govoplan_mail/backend/migrations/versions/82a3b4c5d6e7_mail_delivery_outbox.py b/src/govoplan_mail/backend/migrations/versions/82a3b4c5d6e7_mail_delivery_outbox.py new file mode 100644 index 0000000..b3de117 --- /dev/null +++ b/src/govoplan_mail/backend/migrations/versions/82a3b4c5d6e7_mail_delivery_outbox.py @@ -0,0 +1,243 @@ +"""add durable mail delivery outbox + +Revision ID: 82a3b4c5d6e7 +Revises: 7192a3bcdef0 +Create Date: 2026-07-30 00:00:00.000000 +""" +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "82a3b4c5d6e7" +down_revision = "7192a3bcdef0" +branch_labels = None +depends_on = "c91f0a72be34" + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + tables = set(inspector.get_table_names()) + if "mail_delivery_commands" not in tables: + op.create_table( + "mail_delivery_commands", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=False), + sa.Column("command_type", sa.String(length=60), nullable=False), + sa.Column("source_module", sa.String(length=80), nullable=False), + sa.Column("source_resource_type", sa.String(length=80), nullable=False), + sa.Column("source_resource_id", sa.String(length=120), nullable=True), + sa.Column("source_version_id", sa.String(length=120), nullable=True), + sa.Column("idempotency_key", sa.String(length=200), nullable=False), + sa.Column("canonical_request_hash", sa.String(length=64), nullable=False), + sa.Column("profile_id", sa.String(length=36), nullable=False), + sa.Column("smtp_server_id", sa.String(length=36), nullable=True), + sa.Column("smtp_credential_id", sa.String(length=36), nullable=True), + sa.Column( + "expected_smtp_transport_revision", + sa.String(length=120), + nullable=False, + ), + sa.Column("envelope_from_encrypted", sa.Text(), nullable=True), + sa.Column("envelope_recipients_encrypted", sa.Text(), nullable=True), + sa.Column("from_header_encrypted", sa.Text(), nullable=True), + sa.Column("message_encrypted", sa.Text(), nullable=True), + sa.Column("message_sha256", sa.String(length=64), nullable=False), + sa.Column("message_size_bytes", sa.BigInteger(), nullable=False), + sa.Column("recipient_count", sa.Integer(), nullable=False), + sa.Column("status", sa.String(length=40), nullable=False), + sa.Column("attempt_count", sa.Integer(), nullable=False), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("claimed_by", sa.String(length=255), nullable=True), + sa.Column("claimed_at", sa.DateTime(timezone=True), 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("accepted_count", sa.Integer(), nullable=False), + sa.Column("refused_count", sa.Integer(), nullable=False), + sa.Column("refusal_summary", sa.JSON(), nullable=False), + sa.Column("refusal_details_encrypted", sa.Text(), nullable=True), + sa.Column("failure_code", sa.String(length=80), nullable=True), + sa.Column("failure_summary", sa.String(length=500), nullable=True), + sa.Column("created_by_user_id", sa.String(length=36), nullable=True), + sa.Column("supersedes_command_id", sa.String(length=36), nullable=True), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("payload_purged_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( + ["profile_id"], + ["mail_server_profiles.id"], + name=op.f( + "fk_mail_delivery_commands_profile_id_mail_server_profiles" + ), + ondelete="RESTRICT", + ), + sa.ForeignKeyConstraint( + ["created_by_user_id"], + ["access_users.id"], + name=op.f( + "fk_mail_delivery_commands_created_by_user_id_access_users" + ), + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint( + ["supersedes_command_id"], + ["mail_delivery_commands.id"], + name=op.f( + "fk_mail_delivery_commands_supersedes_command_id_mail_delivery_commands" + ), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_mail_delivery_commands"), + ), + sa.UniqueConstraint( + "tenant_id", + "command_type", + "idempotency_key", + name="uq_mail_delivery_commands_idempotency", + ), + ) + op.create_index( + "ix_mail_delivery_commands_dispatch", + "mail_delivery_commands", + ["status", "next_attempt_at", "created_at"], + unique=False, + ) + op.create_index( + "ix_mail_delivery_commands_source", + "mail_delivery_commands", + [ + "tenant_id", + "source_module", + "source_resource_type", + "source_resource_id", + ], + unique=False, + ) + for column in ( + "tenant_id", + "command_type", + "source_module", + "profile_id", + "status", + "next_attempt_at", + "created_by_user_id", + "supersedes_command_id", + "expires_at", + ): + op.create_index( + op.f(f"ix_mail_delivery_commands_{column}"), + "mail_delivery_commands", + [column], + unique=False, + ) + + inspector = sa.inspect(op.get_bind()) + if "mail_delivery_attempts" not in inspector.get_table_names(): + op.create_table( + "mail_delivery_attempts", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("command_id", sa.String(length=36), nullable=False), + sa.Column("attempt_number", sa.Integer(), nullable=False), + sa.Column("worker_id", sa.String(length=255), nullable=True), + sa.Column("status", sa.String(length=40), 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( + ["command_id"], + ["mail_delivery_commands.id"], + name=op.f( + "fk_mail_delivery_attempts_command_id_mail_delivery_commands" + ), + ondelete="CASCADE", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_mail_delivery_attempts"), + ), + sa.UniqueConstraint( + "command_id", + "attempt_number", + name="uq_mail_delivery_attempts_number", + ), + ) + op.create_index( + "ix_mail_delivery_attempts_command_started", + "mail_delivery_attempts", + ["command_id", "started_at"], + unique=False, + ) + for column in ("command_id", "status"): + op.create_index( + op.f(f"ix_mail_delivery_attempts_{column}"), + "mail_delivery_attempts", + [column], + unique=False, + ) + + inspector = sa.inspect(op.get_bind()) + if "mail_delivery_reconciliations" not in inspector.get_table_names(): + op.create_table( + "mail_delivery_reconciliations", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("command_id", sa.String(length=36), nullable=False), + sa.Column("decision", sa.String(length=40), nullable=False), + sa.Column("evidence_reference", sa.String(length=500), nullable=False), + sa.Column("note_encrypted", sa.Text(), nullable=True), + sa.Column("created_by_user_id", sa.String(length=36), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["command_id"], + ["mail_delivery_commands.id"], + name=op.f( + "fk_mail_delivery_reconciliations_command_id_mail_delivery_commands" + ), + ondelete="CASCADE", + ), + sa.ForeignKeyConstraint( + ["created_by_user_id"], + ["access_users.id"], + name=op.f( + "fk_mail_delivery_reconciliations_created_by_user_id_access_users" + ), + ondelete="SET NULL", + ), + sa.PrimaryKeyConstraint( + "id", + name=op.f("pk_mail_delivery_reconciliations"), + ), + ) + op.create_index( + "ix_mail_delivery_reconciliations_command_created", + "mail_delivery_reconciliations", + ["command_id", "created_at"], + unique=False, + ) + for column in ("command_id", "created_by_user_id"): + op.create_index( + op.f(f"ix_mail_delivery_reconciliations_{column}"), + "mail_delivery_reconciliations", + [column], + unique=False, + ) + + +def downgrade() -> None: + tables = set(sa.inspect(op.get_bind()).get_table_names()) + if "mail_delivery_reconciliations" in tables: + op.drop_table("mail_delivery_reconciliations") + if "mail_delivery_attempts" in tables: + op.drop_table("mail_delivery_attempts") + if "mail_delivery_commands" in tables: + op.drop_table("mail_delivery_commands") diff --git a/src/govoplan_mail/backend/router.py b/src/govoplan_mail/backend/router.py index e5a3d85..f332e45 100644 --- a/src/govoplan_mail/backend/router.py +++ b/src/govoplan_mail/backend/router.py @@ -20,6 +20,9 @@ from govoplan_mail.backend.schemas import ( MailCredentialUpdateRequest, MailImapFolderListResponse, MailImapFolderResponse, + MailDeliveryCommandResponse, + MailDeliveryReconcileRequest, + MailDeliveryResendRequest, MailMailboxBootstrapResponse, MailImapTestRequest, MailMailboxAttachmentResponse, @@ -79,6 +82,14 @@ from govoplan_mail.backend.mail_profiles import ( ) from govoplan_mail.backend.config import ImapConfig, SmtpConfig from govoplan_mail.backend.runtime import get_registry +from govoplan_mail.backend.delivery_outbox import ( + MailDeliveryError, + delivery_command_diagnostics, + delivery_command_summary, + get_delivery_command, + reconcile_delivery_command, + resend_delivery_command, +) from govoplan_mail.backend.server_hierarchy import ( MailHierarchyContext, MailServerHierarchyError, @@ -120,6 +131,114 @@ DEFAULT_MAILBOX_MESSAGE_LIMIT = 50 CAPABILITY_ADDRESSES_LOOKUP = "addresses.lookup" +@router.get( + "/delivery-commands/{command_id}", + response_model=MailDeliveryCommandResponse, +) +def get_mail_delivery_command( + command_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + """Return the business-safe status of one tenant Mail command.""" + + _require_scope(principal, "mail:profile:read") + try: + command = get_delivery_command( + session, + tenant_id=principal.tenant_id, + command_id=command_id, + ) + except MailDeliveryError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + return MailDeliveryCommandResponse(result=delivery_command_summary(command)) + + +@router.get( + "/delivery-commands/{command_id}/diagnostics", + response_model=MailDeliveryCommandResponse, +) +def get_mail_delivery_command_diagnostics( + command_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + """Return bounded recipient and attempt evidence to diagnostic actors.""" + + _require_scope(principal, "mail:delivery:diagnostic") + try: + result = delivery_command_diagnostics( + session, + tenant_id=principal.tenant_id, + command_id=command_id, + ) + except MailDeliveryError as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(exc), + ) from exc + return MailDeliveryCommandResponse(result=result) + + +@router.post( + "/delivery-commands/{command_id}/reconcile", + response_model=MailDeliveryCommandResponse, +) +def reconcile_mail_delivery_command( + command_id: str, + payload: MailDeliveryReconcileRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + _require_scope(principal, "mail:delivery:reconcile") + try: + result = reconcile_delivery_command( + session, + tenant_id=principal.tenant_id, + command_id=command_id, + decision=payload.decision, + evidence_reference=payload.evidence_reference, + note=payload.note, + user_id=principal.user.id, + ) + except MailDeliveryError as exc: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=str(exc), + ) from exc + return MailDeliveryCommandResponse(result=result) + + +@router.post( + "/delivery-commands/{command_id}/resend", + response_model=MailDeliveryCommandResponse, +) +def resend_mail_delivery_command( + command_id: str, + payload: MailDeliveryResendRequest, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +): + _require_scope(principal, "mail:delivery:reconcile") + try: + result = resend_delivery_command( + session, + tenant_id=principal.tenant_id, + command_id=command_id, + idempotency_key=payload.idempotency_key, + user_id=principal.user.id, + ) + except MailDeliveryError as exc: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=str(exc), + ) from exc + return MailDeliveryCommandResponse(result=result) + + def _capability_payload(value: object) -> dict[str, Any]: if dataclasses.is_dataclass(value): return dataclasses.asdict(value) diff --git a/src/govoplan_mail/backend/schemas.py b/src/govoplan_mail/backend/schemas.py index 1d0bbad..b85d47c 100644 --- a/src/govoplan_mail/backend/schemas.py +++ b/src/govoplan_mail/backend/schemas.py @@ -436,3 +436,21 @@ class MailMailboxMessageResponse(BaseModel): port: int | None = None security: str | None = None message: MailMailboxMessageDetailResponse + + +class MailDeliveryCommandResponse(BaseModel): + result: dict[str, Any] + + +class MailDeliveryReconcileRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + decision: Literal["accepted", "not_accepted"] + evidence_reference: str = Field(min_length=1, max_length=500) + note: str | None = Field(default=None, max_length=4000) + + +class MailDeliveryResendRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + idempotency_key: str = Field(min_length=1, max_length=200) diff --git a/tests/test_delivery_outbox.py b/tests/test_delivery_outbox.py new file mode 100644 index 0000000..c78c848 --- /dev/null +++ b/tests/test_delivery_outbox.py @@ -0,0 +1,294 @@ +from __future__ import annotations + +import unittest +from datetime import timedelta +from types import SimpleNamespace +from unittest.mock import patch + +from sqlalchemy import Column, String, Table, create_engine +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from govoplan_core.db.base import Base +from govoplan_mail.backend.db.models import ( + MailDeliveryAttempt, + MailDeliveryCommand, + MailDeliveryReconciliation, + MailServerProfile, +) +from govoplan_mail.backend.delivery_outbox import ( + MailDeliveryIdempotencyConflict, + delivery_command_diagnostics, + dispatch_due, + purge_expired, + submit_delivery_command, + utcnow, +) +from govoplan_mail.backend.sending.smtp import SmtpSendError + + +class MailDeliveryOutboxTests(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), + ) + Base.metadata.create_all( + self.engine, + tables=[ + access_users, + MailServerProfile.__table__, + MailDeliveryCommand.__table__, + MailDeliveryAttempt.__table__, + MailDeliveryReconciliation.__table__, + ], + ) + self.SessionLocal = sessionmaker( + bind=self.engine, + class_=Session, + expire_on_commit=False, + ) + with self.SessionLocal() as session: + session.add( + MailServerProfile( + id="profile-1", + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + name="Delivery", + slug="delivery", + smtp_config={"host": "smtp.example.test", "port": 25}, + ) + ) + session.commit() + self.audit = patch( + "govoplan_mail.backend.delivery_outbox.audit_event" + ) + self.audit.start() + self.addCleanup(self.audit.stop) + self.addCleanup(self.engine.dispose) + + def _submit( + self, + session: Session, + *, + key: str = "request-1", + recipients: list[str] | None = None, + ) -> dict[str, object]: + return submit_delivery_command( + session, + tenant_id="tenant-1", + command_type="campaign_report", + source_module="campaigns", + source_resource_type="campaign", + source_resource_id="campaign-1", + source_version_id="version-1", + idempotency_key=key, + profile_id="profile-1", + message_bytes=b"Subject: report\r\n\r\ncontent", + envelope_from="sender@example.test", + envelope_recipients=recipients or ["recipient@example.test"], + from_header="Sender ", + expected_smtp_transport_revision="revision-1", + created_by_user_id=None, + ) + + def test_submission_is_idempotent_and_conflicts_on_changed_intent(self) -> None: + with self.SessionLocal() as session: + first = self._submit(session) + session.commit() + repeated = self._submit(session) + + self.assertEqual(first["id"], repeated["id"]) + self.assertTrue(repeated["duplicate"]) + with self.assertRaises(MailDeliveryIdempotencyConflict): + self._submit( + session, + recipients=["different@example.test"], + ) + + def test_attempt_is_committed_before_effect_and_accepted_is_not_resent(self) -> None: + observed: dict[str, object] = {} + with self.SessionLocal() as session: + command_id = str(self._submit(session)["id"]) + session.commit() + + def provider_effect(session: Session, **_kwargs): + command = session.get(MailDeliveryCommand, command_id) + attempt = session.query(MailDeliveryAttempt).one() + observed.update( + command_status=command.status, + effect_started=command.effect_started_at is not None, + attempt_status=attempt.status, + ) + return SimpleNamespace( + accepted_count=1, + refused_recipients={}, + ) + + with ( + self.SessionLocal() as session, + patch( + "govoplan_mail.backend.delivery_outbox.send_campaign_email_bytes", + side_effect=provider_effect, + ) as send, + ): + result = dispatch_due(session, worker_id="worker-1") + repeated = dispatch_due(session, worker_id="worker-2") + + self.assertEqual(result["accepted"], 1) + self.assertEqual(repeated["selected"], 0) + self.assertEqual(send.call_count, 1) + self.assertEqual( + observed, + { + "command_status": "in_progress", + "effect_started": True, + "attempt_status": "in_progress", + }, + ) + + def test_partial_refusal_is_terminal_and_diagnostics_are_separate(self) -> None: + with self.SessionLocal() as session: + command_id = str( + self._submit( + session, + recipients=["accepted@example.test", "blocked@example.test"], + )["id"] + ) + session.commit() + with patch( + "govoplan_mail.backend.delivery_outbox.send_campaign_email_bytes", + return_value=SimpleNamespace( + accepted_count=1, + refused_recipients={ + "blocked@example.test": { + "status_code": 550, + "classification": "permanent", + "message": "Permanent recipient rejection", + } + }, + ), + ): + result = dispatch_due(session) + command = session.get(MailDeliveryCommand, command_id) + assert command is not None + self.assertEqual(result["partially_refused"], 1) + self.assertNotIn("blocked@example.test", repr({ + "status": command.status, + "summary": command.refusal_summary, + })) + diagnostics = delivery_command_diagnostics( + session, + tenant_id="tenant-1", + command_id=command_id, + ) + self.assertIn( + "blocked@example.test", + diagnostics["refused_recipients"], + ) + + def test_unknown_outcome_is_never_automatically_retried(self) -> None: + with self.SessionLocal() as session: + command_id = str(self._submit(session)["id"]) + session.commit() + with patch( + "govoplan_mail.backend.delivery_outbox.send_campaign_email_bytes", + side_effect=SmtpSendError( + "unknown", + outcome_unknown=True, + ), + ) as send: + first = dispatch_due(session) + second = dispatch_due(session) + command = session.get(MailDeliveryCommand, command_id) + + assert command is not None + self.assertEqual(first["outcome_unknown"], 1) + self.assertEqual(second["selected"], 0) + self.assertEqual(command.status, "outcome_unknown") + self.assertEqual(send.call_count, 1) + + def test_audit_failure_after_acceptance_cannot_make_command_retryable(self) -> None: + with self.SessionLocal() as session: + command_id = str(self._submit(session)["id"]) + session.commit() + + with ( + self.SessionLocal() as session, + patch( + "govoplan_mail.backend.delivery_outbox.send_campaign_email_bytes", + return_value=SimpleNamespace( + accepted_count=1, + refused_recipients={}, + ), + ) as send, + patch( + "govoplan_mail.backend.delivery_outbox.audit_event", + side_effect=RuntimeError("audit unavailable"), + ), + ): + with self.assertRaisesRegex(RuntimeError, "audit unavailable"): + dispatch_due(session) + + with self.SessionLocal() as session: + command = session.get(MailDeliveryCommand, command_id) + assert command is not None + self.assertEqual(command.status, "accepted") + self.assertEqual(dispatch_due(session)["selected"], 0) + self.assertEqual(send.call_count, 1) + + def test_stale_effect_started_command_becomes_unknown_without_send(self) -> None: + with self.SessionLocal() as session: + command_id = str(self._submit(session)["id"]) + session.commit() + command = session.get(MailDeliveryCommand, command_id) + assert command is not None + command.status = "in_progress" + command.attempt_count = 1 + command.claimed_at = utcnow() - timedelta(hours=1) + command.effect_started_at = utcnow() - timedelta(hours=1) + session.add( + MailDeliveryAttempt( + command_id=command.id, + attempt_number=1, + status="in_progress", + started_at=utcnow() - timedelta(hours=1), + effect_started_at=utcnow() - timedelta(hours=1), + ) + ) + session.commit() + with patch( + "govoplan_mail.backend.delivery_outbox.send_campaign_email_bytes" + ) as send: + result = dispatch_due(session) + self.assertEqual(result["outcome_unknown"], 1) + self.assertEqual(command.status, "outcome_unknown") + send.assert_not_called() + + def test_expired_payload_is_minimized_but_evidence_remains(self) -> None: + with self.SessionLocal() as session: + command_id = str(self._submit(session)["id"]) + session.commit() + command = session.get(MailDeliveryCommand, command_id) + assert command is not None + command.expires_at = utcnow() - timedelta(seconds=1) + session.commit() + self.assertEqual(purge_expired(session), {"purged": 1}) + session.refresh(command) + self.assertIsNone(command.message_encrypted) + self.assertIsNone(command.envelope_recipients_encrypted) + self.assertIsNotNone(command.payload_purged_at) + self.assertEqual(command.message_sha256, command.message_sha256) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_documentation.py b/tests/test_documentation.py index 1d00298..7272c3a 100644 --- a/tests/test_documentation.py +++ b/tests/test_documentation.py @@ -7,7 +7,10 @@ from unittest.mock import patch from sqlalchemy.orm import Session from govoplan_core.core.modules import DocumentationContext -from govoplan_mail.backend.documentation import documentation_topics +from govoplan_mail.backend.documentation import ( + documentation_configuration_states, + documentation_topics, +) from govoplan_mail.backend.mail_profiles import EffectiveMailProfilePolicy, MailProfileError @@ -181,6 +184,48 @@ class MailRuntimeDocumentationTests(unittest.TestCase): self.assertEqual(topics["mail.workflow.choose-and-test-profile"].metadata["help_contexts"], ["mail.profiles", "app.settings"]) self.assertEqual(topics["mail.workflow.read-mailbox"].metadata["help_contexts"], ["mail.list"]) + def test_configuration_provider_exposes_only_explicit_or_inherited_state(self) -> None: + inherited = EffectiveMailProfilePolicy( + source_policies=[ + {"scope_type": "system", "applied_fields": ["defaults"]}, + {"scope_type": "tenant", "applied_fields": []}, + ] + ) + explicit = EffectiveMailProfilePolicy( + source_policies=[ + {"scope_type": "system", "applied_fields": ["defaults"]}, + {"scope_type": "tenant", "applied_fields": ["smtp_hosts"]}, + ] + ) + context = self.context({"mail:profile:read"}, documentation_type="admin") + + with patch( + "govoplan_mail.backend.documentation.effective_mail_profile_policy_for_scope", + return_value=inherited, + ): + inherited_state = documentation_configuration_states( + context, + ("mail_profile_policy",), + ) + with patch( + "govoplan_mail.backend.documentation.effective_mail_profile_policy_for_scope", + return_value=explicit, + ): + explicit_state = documentation_configuration_states( + context, + ("mail_profile_policy",), + ) + + self.assertEqual( + inherited_state["mail_profile_policy"].state, + "inherited", + ) + self.assertEqual( + explicit_state["mail_profile_policy"].state, + "enabled", + ) + self.assertNotIn("smtp_hosts", repr(explicit_state)) + if __name__ == "__main__": unittest.main()