⚡️(backend) send recording notification emails from a Celery task

Emails were rendered and sent inline in the LiveKit webhook, so a
slow SMTP server blocked the request and a transient failure lost
the email. Delivery now runs in a worker, with retries and backoff.

The task is generic (template, subject, recipients, context,
language) so other emails can reuse it. The subject is translated
in the worker, and dates are formatted per owner before queuing,
so the arguments stay JSON-serializable.

Without Celery, the task still runs synchronously. Success now
means "queued", not "delivered".

src/backend/core/recording/event/notification.py

@@ -7,18 +7,18 @@ from datetime import datetime, timezone
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from django.conf import settings
-from django.core.mail import send_mail
-from django.template.loader import render_to_string
-from django.utils.translation import get_language, gettext, override
+from django.utils.translation import get_language, gettext, gettext_noop, override
 from django.utils.translation import gettext_lazy as _
 
 import aiohttp
 import requests
 from asgiref.sync import async_to_sync
+from kombu.exceptions import OperationalError as KombuOperationalError
 from livekit import api as livekit_api
 
 from core import models, utils
 from core.analytics import UserFeatureFlag, is_user_feature_flag_enabled
+from core.tasks.email import send_email
 from core.utils import generate_download_s3_url
 
 logger = logging.getLogger(__name__)
