feat(messaging): send the workflow version a template send graded

products/messaging/backend/facade/contracts.py

@@ -135,6 +135,10 @@ class Rejected:
 class Unavailable:
     """The email pipeline did not confirm the send. It may still go out."""
 
+    detail: str | None = None
+    why: str | None = None
+    fix: str | None = None
+
 
 EmailSendOutcome = Queued | Rejected | Unavailable
 

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

@@ -171,6 +171,12 @@ class TransactionalEmailUnavailable(exceptions.APIException):
     default_code = "delivery_unavailable"
 
 
+class ExplainedUnavailable(TransactionalEmailUnavailable):
+    def __init__(self, refusal: TemplateRefusal) -> None:
+        self.refusal = refusal
+        super().__init__(detail=refusal.detail, code=refusal.code)
+
+
 class TransactionalEmailBurstThrottle(PersonalOrProjectSecretApiKeyRateThrottle):
     scope = "transactional_email_burst"
     rate = "600/minute"
@@ -239,7 +245,7 @@ class TransactionalEmailViewSetMixin(TeamAndOrgViewSetMixin):
     unconfirmed_send_error: type[TransactionalEmailUnavailable] = TransactionalEmailUnavailable
 
     def handle_exception(self, exc: Exception) -> Response:
-        if isinstance(exc, SendCheckFailed):
+        if isinstance(exc, SendCheckFailed | ExplainedUnavailable):
             return Response(TemplateErrorSerializer(exc.refusal).data, status=exc.status_code)
         return super().handle_exception(exc)
 
