Send branded HTML emails and harden SMTP2GO webhooks.

Fix campaign stuck on sending under async queue, and stop UUID ValidationError when SMTP2GO tests send "Headers Unavailable".

site/messaging/README.md

@@ -26,7 +26,8 @@ SMTP2GO → **Settings → Webhooks** (email and SMS stay separate).
 | SMS events | leave unchecked |
 
 `X-Monica-Message-Id` is set on every campaign email send and is required so webhook
-events match the correct recipient row.
+events match the correct recipient row. Invalid / missing header values no longer
+500 the endpoint (SMTP2GO “Test this webhook” often sends a sample non-UUID).
 
 Beta / other hosts: swap the hostname, keep the path.
 

site/messaging/providers/email/smtp2go.py

@@ -2,9 +2,11 @@
 
 from django.conf import settings
 from django.core.mail import EmailMultiAlternatives
+from django.template.loader import get_template
 
 from contacts.models import Channel
 from messaging.services import one_click_unsubscribe_url, preferences_url
+from public.email_branding import email_brand_context, plain_text_to_email_html
 
 # Reported back on SMTP2GO webhooks when this header is selected in webhook settings.
 MONICA_MESSAGE_HEADER = "X-Monica-Message-Id"
@@ -28,15 +30,20 @@ def send_email(message) -> str:
     one_click_path = one_click_unsubscribe_url(str(contact.pk), Channel.EMAIL)
     prefs_url = f"{site}{prefs_path}" if site else prefs_path
     one_click_url = f"{site}{one_click_path}" if site else one_click_path
