fase 4

worker/app/jobs/send_pec.py

@@ -0,0 +1,267 @@
+"""
+Job arq: send_pec – invio PEC con retry esponenziale.
+
+Flusso completo:
+  1. API backend crea Message(state=queued) + SendJob(status=pending)
+  2. Backend enqueue send_pec via arq
+  3. send_pec legge il job dal DB, tenta l'invio SMTP
+  4. Successo → Message(state=sent), SendJob(status=sent), upload EML su MinIO,
+                 enqueue watch_receipt con defer 24h
+  5. Fallimento transitorio → status=retrying, re-enqueue con backoff
+  6. Fallimento definitivo (max_attempts raggiunto) → status=failed,
+                 Message(state=failed), evento WS
+
+Retry backoff (max 5 tentativi totali):
+  Tentativo 1 fallisce → attendi  1 min  → tentativo 2
+  Tentativo 2 fallisce → attendi  5 min  → tentativo 3
+  Tentativo 3 fallisce → attendi 15 min  → tentativo 4
+  Tentativo 4 fallisce → attendi  1 ora  → tentativo 5
+  Tentativo 5 fallisce → FAILED (no retry)
+"""
+
+import io
+import json
+import logging
+import uuid as uuid_module
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from sqlalchemy import select
+
+from app.database import AsyncSessionLocal
+from app.models import Mailbox, Message, SendJob
+from app.smtp.sender import SmtpSender
+from app.storage.minio_client import upload_outbound_eml
+
+logger = logging.getLogger(__name__)
+
+# ─── Configurazione retry ─────────────────────────────────────────────────────
+
+MAX_ATTEMPTS = 5
+
+# Delay in secondi dopo il fallimento del tentativo N (0-based)
+RETRY_DELAYS = [
+    60,      # dopo tentativo 1 fallisce → 1 min
+    300,     # dopo tentativo 2 fallisce → 5 min
+    900,     # dopo tentativo 3 fallisce → 15 min
+    3600,    # dopo tentativo 4 fallisce → 1 ora
+    # tentativo 5 = ultimo → nessun retry
+]
+
+
+# ─── Job principale ───────────────────────────────────────────────────────────
+
+async def send_pec(ctx: dict[str, Any], send_job_id: str) -> dict:
+    """
+    Job arq: invia una PEC.
+
+    Args:
+        ctx:          contesto arq (contiene redis client arq per re-enqueue)
+        send_job_id:  UUID del SendJob da processare
+
+    Returns:
+        dict con esito: status, message_id, attempt
+    """
+    redis_client = ctx.get("redis")
+
+    async with AsyncSessionLocal() as db:
+        # ── Carica dati ──────────────────────────────────────────────────────
+        job = await db.get(SendJob, uuid_module.UUID(send_job_id))
+        if not job:
+            logger.error(f"[send_pec] SendJob {send_job_id} non trovato")
+            return {"status": "error", "message": "SendJob non trovato"}
+
+        if job.status in ("sent", "failed"):
+            logger.warning(
+                f"[send_pec] SendJob {send_job_id} già in stato {job.status!r}, skip"
+            )
+            return {"status": "skipped", "current_status": job.status}
+
+        msg = await db.get(Message, job.message_id)
+        if not msg:
+            logger.error(f"[send_pec] Messaggio {job.message_id} non trovato")
+            job.status = "failed"
+            job.last_error = "Messaggio associato non trovato"
+            await db.commit()
+            return {"status": "error", "message": "Messaggio non trovato"}
+
+        mailbox = await db.get(Mailbox, job.mailbox_id)
+        if not mailbox:
+            logger.error(f"[send_pec] Casella {job.mailbox_id} non trovata")
+            job.status = "failed"
+            job.last_error = "Casella mittente non trovata"
+            msg.state = "failed"
+            await db.commit()
+            return {"status": "error", "message": "Casella non trovata"}
+
+        # ── Aggiorna contatori ────────────────────────────────────────────────
+        job.status = "sending"
+        job.attempt_count += 1
+        current_attempt = job.attempt_count
+        await db.flush()
+
+        logger.info(
+            f"[send_pec] Tentativo {current_attempt}/{MAX_ATTEMPTS} "
+            f"per job {send_job_id} → {msg.to_addresses}"
+        )
+
+        # ── Tenta invio SMTP ──────────────────────────────────────────────────
+        try:
+            sender = SmtpSender(mailbox)
+            message_id_header, raw_eml = await sender.send(
+                to_addresses=list(msg.to_addresses or []),
+                cc_addresses=list(msg.cc_addresses or []),
+                subject=msg.subject or "",
+                body_text=msg.body_text or "",
+                body_html=msg.body_html,
+                attachments=None,  # allegati in fase successiva (Fase 5)
+            )
+
+            # ── Successo: aggiorna DB ─────────────────────────────────────────
+            now = datetime.now(tz=timezone.utc)
+            msg.message_id_header = message_id_header
+            msg.state = "sent"
+            msg.sent_at = now
+
+            job.status = "sent"
+            job.sent_at = now
+            job.message_id = msg.id
+
+            # Upload raw EML su MinIO
+            try:
+                eml_path = await upload_outbound_eml(
+                    tenant_id=str(msg.tenant_id),
+                    mailbox_id=str(msg.mailbox_id),
+                    message_id=str(msg.id),
+                    eml_bytes=raw_eml,
+                )
+                msg.raw_eml_path = eml_path
+                logger.debug(f"[send_pec] EML salvato: {eml_path}")
+            except Exception as minio_err:
+                logger.warning(f"[send_pec] Upload MinIO fallito (non critico): {minio_err}")
+
+            await db.commit()
+
+            # ── Pubblica evento WS ────────────────────────────────────────────
+            if redis_client:
+                await _publish_ws_event(redis_client, msg.tenant_id, {
+                    "type": "message:sent",
+                    "message_id": str(msg.id),
+                    "mailbox_id": str(msg.mailbox_id),
+                    "subject": msg.subject,
+                    "message_id_header": message_id_header,
+                })
+
+            # ── Enqueue watch_receipt dopo 24h ────────────────────────────────
+            try:
+                await redis_client.enqueue_job(
+                    "watch_receipt",
+                    str(msg.id),
+                    _defer_by=timedelta(hours=24),
+                )
+                logger.info(
+                    f"[send_pec] watch_receipt schedulato per {msg.id} "
+                    f"tra 24h"
+                )
+            except Exception as e:
+                logger.warning(f"[send_pec] Errore enqueue watch_receipt: {e}")
+
+            logger.info(
+                f"[send_pec] ✅ PEC inviata: job={send_job_id} "
+                f"message_id_header={message_id_header}"
+            )
+            return {
+                "status": "sent",
+                "send_job_id": send_job_id,
+                "message_id": str(msg.id),
+                "message_id_header": message_id_header,
+                "attempt": current_attempt,
+            }
+
+        except Exception as smtp_error:
+            error_msg = str(smtp_error)
+            logger.warning(
+                f"[send_pec] Tentativo {current_attempt} fallito: {error_msg}"
+            )
+
+            # ── Gestione retry / failure ──────────────────────────────────────
+            if current_attempt >= MAX_ATTEMPTS:
+                # Esauriti tutti i tentativi → FAILED
+                job.status = "failed"
+                job.last_error = error_msg
+                msg.state = "failed"
+                await db.commit()
+
+                # Pubblica evento WS: invio fallito
+                if redis_client:
+                    await _publish_ws_event(redis_client, msg.tenant_id, {
+                        "type": "message:send_failed",
+                        "message_id": str(msg.id),
+                        "mailbox_id": str(msg.mailbox_id),
+                        "subject": msg.subject,
+                        "error": error_msg,
+                        "attempts": current_attempt,
+                    })
+
+                logger.error(
+                    f"[send_pec] ❌ Invio FALLITO definitivamente: "
+                    f"job={send_job_id}, errore: {error_msg}"
+                )
+                return {
+                    "status": "failed",
+                    "send_job_id": send_job_id,
+                    "error": error_msg,
+                    "attempts": current_attempt,
+                }
+
+            else:
+                # Retry con backoff
+                delay_seconds = RETRY_DELAYS[current_attempt - 1]
+                next_retry = datetime.now(tz=timezone.utc) + timedelta(seconds=delay_seconds)
+
+                job.status = "retrying"
+                job.last_error = error_msg
+                job.next_retry_at = next_retry
+                msg.state = "queued"  # torna in coda
+                await db.commit()
+
+                # Re-enqueue con defer
+                try:
+                    await redis_client.enqueue_job(
+                        "send_pec",
+                        send_job_id,
+                        _defer_by=timedelta(seconds=delay_seconds),
+                    )
+                    logger.info(
+                        f"[send_pec] Retry {current_attempt} schedulato "
+                        f"in {delay_seconds}s per job {send_job_id}"
+                    )
+                except Exception as enqueue_err:
+                    logger.error(
+                        f"[send_pec] Impossibile re-enqueue job {send_job_id}: "
+                        f"{enqueue_err}"
+                    )
+
+                return {
+                    "status": "retrying",
+                    "send_job_id": send_job_id,
+                    "attempt": current_attempt,
+                    "next_retry_at": next_retry.isoformat(),
+                    "delay_seconds": delay_seconds,
+                    "error": error_msg,
+                }
+
+
+# ─── Helpers ──────────────────────────────────────────────────────────────────
+
+async def _publish_ws_event(
+    redis_client: Any,
+    tenant_id: uuid_module.UUID,
+    event: dict,
+) -> None:
+    """Pubblica un evento WebSocket per il tenant tramite Redis pub/sub."""
+    try:
+        channel = f"ws:tenant:{tenant_id}"
+        await redis_client.publish(channel, json.dumps(event, default=str))
+    except Exception as e:
+        logger.warning(f"[send_pec] Errore pubblicazione WS: {e}")

