diff --git a/posthog/email.py b/posthog/email.py index 86de9fecdf18..0beaa96d52ec 100644 --- a/posthog/email.py +++ b/posthog/email.py @@ -1,6 +1,7 @@ import re import html import uuid +import hashlib import smtplib import datetime import dataclasses @@ -12,12 +13,14 @@ from django.conf import settings from django.core import exceptions, mail +from django.core.cache import cache from django.core.mail.backends.smtp import EmailBackend from django.db import transaction from django.template.loader import get_template from django.utils import timezone from django.utils.module_loading import import_string +import urllib3 import requests import structlog import css_inline @@ -31,6 +34,8 @@ from posthog.helpers.email_utils import sanitize_display_name, sanitize_email_string from posthog.models.instance_setting import get_instance_setting from posthog.models.messaging import MessagingRecord, get_email_hashes +from posthog.ph_client import ScopedCapture, ph_scoped_capture +from posthog.utils import get_instance_region logger = structlog.get_logger(__name__) @@ -54,7 +59,7 @@ def inline_css(value: str) -> str: def is_http_email_service_available() -> bool: """ Returns whether HTTP email services are available on this instance (i.e. settings are in place). - This currently only supports Customer.io. + This currently only supports Customer.io. Workflows only takes over emails that this path sends. """ return bool(settings.CUSTOMER_IO_API_KEY) @@ -164,6 +169,25 @@ def is_email_available(with_absolute_urls: bool = False) -> bool: "loop_run_summary": "77", } +# Message template ids in the US project, for templates that can send through Workflows instead of Customer.io. +# A template sends through Workflows only for recipients the flag below matches. +WORKFLOWS_TEMPLATE_ID_MAP: dict[str, str] = { + "password_reset": "01a1597a-5d91-0000-99fa-9eeb18d293c5", +} + +# Release conditions match the `email` and `transactional_email_template` person properties passed at evaluation, +# so one flag picks both the recipients and the templates that move. +WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG = "transactional-emails-via-workflows" + +# Matches how long the send API keeps an idempotency key, which outlasts every Celery retry. +WORKFLOWS_PIN_TTL_SECONDS = 24 * 60 * 60 + +EMAIL_WORKFLOWS_FALLBACK_COUNTER = Counter( + "posthog_email_workflows_fallback_total", + "Emails that Workflows refused and Customer.io sent instead, by template.", + labelnames=["template_name"], +) + def get_email_team_and_org_context( team: Optional["Team"] = None, @@ -204,27 +228,277 @@ def get_customer_io_template_id(template_name: str) -> str: return template_id -# Note: this http sender is only configure for customer.io right now and it's set up to send -# via templates so all the configuration is done in the customer.io - i.e. no subject, body, etc. +class WorkflowsEmailError(Exception): + """Workflows did not queue the email, or the caller could not tell whether it did. + + `code` and `status_code` match posthog-python's `EmailError`, so the fallback rule keeps working when the + SDK sends instead. `status_code` is None when no response came back. + """ + + def __init__(self, message: str, *, code: str, status_code: int | None = None) -> None: + super().__init__(message) + self.code = code + self.status_code = status_code + + +def _workflows_error_allows_fallback(err: WorkflowsEmailError) -> bool: + """Whether Customer.io can send the email without the risk of a duplicate.""" + if err.status_code is None: + # A connection that was never made and a client that queued nothing sent nothing. Any other network + # failure can come after Workflows queued the email. + return err.code in ("connection_error", "not_queued") + # Redirects and 4xx answers come before anything is queued, apart from 408 and the two idempotency answers: + # 409 (an attempt with this key is in flight) and 422 (this key already queued an email). A 5xx can follow a + # queued email, because the send API also returns 503 when the workflow service queued it but failed to answer. + return 300 <= err.status_code < 500 and err.status_code not in (408, 409, 422) + + +def _never_connected(err: requests.ConnectionError) -> bool: + """Whether the request failed before a connection existed, so Workflows cannot have received it.""" + if isinstance(err, requests.ConnectTimeout): + return True + reason = err.args[0] if err.args else None + if isinstance(reason, urllib3.exceptions.MaxRetryError): + reason = reason.reason + # A reset after the request went out arrives as a ProtocolError instead. + return isinstance(reason, urllib3.exceptions.NewConnectionError) + + +def get_workflows_template_id( + template_name: str, dest: dict[str, str], pinned_template_id: str | None = None +) -> str | None: + """The message template to send this email with through Workflows, or None to send it through Customer.io. + + A pinned template skips only the flag. The settings checks still apply, because they turn Workflows off. + """ + template_id = pinned_template_id or WORKFLOWS_TEMPLATE_ID_MAP.get(template_name) + if not template_id or not settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY: + return None + if not settings.POSTHOG_HOST.startswith("https://") and not settings.DEBUG: + # The request carries the secret key and the email's contents, such as a password reset link. + logger.warning("email_workflows_insecure_host", template_name=template_name) + return None + if pinned_template_id: + return pinned_template_id + try: + enabled = posthoganalytics.feature_enabled( + WORKFLOWS_TRANSACTIONAL_EMAIL_FLAG, + dest.get("distinct_id") or dest["raw_email"], + person_properties={"email": dest["raw_email"], "transactional_email_template": template_name}, + send_feature_flag_events=False, + ) + except Exception as err: + logger.warning("email_workflows_flag_check_failed", template_name=template_name, error=str(err)) + return None + return template_id if enabled else None + + +def _unescape_for_liquid(value: Any, key: Any = None) -> Any: + """Undo the html.escape in sanitize_email_properties, and keep the URL defang. + + Customer.io outputs Liquid values as they are, so the properties arrive pre-escaped. Workflows escapes Liquid + output itself, so a pre-escaped value renders as `&amp;` and shows the entity to the reader. + """ + if key is not None and _classify_key(key) == "passthrough": + return value + if isinstance(value, str): + return html.unescape(value) + if isinstance(value, dict): + return {k: _unescape_for_liquid(v, key=k) for k, v in value.items()} + if isinstance(value, list): + return [_unescape_for_liquid(item) for item in value] + return value + + +def _queue_workflows_email( + *, + to: str, + template_id: str, + template_data: dict[str, Any], + distinct_id: str | None = None, + idempotency_key: str, +) -> dict[str, Any]: + """Queue one email from a Workflows transactional template and return its `message_id` and `replayed`. + + Takes the same arguments as posthog-python's `Emails.send`, so the SDK can replace this body once it ships. + """ + payload: dict[str, Any] = {"to": to, "message_data": template_data} + if distinct_id: + payload["identifiers"] = {"distinct_id": distinct_id} + try: + response = requests.post( + # The project secret API key names the project, so `@current` resolves to it. + f"{settings.POSTHOG_HOST.rstrip('/')}/api/projects/@current/messaging_templates/{template_id}/send/", + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {settings.POSTHOG_TRANSACTIONAL_EMAIL_API_KEY}", + "Idempotency-Key": idempotency_key, + }, + json=payload, + # The caller holds the MessagingRecord row lock for the whole request, so keep it short. + timeout=10, + allow_redirects=False, + ) + except requests.ConnectionError as err: + code = "connection_error" if _never_connected(err) else "network_error" + raise WorkflowsEmailError(str(err), code=code) from err + except requests.Timeout as err: + raise WorkflowsEmailError(str(err), code="timeout") from err + + if response.status_code != 202: + try: + code = response.json().get("code") or "api_error" + except ValueError: + code = "api_error" + raise WorkflowsEmailError( + f"{response.status_code} - {response.text}", code=code, status_code=response.status_code + ) + try: + body = response.json() + except ValueError: + body = {} + if not body.get("message_id"): + # A 202 means Workflows may have queued the email, so this must not fall back to Customer.io. + raise WorkflowsEmailError("202 without a message_id", code="invalid_response", status_code=response.status_code) + return {"message_id": body["message_id"], "replayed": bool(body.get("replayed", False))} + + +def _workflows_idempotency_key(campaign_key: str, raw_email: str) -> str: + # Derived from the email itself, so a Celery retry of the whole send cannot queue it twice. + return hashlib.sha256(f"{campaign_key}\n{raw_email}".encode()).hexdigest() + + +def _pinned_workflows_template_id(idempotency_key: str) -> str | None: + """The template of an earlier attempt that Workflows may have queued, so a retry sends it the same way.""" + try: + return cache.get(f"email_workflows_pin:{idempotency_key}") + except Exception as err: + logger.warning("email_workflows_pin_read_failed", error=str(err)) + return None + + +def _pin_to_workflows(idempotency_key: str, template_id: str) -> None: + try: + cache.set(f"email_workflows_pin:{idempotency_key}", template_id, timeout=WORKFLOWS_PIN_TTL_SECONDS) + except Exception as err: + logger.warning("email_workflows_pin_write_failed", error=str(err)) + + +def _send_one_via_workflows( + dest: dict[str, str], campaign_key: str, template_id: str, properties: dict +) -> dict[str, Any]: + # Workflows templates read only `variables.*`, so the Customer.io `customer.*` attributes travel in the data. + customer = {"email": dest["raw_email"]} + if dest.get("distinct_id"): + customer["id"] = dest["distinct_id"] + if dest.get("first_name"): + customer["first_name"] = sanitize_email_string(dest["first_name"]) + + result = _queue_workflows_email( + to=dest["raw_email"], + template_id=template_id, + template_data=_unescape_for_liquid({**properties, "customer": customer}), + distinct_id=dest.get("distinct_id") or None, + idempotency_key=_workflows_idempotency_key(campaign_key, dest["raw_email"]), + ) + if not result["message_id"]: + # A disabled SDK client queues nothing and returns no ID. + raise WorkflowsEmailError("Workflows queued nothing", code="not_queued") + return result + + +def _send_one_via_customer_io(dest: dict[str, str], template_name: str, properties: dict) -> dict[str, Any]: + if not settings.CUSTOMER_IO_API_KEY: + raise Exception("Missing Customer.io API key") + + identifiers: dict[str, str] = {"email": dest["raw_email"]} + if dest.get("distinct_id"): + identifiers["id"] = dest["distinct_id"] + + payload = { + "transactional_message_id": get_customer_io_template_id(template_name), + "to": dest["raw_email"], + "identifiers": identifiers, + "message_data": properties, + } + + response = requests.post( + f"{settings.CUSTOMER_IO_API_URL}/v1/send/email", + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {settings.CUSTOMER_IO_API_KEY}", + }, + json=payload, + timeout=30, + ) + + if response.status_code != 200: + raise Exception(f"Customer.io API error: {response.status_code} - {response.text}") + + return response.json() + + +def _capture_email_event(capture: ScopedCapture, dest: dict[str, str], event: str, properties: dict[str, Any]) -> None: + try: + capture( + distinct_id=dest.get("distinct_id") or dest["raw_email"], + event=event, + properties={"recipient_email": dest["raw_email"], **properties}, + ) + except Exception as analytics_err: + logger.warning("email_send_analytics_capture_failed", error=str(analytics_err)) + + +def _send_one( + capture: ScopedCapture, dest: dict[str, str], campaign_key: str, template_name: str, properties: dict +) -> tuple[str, dict]: + """Send one email and return the provider that sent it with the provider's response.""" + idempotency_key = _workflows_idempotency_key(campaign_key, dest["raw_email"]) + # After an attempt Workflows may have queued, a retry skips the flag and never falls back. Otherwise a flag + # change or a different error on the retry would send the same email again through Customer.io. + pinned_template_id = _pinned_workflows_template_id(idempotency_key) + workflows_template_id = get_workflows_template_id(template_name, dest, pinned_template_id) + if workflows_template_id: + try: + return "workflows", _send_one_via_workflows(dest, campaign_key, workflows_template_id, properties) + except WorkflowsEmailError as err: + if pinned_template_id or not _workflows_error_allows_fallback(err): + _pin_to_workflows(idempotency_key, workflows_template_id) + raise + logger.warning("email_workflows_send_rejected", template_name=template_name, error=str(err)) + EMAIL_WORKFLOWS_FALLBACK_COUNTER.labels(template_name=template_name).inc() + _capture_email_event( + capture, + dest, + "transactional email workflows rejected", + {"template_name": template_name, "campaign_key": campaign_key, "error": str(err)[:500]}, + ) + return "customer_io", _send_one_via_customer_io(dest, template_name, properties) + + +# Note: this http sender is set up to send via templates, so all the configuration is done in Customer.io or in +# the Workflows message template - i.e. no subject, body, etc. def _send_via_http( to: list[dict[str, str]], campaign_key: str, template_name: str, properties: dict, ) -> None: - """Sends emails using Customer.io API""" - customerio_api_key = settings.CUSTOMER_IO_API_KEY - - if not customerio_api_key: + """Sends emails using the Customer.io API, or the Workflows transactional email API where the flag matches""" + if not is_http_email_service_available(): raise Exception("Missing Customer.io API key") - headers = { - "Content-Type": "application/json", - "Authorization": f"Bearer {customerio_api_key}", - } + # A dedicated client, because the global one may never flush before a Celery worker exits. + with ph_scoped_capture(event_region=get_instance_region()) as capture: + _send_via_http_batch(capture, to, campaign_key, template_name, properties) + +def _send_via_http_batch( + capture: ScopedCapture, to: list[dict[str, str]], campaign_key: str, template_name: str, properties: dict +) -> None: sent_count = 0 already_sent_count = 0 # delivered in a prior run — not a failure if the batch later aborts + dest: dict[str, str] | None = None try: for dest in to: with transaction.atomic(): @@ -237,25 +511,7 @@ def _send_via_http( already_sent_count += 1 continue - identifiers: dict[str, str] = {"email": dest["raw_email"]} - if dest.get("distinct_id"): - identifiers["id"] = dest["distinct_id"] - - payload = { - "transactional_message_id": get_customer_io_template_id(template_name), - "to": dest["raw_email"], - "identifiers": identifiers, - "message_data": properties, - } - - response = requests.post( - f"{settings.CUSTOMER_IO_API_URL}/v1/send/email", headers=headers, json=payload, timeout=30 - ) - - if response.status_code != 200: - raise Exception(f"Customer.io API error: {response.status_code} - {response.text}") - - provider_response = response.json() + provider, provider_response = _send_one(capture, dest, campaign_key, template_name, properties) # Mark delivery before any non-essential step so a delivered email is never # recorded as rejected (raise_if_delivery_rejected keys off sent_at). Analytics @@ -263,28 +519,45 @@ def _send_via_http( record.sent_at = timezone.now() record.save() - try: - posthoganalytics.capture( - distinct_id=dest.get("distinct_id") or dest["raw_email"], - event="transactional email triggered", - properties={ - "template_name": template_name, - "campaign_key": campaign_key, - "recipient_email": dest["raw_email"], - **provider_response, - }, - ) - except Exception as analytics_err: - logger.warning("email_send_analytics_capture_failed", error=str(analytics_err)) + _capture_email_event( + capture, + dest, + "transactional email triggered", + { + "template_name": template_name, + "campaign_key": campaign_key, + "provider": provider, + **provider_response, + }, + ) EMAIL_SEND_COUNTER.labels(outcome="sent", transport="http").inc() sent_count += 1 except Exception as err: capture_exception(err) # already logs the traceback via logger.exception + # The exception carries no template name, so this event is what lets a dashboard count failures per template. + retrying = isinstance(err, WorkflowsEmailError) + if dest: + _capture_email_event( + capture, + dest, + "transactional email failed", + { + "template_name": template_name, + "campaign_key": campaign_key, + "error_type": type(err).__name__, + "error": str(err)[:500], + "retrying": retrying, + }, + ) # Count every recipient that did not get through (the failing one + any not yet attempted), # so `failed` shares the per-recipient unit with `sent` instead of registering once per batch. EMAIL_SEND_COUNTER.labels(outcome="failed", transport="http").inc(len(to) - already_sent_count - sent_count) + if retrying: + # Workflows may have queued this email. The Celery retry sends the same idempotency key, so Workflows + # returns the queued email instead of sending a second one, and recipients already sent are skipped. + raise def _send_via_smtp( @@ -518,14 +791,34 @@ def _send_email_now( reply_to: Optional[str] = None, use_http: Optional[bool] = False, properties: Optional[dict] = None, + retry_in_worker: bool = False, ) -> None: if use_http and is_http_email_service_available(): - _send_via_http( - to=to, - campaign_key=campaign_key, - template_name=template_name, - properties=properties or {}, - ) + try: + _send_via_http( + to=to, + campaign_key=campaign_key, + template_name=template_name, + properties=properties or {}, + ) + except WorkflowsEmailError: + if not retry_in_worker: + raise + # The retry sends the same idempotency key, so Workflows does not send the email twice. + _send_email.apply_async( + kwargs={ + "campaign_key": campaign_key, + "to": to, + "subject": subject, + "headers": headers, + "txt_body": txt_body, + "html_body": html_body, + "template_name": template_name, + "reply_to": reply_to, + "use_http": use_http, + "properties": properties, + } + ) elif is_smtp_email_service_available(): _send_via_smtp( to=to, @@ -584,6 +877,8 @@ def add_recipient(self, email: str, name: Optional[str] = None, distinct_id: Opt } if distinct_id: recipient_data["distinct_id"] = distinct_id + if name: + recipient_data["first_name"] = name self.to.append(recipient_data) def add_user_recipient(self, user: "User", email_override: Optional[str] = None) -> None: @@ -615,7 +910,8 @@ def send(self, send_async: bool = True, *, retry: bool = True) -> None: if send_async: _send_email.apply_async(kwargs=kwargs) elif retry: - _send_email.apply(kwargs=kwargs) + # Celery runs eager retries at once, so each Workflows timeout would hold the request for another 10s. + _send_email.apply(kwargs={**kwargs, "retry_in_worker": True}) else: _send_email_now( campaign_key=self.campaign_key, diff --git a/posthog/test/test_email.py b/posthog/test/test_email.py index dc4d84775623..c17d1891f063 100644 --- a/posthog/test/test_email.py +++ b/posthog/test/test_email.py @@ -13,16 +13,20 @@ from django.conf import settings from django.core import mail +from django.core.cache import cache from django.core.exceptions import ImproperlyConfigured from django.test import override_settings from django.utils import timezone +import urllib3 +import requests from parameterized import parameterized from prometheus_client import REGISTRY from posthog.email import ( CUSTOMER_IO_TEMPLATE_ID_MAP, EmailMessage, + WorkflowsEmailError, _send_email, _send_via_http, _send_via_smtp, @@ -88,6 +92,7 @@ def create_person(self, team: Team, base_distinct_id: str = "") -> Person: @time_machine.travel("2020-09-21", tick=False) def setUp(self): super().setUp() + cache.clear() self.organization = Organization.objects.create() self.team = Team.objects.create(organization=self.organization, name="The Bakery") self.user = User.objects.create(email="test@posthog.com") @@ -302,9 +307,10 @@ def test_send_via_http_failed_metric_counts_undelivered_recipients(self, mock_po after = REGISTRY.get_sample_value("posthog_email_send_total", failed_labels) or 0.0 self.assertEqual(after - before, 2.0) # b failed + c never attempted; a succeeded - @patch("posthoganalytics.capture") + @patch("posthog.email.ph_scoped_capture") @patch("posthog.email.requests.post") - def test_send_via_http_success(self, mock_post, mock_capture) -> None: + def test_send_via_http_success(self, mock_post, mock_scoped_capture) -> None: + mock_capture = mock_scoped_capture.return_value.__enter__.return_value mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = {"delivery_id": "test_delivery_id", "queued_at": 1604977406} @@ -339,6 +345,7 @@ def test_send_via_http_success(self, mock_post, mock_capture) -> None: "template_name": "2fa_enabled", "campaign_key": "test_campaign", "recipient_email": "test@posthog.com", + "provider": "customer_io", "delivery_id": "test_delivery_id", "queued_at": 1604977406, }, @@ -399,6 +406,278 @@ def test_send_via_http_api_error(self, mock_post) -> None: record = MessagingRecord.objects.filter(campaign_key="test_campaign").first() self.assertIsNone(record) + @parameterized.expand( + [ + ("workflows_accepts", True, [202], ["workflows"], "workflows", ["transactional email triggered"]), + ( + "workflows_rejects_falls_back", + True, + [400, 200], + ["workflows", "customer_io"], + "customer_io", + ["transactional email workflows rejected", "transactional email triggered"], + ), + ("workflows_in_flight_sends_nothing", True, [409], ["workflows"], None, ["transactional email failed"]), + ("workflows_server_error_sends_nothing", True, [503], ["workflows"], None, ["transactional email failed"]), + ( + "workflows_accepts_without_message_id_sends_nothing", + True, + [(202, {})], + ["workflows"], + None, + ["transactional email failed"], + ), + ( + "workflows_unreachable_falls_back", + True, + [ + requests.ConnectionError( + urllib3.exceptions.MaxRetryError( + urllib3.HTTPConnectionPool("us.posthog.com"), + "/", + urllib3.exceptions.NewConnectionError(None, "refused"), + ) + ), + 200, + ], + ["workflows", "customer_io"], + "customer_io", + ["transactional email workflows rejected", "transactional email triggered"], + ), + ( + "workflows_reset_after_send_sends_nothing", + True, + [requests.ConnectionError(urllib3.exceptions.ProtocolError("Connection aborted."))], + ["workflows"], + None, + ["transactional email failed"], + ), + ( + "workflows_timeout_sends_nothing", + True, + [requests.ReadTimeout("no response")], + ["workflows"], + None, + ["transactional email failed"], + ), + ( + "flag_off_uses_customer_io", + False, + [200], + ["customer_io"], + "customer_io", + ["transactional email triggered"], + ), + ] + ) + @patch("posthog.email.ph_scoped_capture") + @patch("posthog.email.posthoganalytics.feature_enabled") + @patch("posthog.email.requests.post") + def test_send_via_http_routes_through_workflows_when_flag_matches( + self, + _name: str, + flag_enabled: bool, + status_codes: list[int | tuple[int, dict] | Exception], + expected_providers: list[str], + sent_by: str | None, + expected_events: list[str], + mock_post: MagicMock, + mock_feature_enabled: MagicMock, + mock_scoped_capture: MagicMock, + ) -> None: + mock_capture = mock_scoped_capture.return_value.__enter__.return_value + mock_feature_enabled.return_value = flag_enabled + responses: list[MagicMock | Exception] = [] + for status_code in status_codes: + if isinstance(status_code, Exception): + responses.append(status_code) + continue + if isinstance(status_code, tuple): + status_code, body = status_code + else: + body = {"message_id": "m1"} if status_code == 202 else {"delivery_id": "d1"} + response = MagicMock(status_code=status_code, text="error") + response.json.return_value = body + responses.append(response) + mock_post.side_effect = responses + + with ( + override_instance_config("EMAIL_HOST", "localhost"), + self.settings( + CUSTOMER_IO_API_KEY="cio-key", + POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test", + POSTHOG_HOST="https://us.posthog.com", + ), + patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"2fa_enabled": "tmpl-1"}), + ): + message = EmailMessage( + campaign_key="routing_campaign", + subject="Test subject", + template_name="2fa_enabled", + template_context={"team_name": "O'Brien & Co", "url": "https://app.posthog.com/r?a=1&b=2"}, + use_http=True, + ) + message.add_recipient("test@posthog.com", "Test", distinct_id="user-1") + if sent_by is None: + # Workflows may have queued the email, so the error reaches Celery, which retries with the same key. + with self.assertRaises(WorkflowsEmailError): + message.send(send_async=False, retry=False) + else: + message.send(send_async=False, retry=False) + + providers = [ + "workflows" if call.args[0].startswith("https://us.posthog.com") else "customer_io" + for call in mock_post.call_args_list + ] + self.assertEqual(providers, expected_providers) + mock_feature_enabled.assert_called_once_with( + "transactional-emails-via-workflows", + "user-1", + person_properties={"email": "test@posthog.com", "transactional_email_template": "2fa_enabled"}, + send_feature_flag_events=False, + ) + + workflows_call = next((c for c in mock_post.call_args_list if c.args[0].startswith("https://us.")), None) + if workflows_call is not None: + self.assertEqual( + workflows_call.args[0], "https://us.posthog.com/api/projects/@current/messaging_templates/tmpl-1/send/" + ) + self.assertEqual(workflows_call.kwargs["headers"]["Authorization"], "Bearer phs_test") + self.assertIn("Idempotency-Key", workflows_call.kwargs["headers"]) + self.assertFalse(workflows_call.kwargs["allow_redirects"]) + payload = workflows_call.kwargs["json"] + self.assertEqual(payload["identifiers"], {"distinct_id": "user-1"}) + # Workflows escapes Liquid output itself, so values must arrive unescaped or they render as &amp; + self.assertEqual(payload["message_data"]["team_name"], "O'Brien & Co") + self.assertEqual(payload["message_data"]["url"], "https://app.posthog.com/r?a=1&b=2") + self.assertEqual( + payload["message_data"]["customer"], + {"email": "test@posthog.com", "id": "user-1", "first_name": "Test"}, + ) + + self.assertEqual([c.kwargs["event"] for c in mock_capture.call_args_list], expected_events) + self.assertTrue( + all(c.kwargs["properties"]["template_name"] == "2fa_enabled" for c in mock_capture.call_args_list) + ) + record = MessagingRecord.objects.filter(campaign_key="routing_campaign").first() + if sent_by is None: + self.assertTrue(record is None or record.sent_at is None) + else: + assert record is not None + self.assertIsNotNone(record.sent_at) + self.assertEqual(mock_capture.call_args.kwargs["properties"]["provider"], sent_by) + + @parameterized.expand( + [ + ( + "flag_turned_off", + False, + "phs_test", + MagicMock(status_code=202, json=MagicMock(return_value={"message_id": "m1"})), + ["workflows", "workflows"], + ), + ( + "workflows_unreachable", + True, + "phs_test", + requests.ConnectionError( + urllib3.exceptions.MaxRetryError( + urllib3.HTTPConnectionPool("us.posthog.com"), + "/", + urllib3.exceptions.NewConnectionError(None, "refused"), + ) + ), + ["workflows", "workflows"], + ), + # Removing the key turns Workflows off, so a pinned retry must not post the email without it. + ( + "api_key_removed", + True, + "", + MagicMock(status_code=200, json=MagicMock(return_value={"delivery_id": "d1"})), + ["workflows", "customer_io"], + ), + ] + ) + @patch("posthog.email.ph_scoped_capture") + @patch("posthog.email.posthoganalytics.feature_enabled") + @patch("posthog.email.requests.post") + def test_retry_after_an_uncertain_workflows_send_stays_on_workflows_while_it_is_configured( + self, + _name: str, + flag_on_retry: bool, + api_key_on_retry: str, + retry_response: MagicMock | Exception, + expected_providers: list[str], + mock_post: MagicMock, + mock_feature_enabled: MagicMock, + _mock_capture: MagicMock, + ) -> None: + mock_feature_enabled.side_effect = [True, flag_on_retry] + mock_post.side_effect = [requests.ReadTimeout("no response"), retry_response] + + with ( + override_instance_config("EMAIL_HOST", "localhost"), + self.settings( + CUSTOMER_IO_API_KEY="cio-key", + POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test", + POSTHOG_HOST="https://us.posthog.com", + ), + patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"password_reset": "tmpl-1"}), + ): + message = EmailMessage( + campaign_key="pinned_campaign", subject="Test subject", template_name="password_reset", use_http=True + ) + message.add_recipient("test@posthog.com") + with self.assertRaises(WorkflowsEmailError): + message.send(send_async=False, retry=False) + with self.settings(POSTHOG_TRANSACTIONAL_EMAIL_API_KEY=api_key_on_retry): + try: + message.send(send_async=False, retry=False) + except WorkflowsEmailError: + pass + + # The first attempt may have queued the email, so Customer.io sending the retry would deliver it twice. + providers = [ + "workflows" if call.args[0].startswith("https://us.posthog.com") else "customer_io" + for call in mock_post.call_args_list + ] + self.assertEqual(providers, expected_providers) + + @patch.object(_send_email, "apply_async") + @patch("posthog.email.ph_scoped_capture") + @patch("posthog.email.posthoganalytics.feature_enabled", return_value=True) + @patch("posthog.email.requests.post", side_effect=requests.ReadTimeout("no response")) + def test_inline_send_hands_an_uncertain_workflows_send_to_a_worker( + self, + mock_post: MagicMock, + _mock_feature_enabled: MagicMock, + _mock_capture: MagicMock, + mock_apply_async: MagicMock, + ) -> None: + with ( + override_instance_config("EMAIL_HOST", "localhost"), + self.settings( + CUSTOMER_IO_API_KEY="cio-key", + POSTHOG_TRANSACTIONAL_EMAIL_API_KEY="phs_test", + POSTHOG_HOST="https://us.posthog.com", + ), + patch.dict("posthog.email.WORKFLOWS_TEMPLATE_ID_MAP", {"password_reset": "tmpl-1"}), + ): + message = EmailMessage( + campaign_key="inline_campaign", subject="Test subject", template_name="password_reset", use_http=True + ) + message.add_recipient("test@posthog.com") + message.send(send_async=False) + + # Eager retries would post once per retry inside the request, each up to the 10s timeout. + self.assertEqual(mock_post.call_count, 1) + mock_apply_async.assert_called_once() + retry_kwargs = mock_apply_async.call_args.kwargs["kwargs"] + self.assertEqual(retry_kwargs["campaign_key"], "inline_campaign") + # A worker still on the previous release rejects an argument it does not know. + self.assertNotIn("retry_in_worker", retry_kwargs) + def test_sanitize_email_properties(self) -> None: # Test with various types of input including potential HTML injection properties = {