from __future__ import annotations import base64 import hashlib import json from collections import Counter from datetime import datetime, timedelta, timezone from email import policy from email.parser import BytesParser from typing import Any from sqlalchemy import 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.recovery import ( MailRecoveryError, reconcile_outbox_provider_effect, ) 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 _rfc_message_id(message: bytes) -> str | None: try: value = BytesParser(policy=policy.default).parsebytes( message, headersonly=True, ).get("Message-ID") except Exception: return None normalized = " ".join(str(value or "").split()) return normalized[:998] or None 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() rfc_message_id = _rfc_message_id(message_bytes) 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, rfc_message_id=rfc_message_id, 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, "rfc_message_id": command.rfc_message_id, "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, recovery_effect_id=( f"outbox:{command.id}:smtp-attempt:{attempt.attempt_number}" ), recovery_resource_type="mail_delivery_command", recovery_resource_id=command.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") try: reconcile_outbox_provider_effect( command_id=command.id, effect_occurred=clean_decision == "accepted", evidence_reference=clean_evidence, user_id=user_id, ) except MailRecoveryError as exc: raise MailDeliveryStateError(str(exc)) from exc 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, *, tenant_id: str | None = None, limit: int = 250, ) -> dict[str, object]: now = utcnow() clauses = [ MailDeliveryCommand.expires_at <= now, MailDeliveryCommand.payload_purged_at.is_(None), ] if tenant_id: clauses.append(MailDeliveryCommand.tenant_id == tenant_id) commands = session.scalars( select(MailDeliveryCommand) .where(*clauses) .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)