feat(messaging): send the workflow version a template send by key was checked against

products/messaging/backend/facade/contracts.py

@@ -134,6 +134,10 @@ class DeliveryUnavailable(Exception):
     """The email pipeline did not accept the send. Safe to retry."""
 
 
+class WorkflowVersionBehind(DeliveryUnavailable):
+    pass
+
+
 @frozen
 class QueuedEmail:
     message_id: str

products/messaging/backend/facade/transactional_email.py

@@ -9,6 +9,7 @@
     TemplateOutdated,
     TemplateSendOptions,
     TransactionalEmailError,
+    WorkflowVersionBehind,
 )
 from products.messaging.backend.services.transactional_email import (
     SENDER_ADDRESS_MAX_LENGTH,
@@ -38,6 +39,7 @@
     "TemplateSendOptions",
     "TransactionalEmailError",
     "TransactionalEmailStage",
+    "WorkflowVersionBehind",
     "get_email_status",
     "make_transactional",
     "observe_transactional_email_stage",

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

@@ -96,6 +96,7 @@
     TemplateSendOptions,
     TransactionalEmailError,
     TransactionalEmailStage,
+    WorkflowVersionBehind,
     make_transactional,
     observe_transactional_email_stage,
     report_first_code_send,
@@ -744,6 +745,13 @@ class TransactionalEmailUnavailable(exceptions.APIException):
     default_code = "delivery_unavailable"
 
 
+class PipelineHoldsAnOlderWorkflow(TransactionalEmailUnavailable):
+    default_detail = (
+        "This template's newest version hasn't reached the email pipeline yet, so nothing was sent. Retry in a few "
+        "seconds, with the same Idempotency-Key if you sent one."
+    )
+
+
 class UnconfirmedEmailSend(TransactionalEmailUnavailable):
     default_detail = (
         "The email pipeline did not confirm the send, so the email may still go out. To retry, send it again "
@@ -956,6 +964,9 @@ def _queue_send(
         except TransactionalEmailError as e:
             outcome.update({"outcome": "rejected", "error_code": e.code})
             raise TransactionalEmailRejected(e)
+        except WorkflowVersionBehind:
+            outcome["outcome"] = "unavailable"
+            raise PipelineHoldsAnOlderWorkflow() from None
         except DeliveryUnavailable:
             outcome["outcome"] = "unavailable"
             raise unavailable() from None
@@ -1668,7 +1679,7 @@ def email(self, request: Any, *args: Any, **kwargs: Any) -> Response:
                 raise NotFound()
             self.check_object_permissions(request, template)
         return self._queue_transactional_email(
-            request, template, route="send_email", options=_template_send_options(request.data, request.validated_data)
+            request, template, route="send_email", options=_id_send_options(request.data, request.validated_data)
         )
 
     def _refuse_engagement_capture_outside_the_beta(self, sent: Mapping[str, Any]) -> None:
@@ -1692,7 +1703,7 @@ def _send_template_by_key(self, request: Any) -> Response:
         with transactional_email_stage("workflow_access"):
             if template.transactional_hog_flow_id:
                 self._require_workflow_editor(template.transactional_hog_flow_id)
-        options = _template_send_options(request.data, data)
+        options = _template_send_options(data)
 
         def send(idempotency_key: str | None) -> QueuedEmail:
             if found.hash_state == ContentHashState.OUTDATED:
@@ -1760,16 +1771,17 @@ def _send_code_email(self, request: Any) -> Response:
     "PostHog doesn't hold the content `template.hash` describes, so nothing was sent. Upload the template, then send "
     "again."
 )
-TEMPLATE_SEND_OPTION_FIELDS = ("from", "reply_to", "capture_engagement")
 ENGAGEMENT_CAPTURE_OUTSIDE_THE_BETA = (
     "A send by `template.id` takes this field only in the beta for sending email from code. Remove it to send the "
     "email."
 )
 
 
-def _template_send_options(sent: Any, data: Mapping[str, Any]) -> TemplateSendOptions | None:
-    if not isinstance(sent, Mapping) or not any(field in sent for field in TEMPLATE_SEND_OPTION_FIELDS):
-        return None
+def _id_send_options(sent: Mapping[str, Any], data: Mapping[str, Any]) -> TemplateSendOptions | None:
+    return _template_send_options(data) if "capture_engagement" in sent else None
+
+
+def _template_send_options(data: Mapping[str, Any]) -> TemplateSendOptions:
     return TemplateSendOptions(
         sender=data.get("from"),
         reply_to=tuple(data.get("reply_to", ())),

products/messaging/backend/services/transactional_email.py

@@ -46,13 +46,14 @@
     TemplateSendOptions,
     TransactionalEmailError,
     TransactionalEmailPaused,
+    WorkflowVersionBehind,
 )
 from products.messaging.backend.models import FirstCodeSend, MessageCategory, MessageTemplate, TransactionalEmailKey
 from products.messaging.backend.models.first_code_send import FIRST_CODE_SEND_CONSTRAINT
 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,
@@ -479,7 +480,10 @@ def send_transactional_email(
     payload = _send_payload(to, distinct_id, message_data, idempotency_key)
     if options is None:
         return _queue(team.id, hog_flow_id, payload, run_transactional_workflow, timeout=timeout)
-    accepted_send = _accepted_template_send(team.id, options)
+    accepted_send = {
+        **_accepted_template_send(team.id, options),
+        "hogFlowVersion": _graded_version(team.id, hog_flow_id),
+    }
     return _queue(
         team.id, hog_flow_id, {**payload, "send": accepted_send}, send_accepted_transactional_email, timeout=timeout
     )
@@ -539,6 +543,17 @@ def _accepted_send(team_id: int, email: CodeEmail) -> dict[str, Any]:
     return accepted
 
 
+def _graded_version(team_id: int, hog_flow_id: UUID) -> int:
+    try:
+        return get_workflow_version(team_id=team_id, workflow_id=str(hog_flow_id))
+    except WorkflowNotFound:
+        workflow_deleted_since_the_lookup = DeliveryUnavailable()
+    logger.error(
+        "transactional_email_queue_rejected", team_id=team_id, hog_flow_id=str(hog_flow_id), reason="workflow_not_found"
+    )
+    raise workflow_deleted_since_the_lookup
+
+
 def _accepted_template_send(team_id: int, options: TemplateSendOptions) -> dict[str, Any]:
     accepted: dict[str, Any] = {"tracking": {"captureEngagement": options.capture_engagement}}
     if options.sender:
@@ -604,6 +619,8 @@ def _queue(
             status_code=response.status_code,
             refusal_code=refusal_code,
         )
+        if refusal_code == "workflow_version_behind":
+            raise WorkflowVersionBehind()
         raise DeliveryUnavailable()
 
     body = response.json()

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

@@ -615,7 +615,8 @@ def test_a_send_by_key_with_a_hash_only_ever_sends_the_published_content(
             post.assert_not_called()
         else:
             assert [warning["code"] for warning in response.json().get("warnings", [])] == expected_warnings
-            assert post.call_args.args[0].endswith(f"/hog_flows/{hog_flow_id}/transactional_email")
+            assert post.call_args.args[0].endswith(f"/hog_flows/{hog_flow_id}/transactional_email/send")
+            assert post.call_args.kwargs["json"]["send"]["hogFlowVersion"] == 2
         assert self._template(published["id"])["content_hash"] == published["content_hash"]
         assert self._sent_email(hog_flow_id)["subject"] == "Welcome aboard, {{ variables.first_name }}"
         assert self._workflow(hog_flow_id)["version"] == 2

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

@@ -6,6 +6,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
 
@@ -60,6 +61,11 @@ def _stage_counts() -> dict[str, float]:
     }
 
 
+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.message_templates.posthoganalytics.capture")
 @patch(
     "products.messaging.backend.presentation.views.message_templates.posthoganalytics.feature_enabled",
@@ -566,7 +572,7 @@ def test_refuses_on_a_send_by_template_id_what_its_template_sets(
                 {"capture_engagement": False},
                 BOTH_EMAIL_FLAGS,
                 "transactional_email/send",
-                {"tracking": {"captureEngagement": False}},
+                {"tracking": {"captureEngagement": False}, "hogFlowVersion": 1},
             ),
             ("by_default_while_the_code_flag_is_off", {}, {TRANSACTIONAL_EMAIL_API_FLAG}, "transactional_email", None),
         ]
@@ -631,13 +637,25 @@ def _keyed(self) -> dict[str, Any]:
 
     @parameterized.expand(
         [
-            ("without_a_hash", {}, "transactional_email", None, None),
-            ("with_the_current_hash", {"hash": PASSWORD_RESET_HASH}, "transactional_email", None, None),
+            (
+                "without_a_hash",
+                {},
+                "transactional_email/send",
+                {"tracking": {"captureEngagement": True}, "hogFlowVersion": 1},
+                None,
+            ),
+            (
+                "with_the_current_hash",
+                {"hash": PASSWORD_RESET_HASH},
+                "transactional_email/send",
+                {"tracking": {"captureEngagement": True}, "hogFlowVersion": 1},
+                None,
+            ),
             (
                 "with_a_superseded_hash",
                 {"hash": SUPERSEDED_HASH},
-                "transactional_email",
-                None,
+                "transactional_email/send",
+                {"tracking": {"captureEngagement": True}, "hogFlowVersion": 1},
                 [
                     {
                         "code": "template_superseded",
@@ -655,14 +673,15 @@ def _keyed(self) -> dict[str, Any]:
                     "from": {"email": "hello@example.com", "name": "Example"},
                     "replyTo": ["help@example.com"],
                     "tracking": {"captureEngagement": True},
+                    "hogFlowVersion": 1,
                 },
                 None,
             ),
             (
                 "without_engagement_capture",
                 {"capture_engagement": False},
                 "transactional_email/send",
-                {"tracking": {"captureEngagement": False}},
+                {"tracking": {"captureEngagement": False}, "hogFlowVersion": 1},
                 None,
             ),
         ]
@@ -1146,6 +1165,67 @@ def test_rejects_an_update_that_would_stop_a_transactional_template_sending(
         assert (after.message_category_id, after.content) == (None, before.content)
         assert self._workflow(self.hog_flow_id)["version"] == 1
 
+    @patch("products.messaging.backend.services.transactional_email.reload_hog_flows_on_workers")
+    def test_a_send_after_a_save_carries_the_workflow_version_that_holds_the_saved_content(
+        self, _reload, mock_post, _flag, _capture
+    ):
+        self._cdp_accepts(mock_post)
+        body = self._keyed()
+        with self.captureOnCommitCallbacks(execute=True):
+            MessageTemplate.objects.filter(id=self.template.id).update(managed_by=None)
+            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",
+            )
+            MessageTemplate.objects.filter(id=self.template.id).update(managed_by=TemplateManagedBy.CODE)
+        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, body, 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_send_the_pipeline_holds_an_older_workflow_for_asks_for_a_retry(self, mock_post, _flag, capture):
+        mock_post.return_value = MagicMock(status_code=503, json=lambda: {"error": "workflow_version_behind"})
+
+        response = self._psak_client(["hog_flow:write"]).post(self.send_email_url, self._keyed(), format="json")
+
+        assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
+        assert (response.json()["code"], response.json()["detail"]) == (
+            "delivery_unavailable",
+            "This template's newest version hasn't reached the email pipeline yet, so nothing was sent. Retry in a "
+            "few seconds, with the same Idempotency-Key if you sent one.",
+        )
+        assert "workflow_version_behind" not in response.content.decode()
+        assert capture.call_args.kwargs["properties"]["outcome"] == "unavailable"
+
+    def test_a_send_whose_workflow_is_deleted_before_its_version_read_is_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)
+
+        with patch.object(
+            transactional_email, "get_workflow_version", side_effect=delete_the_workflow_then_read_its_version
+        ):
+            response = self._psak_client(["hog_flow:write"]).post(self.send_email_url, self._keyed(), format="json")
+
+        assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE
+        assert response.json()["code"] == "delivery_unavailable"
+        mock_post.assert_not_called()
+
     def _sender_client(self, sender: str) -> APIClient:
         if sender == "personal_api_key":
             token = generate_random_token_personal()

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
@@ -222,6 +223,18 @@ def get_workflow_names(*, team_id: int, workflow_ids: Iterable[str]) -> dict[str
     }
 
 
+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 get_workflow_summary(*, team_id: int, workflow_id: str) -> WorkflowSummary:
     row = HogFlow.objects.filter(team_id=team_id, id=workflow_id).values("id", "name", "status").first()
     if row is None: