feat(messaging): send each email once per idempotency key

products/messaging/backend/facade/contracts.py

@@ -80,6 +80,7 @@ class SendFrom:
     integration_id: int
     email: str | None
     name: str | None
+    is_default: bool = False
 
 
 @frozen

products/messaging/backend/presentation/views/messaging_send.py

@@ -5,7 +5,8 @@
 from django.core.exceptions import ValidationError as DjangoValidationError
 from django.core.validators import validate_email
 
-from drf_spectacular.utils import OpenApiResponse, extend_schema
+from drf_spectacular.types import OpenApiTypes
+from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema
 from rest_framework import serializers, status, viewsets
 from rest_framework.decorators import action
 from rest_framework.permissions import IsAuthenticated
@@ -236,8 +237,8 @@ class SendEmailWarningSerializer(serializers.Serializer):
 
 class SendEmailResponseSerializer(SendTransactionalEmailResponseSerializer):
     replayed = serializers.BooleanField(
-        help_text="Always false on this route. It doesn't read the `Idempotency-Key` header, so a repeated request "
-        "sends another email."
+        help_text="True when an earlier request with the same `Idempotency-Key` and the same request already queued "
+        "this email, so this request sent nothing new and `message_id` is the first send's."
     )
     warnings = SendEmailWarningSerializer(
         many=True, help_text="Things to know about the queued email. Empty in the normal case."
@@ -260,11 +261,12 @@ class SendWarning:
 
 TEMPLATE_OUTDATED = TemplateRefusal(
     code="template_outdated",
-    detail="PostHog does not hold the content this hash describes, so nothing was sent.",
+    detail="PostHog does not hold the content this hash describes, so this request sent nothing.",
     attr="template__hash",
     why="PostHog does not hold this content as the template's current content or one of the last 100 different "
     "contents it replaced, so sending the current content could send the wrong email.",
-    fix="Upload the template, then send again.",
+    fix="Upload the template, then send again. If you sent an `Idempotency-Key`, keep it: if an earlier request "
+    "with it already sent the email, PostHog answers with that email's `message_id` and sends nothing new.",
 )
 
 
@@ -369,9 +371,28 @@ def unusable_sender(error: SenderNotVerified | SenderNotAllowed) -> TemplateRefu
     )
 
 
+SEND_EMAIL_IDEMPOTENCY_KEY_PARAMETER = OpenApiParameter(
+    name="Idempotency-Key",
+    type=OpenApiTypes.STR,
+    location=OpenApiParameter.HEADER,
+    required=False,
+    description="A unique key for this email, 1 to 255 visible ASCII characters. A retry with the same key and the "
+    "same request within 24 hours sends nothing new and answers with the first send's `message_id` and "
+    "`replayed: true`. The key binds the template or `template_key`, `to`, `identifiers.distinct_id`, "
+    "`message_data`, the content you pass, a `from` you name, `reply_to` and `capture_engagement`. It doesn't bind "
+    "`template.hash`, the template's content or the default sender, so a retry still replays after those change, "
+    "and after the email's workflow is paused. A `template.hash` PostHog doesn't hold answers `409 template_outdated` "
+    "before the key is checked. Upload the template, then retry with the same key: if an earlier request with it "
+    "already sent the email, PostHog answers with that email's `message_id` and sends nothing new. Reusing the key "
+    "for a different "
+    "request, or a key used on `messaging_templates/{id}/send/`, answers `422 idempotency_key_reused`.",
+)
+
+
 class UnconfirmedEmailSend(TransactionalEmailUnavailable):
     default_detail = (
-        "The email pipeline did not confirm the send. The email may still go out, so check before you send it again."
+        "The email pipeline did not confirm the send. The email may still go out. If you sent an Idempotency-Key, "
+        "retry with the same key. Otherwise check before you send it again."
     )
 
 
@@ -404,16 +425,27 @@ class MessagingSendViewSet(TransactionalEmailViewSetMixin, viewsets.ViewSet):
                 response=TemplateErrorSerializer,
                 description="`template_outdated`: PostHog does not hold the content `template.hash` describes, as the "
                 "current content or one of the last 100 different contents it replaced, or no template has this key "