-    body_with_unsub = (
-        f"{body}\n\n---\n"
-        f"Manage preferences: {prefs_url}\n"
-        f"Unsubscribe from email: {one_click_url}"
+
+    ctx = email_brand_context(
+        title=subject,
+        content=body,
+        content_html=plain_text_to_email_html(body),
+        prefs_url=prefs_url,
+        one_click_url=one_click_url,
     )
+    text_content = get_template("emails/marketing_email.txt").render(ctx)
+    html_content = get_template("emails/marketing_email.html").render(ctx)
 
     email = EmailMultiAlternatives(
         subject=subject,
-        body=body_with_unsub,
+        body=text_content,
         from_email=settings.DEFAULT_FROM_EMAIL,
         to=[contact.email],
         headers={
@@ -45,6 +52,7 @@ def send_email(message) -> str:
             MONICA_MESSAGE_HEADER: str(message.pk),
         },
     )
+    email.attach_alternative(html_content, "text/html")
     email.send(fail_silently=False)
     # Placeholder until SMTP2GO webhook supplies the real email_id.
     return f"smtp-{message.pk}"

site/messaging/services.py

@@ -256,8 +256,12 @@ def send_campaign_completion_notify(campaign: Campaign) -> bool:
     from django.conf import settings
     from django.core.mail import EmailMultiAlternatives
     from django.db.models import Count, Q
+    from django.template.loader import get_template
     from django.urls import reverse
 
+    from public.email_branding import email_brand_context
+
+    campaign.refresh_from_db()
     if campaign.notify_sent_at:
         return True
     if campaign.status != Campaign.Status.COMPLETED:
@@ -267,6 +271,17 @@ def send_campaign_completion_notify(campaign: Campaign) -> bool:
     if not to_email:
         return False
 
+    # Claim the notify slot atomically so concurrent refresh calls only send once.
+    now = timezone.now()
+    claimed = Campaign.objects.filter(
+        pk=campaign.pk,
+        status=Campaign.Status.COMPLETED,
+        notify_sent_at__isnull=True,
+    ).update(notify_sent_at=now)
+    if not claimed:
+        return True
+    campaign.notify_sent_at = now
+
     counts = campaign.messages.aggregate(
         sent=Count("id", filter=Q(status=Message.Status.SENT)),
         delivered=Count("id", filter=Q(status=Message.Status.DELIVERED)),
@@ -287,22 +302,26 @@ def send_campaign_completion_notify(campaign: Campaign) -> bool:
     report_url = f"{public}{report_path}" if public else report_path
 
     subject = f"Campaign sent: {campaign.name}"
-    body = (
-        f"Your {campaign.get_channel_display()} campaign “{campaign.name}” "
-        f"has finished sending.\n\n"
-        f"Recipients: {counts['total']}\n"
-        f"Sent: {counts['sent']}\n"
-        f"Delivered: {counts['delivered']}\n"
-        f"Failed / bounced: {counts['failed']}\n"
-        f"Suppressed: {counts['suppressed']}\n\n"
-        f"Report: {report_url}\n"
+    ctx = email_brand_context(
+        subject=subject,
+        campaign_name=campaign.name,
+        channel_display=campaign.get_channel_display(),
+        total=counts["total"],
+        sent=counts["sent"],
+        delivered=counts["delivered"],
+        failed=counts["failed"],
+        suppressed=counts["suppressed"],
+        report_url=report_url,
     )
+    text_content = get_template("emails/campaign_complete.txt").render(ctx)
+    html_content = get_template("emails/campaign_complete.html").render(ctx)
     email = EmailMultiAlternatives(
         subject=subject,
-        body=body,
+        body=text_content,
         from_email=settings.DEFAULT_FROM_EMAIL,
         to=[to_email],
     )
+    email.attach_alternative(html_content, "text/html")
     try:
         email.send(fail_silently=False)
     except Exception:  # noqa: BLE001 — don't block completion on mail errors
@@ -311,15 +330,16 @@ def send_campaign_completion_notify(campaign: Campaign) -> bool:
         logging.getLogger(__name__).exception(
             "Campaign completion notify failed for %s", campaign.pk
         )
+        Campaign.objects.filter(pk=campaign.pk).update(notify_sent_at=None)
+        campaign.notify_sent_at = None
         return False
 
-    campaign.notify_sent_at = timezone.now()
-    campaign.save(update_fields=["notify_sent_at", "updated_at"])
     return True
 
 
 def refresh_campaign_status(campaign: Campaign) -> Campaign:
     """Set campaign to completed when no messages remain pending."""
+    campaign.refresh_from_db()
     pending = campaign.messages.filter(
         status__in=[
             Message.Status.DRAFT,
@@ -377,6 +397,9 @@ def send_campaign_test_email(campaign: Campaign, to_email: str) -> None:
     """Send one preview copy to ``to_email`` without touching recipient rows."""
     from django.conf import settings
     from django.core.mail import EmailMultiAlternatives
+    from django.template.loader import get_template
+
+    from public.email_branding import email_brand_context, plain_text_to_email_html
 
     if campaign.channel != Channel.EMAIL:
         raise ValueError("Test send is only available for email campaigns.")
@@ -391,14 +414,25 @@ def send_campaign_test_email(campaign: Campaign, to_email: str) -> None:
     if not body.strip():
         raise ValueError("Campaign has no body.")
 
+    notice = (
+        "This is a test send from the Monica portal. "
+        "Recipient list was not notified."
+    )
+    ctx = email_brand_context(
+        title=f"[TEST] {subject}",
+        content=f"{body}\n\n{notice}",
+        content_html=(
+            f"{plain_text_to_email_html(body)}"
+            f'<p style="margin:24px 0 0;color:#6b7280;font-size:13px;">{notice}</p>'
+        ),
+    )
+    text_content = get_template("emails/marketing_email.txt").render(ctx)
+    html_content = get_template("emails/marketing_email.html").render(ctx)
     email = EmailMultiAlternatives(
         subject=f"[TEST] {subject}",
-        body=(
-            f"{body}\n\n---\n"
-            "This is a test send from the Monica portal. "
-            "Recipient list was not notified."
-        ),
+        body=text_content,
         from_email=settings.DEFAULT_FROM_EMAIL,
         to=[to_email],
     )
+    email.attach_alternative(html_content, "text/html")
     email.send(fail_silently=False)

site/messaging/tasks.py

@@ -3,11 +3,12 @@ from django.utils import timezone
 
 from messaging.channels import dispatch_message
 from messaging.models import Message
-from messaging.services import contact_may_receive
+from messaging.services import contact_may_receive, refresh_campaign_status
 
 
 @task
 def send_campaign_message(message_id: str) -> None:
+    campaign = None
     try:
         message = Message.objects.select_related("contact", "campaign").get(
             pk=message_id
@@ -15,10 +16,13 @@ def send_campaign_message(message_id: str) -> None:
     except Message.DoesNotExist:
         return
 
+    campaign = message.campaign
+
     if not contact_may_receive(message.contact, message.channel):
         message.status = Message.Status.SUPPRESSED
         message.error = "Contact opted out or suppressed"
         message.save(update_fields=["status", "error", "updated_at"])
+        refresh_campaign_status(campaign)
         return
 
     try:
@@ -42,4 +46,7 @@ def send_campaign_message(message_id: str) -> None:
         message.status = Message.Status.FAILED
         message.error = str(exc)[:2000]
         message.save(update_fields=["status", "error", "updated_at"])
+        refresh_campaign_status(campaign)
         raise
+
+    refresh_campaign_status(campaign)

site/messaging/tests.py

@@ -271,6 +271,8 @@ class CampaignSendTests(TestCase):
         self.assertEqual(len(mail.outbox), 1)
         self.assertEqual(mail.outbox[0].to, ["me@example.com"])
         self.assertTrue(mail.outbox[0].subject.startswith("[TEST]"))
+        self.assertTrue(mail.outbox[0].alternatives)
+        self.assertEqual(mail.outbox[0].alternatives[0][1], "text/html")
         # Recipients untouched
         self.assertEqual(
             self.campaign.messages.filter(status=Message.Status.DRAFT).count(), 1
@@ -300,6 +302,10 @@ class CampaignSendTests(TestCase):
         summary = mail.outbox[1]
         self.assertIn("Campaign sent:", summary.subject)
         self.assertEqual(summary.to, [self.user.email])
+        self.assertTrue(mail.outbox[0].alternatives)
+        self.assertEqual(mail.outbox[0].alternatives[0][1], "text/html")
+        self.assertIn("#00626c", mail.outbox[0].alternatives[0][0])
+        self.assertTrue(summary.alternatives)
 
 
 class Smtp2goEmailWebhookTests(TestCase):
@@ -443,6 +449,50 @@ class Smtp2goEmailWebhookTests(TestCase):
             )
             self.assertEqual(ok.status_code, 200)
 
+    def test_invalid_monica_header_does_not_500(self):
+        """SMTP2GO tests often send a non-UUID sample custom header."""
+        import json
+
+        url = reverse("messaging:email_webhook")
+        response = self.client.post(
+            url,
+            data=json.dumps(
+                {
+                    "event": "delivered",
+                    "rcpt": "nobody@example.com",
+                    "X-Monica-Message-Id": "Headers Unavailable",
+                    "email_id": "smtp2go-test-id",
+                }
+            ),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+        self.assertTrue(data["ok"])
+        self.assertFalse(data["matched"])
+        self.assertTrue(
+            ProviderEvent.objects.filter(event_type="delivered").exists()
+        )
+
+    def test_ui_event_label_aliases(self):
+        url = reverse("messaging:email_webhook")
+        response = self.client.post(
+            url,
+            data={
+                "event": "bounced",
+                "bounce": "soft",
+                "X-Monica-Message-Id": str(self.message.pk),
+            },
+        )
+        self.assertEqual(response.status_code, 200)
+        self.message.refresh_from_db()
+        self.assertEqual(self.message.status, Message.Status.BOUNCED)
+        self.assertTrue(
+            ProviderEvent.objects.filter(
+                message=self.message, event_type="bounce"
+            ).exists()
+        )
+
 
 class Smtp2goSmsWebhookTests(TestCase):
     def setUp(self):

site/messaging/views.py

@@ -1,5 +1,6 @@
 import hashlib
 import hmac
+import logging
 
 from django.conf import settings
 from django.contrib import messages
@@ -38,6 +39,8 @@ from messaging.webhooks import (
     process_smtp2go_sms_webhook,
 )
 
+logger = logging.getLogger(__name__)
+
 
 def _audience_choices() -> list[tuple[str, str]]:
     """Labeled audience options with live opted-in counts."""
@@ -256,6 +259,11 @@ def campaign_detail(request, pk):
 def campaign_status_json(request, pk):
     """JSON snapshot for live-updating the campaign report page."""
     campaign = get_object_or_404(Campaign, pk=pk)
+    # Async queue may finish after enqueue; re-evaluate completion on poll.
+    from messaging.services import refresh_campaign_status
+
+    refresh_campaign_status(campaign)
+    campaign.refresh_from_db()
     ctx = _campaign_report(campaign)
     return JsonResponse(
         {
@@ -580,7 +588,11 @@ def email_webhook(request):
     ):
         return HttpResponseForbidden("invalid webhook token")
     payload = parse_webhook_payload(request)
-    event = process_smtp2go_email_webhook(payload)
+    try:
+        event = process_smtp2go_email_webhook(payload)
+    except Exception:  # noqa: BLE001 — never 500 SMTP2GO (they retry for 48h)
+        logger.exception("SMTP2GO email webhook processing failed")
+        return JsonResponse({"ok": False, "error": "processing_failed"}, status=200)
     return JsonResponse(
         {
             "ok": True,

site/messaging/webhooks.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 
 import json
 import logging
+import uuid
 from typing import Any
 
+from django.core.exceptions import ValidationError
 from django.http import HttpRequest
 
 from contacts.models import Channel, Contact
@@ -38,6 +40,45 @@ _MONICA_HEADER_KEYS = (
     "monica-message-id",
 )
 
+# SMTP2GO UI labels → canonical event strings from their docs.
+_EMAIL_EVENT_ALIASES = {
+    "bounced": "bounce",
+    "rejected": "reject",
+    "opened": "open",
+    "clicked": "click",
+    "unsubscribed": "unsubscribe",
+    "resubscribed": "resubscribe",
+}
+
+
+def _as_str(value: Any) -> str:
+    """Coerce webhook field values to a stripped string (lists / None safe)."""
+    if value is None:
+        return ""
+    if isinstance(value, (list, tuple)):
+        if not value:
+            return ""
+        value = value[0]
+    if isinstance(value, bytes):
+        value = value.decode("utf-8", errors="replace")
+    return str(value).strip()
+
+
+def _json_safe(value: Any) -> Any:
+    """Ensure ProviderEvent.payload can be stored as JSON."""
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    if isinstance(value, dict):
+        return {str(k): _json_safe(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_json_safe(v) for v in value]
+    return str(value)
+
+
+def _normalize_email_event(event: str) -> str:
+    event = (event or "").strip().lower()
+    return _EMAIL_EVENT_ALIASES.get(event, event)
+
 
 def parse_webhook_payload(request: HttpRequest) -> dict[str, Any]:
     """Accept JSON or form-encoded SMTP2GO webhook bodies."""
@@ -45,7 +86,7 @@ def parse_webhook_payload(request: HttpRequest) -> dict[str, Any]:
     if "application/json" in content_type:
         try:
             data = json.loads(request.body.decode() or "{}")
-        except json.JSONDecodeError:
+        except (json.JSONDecodeError, UnicodeDecodeError):
             return {}
         return data if isinstance(data, dict) else {}
     # Form-encoded (SMTP2GO default)
@@ -55,37 +96,65 @@ def parse_webhook_payload(request: HttpRequest) -> dict[str, Any]:
 def extract_monica_message_id(payload: dict[str, Any]) -> str:
     """Pull our correlation id from flat keys or a nested headers object."""
     for key in _MONICA_HEADER_KEYS:
-        value = payload.get(key)
+        value = _as_str(payload.get(key))
         if value:
-            return str(value).strip()
+            return value
 
     headers = payload.get("headers") or payload.get("email_headers") or {}
     if isinstance(headers, dict):
         for key in _MONICA_HEADER_KEYS:
-            value = headers.get(key)
+            value = _as_str(headers.get(key))
             if value:
-                return str(value).strip()
+                return value
         # Case-insensitive scan
         lower_map = {str(k).lower(): v for k, v in headers.items()}
         for key in _MONICA_HEADER_KEYS:
-            value = lower_map.get(key.lower())
+            value = _as_str(lower_map.get(key.lower()))
             if value:
-                return str(value).strip()
+                return value
+    elif isinstance(headers, list):
+        # Some ESP shapes send [["X-Monica-Message-Id", "..."], ...]
+        for item in headers:
+            if isinstance(item, (list, tuple)) and len(item) >= 2:
+                if _as_str(item[0]).lower() in {
+                    k.lower() for k in _MONICA_HEADER_KEYS
+                }:
+                    value = _as_str(item[1])
+                    if value:
+                        return value
+            elif isinstance(item, str) and ":" in item:
+                name, _, rest = item.partition(":")
+                if name.strip().lower() in {k.lower() for k in _MONICA_HEADER_KEYS}:
+                    value = rest.strip()
+                    if value:
+                        return value
     return ""
 
 
+def _message_by_pk(pk: str) -> Message | None:
+    """Lookup Message by UUID pk without raising on malformed ids."""
+    try:
+        uuid.UUID(str(pk))
+    except (ValueError, AttributeError, TypeError):
+        return None
+    try:
+        return (
+            Message.objects.select_related("contact", "campaign")
+            .filter(pk=pk)
+            .first()
+        )
+    except (ValidationError, ValueError):
+        return None
+
+
 def find_message_for_email_event(payload: dict[str, Any]) -> Message | None:
     monica_id = extract_monica_message_id(payload)
     if monica_id:
-        message = (
-            Message.objects.select_related("contact", "campaign")
-            .filter(pk=monica_id)
-            .first()
-        )
+        message = _message_by_pk(monica_id)
         if message:
             return message
 
-    email_id = (payload.get("email_id") or payload.get("email-id") or "").strip()
+    email_id = _as_str(payload.get("email_id") or payload.get("email-id"))
     if email_id:
         message = (
             Message.objects.select_related("contact", "campaign")
@@ -95,13 +164,13 @@ def find_message_for_email_event(payload: dict[str, Any]) -> Message | None:
         if message:
             return message
 
-    rcpt = (payload.get("rcpt") or "").strip().lower()
+    rcpt = _as_str(payload.get("rcpt")).lower()
     if not rcpt:
         recipients = payload.get("recipients")
         if isinstance(recipients, str) and recipients.strip():
             rcpt = recipients.split(",")[0].strip().lower()
         elif isinstance(recipients, list) and recipients:
-            rcpt = str(recipients[0]).strip().lower()
+            rcpt = _as_str(recipients[0]).lower()
 
     if not rcpt:
         return None
@@ -157,11 +226,11 @@ def _maybe_upgrade_status(message: Message, new_status: str, *, error: str = "")
 
 
 def _apply_email_event(message: Message, event: str, payload: dict[str, Any]) -> None:
-    event = (event or "").strip().lower()
-    bounce_kind = (payload.get("bounce") or "").strip().lower()
-    err = (payload.get("message") or payload.get("context") or "").strip()
+    event = _normalize_email_event(event)
+    bounce_kind = _as_str(payload.get("bounce")).lower()
+    err = _as_str(payload.get("message") or payload.get("context"))
 
-    email_id = (payload.get("email_id") or payload.get("email-id") or "").strip()
+    email_id = _as_str(payload.get("email_id") or payload.get("email-id"))
     if email_id and message.provider_message_id != email_id:
         message.provider_message_id = email_id
         message.provider = PROVIDER_EMAIL
@@ -233,7 +302,7 @@ def process_smtp2go_email_webhook(payload: dict[str, Any]) -> ProviderEvent | No
 
     Returns the stored event (even if message could not be matched).
     """
-    event = (payload.get("event") or "").strip().lower()
+    event = _normalize_email_event(_as_str(payload.get("event")))
     if not event:
         logger.warning("SMTP2GO webhook missing event: %s", payload)
         return None
@@ -253,8 +322,8 @@ def process_smtp2go_email_webhook(payload: dict[str, Any]) -> ProviderEvent | No
     return ProviderEvent.objects.create(
         message=message,
         provider=PROVIDER_EMAIL,
-        event_type=event,
-        payload=payload,
+        event_type=event[:64],
+        payload=_json_safe(payload) if isinstance(payload, dict) else {},
     )
 
 
@@ -442,11 +511,7 @@ def find_message_for_pcm_event(payload: dict[str, Any]) -> Message | None:
         ext = payload["recipient"].get("extRefNbr") or ""
     ext = str(ext).strip()
     if ext:
-        message = (
-            Message.objects.select_related("contact", "campaign")
-            .filter(pk=ext)
-            .first()
-        )
+        message = _message_by_pk(ext)
         if message:
             return message