@@ -68,10 +68,14 @@ class NotificationService:
     @staticmethod
     def _notify_user_by_email(recording) -> bool:
         """
-        Send an email notification to recording owners when their recording is ready.
+        Queue an email notification to recording owners when their recording is ready.
 
         The email includes a direct link that redirects owners to a dedicated download
         page in the frontend where they can access their specific recording.
+
+        Rendering and SMTP delivery happen in a Celery worker (``send_email``), so
+        this only does one DB query and enqueues one message per owner.
+        Returns True when every email was queued (or sent, if Celery is disabled).
         """
 
         owner_accesses = (
@@ -88,10 +92,6 @@ class NotificationService:
             return False
 
         context = {
-            "brandname": settings.EMAIL_BRAND_NAME,
-            "support_email": settings.EMAIL_SUPPORT_EMAIL,
-            "logo_img": settings.EMAIL_LOGO_IMG,
-            "domain": settings.EMAIL_DOMAIN,
             "room_name": recording.room.name,
             "recording_expiration_days": settings.RECORDING_EXPIRATION_DAYS,
             "link": f"{get_recording_download_base_url()}/{recording.id}",
@@ -104,37 +104,22 @@ class NotificationService:
         # 2. The number of recipients per recording is typically small (not thousands)
         for access in owner_accesses:
             user = access.user
-            language = user.language or get_language()
-            with override(language):
-                personalized_context = {
-                    "recording_date": recording.created_at.astimezone(
-                        user.timezone
-                    ).strftime("%Y-%m-%d"),
-                    "recording_time": recording.created_at.astimezone(
-                        user.timezone
-                    ).strftime("%H:%M"),
-                    **context,
-                }
-                msg_html = render_to_string(
-                    "mail/html/screen_recording.html", personalized_context
+            local_created_at = recording.created_at.astimezone(user.timezone)
+            try:
+                send_email.delay(
+                    template="screen_recording",
+                    subject=gettext_noop("Your recording is ready"),
+                    recipients=[user.email],
+                    language=user.language or get_language(),
+                    context={
+                        **context,
+                        "recording_date": local_created_at.strftime("%Y-%m-%d"),
+                        "recording_time": local_created_at.strftime("%H:%M"),
+                    },
                 )
-                msg_plain = render_to_string(
-                    "mail/text/screen_recording.txt", personalized_context
-                )
-                subject = gettext("Your recording is ready")  # Force translation
-
-                try:
-                    send_mail(
-                        subject.capitalize(),
-                        msg_plain,
-                        settings.EMAIL_FROM,
-                        [user.email],
-                        html_message=msg_html,
-                        fail_silently=False,
-                    )
-                except smtplib.SMTPException as exception:
-                    logger.error("notification could not be sent: %s", exception)
-                    has_failures = True
+            except (smtplib.SMTPException, KombuOperationalError) as exception:
+                logger.error("notification could not be sent: %s", exception)
+                has_failures = True
 
         return not has_failures
 

src/backend/core/tasks/__init__.py

@@ -1,9 +1,11 @@
 """Celery tasks for the core app."""
 
 from core.tasks.connection_test import delete_connection_test_room
+from core.tasks.email import send_email
 from core.tasks.file import process_file_deletion
 
 __all__ = (
     "delete_connection_test_room",
     "process_file_deletion",
+    "send_email",
 )

src/backend/core/tasks/email.py

@@ -0,0 +1,74 @@
+"""Generic tasks to send transactional emails."""
+
+import logging
+import smtplib
+
+from django.conf import settings
+from django.core.mail import send_mail
+from django.template.loader import render_to_string
+from django.utils.translation import get_language, gettext, override
+
+from core.tasks._task import task
+
+logger = logging.getLogger(__name__)
+
+
+def get_base_email_context() -> dict:
+    """Context shared by every email template (branding, support, domain)."""
+    return {
+        "brandname": settings.EMAIL_BRAND_NAME,
+        "support_email": settings.EMAIL_SUPPORT_EMAIL,
+        "logo_img": settings.EMAIL_LOGO_IMG,
+        "domain": settings.EMAIL_DOMAIN,
+    }
+
+
+@task(
+    autoretry_for=(smtplib.SMTPException, ConnectionError, TimeoutError),
+    retry_backoff=True,
+    retry_backoff_max=600,
+    retry_jitter=True,
+    max_retries=5,
+)
+def send_email(
+    *,
+    template: str,
+    subject: str,
+    recipients: list[str],
+    context: dict | None = None,
+    language: str | None = None,
+):
+    """Render a mail template and send it.
+
+    Generic on purpose: any feature can reuse it with its own template and context.
+
+    Args:
+        template: Template base name, e.g. "screen_recording". Both
+            "mail/html/<template>.html" and "mail/text/<template>.txt" must exist.
+        subject: Untranslated subject msgid. Mark it with ``gettext_noop`` at the
+            call site; it is translated here, in ``language``.
+        recipients: Email addresses, all receiving the same rendered message.
+        context: Template context. Must be JSON-serializable (str, int, list, dict…):
+            pre-format dates, cast UUIDs to str. Merged over the base branding context.
+        language: Language to render the email in. Defaults to the active language.
+
+    Raises:
+        smtplib.SMTPException: after retries are exhausted (or immediately when
+            Celery is disabled and the call runs synchronously).
+    """
+    full_context = {**get_base_email_context(), **(context or {})}
+
+    with override(language or get_language()):
+        msg_html = render_to_string(f"mail/html/{template}.html", full_context)
+        msg_plain = render_to_string(f"mail/text/{template}.txt", full_context)
+        translated_subject = gettext(subject)
+
+    send_mail(
+        translated_subject,
+        msg_plain,
+        settings.EMAIL_FROM,
+        recipients,
+        html_message=msg_html,
+        fail_silently=False,
+    )
+    logger.info("Email '%s' sent to %d recipient(s)", template, len(recipients))

src/backend/core/tests/recording/event/test_notification.py

@@ -12,6 +12,7 @@ from unittest import mock
 from django.contrib.sites.models import Site
 
 import pytest
+from kombu.exceptions import OperationalError as KombuOperationalError
 
 from core import factories, models
 from core.analytics import UserFeatureFlag
@@ -142,7 +143,7 @@ def test_notify_user_by_email_success(mocked_current_site, settings):
 
     notification_service = NotificationService()
 
-    with mock.patch("core.recording.event.notification.send_mail") as mock_send_mail:
+    with mock.patch("core.tasks.email.send_mail") as mock_send_mail:
         result = notification_service._notify_user_by_email(recording)
 
         assert result is True
@@ -237,7 +238,7 @@ def test_notify_user_by_email_smtp_exception(mocked_current_site, caplog):
     notification_service = NotificationService()
 
     with mock.patch(
-        "core.recording.event.notification.send_mail",
+        "core.tasks.email.send_mail",
         side_effect=smtplib.SMTPException("SMTP Error"),
     ) as mock_send_mail:
         result = notification_service._notify_user_by_email(recording)
@@ -247,6 +248,60 @@ def test_notify_user_by_email_smtp_exception(mocked_current_site, caplog):
         assert "notification could not be sent:" in caplog.text
 
 
+def test_notify_user_by_email_queues_generic_task(settings):
+    """One generic send_email task is queued per owner, with serializable args."""
+    settings.RECORDING_DOWNLOAD_BASE_URL = "https://acme.com/recordings"
+    settings.RECORDING_EXPIRATION_DAYS = 7
+
+    recording = factories.RecordingFactory(room__name="Conference Room A")
+    recording.created_at = datetime.datetime(
+        2023, 5, 15, 14, 30, 0, tzinfo=datetime.timezone.utc
+    )
+    user = factories.UserFactory(
+        email="franc@test.com", language="fr-fr", timezone="Europe/Paris"
+    )
+    factories.UserRecordingAccessFactory(
+        recording=recording, role=models.RoleChoices.OWNER, user=user
+    )
+
+    with mock.patch("core.recording.event.notification.send_email.delay") as mock_delay:
+        result = notification_service._notify_user_by_email(recording)
+
+    assert result is True
+    mock_delay.assert_called_once_with(
+        template="screen_recording",
+        subject="Your recording is ready",
+        recipients=["franc@test.com"],
+        language="fr-fr",
+        context={
+            "room_name": "Conference Room A",
+            "recording_expiration_days": 7,
+            "link": f"https://acme.com/recordings/{recording.id}",
+            "recording_date": "2023-05-15",
+            "recording_time": "16:30",
+        },
+    )
+    # Celery's JSON serializer must accept every argument
+    json.dumps(mock_delay.call_args.kwargs)
+
+
+def test_notify_user_by_email_broker_unavailable(caplog):
+    """A broker outage is reported as a failure without crashing the caller."""
+    recording = factories.RecordingFactory()
+    factories.UserRecordingAccessFactory(
+        recording=recording, role=models.RoleChoices.OWNER
+    )
+
+    with mock.patch(
+        "core.recording.event.notification.send_email.delay",
+        side_effect=KombuOperationalError("broker down"),
+    ):
+        result = notification_service._notify_user_by_email(recording)
+
+    assert result is False
+    assert "notification could not be sent: broker down" in caplog.text
+
+
 @mock.patch("core.recording.event.notification.requests.post")
 @mock.patch("core.recording.event.notification.generate_download_s3_url")
 @mock.patch.object(

src/backend/core/tests/tasks/test_email.py

@@ -0,0 +1,89 @@
+"""Tests for the generic send_email task."""
+
+# pylint: disable=unused-argument,redefined-outer-name
+
+import smtplib
+from unittest import mock
+
+from django.utils.translation import gettext_noop
+
+import pytest
+
+from core.tasks.email import send_email
+
+
+@pytest.fixture
+def email_settings(settings):
+    """Branding settings shared by every email."""
+    settings.EMAIL_BRAND_NAME = "ACME"
+    settings.EMAIL_SUPPORT_EMAIL = "support@acme.com"
+    settings.EMAIL_LOGO_IMG = "https://acme.com/logo"
+    settings.EMAIL_DOMAIN = "acme.com"
+    settings.EMAIL_FROM = "notifications@acme.com"
+    return settings
+
+
+@mock.patch("core.tasks.email.send_mail")
+@mock.patch("core.tasks.email.render_to_string", side_effect=["<p>html</p>", "text"])
+def test_send_email_renders_template_and_sends(
+    mock_render, mock_send_mail, email_settings
+):
+    """Both template variants are rendered with base + caller context, then sent."""
+    send_email(
+        template="some_template",
+        subject="Any subject",
+        recipients=["a@test.com", "b@test.com"],
+        context={"foo": "bar", "brandname": "Overridden"},
+    )
+
+    expected_context = {
+        "brandname": "Overridden",
+        "support_email": "support@acme.com",
+        "logo_img": "https://acme.com/logo",
+        "domain": "acme.com",
+        "foo": "bar",
+    }
+    assert mock_render.call_args_list == [
+        mock.call("mail/html/some_template.html", expected_context),
+        mock.call("mail/text/some_template.txt", expected_context),
+    ]
+    mock_send_mail.assert_called_once_with(
+        "Any subject",
+        "text",
+        "notifications@acme.com",
+        ["a@test.com", "b@test.com"],
+        html_message="<p>html</p>",
+        fail_silently=False,
+    )
+
+
+@mock.patch("core.tasks.email.send_mail")
+def test_send_email_translates_subject_in_requested_language(
+    mock_send_mail, email_settings
+):
+    """The subject msgid is translated in the worker, in the recipient language."""
+    send_email(
+        template="screen_recording",
+        subject=gettext_noop("Your recording is ready"),
+        recipients=["franc@test.com"],
+        context={"room_name": "Room", "link": "https://acme.com/r/1"},
+        language="fr-fr",
+    )
+
+    subject, body, *_ = mock_send_mail.call_args[0]
+    assert subject == "Votre enregistrement est prêt"
+    assert "Votre enregistrement est prêt !" in body
+
+
+@mock.patch(
+    "core.tasks.email.send_mail", side_effect=smtplib.SMTPException("SMTP Error")
+)
+@mock.patch("core.tasks.email.render_to_string", return_value="content")
+def test_send_email_raises_on_smtp_error(mock_render, mock_send_mail, email_settings):
+    """SMTP errors propagate so Celery can retry the task."""
+    with pytest.raises(smtplib.SMTPException):
+        send_email(
+            template="some_template",
+            subject="Any subject",
+            recipients=["a@test.com"],
+        )