-                "yet. Nothing was sent",
+                "yet. This request sent nothing. "
+                "Upload the template, then send again. If you sent an `Idempotency-Key`, keep it: if an earlier "
+                "request with it already sent the email, PostHog answers with that email's `message_id` and sends "
+                "nothing new. "
+                "`idempotency_key_in_flight`: a request with the same `Idempotency-Key` is still in progress. Retry in "
+                "a few seconds",
+            ),
+            422: OpenApiResponse(
+                response=TemplateErrorSerializer,
+                description="`idempotency_key_reused`: the `Idempotency-Key` was already used for a different request. "
+                "Nothing was sent. Use a new key for a new email",
             ),
             429: OpenApiResponse(
                 response=TemplateErrorSerializer,
                 description="`throttled`: rate limited, retry later. `quota_limited`: the project is over its email "
                 "quota, and a retry sends nothing",
             ),
             503: OpenApiResponse(
-                description="The email pipeline did not confirm the send. The email may still go out, so sending "
-                "it again can deliver it twice"
+                description="The email pipeline did not confirm the send. The email may still go out. Retry with the "
+                "same `Idempotency-Key` within 24 hours: if the first request queued the email, the retry sends "
+                "nothing new and answers with its `message_id`. Without a key, sending it again can deliver it twice"
             ),
         },
         summary="Send a transactional email",
@@ -424,8 +456,10 @@ class MessagingSendViewSet(TransactionalEmailViewSetMixin, viewsets.ViewSet):
             "`template_outdated`, and content that newer content replaced sends the current content with a "
             "`template_superseded` warning. Or content your code rendered, named by `template_key`: PostHog sends it "
             "as is and never stores it. The first send under a key makes an unlisted workflow for it, so its metrics "
-            "and run history group under the key. Transactional emails skip opt-outs and frequency caps."
+            "and run history group under the key. Transactional emails skip opt-outs and frequency caps. Send an "
+            "`Idempotency-Key` header to retry without sending the email twice."
         ),
+        parameters=[SEND_EMAIL_IDEMPOTENCY_KEY_PARAMETER],
     )
     @action(detail=False, methods=["POST"], url_path="send/email")
     def send_email(self, request: Request, *args: Any, **kwargs: Any) -> Response:
@@ -463,7 +497,14 @@ def _send_template(self, request: Request, data: dict[str, Any]) -> Response:
             )
         except (SenderNotVerified, SenderNotAllowed) as e:
             return self._refuse(unusable_sender(e), usage=usage)
-        queued = self._send_transactional_email(request, target.template, data, usage=usage, send=send)
+        queued = self._send_transactional_email(
+            request,
+            target.template,
+            data,
+            usage=usage,
+            send=send,
+            idempotency_key=request.headers.get("Idempotency-Key"),
+        )
         warnings = [TEMPLATE_SUPERSEDED] if target.hash_state == ContentHashState.SUPERSEDED else []
         return self._queued(queued, warnings)
 
@@ -498,6 +539,7 @@ def _send_rendered_content(self, request: Request, data: dict[str, Any]) -> Resp
             to=data["to"],
             distinct_id=self._distinct_id(data),
             source=RenderedEmail(key=key, send=send),
+            idempotency_key=request.headers.get("Idempotency-Key"),
         )
         return self._queued(self._send(request, command, report=usage), [])
 

products/messaging/backend/services/transactional_email.py

@@ -244,7 +244,7 @@ def sync_transactional_workflow(team_id: int, template_id: UUID, *, user_id: int
         raise SenderUnavailable("No verified email sender. Verify a sending domain first.") from e
 
 
-def _transactional_workflow_id(team_id: int, template_id: UUID) -> UUID:
+def _transactional_workflow_id(team_id: int, template_id: UUID, *, require_active: bool) -> UUID:
     template = MessageTemplate.objects.get(team_id=team_id, id=template_id, deleted=False)
     if not template.transactional_hog_flow_id:
         raise TemplateNotTransactional("The template isn't transactional. Make it transactional to send it.")
@@ -255,29 +255,29 @@ def _transactional_workflow_id(team_id: int, template_id: UUID) -> UUID:
         content=template.content,
         message_category_id=template.message_category_id,
     )
-    _require_active(team_id, template.transactional_hog_flow_id)
+    _require_workflow(team_id, template.transactional_hog_flow_id, require_active=require_active)
     return template.transactional_hog_flow_id
 
 