worker/app/main.py

@@ -24,7 +24,9 @@ from arq.connections import RedisSettings
 
 from app.config import get_settings
 from app.imap.pool import MailboxPool
+from app.jobs.send_pec import send_pec
 from app.jobs.sync_mailbox import sync_mailbox
+from app.smtp.receipt_watcher import watch_receipt
 from app.storage.minio_client import ensure_bucket_exists
 
 settings = get_settings()
@@ -127,7 +129,7 @@ class WorkerSettings:
     """Configurazione del worker arq."""
 
     # Funzioni/job registrati
-    functions = [sync_mailbox, health_check]
+    functions = [sync_mailbox, send_pec, watch_receipt, health_check]
 
     # Callbacks lifecycle
     on_startup = on_startup
@@ -140,7 +142,8 @@ class WorkerSettings:
     max_jobs = 20
 
     # Timeout per ogni job (secondi)
-    job_timeout = 300
+    # send_pec può richiedere più tempo su SMTP lenti
+    job_timeout = 120
 
     # Retry automatico in caso di errore
     max_tries = 3

worker/app/models.py

@@ -133,6 +133,44 @@ class Message(Base):
     )
 
 
+SendJobStatus = Enum(
+    "pending", "sending", "sent", "failed", "retrying",
+    name="send_job_status", create_type=False,
+)
+
+
+class SendJob(Base):
+    """
+    Job di invio PEC – traccia ogni tentativo di invio SMTP.
+    Corrisponde alla tabella `send_jobs` nel DB.
+    """
+
+    __tablename__ = "send_jobs"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4
+    )
+    tenant_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
+    mailbox_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False)
+    message_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("messages.id"),
+        nullable=True,
+    )
+    status: Mapped[str] = mapped_column(SendJobStatus, nullable=False, default="pending")
+    attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    max_attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=5)
+    next_retry_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    queued_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now()
+    )
+    sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_by: Mapped[uuid.UUID | None] = mapped_column(UUID(as_uuid=True), nullable=True)
+
+
 class Attachment(Base):
     """
     Allegato di un messaggio PEC.

worker/app/smtp/__init__.py

@@ -0,0 +1 @@
+"""Package SMTP – invio PEC via aiosmtplib."""

worker/app/smtp/receipt_watcher.py

@@ -0,0 +1,94 @@
+"""
+Job arq: watch_receipt – attende la ricevuta di accettazione per una PEC inviata.
+
+Viene enqueued da send_pec dopo un invio riuscito con un defer di 24 ore.
+Se dopo 24h nessuna ricevuta (accettazione o avvenuta_consegna) è arrivata
+tramite IMAP sync, imposta lo stato del messaggio a 'anomaly' e pubblica
+un evento WebSocket all'admin del tenant.
+
+Flow:
+  send_pec → invio OK → enqueue watch_receipt (defer 24h)
+  IMAP sync → ricevuta arriva → aggiorna Message.state a 'accepted'/'delivered'
+  watch_receipt (dopo 24h) → verifica se state == 'accepted'/'delivered'
+                           → no → state = 'anomaly' + WS event
+"""
+
+import json
+import logging
+import uuid as uuid_module
+from typing import Any
+
+from sqlalchemy import select
+
+from app.database import AsyncSessionLocal
+from app.models import Message
+
+logger = logging.getLogger(__name__)
+
+# Stati che indicano ricezione ricevuta (impostati da IMAP sync via pec_parser)
+_ACCEPTED_STATES = {"accepted", "delivered"}
+
+
+async def watch_receipt(ctx: dict[str, Any], message_id: str) -> dict:
+    """
+    Job arq: verifica se il messaggio outbound ha ricevuto accettazione.
+
+    Args:
+        ctx:        contesto arq (redis, ecc.)
+        message_id: UUID del messaggio outbound da monitorare
+
+    Returns:
+        dict con esito del controllo
+    """
+    redis_client = ctx.get("redis")
+
+    async with AsyncSessionLocal() as db:
+        msg = await db.get(Message, uuid_module.UUID(message_id))
+
+        if not msg:
+            logger.warning(f"[watch_receipt] Messaggio {message_id} non trovato")
+            return {"status": "error", "message": "Messaggio non trovato"}
+
+        if msg.direction != "outbound":
+            return {"status": "skipped", "message": "Non è un messaggio outbound"}
+
+        if msg.state in _ACCEPTED_STATES:
+            # Ricevuta già arrivata tramite IMAP sync: OK
+            logger.info(
+                f"[watch_receipt] Messaggio {message_id} ha ricevuto "
+                f"accettazione (state={msg.state!r})"
+            )
+            return {"status": "ok", "state": msg.state}
+
+        # Nessuna ricevuta in 24h → anomalia
+        logger.warning(
+            f"[watch_receipt] Nessuna accettazione in 24h per {message_id} "
+            f"(state={msg.state!r}, mailbox={msg.mailbox_id})"
+        )
+
+        prev_state = msg.state
+        msg.state = "anomaly"
+        await db.commit()
+
+        # Pubblica evento WebSocket al tenant
+        if redis_client:
+            event = {
+                "type": "message:anomaly",
+                "message_id": message_id,
+                "mailbox_id": str(msg.mailbox_id),
+                "subject": msg.subject,
+                "reason": "Nessuna ricevuta di accettazione entro 24 ore",
+                "previous_state": prev_state,
+            }
+            channel = f"ws:tenant:{msg.tenant_id}"
+            try:
+                await redis_client.publish(channel, json.dumps(event, default=str))
+                logger.debug(f"[watch_receipt] Evento anomalia pubblicato su {channel}")
+            except Exception as e:
+                logger.error(f"[watch_receipt] Errore pubblicazione Redis: {e}")
+
+        return {
+            "status": "anomaly",
+            "message_id": message_id,
+            "reason": "Nessuna ricevuta di accettazione entro 24 ore",
+        }

worker/app/smtp/sender.py

@@ -0,0 +1,256 @@
+"""
+SmtpSender – invio PEC via SMTP (SSL/STARTTLS) con aiosmtplib.
+
+Costruisce il messaggio MIME, si connette al server SMTP della casella,
+invia e restituisce il Message-ID e i byte raw EML per l'archiviazione.
+
+Porta 465 → SSL diretto (use_tls=True, start_tls=False)
+Porta 587 → STARTTLS (use_tls=False, start_tls=True)
+Porta 25  → plain (use_tls=False, start_tls=False) – deprecato, non usato
+"""
+
+import base64
+import io
+import logging
+import uuid
+from email import encoders
+from email.headerregistry import Address
+from email.mime.base import MIMEBase
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from email.utils import formatdate, make_msgid
+
+import aiosmtplib
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+from app.config import get_settings
+from app.models import Mailbox
+
+logger = logging.getLogger(__name__)
+
+
+# ─── Helpers ──────────────────────────────────────────────────────────────────
+
+def _decrypt(enc_value: str) -> str:
+    """
+    Decifra un campo credenziale cifrato con AES-256-GCM (ADR-002).
+
+    Usa get_settings() in modo lazy (non module-level) per permettere
+    ai test di iniettare la chiave tramite env var o mock.
+    """
+    raw = base64.b64decode(enc_value)
+    nonce, ciphertext_tag = raw[:12], raw[12:]
+    aesgcm = AESGCM(get_settings().encryption_key_bytes)
+    return aesgcm.decrypt(nonce, ciphertext_tag, None).decode("utf-8")
+
+
+def decrypt_smtp_credentials(mailbox: Mailbox) -> dict:
+    """Restituisce le credenziali SMTP in chiaro della casella."""
+    return {
+        "host": _decrypt(mailbox.smtp_host_enc),
+        "port": int(_decrypt(mailbox.smtp_port_enc)),
+        "user": _decrypt(mailbox.smtp_user_enc),
+        "password": _decrypt(mailbox.smtp_pass_enc),
+        "use_tls": mailbox.smtp_use_tls,
+    }
+
+
+# ─── SmtpSender ───────────────────────────────────────────────────────────────
+
+class SmtpSender:
+    """
+    Gestisce la connessione SMTP e l'invio di un singolo messaggio PEC.
+
+    Esempio di utilizzo::
+
+        sender = SmtpSender(mailbox)
+        msg_id, raw_eml = await sender.send(
+            to_addresses=["dest@pec.it"],
+            cc_addresses=[],
+            subject="Test",
+            body_text="Corpo del messaggio",
+        )
+    """
+
+    def __init__(self, mailbox: Mailbox) -> None:
+        self.mailbox = mailbox
+        self._creds = decrypt_smtp_credentials(mailbox)
+
+    # ── Costruzione MIME ──────────────────────────────────────────────────────
+
+    def build_mime_message(
+        self,
+        to_addresses: list[str],
+        cc_addresses: list[str],
+        subject: str,
+        body_text: str,
+        body_html: str | None = None,
+        attachments: list[dict] | None = None,
+    ) -> tuple[MIMEMultipart, str]:
+        """
+        Costruisce il messaggio MIME per la PEC.
+
+        Args:
+            to_addresses:  destinatari principali
+            cc_addresses:  destinatari in copia (può essere vuoto)
+            subject:       oggetto del messaggio
+            body_text:     corpo in testo semplice
+            body_html:     corpo HTML opzionale
+            attachments:   lista di dict {filename, content: bytes, content_type}
+
+        Returns:
+            (msg MIME, message_id_header)
+        """
+        attachments = attachments or []
+
+        # Struttura MIME
+        if attachments:
+            msg = MIMEMultipart("mixed")
+            body_container = MIMEMultipart("alternative")
+        elif body_html:
+            msg = MIMEMultipart("alternative")
+            body_container = msg
+        else:
+            msg = MIMEMultipart("mixed")
+            body_container = msg
+
+        # Headers obbligatori
+        message_id = make_msgid(domain="pecflow.local")
+        msg["From"] = self.mailbox.email_address
+        msg["To"] = ", ".join(to_addresses)
+        if cc_addresses:
+            msg["Cc"] = ", ".join(cc_addresses)
+        msg["Subject"] = subject
+        msg["Date"] = formatdate(localtime=True)
+        msg["Message-ID"] = message_id
+        msg["MIME-Version"] = "1.0"
+
+        # Corpo
+        if body_text:
+            body_container.attach(MIMEText(body_text, "plain", "utf-8"))
+        if body_html:
+            body_container.attach(MIMEText(body_html, "html", "utf-8"))
+        elif not body_text:
+            # Almeno un body vuoto per evitare messaggi malformati
+            body_container.attach(MIMEText("", "plain", "utf-8"))
+
+        # Se la struttura è mixed, aggiungi il body_container come parte
+        if attachments and body_container is not msg:
+            msg.attach(body_container)
+
+        # Allegati
+        for att in attachments:
+            filename: str = att["filename"]
+            content: bytes = att["content"]
+            content_type: str = att.get("content_type", "application/octet-stream")
+
+            try:
+                main_type, sub_type = content_type.split("/", 1)
+            except ValueError:
+                main_type, sub_type = "application", "octet-stream"
+
+            part = MIMEBase(main_type, sub_type)
+            part.set_payload(content)
+            encoders.encode_base64(part)
+            part.add_header(
+                "Content-Disposition",
+                "attachment",
+                filename=filename,
+            )
+            msg.attach(part)
+
+        return msg, message_id
+
+    # ── Invio SMTP ────────────────────────────────────────────────────────────
+
+    async def send(
+        self,
+        to_addresses: list[str],
+        cc_addresses: list[str],
+        subject: str,
+        body_text: str,
+        body_html: str | None = None,
+        attachments: list[dict] | None = None,
+    ) -> tuple[str, bytes]:
+        """
+        Invia la PEC via SMTP.
+
+        Supporta:
+        - Porta 465 con SSL diretto (use_tls=True)
+        - Porta 587 con STARTTLS (use_tls=False, porta 587)
+        - Porta 25  plain (uso sconsigliato)
+
+        Returns:
+            (message_id_header, raw_eml_bytes)
+
+        Raises:
+            aiosmtplib.SMTPException: su errore SMTP non recuperabile
+            aiosmtplib.SMTPConnectError: su timeout/connessione fallita
+        """
+        msg, message_id = self.build_mime_message(
+            to_addresses=to_addresses,
+            cc_addresses=cc_addresses,
+            subject=subject,
+            body_text=body_text,
+            body_html=body_html,
+            attachments=attachments,
+        )
+
+        raw_eml: bytes = msg.as_bytes()
+        creds = self._creds
+        all_recipients = list(to_addresses) + list(cc_addresses)
+
+        # Determina la modalità di connessione in base alla porta e al flag
+        port: int = creds["port"]
+        use_tls: bool = creds["use_tls"]
+        start_tls: bool = False
+
+        if port == 587:
+            # STARTTLS tipico
+            use_tls = False
+            start_tls = True
+        elif port == 465:
+            # SSL diretto
+            use_tls = True
+            start_tls = False
+        # porta 25 → plain (entrambi False)
+
+        logger.debug(
+            f"SMTP connect: {creds['host']}:{port} "
+            f"(use_tls={use_tls}, start_tls={start_tls})"
+        )
+
+        smtp = aiosmtplib.SMTP(
+            hostname=creds["host"],
+            port=port,
+            use_tls=use_tls,
+            start_tls=start_tls,
+            timeout=30,
+        )
+
+        try:
+            await smtp.connect()
+            await smtp.login(creds["user"], creds["password"])
+            errors, response = await smtp.sendmail(
+                sender=self.mailbox.email_address,
+                recipients=all_recipients,
+                message=raw_eml,
+            )
+            if errors:
+                failed = ", ".join(f"{addr}: {err}" for addr, err in errors.items())
+                raise aiosmtplib.SMTPRecipientsRefused(
+                    recipients={a: (code, msg_b) for a, (code, msg_b) in errors.items()}
+                )
+            await smtp.quit()
+        except Exception:
+            try:
+                smtp.close()
+            except Exception:
+                pass
+            raise
+
+        logger.info(
+            f"PEC inviata: {message_id} da {self.mailbox.email_address} "
+            f"→ {all_recipients} ({len(raw_eml)} bytes)"
+        )
+        return message_id, raw_eml

worker/app/storage/minio_client.py

@@ -134,6 +134,52 @@ def _sanitize_filename(filename: str) -> str:
     return safe or "attachment"
 
 
+async def upload_outbound_eml(
+    tenant_id: str,
+    mailbox_id: str,
+    message_id: str,
+    eml_bytes: bytes,
+) -> str:
+    """
+    Carica il raw EML di un messaggio outbound su MinIO.
+
+    Percorso: tenants/{tenant_id}/mailboxes/{mailbox_id}/outbound/{message_id}.eml
+
+    Args:
+        tenant_id:  UUID del tenant
+        mailbox_id: UUID della casella mittente
+        message_id: UUID del messaggio
+        eml_bytes:  byte del raw EML
+
+    Returns:
+        Percorso oggetto su MinIO (senza bucket name)
+    """
+    client = get_minio_client()
+    bucket = settings.minio_bucket
+    object_path = (
+        f"tenants/{tenant_id}/mailboxes/{mailbox_id}/outbound/{message_id}.eml"
+    )
+
+    try:
+        import io as _io
+        data_stream = _io.BytesIO(eml_bytes)
+        await client.put_object(
+            bucket_name=bucket,
+            object_name=object_path,
+            data=data_stream,
+            length=len(eml_bytes),
+            content_type="message/rfc822",
+        )
+        logger.debug(
+            f"EML outbound caricato: s3://{bucket}/{object_path} "
+            f"({len(eml_bytes)} bytes)"
+        )
+        return object_path
+    except Exception as e:
+        logger.error(f"Errore upload EML outbound {object_path}: {e}")
+        raise
+
+
 async def ensure_bucket_exists() -> None:
     """Verifica che il bucket MinIO esista, altrimenti lo crea."""
     client = get_minio_client()