@@ -326,6 +332,17 @@ def _send(self, request: Any, command: EmailSendCommand, *, report: dict[str, st
                 raise SendCheckFailed(TemplateRefusal(code=code, detail=detail, attr=attr, why=why, fix=fix))
             case Rejected(code=code, detail=detail):
                 raise TransactionalEmailRejected(code, detail)
+            case Unavailable(detail=str() as detail, why=str() as why, fix=str() as fix):
+                raise ExplainedUnavailable(
+                    TemplateRefusal(
+                        code=TransactionalEmailUnavailable.default_code,
+                        detail=detail,
+                        attr=None,
+                        why=why,
+                        fix=fix,
+                        type="server_error",
+                    )
+                )
             case Unavailable():
                 raise self.unconfirmed_send_error()
 

products/messaging/backend/services/transactional_email.py

@@ -45,7 +45,7 @@
 from products.messaging.backend.models.message_category import MessageCategoryType
 from products.messaging.backend.models.transactional_email_key import TRANSACTIONAL_EMAIL_KEY_CONSTRAINT
 from products.messaging.backend.services.html_text import text_from_html
-from products.workflows.backend.facade.api import WorkflowNotFound, get_workflow_summary
+from products.workflows.backend.facade.api import WorkflowNotFound, get_workflow_summary, get_workflow_version
 from products.workflows.backend.facade.transactional_emails import (
     TransactionalEmailContent,
     TransactionalEmailContentInvalid,
@@ -381,6 +381,8 @@ def send_email(command: EmailSendCommand) -> EmailSendOutcome:
         payload["idempotency_key"] = command.idempotency_key
     try:
         response = _queue(command.team_id, str(hog_flow_id), command.source, payload)
+    except WorkflowNotFound:
+        return _rejected_for_a_deleted_workflow(command.team_id, hog_flow_id)
     except Exception:
         logger.exception("transactional_email_queue_failed", team_id=command.team_id, hog_flow_id=str(hog_flow_id))
         return Unavailable()
@@ -404,6 +406,8 @@ def send_email(command: EmailSendCommand) -> EmailSendOutcome:
                 code="idempotency_key_reused",
                 detail="This Idempotency-Key was already used for a different request. Use a new key for a new email.",
             )
+        if _pipeline_holds_an_older_workflow(response):
+            return _unavailable_until_the_pipeline_catches_up(command.team_id, hog_flow_id)
         logger.error(
             "transactional_email_queue_rejected",
             team_id=command.team_id,
@@ -416,6 +420,41 @@ def send_email(command: EmailSendCommand) -> EmailSendOutcome:
     return Queued(message_id=str(body["invocation_id"]), replayed=bool(body.get("replayed", False)))
 
 
+def _rejected_for_a_deleted_workflow(team_id: int, hog_flow_id: UUID) -> Unavailable:
+    logger.error(
+        "transactional_email_queue_rejected",
+        team_id=team_id,
+        hog_flow_id=str(hog_flow_id),
+        reason="workflow_not_found",
+    )
+    return Unavailable()
+
+
+def _pipeline_holds_an_older_workflow(response: requests.Response) -> bool:
+    if response.status_code != 503:
+        return False
+    try:
+        body = response.json()
+    except ValueError:
+        return False
+    return isinstance(body, dict) and body.get("error") == "workflow_version_behind"
+
+
+def _unavailable_until_the_pipeline_catches_up(team_id: int, hog_flow_id: UUID) -> Unavailable:
+    logger.error(
+        "transactional_email_queue_rejected",
+        team_id=team_id,
+        hog_flow_id=str(hog_flow_id),
+        reason="workflow_version_behind",
+    )
+    return Unavailable(
+        detail="This template's newest version hasn't reached the email pipeline yet.",
+        why="The email pipeline still holds an older version than the one PostHog checked the send against, "
+        "so nothing was sent.",
+        fix="Retry in a few seconds. If you sent an Idempotency-Key, use the same key.",
+    )
+
+
 SEND_CHECK_ATTR = {"invalid_recipient": "to", "recipient_suppressed": "to"}
 NAMED_SENDER_CHECK_ATTR = {**SEND_CHECK_ATTR, "sender_not_verified": "from"}
 
@@ -459,7 +498,10 @@ def _queue(
 ) -> requests.Response:
     if source.send is None:
         return run_transactional_workflow(team_id, hog_flow_id, payload)
-    return send_accepted_transactional_email(team_id, hog_flow_id, {**payload, "send": _send_payload(source.send)})
+    send = _send_payload(source.send)
+    if isinstance(source, StoredTemplate):
+        send["hogFlowVersion"] = get_workflow_version(team_id=team_id, workflow_id=hog_flow_id)
+    return send_accepted_transactional_email(team_id, hog_flow_id, {**payload, "send": send})
 
 
 def _send_payload(send: AcceptedSend) -> dict[str, Any]:

products/messaging/backend/tests/api/test_transactional_email.py

@@ -5,6 +5,7 @@
 from posthog.test.base import APIBaseTest
 from unittest.mock import MagicMock, patch
 
+from django.conf import settings
 from django.db import IntegrityError, connection, transaction
 from django.test.utils import CaptureQueriesContext
 
@@ -41,6 +42,11 @@
 PASSWORD_RESET_HASH = "v1:973d41bbe3fdca5c52bb951cd4b5fd7159ae630b56c8f3997cbb31afd1ea76fd"
 
 
+class _HogFlowReadsFromMissingReplicaRouter:
+    def db_for_read(self, model: type, **hints: Any) -> str | None:
+        return "unavailable_replica" if model is HogFlow else None
+
+
 @patch("products.messaging.backend.presentation.views.transactional_email_api.posthoganalytics.capture")
 @patch(
     "products.messaging.backend.presentation.views.transactional_email_api.posthoganalytics.feature_enabled",
@@ -489,7 +495,7 @@ def test_both_send_routes_share_one_rate_limit_per_key(self, _rate_limit, mock_p
 
     @parameterized.expand(
         [
-            (f"{name}_{addressed_by}", addressed_by, outcome, detail)
+            (f"{name}_{addressed_by}", addressed_by, outcome, detail, None, None)
             for addressed_by, detail in (
                 ("url_id", "The email pipeline is unavailable. Try again shortly."),
                 (
@@ -513,9 +519,20 @@ def test_both_send_routes_share_one_rate_limit_per_key(self, _rate_limit, mock_p
                 ),
             ]
         ]
+        + [
+            (
+                "cdp_api_holding_an_older_workflow_version",
+                "id",
+                MagicMock(status_code=503, json=lambda: {"error": "workflow_version_behind"}),
+                "This template's newest version hasn't reached the email pipeline yet.",
+                "The email pipeline still holds an older version than the one PostHog checked the send against, "
+                "so nothing was sent.",
+                "Retry in a few seconds. If you sent an Idempotency-Key, use the same key.",
+            )
+        ]
     )
     def test_a_send_the_pipeline_does_not_confirm_is_unavailable(
-        self, mock_post, _flag, capture, _name, addressed_by, outcome, detail
+        self, mock_post, _flag, capture, _name, addressed_by, outcome, detail, why, fix
     ):
         if isinstance(outcome, Exception):
             mock_post.side_effect = outcome
@@ -525,8 +542,15 @@ def test_a_send_the_pipeline_does_not_confirm_is_unavailable(
 
         response = self.client.post(url, body, format="json")
 
+        body = response.json()
+        explanation = {field: body[field] for field in ("why", "fix") if field in body}
         assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
-        assert (response.json()["code"], response.json()["detail"]) == ("delivery_unavailable", detail)
+        assert (body["code"], body["detail"], explanation) == (
+            "delivery_unavailable",
+            detail,
+            {"why": why, "fix": fix} if why else {},
+        )
+        assert "workflow_version_behind" not in response.content.decode()
         assert capture.call_args.kwargs["properties"]["outcome"] == "unavailable"
 
     def test_a_send_that_breaks_after_the_pipeline_answers_is_counted_as_an_error(self, mock_post, _flag, capture):
@@ -834,19 +858,19 @@ def test_sends_rendered_content_as_is_under_the_keys_workflow(self, mock_post, _
                     "tracking": {"captureEngagement": True},
                 },
             ),
-            ("template_by_key", "key", {}, {"tracking": {"captureEngagement": True}}),
-            ("template_by_id", "id", {}, {"tracking": {"captureEngagement": True}}),
+            ("template_by_key", "key", {}, {"tracking": {"captureEngagement": True}, "hogFlowVersion": 1}),
+            ("template_by_id", "id", {}, {"tracking": {"captureEngagement": True}, "hogFlowVersion": 1}),
             (
                 "template_by_id_without_capture",
                 "id",
                 {"capture_engagement": False},
-                {"tracking": {"captureEngagement": False}},
+                {"tracking": {"captureEngagement": False}, "hogFlowVersion": 1},
             ),
             (
                 "template_by_key_with_reply_to",
                 "key",
                 {"reply_to": ["help@example.com"]},
-                {"replyTo": ["help@example.com"], "tracking": {"captureEngagement": True}},
+                {"replyTo": ["help@example.com"], "tracking": {"captureEngagement": True}, "hogFlowVersion": 1},
             ),
             (
                 "template_by_key_with_from",
@@ -855,6 +879,7 @@ def test_sends_rendered_content_as_is_under_the_keys_workflow(self, mock_post, _
                 {
                     "from": {"email": "security@example.com", "name": "Example"},
                     "tracking": {"captureEngagement": True},
+                    "hogFlowVersion": 1,
                 },
             ),
         ]
@@ -1197,6 +1222,61 @@ def test_rejects_making_a_template_transactional_it_cannot_send(
         template.refresh_from_db()
         assert template.transactional_hog_flow_id is None
 
+    @patch("products.messaging.backend.services.transactional_email.reload_hog_flows_on_workers")
+    def test_a_template_send_after_a_save_carries_the_workflow_version_that_holds_the_saved_content(
+        self, _reload, mock_post, _flag, _capture
+    ):
+        self._cdp_accepts(mock_post)
+        with self.captureOnCommitCallbacks(execute=True):
+            saved = self.client.patch(
+                f"/api/projects/{self.team.id}/messaging_templates/{self.template.id}/",
+                {"content": {"email": {"subject": "A new subject", "html": "<p>New</p>"}}},
+                format="json",
+            )
+        assert saved.status_code == status.HTTP_200_OK, saved.json()
+
+        read_version = transactional_email.get_workflow_version
+
+        def read_version_with_hog_flow_reads_on_a_missing_replica(**kwargs: Any) -> int:
+            with self.settings(DATABASE_ROUTERS=[_HogFlowReadsFromMissingReplicaRouter(), *settings.DATABASE_ROUTERS]):
+                return read_version(**kwargs)
+
+        with patch.object(
+            transactional_email,
+            "get_workflow_version",
+            side_effect=read_version_with_hog_flow_reads_on_a_missing_replica,
+        ):
+            response = self._psak_client(["hog_flow:write"]).post(
+                self.send_email_url, self._send_body("id"), format="json"
+            )
+
+        assert response.status_code == status.HTTP_202_ACCEPTED, response.json()
+        assert mock_post.call_args.kwargs["json"]["send"]["hogFlowVersion"] == 2
+
+    def test_a_template_send_whose_workflow_is_deleted_before_its_version_read_is_rejected_as_unavailable(
+        self, mock_post, _flag, capture
+    ):
+        read_version = transactional_email.get_workflow_version
+
+        def delete_the_workflow_then_read_its_version(**kwargs: Any) -> int:
+            HogFlow.objects.filter(id=kwargs["workflow_id"]).delete()
+            return read_version(**kwargs)
+
+        url, body = self._send_request("key", self.body)
+        with (
+            patch.object(
+                transactional_email, "get_workflow_version", side_effect=delete_the_workflow_then_read_its_version
+            ),
+            patch.object(transactional_email, "logger") as logger,
+        ):
+            response = self.client.post(url, body, format="json")
+
+        assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
+        assert response.json()["code"] == "delivery_unavailable"
+        assert [call.args[0] for call in logger.error.call_args_list] == ["transactional_email_queue_rejected"]
+        logger.exception.assert_not_called()
+        mock_post.assert_not_called()
+
     @patch("products.messaging.backend.services.transactional_email.reload_hog_flows_on_workers")
     def test_saving_a_transactional_template_writes_a_new_workflow_version(self, reload, mock_post, _flag, _capture):
         MessageTemplate.objects.filter(id=self.template.id).update(draft_content={"email": {"subject": "Draft"}})

products/workflows/backend/facade/api.py

@@ -2,6 +2,7 @@
 from typing import Any
 from uuid import UUID
 
+from django.db import router
 from django.db.models import F
 
 from posthog.helpers.full_text_search import build_rank
@@ -224,6 +225,18 @@ def get_workflow_summary(*, team_id: int, workflow_id: str) -> WorkflowSummary:
     return WorkflowSummary(id=str(row["id"]), name=row["name"] or "", status=row["status"])
 
 
+def get_workflow_version(*, team_id: int, workflow_id: str) -> int:
+    version = (
+        HogFlow.objects.using(router.db_for_write(HogFlow))
+        .filter(team_id=team_id, id=workflow_id)
+        .values_list("version", flat=True)
+        .first()
+    )
+    if version is None:
+        raise WorkflowNotFound()
+    return version
+
+
 def has_active_workflows(*, team_id: int) -> bool:
     return HogFlow.objects.filter(team_id=team_id, status=HogFlow.State.ACTIVE).exists()