""" Ingesta de los eventos de entrega que manda el proveedor. SendGrid agrupa varios eventos por peticion y no garantiza ni el orden ni la entrega unica: el mismo evento puede llegar varias veces y uno posterior puede adelantar a otro anterior. El procesamiento tiene eso en cuenta: * la unicidad de `sg_event_id` en la base de datos descarta los duplicados; * un estado final (entregado, rebotado, descartado, spam) no se sobrescribe con uno anterior que llegue tarde. """ from __future__ import annotations import datetime as dt import logging from django.db import transaction from django.utils import timezone from notifications.channels.email.sendgrid import CUSTOM_ARG_MESSAGE_ID from notifications.models.event import NotificationEvent from notifications.models.message import ( TERMINAL_STATUSES, MessageStatus, NotificationMessage, ) logger = logging.getLogger("notifications.events") #: Eventos que cambian el estado del envio. El resto (aperturas, clics, bajas) #: se guardan como historial pero no alteran el estado. _EVENT_TO_STATUS = { "delivered": MessageStatus.DELIVERED, "bounce": MessageStatus.BOUNCED, "dropped": MessageStatus.DROPPED, "deferred": MessageStatus.DEFERRED, "spamreport": MessageStatus.SPAM, } def ingest_sendgrid_events(events: list[dict]) -> int: """ Guarda un lote de eventos y actualiza el estado de los envios. Devuelve cuantos eventos se guardaron (los duplicados no cuentan). """ stored = 0 for raw in events: if not isinstance(raw, dict): continue try: if _ingest_one(raw): stored += 1 except Exception: # noqa: BLE001 - un evento malo no debe parar el lote logger.exception( "No se pudo procesar un evento del webhook", extra={ "event": raw.get("event"), "sg_event_id": raw.get("sg_event_id"), }, ) return stored @transaction.atomic def _ingest_one(raw: dict) -> bool: event_id = str(raw.get("sg_event_id") or "").strip() if not event_id: logger.warning("Evento sin sg_event_id; se descarta por no ser idempotente") return False event_name = str(raw.get("event") or "").strip().lower() message = _resolve_message(raw) occurred_at = _parse_timestamp(raw.get("timestamp")) _, created = NotificationEvent.objects.get_or_create( provider_event_id=event_id[:128], defaults={ "message": message, "provider": "sendgrid", "event": event_name[:40], "recipient": _safe_recipient(raw.get("email")), "provider_message_id": str(raw.get("sg_message_id") or "")[:255], "occurred_at": occurred_at, "reason": str(raw.get("reason") or raw.get("response") or "")[:2000], "status_code": str(raw.get("status") or "")[:32], "payload": raw, }, ) if created and message is not None: _apply_status(message, event_name, occurred_at) return created def _resolve_message(raw: dict) -> NotificationMessage | None: """ Localiza el envio al que pertenece el evento. Se prefiere el argumento propio que se manda en cada envio: es exacto. El `sg_message_id` sirve de respaldo, pero llega con un sufijo que hay que recortar (`.filterdrecv-...`). """ api_message_id = raw.get(CUSTOM_ARG_MESSAGE_ID) if api_message_id: message = NotificationMessage.objects.filter(pk=str(api_message_id)).first() if message is not None: return message sg_message_id = str(raw.get("sg_message_id") or "") if sg_message_id: base_id = sg_message_id.split(".", 1)[0] if base_id: return NotificationMessage.objects.filter( provider_message_id__startswith=base_id ).first() return None def _apply_status( message: NotificationMessage, event_name: str, occurred_at: dt.datetime ) -> None: """Actualiza el estado del envio si el evento lo justifica.""" message.last_event_at = occurred_at update_fields = ["last_event_at", "updated_at"] new_status = _EVENT_TO_STATUS.get(event_name) if new_status is not None and message.status not in TERMINAL_STATUSES: message.status = new_status update_fields.append("status") if event_name in {"bounce", "dropped"}: message.error_code = event_name update_fields.append("error_code") message.save(update_fields=update_fields) def _parse_timestamp(value) -> dt.datetime: """Convierte la marca de tiempo de SendGrid (segundos UNIX) a datetime.""" try: return dt.datetime.fromtimestamp(int(value), tz=dt.UTC) except (TypeError, ValueError, OSError, OverflowError): from django.utils import timezone return timezone.now() def _safe_recipient(value) -> str: """Recorta y limpia la direccion que viene en el evento.""" recipient = str(value or "").strip() if len(recipient) > 254 or "\n" in recipient or "\r" in recipient: return "" return recipient # --------------------------------------------------------------------------- # Twilio: avisos de cambio de estado # --------------------------------------------------------------------------- #: Estados que manda Twilio en el callback, traducidos a los propios. #: https://www.twilio.com/docs/messaging/api/message-resource#message-status-values _TWILIO_A_ESTADO = { "queued": MessageStatus.QUEUED, "sending": MessageStatus.SENDING, "sent": MessageStatus.SENT, "delivered": MessageStatus.DELIVERED, "read": MessageStatus.READ, "undelivered": MessageStatus.BOUNCED, "failed": MessageStatus.FAILED, "canceled": MessageStatus.DROPPED, } @transaction.atomic def ingest_twilio_status(message_id: str, payload: dict) -> bool: """ Registra un cambio de estado notificado por Twilio. Twilio reintenta el callback si no respondemos rapido, asi que el evento se identifica por (SID del mensaje + estado): el mismo aviso repetido no genera una fila nueva. """ sid = str(payload.get("MessageSid") or payload.get("SmsSid") or "").strip() estado_twilio = str(payload.get("MessageStatus") or payload.get("SmsStatus") or "").strip().lower() if not sid or not estado_twilio: logger.warning( "Aviso de Twilio sin SID o sin estado; se descarta", extra={"message_id": message_id}, ) return False message = NotificationMessage.objects.filter(pk=message_id).first() if message is None: logger.warning( "Aviso de Twilio para un envio inexistente", extra={"message_id": message_id, "sid": sid}, ) return False _, created = NotificationEvent.objects.get_or_create( provider_event_id=f"twilio:{sid}:{estado_twilio}"[:128], defaults={ "message": message, "provider": "twilio", "event": estado_twilio[:40], "recipient": (message.to or [""])[0][:254], "provider_message_id": sid[:255], "occurred_at": timezone.now(), "reason": str(payload.get("ErrorMessage") or "")[:2000], "status_code": str(payload.get("ErrorCode") or "")[:32], "payload": payload, }, ) if not created: return False nuevo_estado = _TWILIO_A_ESTADO.get(estado_twilio) campos = ["last_event_at", "updated_at"] message.last_event_at = timezone.now() # Un estado final no se degrada con uno anterior que llegue tarde, y # tampoco se retrocede de 'sent' a 'queued' si los avisos se cruzan. if nuevo_estado is not None and message.status not in TERMINAL_STATUSES: if not ( message.status == MessageStatus.SENT and nuevo_estado in {MessageStatus.QUEUED, MessageStatus.SENDING} ): message.status = nuevo_estado campos.append("status") if nuevo_estado in {MessageStatus.BOUNCED, MessageStatus.FAILED}: message.error_code = str(payload.get("ErrorCode") or estado_twilio)[:64] message.error_message = str(payload.get("ErrorMessage") or "")[:2000] campos.extend(["error_code", "error_message"]) message.save(update_fields=campos) return True