-def _require_active(team_id: int, hog_flow_id: UUID) -> None:
+def _require_workflow(team_id: int, hog_flow_id: UUID, *, require_active: bool) -> None:
     try:
         workflow = get_workflow_summary(team_id=team_id, workflow_id=str(hog_flow_id))
     except WorkflowNotFound as e:
         raise TransactionalEmailPaused("The template's transactional email no longer exists.") from e
-    if workflow.status != "active":
+    if require_active and workflow.status != "active":
         raise TransactionalEmailPaused("The template's transactional email isn't active.")
 
 
 def _template_an_upload_claimed(team_id: int, key: str) -> MessageTemplate | None:
     return MessageTemplate.objects.filter(team_id=team_id, key=key, deleted=False, managed_by__isnull=False).first()
 
 
-def _code_email_workflow_id(team_id: int, key: str) -> UUID:
+def _code_email_workflow_id(team_id: int, key: str, *, require_active: bool) -> UUID:
     template = _template_an_upload_claimed(team_id, key)
     if template is not None:
-        return _transactional_workflow_id(team_id, template.id)
+        return _transactional_workflow_id(team_id, template.id, require_active=require_active)
     hog_flow_id = _key_workflow_id(team_id, key)
-    _require_active(team_id, hog_flow_id)
+    _require_workflow(team_id, hog_flow_id, require_active=require_active)
     return hog_flow_id
 
 
@@ -332,7 +332,7 @@ def code_send_from(team_id: int, address: str | None) -> SendFrom:
         if address:
             raise SenderNotAllowed(f"No verified sender in this project can send from `{email}`.") from e
         raise SenderNotVerified("The project has no verified default email sender.") from e
-    return SendFrom(integration_id=sender.integration_id, email=sender.email, name=sender.name)
+    return SendFrom(integration_id=sender.integration_id, email=sender.email, name=sender.name, is_default=not address)
 
 
 def rendered_content(*, subject: str, html: str, text: str) -> RenderedContent:
@@ -349,7 +349,7 @@ def send_email(command: EmailSendCommand) -> EmailSendOutcome:
     if "{" in command.to or "}" in command.to:
         return Rejected(code="invalid_recipient", detail="The recipient address can't contain { or }.")
     try:
-        hog_flow_id = _sendable_workflow_id(command.team_id, command.source)
+        hog_flow_id = _sendable_workflow_id(command)
     except TransactionalEmailError as e:
         return Rejected(code=e.code, detail=str(e))
     payload: dict[str, Any] = {
@@ -423,12 +423,17 @@ def _send_check_attr(body: dict[str, Any], code: str, source: StoredTemplate | R
     return (NAMED_SENDER_CHECK_ATTR if names_sender else SEND_CHECK_ATTR).get(code)
 
 
-def _sendable_workflow_id(team_id: int, source: StoredTemplate | RenderedEmail) -> UUID:
-    match source:
+def _sendable_workflow_id(command: EmailSendCommand) -> UUID:
+    require_active = not _pipeline_checks_active_after_the_key(command)
+    match command.source:
         case StoredTemplate(template_id=template_id):
-            return _transactional_workflow_id(team_id, template_id)
+            return _transactional_workflow_id(command.team_id, template_id, require_active=require_active)
         case RenderedEmail(key=key):
-            return _code_email_workflow_id(team_id, key)
+            return _code_email_workflow_id(command.team_id, key, require_active=require_active)
+
+
+def _pipeline_checks_active_after_the_key(command: EmailSendCommand) -> bool:
+    return command.source.send is not None and command.idempotency_key is not None
 
 
 def _queue(
@@ -447,7 +452,12 @@ def _send_payload(send: AcceptedSend) -> dict[str, Any]:
         )
     if send.sender is not None:
         payload["from"] = _without_empty(
-            {"integrationId": send.sender.integration_id, "email": send.sender.email, "name": send.sender.name}
+            {
+                "integrationId": send.sender.integration_id,
+                "email": send.sender.email,
+                "name": send.sender.name,
+                "isDefault": send.sender.is_default,
+            }
         )
     if send.reply_to:
         payload["replyTo"] = list(send.reply_to)