""" Orquestacion del envio de correo. Es el unico sitio que sabe como encajan validacion, auditoria, cola y proveedor. Las vistas se limitan a validar la entrada y llamar aqui. Sustituye a `notifications/utils/mails.py`, que resolvia el envio haciendo una peticion HTTP a la API `api-mails` con una clave escrita en el codigo. Esa llamada anadia una latencia completa, un punto de falla y una credencial que no se podia rotar sin desplegar. Ahora el correo sale directamente desde aqui. Flujo de un envio: 1. Se crea la fila de auditoria en estado `queued` y se guarda el contenido. 2. Se encola la tarea (o se envia en el acto si se pidio `send_async=false`). 3. El worker recupera el contenido, llama al proveedor y actualiza el estado. 4. El contenido temporal se borra al llegar a un estado final. Los pasos 1 y 2 van en una transaccion con `on_commit`: sin eso, el worker puede empezar a procesar la tarea antes de que la fila exista. """ from __future__ import annotations import logging from django.conf import settings from django.db import transaction from core.utils.http import get_client_ip from notifications.channels.email.base import EmailPayload, SendResult from notifications.channels.email.registry import get_provider from notifications.channels.email.sendgrid import CUSTOM_ARG_MESSAGE_ID from notifications.models.message import Channel, MessageStatus, NotificationMessage from notifications.models.payload import MessagePayloadBlob from notifications.services.serialization import payload_from_bytes, payload_to_bytes logger = logging.getLogger("notifications.mailer") def build_payload(data: dict, message_id: str) -> EmailPayload: """Convierte los datos ya validados en el payload neutro del proveedor.""" return EmailPayload( from_email=data["from_email"], from_name=data.get("from_name", ""), reply_to=data.get("reply_to", ""), to=tuple(data["to"]), cc=tuple(data.get("cc") or ()), bcc=tuple(data.get("bcc") or ()), subject=data.get("subject", ""), html=data.get("html", ""), text=data.get("text", ""), template_id=data.get("template_id", ""), template_data=data.get("template_data") or {}, categories=tuple(data.get("categories") or ()), headers=data.get("headers") or {}, attachments=data.get("_attachments") or (), # Viaja hasta los eventos del webhook: es lo que permite enlazar un # rebote con la fila de auditoria correcta. custom_args={CUSTOM_ARG_MESSAGE_ID: message_id}, ) @transaction.atomic def register_message( data: dict, *, provider_name: str, request=None, api_key=None, ) -> tuple[NotificationMessage, EmailPayload]: """Crea la fila de auditoria y guarda el contenido pendiente de enviar.""" attachments = data.get("_attachments") or () message = NotificationMessage( channel=Channel.EMAIL, api_key=api_key, provider=provider_name, status=MessageStatus.QUEUED, subject=data.get("subject", ""), sender=data["from_email"], sender_name=data.get("from_name", ""), reply_to=data.get("reply_to", ""), to=list(data["to"]), cc=list(data.get("cc") or []), bcc=list(data.get("bcc") or []), template_id=data.get("template_id", ""), attachment_count=len(attachments), attachment_bytes=sum(a.size for a in attachments), request_id=getattr(request, "request_id", "") if request else "", source_ip=get_client_ip(request) if request else None, ) message.recipient_count = len(message.to) + len(message.cc) + len(message.bcc) if settings.NOTIFICATIONS_STORE_BODY: message.body_html = data.get("html", "") message.body_text = data.get("text", "") message.save() payload = build_payload(data, str(message.id)) MessagePayloadBlob.objects.create(message=message, data=payload_to_bytes(payload)) return message, payload def dispatch( message: NotificationMessage, *, send_async: bool, request_id: str = "" ) -> None: """ Entrega el mensaje al worker o lo envia en el acto. En modo asincrono la tarea se encola despues de confirmar la transaccion, para que el worker nunca vea una fila que aun no existe. """ if send_async: from notifications.tasks import send_email_task transaction.on_commit( lambda: send_email_task.delay( str(message.id), request_id=request_id or message.request_id ) ) return deliver(message) def deliver(message: NotificationMessage) -> SendResult: """ Envia el mensaje a traves de su proveedor y actualiza la auditoria. Devuelve el resultado para que quien llame (la vista en modo sincrono o la tarea de Celery) decida que hacer: responder al cliente o reintentar. """ blob = MessagePayloadBlob.objects.filter(message=message).first() if blob is None: logger.error( "No se encontro el contenido del envio", extra={"message_id": str(message.id)}, ) message.mark_failed( code="payload_missing", message="El contenido del mensaje ya no esta disponible.", ) return SendResult.failure( message.provider, error_code="payload_missing", error_message="Contenido no disponible.", retryable=False, ) try: payload = payload_from_bytes(blob.data) except Exception as exc: # noqa: BLE001 - contenido corrupto logger.exception( "El contenido guardado del envio no se pudo leer", extra={"message_id": str(message.id)}, ) message.mark_failed(code="payload_corrupt", message=str(exc)) _discard_payload(message) return SendResult.failure( message.provider, error_code="payload_corrupt", error_message=str(exc), retryable=False, ) message.mark_sending() try: provider = get_provider(message.provider) except Exception as exc: # noqa: BLE001 - proveedor mal configurado logger.error( "El proveedor '%s' no esta disponible: %s", message.provider, exc, extra={"message_id": str(message.id), "provider": message.provider}, ) message.mark_failed(code="provider_unavailable", message=str(exc)) return SendResult.failure( message.provider, error_code="provider_unavailable", error_message=str(exc), retryable=True, ) result = provider.send(payload) if result.success: message.mark_sent(result.message_id) _discard_payload(message) logger.info( "Envio de correo completado", extra={ "message_id": str(message.id), "provider": message.provider, "provider_message_id": result.message_id, "recipients": message.recipient_count, }, ) else: message.mark_failed(result.error_code, result.error_message) if not result.retryable: # No se va a reintentar: el contenido ya no hace falta. _discard_payload(message) logger.warning( "Envio de correo fallido", extra={ "message_id": str(message.id), "provider": message.provider, "error_code": result.error_code, "retryable": result.retryable, "attempts": message.attempts, }, ) return result def _discard_payload(message: NotificationMessage) -> None: """Borra el contenido temporal del envio.""" MessagePayloadBlob.objects.filter(message=message).delete()