From 19b66db9daaa3eaa7ded814c55f219c243ee75c6 Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:24:45 +0000 Subject: [PATCH 1/5] feat(messaging): read a transactional email's status by message id GET /api/projects/:project_id/messaging/emails/:message_id/ returns { id, status, created_at, status_at, reason } for a transactional send. - queued: the send API records the run before it answers 202, so a real send has a status from the start. - sent: the email worker's record of every email the provider accepts (message_assets). - skipped: the run finished without a send record, after a short grace for the record to arrive. - failed: the run failed, or a $workflows_email_failed event exists. - delivered, bounced, complained: from $workflows_email_* events keyed by $workflow_invocation_id, which need engagement capture and the change in #123902. Reads go through the cdp facade (fetch_invocation_lifecycle) and the workflows facade (fetch_sent_email_asset), scoped to the project's transactional workflows so each ClickHouse read stays on its sort key. Reads have their own rate-limit buckets and need hog_flow:read. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: e72a2c10-f145-4f7c-8754-a4fcc06a18be --- .github/new-events-schema-targets.txt | 1 + products/cdp/backend/facade/api.py | 8 + .../backend/services/invocation_lifecycle.py | 78 +++++++ .../messaging/backend/facade/contracts.py | 22 ++ .../backend/facade/transactional_email.py | 12 +- .../presentation/views/message_templates.py | 20 +- .../views/transactional_email_status.py | 107 +++++++++ products/messaging/backend/routes.py | 2 + .../services/transactional_email_status.py | 138 ++++++++++++ .../api/test_transactional_email_status.py | 205 ++++++++++++++++++ .../frontend/generated/api.schemas.ts | 45 ++++ products/messaging/frontend/generated/api.ts | 20 ++ .../backend/facade/message_assets.py | 2 + products/workflows/backend/facade/testing.py | 35 +++ .../backend/services/message_assets.py | 38 ++++ .../backend/tests/api/test_message_assets.py | 42 ++++ services/mcp/src/api/generated.ts | 46 ++++ tach.toml | 7 +- 18 files changed, 816 insertions(+), 12 deletions(-) create mode 100644 products/cdp/backend/services/invocation_lifecycle.py create mode 100644 products/messaging/backend/presentation/views/transactional_email_status.py create mode 100644 products/messaging/backend/services/transactional_email_status.py create mode 100644 products/messaging/backend/tests/api/test_transactional_email_status.py diff --git a/.github/new-events-schema-targets.txt b/.github/new-events-schema-targets.txt index cdbdda95b3a7..1b878ed6cee7 100644 --- a/.github/new-events-schema-targets.txt +++ b/.github/new-events-schema-targets.txt @@ -43,6 +43,7 @@ products/mcp_analytics/backend/tests/test_model_breakdown.py products/mcp_analytics/backend/tests/test_protocol_version_breakdown.py products/mcp_analytics/backend/tests/test_tool_quality_tables.py products/mcp_analytics/backend/tests/test_tool_tables.py +products/messaging/backend/tests/api/test_transactional_email_status.py products/product_analytics/backend/hogql_queries products/product_analytics/backend/tests/test_insight_data_checks.py products/replay_vision/backend/tests/test_session_identity.py diff --git a/products/cdp/backend/facade/api.py b/products/cdp/backend/facade/api.py index c8f0e1f1971e..8ee666f88796 100644 --- a/products/cdp/backend/facade/api.py +++ b/products/cdp/backend/facade/api.py @@ -21,6 +21,11 @@ from products.cdp.backend.flag_gated_templates import FLAG_GATED_TEMPLATE_IDS, gated_template_enabled from products.cdp.backend.models.hog_function_template import HogFunctionTemplate from products.cdp.backend.models.hog_functions import HogFunction +from products.cdp.backend.services.invocation_lifecycle import ( + InvocationLifecycle, + InvocationLifecycleStatus, + fetch_invocation_lifecycle, +) _B = "products.cdp.backend." @@ -28,7 +33,10 @@ __all__ = [ *sorted(_LAZY), + "InvocationLifecycle", + "InvocationLifecycleStatus", "create_hog_functions", + "fetch_invocation_lifecycle", "has_enabled_hog_function_for_warehouse_table", "is_hog_function_template_available", "warehouse_tables_with_enabled_hog_functions", diff --git a/products/cdp/backend/services/invocation_lifecycle.py b/products/cdp/backend/services/invocation_lifecycle.py new file mode 100644 index 000000000000..06a7eee0a772 --- /dev/null +++ b/products/cdp/backend/services/invocation_lifecycle.py @@ -0,0 +1,78 @@ +"""The latest lifecycle state of one hog flow or hog function run, read from hog_invocation_results.""" + +from collections.abc import Sequence +from datetime import UTC, datetime +from typing import Literal, Optional, cast + +from posthog.clickhouse.client.connection import Workload +from posthog.clickhouse.client.execute import sync_execute +from posthog.dataclasses import frozen + +InvocationLifecycleStatus = Literal["running", "succeeded", "failed", "canceled"] + + +@frozen +class InvocationLifecycle: + invocation_id: str + status: InvocationLifecycleStatus + scheduled_at: datetime + # None while the run waits on the queue. The producer writes a running row with no start when it queues a run. + started_at: Optional[datetime] + finished_at: Optional[datetime] + error_kind: str + + +def fetch_invocation_lifecycle( + team_id: int, + function_kind: str, + function_ids: Sequence[str], + invocation_id: str, + scheduled_after: datetime, +) -> Optional[InvocationLifecycle]: + """The run's latest state, or None. Only searches the given functions, so the read stays on the sort key.""" + if not function_ids: + return None + kwargs = { + "team_id": team_id, + "function_kind": function_kind, + "function_ids": list(function_ids), + "invocation_id": invocation_id, + "scheduled_after": scheduled_after.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + } + # Lifecycle rows collapse to the latest version. The aliases differ from the column names, because ClickHouse + # would resolve a same-named alias in WHERE to the aggregate. + query = """ + SELECT invocation_id, latest_status, latest_scheduled_at, latest_started_at, latest_finished_at, latest_error_kind + FROM ( + SELECT + invocation_id, + argMax(status, version) AS latest_status, + min(scheduled_at) AS latest_scheduled_at, + argMax(started_at, version) AS latest_started_at, + argMax(finished_at, version) AS latest_finished_at, + argMax(error_kind, version) AS latest_error_kind, + argMax(is_deleted, version) AS latest_is_deleted + FROM hog_invocation_results + WHERE team_id = %(team_id)s + AND function_kind = %(function_kind)s + AND function_id IN %(function_ids)s + AND invocation_id = %(invocation_id)s + AND scheduled_at >= toDateTime64(%(scheduled_after)s, 6) + GROUP BY invocation_id + ) + WHERE latest_is_deleted = 0 + LIMIT 1 + """ + # Called for API-key reads that a server polls. No person waits on screen. + results = cast(list, sync_execute(query, kwargs, workload=Workload.OFFLINE, team_id=team_id)) + if not results: + return None + row = results[0] + return InvocationLifecycle( + invocation_id=row[0], + status=row[1], + scheduled_at=row[2], + started_at=row[3], + finished_at=row[4], + error_kind=row[5], + ) diff --git a/products/messaging/backend/facade/contracts.py b/products/messaging/backend/facade/contracts.py index 666cdc1dea14..999bff31092e 100644 --- a/products/messaging/backend/facade/contracts.py +++ b/products/messaging/backend/facade/contracts.py @@ -1,8 +1,10 @@ from dataclasses import dataclass +from datetime import datetime from typing import Any from uuid import UUID from posthog.dataclasses import frozen +from posthog.enums import LabeledStrEnum class UnlayerError(Exception): @@ -105,3 +107,23 @@ class QueuedEmail: message_id: str # True when an earlier request with the same idempotency key already queued this email. replayed: bool = False + + +class EmailDeliveryStatus(LabeledStrEnum): + QUEUED = "queued", "Queued" + SKIPPED = "skipped", "Skipped" + SENT = "sent", "Sent" + DELIVERED = "delivered", "Delivered" + BOUNCED = "bounced", "Bounced" + COMPLAINED = "complained", "Complained" + FAILED = "failed", "Failed" + + +@frozen +class EmailStatus: + message_id: str + status: EmailDeliveryStatus + created_at: datetime + status_at: datetime + # The bounce type for a bounced email, such as hard or soft, or the error kind of a failed run. + reason: str | None = None diff --git a/products/messaging/backend/facade/transactional_email.py b/products/messaging/backend/facade/transactional_email.py index 105a55325fae..4e5fe7ec755d 100644 --- a/products/messaging/backend/facade/transactional_email.py +++ b/products/messaging/backend/facade/transactional_email.py @@ -1,6 +1,12 @@ """Transactional email: a template made sendable over the API, backed by a workflow that runs once per send.""" -from products.messaging.backend.facade.contracts import DeliveryUnavailable, QueuedEmail, TransactionalEmailError +from products.messaging.backend.facade.contracts import ( + DeliveryUnavailable, + EmailDeliveryStatus, + EmailStatus, + QueuedEmail, + TransactionalEmailError, +) from products.messaging.backend.services.transactional_email import ( SENDER_ADDRESS_MAX_LENGTH, check_transactional_update, @@ -10,13 +16,17 @@ sender_address_error, sync_transactional_workflow, ) +from products.messaging.backend.services.transactional_email_status import get_email_status __all__ = [ "SENDER_ADDRESS_MAX_LENGTH", "DeliveryUnavailable", + "EmailDeliveryStatus", + "EmailStatus", "QueuedEmail", "TransactionalEmailError", "check_transactional_update", + "get_email_status", "lock_transactional_workflow_id", "make_transactional", "send_transactional_email", diff --git a/products/messaging/backend/presentation/views/message_templates.py b/products/messaging/backend/presentation/views/message_templates.py index cfc7be24241f..c7000d0069e1 100644 --- a/products/messaging/backend/presentation/views/message_templates.py +++ b/products/messaging/backend/presentation/views/message_templates.py @@ -490,6 +490,17 @@ class TransactionalEmailTeamSustainedThrottle(ProjectSecretApiKeyTeamRateThrottl rate = "20000/hour" +def require_transactional_email_api(team: Any) -> None: + if not posthoganalytics.feature_enabled( + TRANSACTIONAL_EMAIL_API_FLAG, + str(team.uuid), + groups={"organization": str(team.organization_id)}, + group_properties={"organization": {"id": str(team.organization_id)}}, + send_feature_flag_events=False, + ): + raise NotFound() + + def transactional_email_throttles() -> list[BaseThrottle]: # Both send routes share these scopes, so a caller can't double its rate by using both. return [ @@ -528,14 +539,7 @@ class TransactionalEmailSendMixin: user_access_control: Any def _require_transactional_email_flag(self) -> None: - if not posthoganalytics.feature_enabled( - TRANSACTIONAL_EMAIL_API_FLAG, - str(self.team.uuid), - groups={"organization": str(self.team.organization_id)}, - group_properties={"organization": {"id": str(self.team.organization_id)}}, - send_feature_flag_events=False, - ): - raise NotFound() + require_transactional_email_api(self.team) def _queue_transactional_email(self, request: Any, template: MessageTemplateRow, *, route: str) -> Response: if template.transactional_hog_flow_id: diff --git a/products/messaging/backend/presentation/views/transactional_email_status.py b/products/messaging/backend/presentation/views/transactional_email_status.py new file mode 100644 index 000000000000..ec8c618fefbd --- /dev/null +++ b/products/messaging/backend/presentation/views/transactional_email_status.py @@ -0,0 +1,107 @@ +from typing import Any +from uuid import UUID + +from drf_spectacular.types import OpenApiTypes +from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema +from rest_framework import serializers, viewsets +from rest_framework.exceptions import NotFound +from rest_framework.permissions import IsAuthenticated +from rest_framework.request import Request +from rest_framework.response import Response +from rest_framework.throttling import BaseThrottle + +from posthog.api.routing import TeamAndOrgViewSetMixin +from posthog.auth import ProjectSecretAPIKeyAuthentication +from posthog.rate_limit import PersonalOrProjectSecretApiKeyRateThrottle, ProjectSecretApiKeyTeamRateThrottle + +from products.messaging.backend.facade.transactional_email import EmailDeliveryStatus, get_email_status +from products.messaging.backend.presentation.views.message_templates import require_transactional_email_api + +MESSAGE_ID_PARAMETER = OpenApiParameter( + name="id", + type=OpenApiTypes.UUID, + location=OpenApiParameter.PATH, + description="The `message_id` that the send returned.", +) + + +class EmailStatusSerializer(serializers.Serializer): + id = serializers.CharField(source="message_id", help_text="The `message_id` that the send returned.") + status = serializers.ChoiceField( + choices=EmailDeliveryStatus.choices, + help_text="What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. " + "`skipped` means it was not sent, for example because the recipient was suppressed after the send. " + "`failed` means PostHog could not send it. `delivered` means the recipient's mail server accepted it. " + "`bounced` and `complained` come later from the recipient's mail provider.", + ) + created_at = serializers.DateTimeField(help_text="When the send was queued.") + status_at = serializers.DateTimeField(help_text="When the email reached its current status.") + reason = serializers.CharField( + allow_null=True, + help_text="The bounce type for a bounced email, such as `hard` or `soft`, or the error kind of a failed " + "email. Null for any other status.", + ) + + +# Reads have their own buckets, so polling for status can't use up the send budget. +class TransactionalEmailReadBurstThrottle(PersonalOrProjectSecretApiKeyRateThrottle): + scope = "transactional_email_read_burst" + rate = "600/minute" + + +class TransactionalEmailReadSustainedThrottle(PersonalOrProjectSecretApiKeyRateThrottle): + scope = "transactional_email_read_sustained" + rate = "20000/hour" + + +class TransactionalEmailReadTeamBurstThrottle(ProjectSecretApiKeyTeamRateThrottle): + scope = "transactional_email_read_psak_team_burst" + rate = "600/minute" + + +class TransactionalEmailReadTeamSustainedThrottle(ProjectSecretApiKeyTeamRateThrottle): + scope = "transactional_email_read_psak_team_sustained" + rate = "20000/hour" + + +class MessagingEmailsViewSet(TeamAndOrgViewSetMixin, viewsets.ViewSet): + scope_object = "hog_flow" + scope_object_read_actions = ["retrieve"] + permission_classes = [IsAuthenticated] + authentication_classes = [ProjectSecretAPIKeyAuthentication] + psak_allowed_actions = ["retrieve"] + + def get_throttles(self) -> list[BaseThrottle]: + return [ + TransactionalEmailReadBurstThrottle(), + TransactionalEmailReadSustainedThrottle(), + TransactionalEmailReadTeamBurstThrottle(), + TransactionalEmailReadTeamSustainedThrottle(), + ] + + @extend_schema( + parameters=[MESSAGE_ID_PARAMETER], + responses={ + 200: EmailStatusSerializer, + 404: OpenApiResponse( + description="No email of this project has this ID. A send can also answer 404 for a second or two " + "right after it was queued." + ), + }, + summary="Get a transactional email's status", + description=( + "Get the status of one transactional email by the `message_id` that the send returned. Every project sees " + "`queued`, `sent`, `skipped` and `failed`. `delivered`, `bounced` and `complained` come from the email's " + "events, so they need engagement capture turned on for the project." + ), + ) + def retrieve(self, request: Request, *args: Any, pk: str | None = None, **kwargs: Any) -> Response: + require_transactional_email_api(self.team) + try: + message_id = UUID(str(pk)) + except ValueError: + raise NotFound() + email_status = get_email_status(self.team, message_id) + if email_status is None: + raise NotFound() + return Response(EmailStatusSerializer(email_status).data) diff --git a/products/messaging/backend/routes.py b/products/messaging/backend/routes.py index 05ddff7799e0..f1f2b78d4b7b 100644 --- a/products/messaging/backend/routes.py +++ b/products/messaging/backend/routes.py @@ -7,6 +7,7 @@ MessageTemplatesViewSet, MessagingSendViewSet, ) +from products.messaging.backend.presentation.views.transactional_email_status import MessagingEmailsViewSet def register_routes(routers: RouterRegistry) -> None: @@ -24,3 +25,4 @@ def register_routes(routers: RouterRegistry) -> None: r"messaging_suppressions", MessageSuppressionViewSet, "project_messaging_suppressions", ["team_id"] ) routers.projects.register(r"messaging/send", MessagingSendViewSet, "project_messaging_send", ["team_id"]) + routers.projects.register(r"messaging/emails", MessagingEmailsViewSet, "project_messaging_emails", ["team_id"]) diff --git a/products/messaging/backend/services/transactional_email_status.py b/products/messaging/backend/services/transactional_email_status.py new file mode 100644 index 000000000000..d1fa3e7a7cbd --- /dev/null +++ b/products/messaging/backend/services/transactional_email_status.py @@ -0,0 +1,138 @@ +"""The status of one transactional email. + +The send API records the run before it answers, and the email worker records every email the provider accepts, so +`queued`, `sent`, `skipped` and a failed run need nothing else. Delivery, bounces and complaints arrive as email +events, which a project only records when it turns on engagement capture. +""" + +from datetime import UTC, datetime, timedelta +from uuid import UUID + +from posthog.hogql import ast +from posthog.hogql.parser import parse_select +from posthog.hogql.query import execute_hogql_query + +from posthog.clickhouse.client.connection import Workload +from posthog.clickhouse.query_tagging import Feature, Product, tag_queries +from posthog.models.team import Team + +from products.cdp.backend.facade.api import fetch_invocation_lifecycle +from products.messaging.backend.facade.contracts import EmailDeliveryStatus, EmailStatus +from products.messaging.backend.models.message_template import MessageTemplate +from products.workflows.backend.facade.message_assets import fetch_sent_email_asset + +_STATUS_BY_EVENT: dict[str, EmailDeliveryStatus] = { + "$workflows_email_sent": EmailDeliveryStatus.SENT, + "$workflows_email_failed": EmailDeliveryStatus.FAILED, + "$workflows_email_delivered": EmailDeliveryStatus.DELIVERED, + # An open or a click proves delivery, also when the delivery event arrives later or not at all. + "$workflows_email_opened": EmailDeliveryStatus.DELIVERED, + "$workflows_email_link_clicked": EmailDeliveryStatus.DELIVERED, + "$workflows_email_bounced": EmailDeliveryStatus.BOUNCED, + # The mailbox provider reported a spam complaint. + "$workflows_email_blocked": EmailDeliveryStatus.COMPLAINED, +} + +# Highest first. A later open does not undo a bounce or a complaint. +_PRECEDENCE = [ + EmailDeliveryStatus.COMPLAINED, + EmailDeliveryStatus.BOUNCED, + EmailDeliveryStatus.FAILED, + EmailDeliveryStatus.DELIVERED, + EmailDeliveryStatus.SENT, +] + +# The provider reports an email's delivery, bounce and complaint within days of the send. +_EVENT_WINDOW = timedelta(days=30) +# The Node API stamps the send's ID with its own clock, which can run ahead of the event timestamps. +_CLOCK_SKEW = timedelta(minutes=5) +# How long after a run finishes its send record can still be on its way to ClickHouse. +_SEND_RECORD_GRACE = timedelta(minutes=1) + + +def queued_at(message_id: UUID) -> datetime | None: + """When the send was queued, or None when the ID is not one the send API returns.""" + # The first 6 bytes of a UUIDT are the Unix time in milliseconds. + queued_ms = int.from_bytes(message_id.bytes[:6], "big") + now = datetime.now(UTC) + if queued_ms > (now + _CLOCK_SKEW).timestamp() * 1000: + return None + return datetime.fromtimestamp(queued_ms / 1000, tz=UTC) + + +def get_email_status(team: Team, message_id: UUID) -> EmailStatus | None: + """The send's status, or None when no send of this project has this ID.""" + created_at = queued_at(message_id) + if created_at is None: + return None + # Every send this API makes runs one of the project's transactional workflows, including ones whose template + # was deleted since. + workflow_ids = [ + str(flow_id) + for flow_id in MessageTemplate.objects.filter(team_id=team.id) + .exclude(transactional_hog_flow_id=None) + .values_list("transactional_hog_flow_id", flat=True) + ] + # The send API records the run before it answers 202, so a send that has no run is not this project's. + run = fetch_invocation_lifecycle(team.id, "hog_flow", workflow_ids, str(message_id), created_at - _CLOCK_SKEW) + if run is None: + return None + + def status_of(status: EmailDeliveryStatus, status_at: datetime, reason: str | None = None) -> EmailStatus: + return EmailStatus( + message_id=str(message_id), status=status, created_at=created_at, status_at=status_at, reason=reason + ) + + if run.status == "running": + return status_of(EmailDeliveryStatus.QUEUED, run.started_at or run.scheduled_at) + + sent = fetch_sent_email_asset(team.id, workflow_ids, str(message_id), created_at - _CLOCK_SKEW) + latest, bounce_type = _engagement(team, message_id, created_at) + if sent is not None and EmailDeliveryStatus.SENT not in latest: + latest[EmailDeliveryStatus.SENT] = sent.sent_at + for status in _PRECEDENCE: + if status in latest: + return status_of(status, latest[status], bounce_type if status == EmailDeliveryStatus.BOUNCED else None) + + finished_at = run.finished_at or run.scheduled_at + if run.status in ("failed", "canceled"): + return status_of(EmailDeliveryStatus.FAILED, finished_at, run.error_kind or None) + # The worker writes the send record next to the finished run, so give it a moment to arrive. + if datetime.now(UTC) - finished_at < _SEND_RECORD_GRACE: + return status_of(EmailDeliveryStatus.QUEUED, finished_at) + # The run finished without sending, for example because the recipient was suppressed after the send queued. + return status_of(EmailDeliveryStatus.SKIPPED, finished_at) + + +def _engagement( + team: Team, message_id: UUID, created_at: datetime +) -> tuple[dict[EmailDeliveryStatus, datetime], str | None]: + query = parse_select( + """ + SELECT event, timestamp, properties.$bounce_type + FROM events + WHERE event IN {events} + AND timestamp >= {start} + AND timestamp < {end} + AND properties.$workflow_invocation_id = {message_id} + ORDER BY timestamp + LIMIT 100 + """, + placeholders={ + "events": ast.Constant(value=list(_STATUS_BY_EVENT)), + "start": ast.Constant(value=created_at - _CLOCK_SKEW), + "end": ast.Constant(value=created_at + _EVENT_WINDOW), + "message_id": ast.Constant(value=str(message_id)), + }, + ) + tag_queries(product=Product.MESSAGING, feature=Feature.QUERY) + # Callers hold a project secret API key and poll from a server. No person waits on screen. + response = execute_hogql_query(query=query, team=team, workload=Workload.OFFLINE) + latest: dict[EmailDeliveryStatus, datetime] = {} + bounce_type: str | None = None + for event, timestamp, event_bounce_type in response.results: + status = _STATUS_BY_EVENT[event] + latest[status] = timestamp + if status == EmailDeliveryStatus.BOUNCED: + bounce_type = event_bounce_type or None + return latest, bounce_type diff --git a/products/messaging/backend/tests/api/test_transactional_email_status.py b/products/messaging/backend/tests/api/test_transactional_email_status.py new file mode 100644 index 000000000000..866e3977b52d --- /dev/null +++ b/products/messaging/backend/tests/api/test_transactional_email_status.py @@ -0,0 +1,205 @@ +from datetime import UTC, datetime, timedelta +from typing import Any +from uuid import uuid4 + +from posthog.test.base import APIBaseTest, ClickhouseTestMixin, _create_event, flush_persons_and_events +from unittest.mock import patch + +from parameterized import parameterized +from rest_framework import status +from rest_framework.test import APIClient + +from posthog.clickhouse.client.execute import sync_execute +from posthog.models import ProjectSecretAPIKey, Team +from posthog.models.personal_api_key import hash_key_value +from posthog.models.utils import generate_random_token_secret +from posthog.uuidt import UUIDT + +from products.cdp.backend.facade.clickhouse import INSERT_HOG_INVOCATION_RESULT_SQL +from products.messaging.backend.models.message_template import MessageTemplate +from products.workflows.backend.facade.testing import create_sent_email_asset_for_test + +SENT = "$workflows_email_sent" +DELIVERED = "$workflows_email_delivered" +OPENED = "$workflows_email_opened" +BOUNCED = "$workflows_email_bounced" +BLOCKED = "$workflows_email_blocked" +FAILED = "$workflows_email_failed" + + +@patch( + "products.messaging.backend.presentation.views.message_templates.posthoganalytics.feature_enabled", + return_value=True, +) +class TestTransactionalEmailStatus(ClickhouseTestMixin, APIBaseTest): + def setUp(self): + super().setUp() + self.queued_at = datetime.now(UTC).replace(microsecond=0) - timedelta(hours=1) + self.message_id = str(UUIDT(unix_time_ms=int(self.queued_at.timestamp() * 1000))) + self.workflow_id = self._transactional_workflow(self.team) + + def _transactional_workflow(self, team: Team) -> str: + workflow_id = str(uuid4()) + MessageTemplate.objects.create( + team=team, name="Password reset", type="email", transactional_hog_flow_id=workflow_id + ) + return workflow_id + + def _url(self, message_id: str) -> str: + return f"/api/projects/{self.team.id}/messaging/emails/{message_id}/" + + def _at(self, seconds: int) -> datetime: + return self.queued_at + timedelta(seconds=seconds) + + def _run( + self, + run_status: str, + *, + team: Team | None = None, + workflow_id: str | None = None, + finished_at: datetime | None = None, + error_kind: str = "", + ) -> None: + params: dict[str, Any] = { + "team_id": (team or self.team).id, + "function_kind": "hog_flow", + "function_id": workflow_id or self.workflow_id, + "invocation_id": self.message_id, + "parent_run_id": "", + "status": run_status, + "attempts": 1, + "is_retry": 0, + "scheduled_at": self.queued_at, + "first_scheduled_at": self.queued_at, + "started_at": None if run_status == "running" else self.queued_at, + "finished_at": None if run_status == "running" else (finished_at or self._at(1)), + "duration_ms": None, + "error_kind": error_kind, + "error_message": "", + "event_uuid": "", + "distinct_id": "user@example.com", + "person_id": "", + "invocation_globals": "{}", + "version": 1, + "is_deleted": 0, + } + sync_execute(INSERT_HOG_INVOCATION_RESULT_SQL, params) + + def _sent_record(self) -> None: + create_sent_email_asset_for_test( + team_id=self.team.id, function_id=self.workflow_id, invocation_id=self.message_id, sent_at=self._at(2) + ) + + def _email_events(self, team: Team, events: list[tuple[str, dict]]) -> None: + for offset, (event, properties) in enumerate(events, start=3): + _create_event( + team=team, + event=event, + distinct_id="user@example.com", + properties={"$workflow_invocation_id": self.message_id, **properties}, + timestamp=self._at(offset), + ) + flush_persons_and_events() + + def _expected(self, email_status: str, status_offset: int, reason: str | None) -> dict: + return { + "id": self.message_id, + "status": email_status, + "created_at": self.queued_at.isoformat().replace("+00:00", "Z"), + "status_at": self._at(status_offset).isoformat().replace("+00:00", "Z"), + "reason": reason, + } + + @parameterized.expand( + [ + ("queued", "running", "", False, [], "queued", 0, None), + ("skipped", "succeeded", "", False, [], "skipped", 1, None), + ("failed_run", "failed", "timeout", False, [], "failed", 1, "timeout"), + # Engagement capture off: the worker's record of the send is all there is. + ("sent_without_engagement_capture", "succeeded", "", True, [], "sent", 2, None), + ("sent", "succeeded", "", True, [(SENT, {})], "sent", 3, None), + ("delivered", "succeeded", "", True, [(SENT, {}), (DELIVERED, {})], "delivered", 4, None), + ("an_open_proves_delivery", "succeeded", "", True, [(SENT, {}), (OPENED, {})], "delivered", 4, None), + ("bounced", "succeeded", "", True, [(SENT, {}), (BOUNCED, {"$bounce_type": "hard"})], "bounced", 4, "hard"), + ( + "a_later_open_keeps_the_complaint", + "succeeded", + "", + True, + [(SENT, {}), (BLOCKED, {}), (OPENED, {})], + "complained", + 4, + None, + ), + ("failed_send", "succeeded", "", False, [(FAILED, {})], "failed", 3, None), + ] + ) + def test_reports_the_status_of_the_send( + self, _flag, _name, run_status, error_kind, sent, events, expected_status, status_offset, reason + ): + self._run(run_status, error_kind=error_kind) + if sent: + self._sent_record() + self._email_events(self.team, events) + + response = self.client.get(self._url(self.message_id)) + + assert response.status_code == status.HTTP_200_OK, response.json() + assert response.json() == self._expected(expected_status, status_offset, reason) + + def test_stays_queued_while_the_send_record_of_a_finished_run_is_on_its_way(self, _flag): + finished_at = datetime.now(UTC).replace(microsecond=0) + self._run("succeeded", finished_at=finished_at) + + response = self.client.get(self._url(self.message_id)) + + assert response.json()["status"] == "queued" + + @parameterized.expand( + [ + ("unknown_id", "message_id", False), + ("another_projects_send", "message_id", True), + ("not_a_uuid", "not-a-uuid", False), + ("an_id_from_the_future", "ffffffff-ffff-4fff-bfff-ffffffffffff", False), + ] + ) + def test_answers_404_for_an_email_it_cannot_report(self, _flag, _name, message_id, other_project): + if other_project: + other = Team.objects.create(organization=self.organization, name="Other") + self._run("succeeded", team=other, workflow_id=self._transactional_workflow(other)) + self._email_events(other, [(SENT, {})]) + + response = self.client.get(self._url(self.message_id if message_id == "message_id" else message_id)) + + assert response.status_code == status.HTTP_404_NOT_FOUND + + @parameterized.expand( + [ + ("read_scope", ["hog_flow:read"], status.HTTP_200_OK), + ("no_workflow_scope", ["feature_flag:read"], status.HTTP_403_FORBIDDEN), + ] + ) + def test_project_secret_api_key_needs_the_read_scope(self, _flag, _name, scopes, expected_status): + self._run("running") + raw_token = generate_random_token_secret() + ProjectSecretAPIKey.objects.create( + team=self.team, + label="email status", + secure_value=hash_key_value(raw_token), + scopes=scopes, + mask_value=f"{raw_token[:4]}...{raw_token[-4:]}", + ) + client = APIClient() + client.credentials(HTTP_AUTHORIZATION=f"Bearer {raw_token}") + + response = client.get(self._url(self.message_id)) + + assert response.status_code == expected_status + + def test_is_hidden_while_the_flag_is_off(self, flag): + flag.return_value = False + self._run("running") + + response = self.client.get(self._url(self.message_id)) + + assert response.status_code == status.HTTP_404_NOT_FOUND diff --git a/products/messaging/frontend/generated/api.schemas.ts b/products/messaging/frontend/generated/api.schemas.ts index 79210791e13f..f4462532267d 100644 --- a/products/messaging/frontend/generated/api.schemas.ts +++ b/products/messaging/frontend/generated/api.schemas.ts @@ -7,6 +7,51 @@ * PostHog API - generated * OpenAPI spec version: 1.0.0 */ +/** + * * `queued` - Queued + * * `skipped` - Skipped + * * `sent` - Sent + * * `delivered` - Delivered + * * `bounced` - Bounced + * * `complained` - Complained + * * `failed` - Failed + */ +export type EmailDeliveryStatusEnumApi = (typeof EmailDeliveryStatusEnumApi)[keyof typeof EmailDeliveryStatusEnumApi] + +export const EmailDeliveryStatusEnumApi = { + Queued: 'queued', + Skipped: 'skipped', + Sent: 'sent', + Delivered: 'delivered', + Bounced: 'bounced', + Complained: 'complained', + Failed: 'failed', +} as const + +export interface EmailStatusApi { + /** The `message_id` that the send returned. */ + id: string + /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means it was not sent, for example because the recipient was suppressed after the send. `failed` means PostHog could not send it. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. + * + * * `queued` - Queued + * * `skipped` - Skipped + * * `sent` - Sent + * * `delivered` - Delivered + * * `bounced` - Bounced + * * `complained` - Complained + * * `failed` - Failed */ + status: EmailDeliveryStatusEnumApi + /** When the send was queued. */ + created_at: string + /** When the email reached its current status. */ + status_at: string + /** + * The bounce type for a bounced email, such as `hard` or `soft`, or the error kind of a failed email. Null for any other status. + * @nullable + */ + reason: string | null +} + /** * Data for the template's Liquid placeholders, available as `variables`. For example `{"reset_link": "https://..."}` fills `{{ variables.reset_link }}`. */ diff --git a/products/messaging/frontend/generated/api.ts b/products/messaging/frontend/generated/api.ts index e867bbf69f03..6e331959a172 100644 --- a/products/messaging/frontend/generated/api.ts +++ b/products/messaging/frontend/generated/api.ts @@ -13,6 +13,7 @@ import type { AddSuppressionRequestApi, BulkAddOptOutsRequestApi, BulkAddOptOutsResultApi, + EmailStatusApi, GenerateLinkRequestApi, MakeTransactionalApi, MessageCategoryApi, @@ -56,6 +57,25 @@ type NonReadonly = [T] extends [UnionToIntersection] } : DistributeReadOnlyOverUnions +export const getMessagingEmailsRetrieveUrl = (projectId: string, id: string) => { + return `/api/projects/${projectId}/messaging/emails/${id}/` +} + +/** + * Get the status of one transactional email by the `message_id` that the send returned. Every project sees `queued`, `sent`, `skipped` and `failed`. `delivered`, `bounced` and `complained` come from the email's events, so they need engagement capture turned on for the project. + * @summary Get a transactional email's status + */ +export const messagingEmailsRetrieve = async ( + projectId: string, + id: string, + options?: RequestInit +): Promise => { + return apiMutator(getMessagingEmailsRetrieveUrl(projectId, id), { + ...options, + method: 'GET', + }) +} + export const getMessagingSendEmailCreateUrl = (projectId: string) => { return `/api/projects/${projectId}/messaging/send/email/` } diff --git a/products/workflows/backend/facade/message_assets.py b/products/workflows/backend/facade/message_assets.py index 590b3f4f061a..a62065a3557e 100644 --- a/products/workflows/backend/facade/message_assets.py +++ b/products/workflows/backend/facade/message_assets.py @@ -4,10 +4,12 @@ fetch_message_asset_html, fetch_message_assets, fetch_message_assets_for_person, + fetch_sent_email_asset, ) __all__ = [ "fetch_message_asset_html", "fetch_message_assets", "fetch_message_assets_for_person", + "fetch_sent_email_asset", ] diff --git a/products/workflows/backend/facade/testing.py b/products/workflows/backend/facade/testing.py index d982da6eb47f..b0b1190eaf17 100644 --- a/products/workflows/backend/facade/testing.py +++ b/products/workflows/backend/facade/testing.py @@ -1,14 +1,49 @@ +from datetime import datetime from typing import Any from asgiref.sync import sync_to_async +from posthog.clickhouse.client.execute import sync_execute from posthog.models.team import Team from products.workflows.backend.facade.contracts import WorkflowSummary from products.workflows.backend.models import HogFlow +from products.workflows.backend.models.message_assets_sql import INSERT_MESSAGE_ASSET_SQL from products.workflows.backend.services.account_audience import get_account_audience_count, get_account_audience_page +def create_sent_email_asset_for_test( + *, + team_id: int, + function_id: str, + invocation_id: str, + sent_at: datetime, + is_deleted: int = 0, +) -> None: + """Seed the record the email worker writes for an email it sent.""" + sync_execute( + INSERT_MESSAGE_ASSET_SQL, + { + "team_id": team_id, + "function_kind": "hog_flow", + "function_id": function_id, + "parent_run_id": "", + "invocation_id": invocation_id, + "action_id": "email-step", + "kind": "email", + "distinct_id": "user@example.com", + "person_id": "", + "recipient": "user@example.com", + "subject": "Reset your password", + "html": "

Reset

", + "status": "sent", + "sent_at": sent_at, + "version": 1, + "is_deleted": is_deleted, + }, + ) + + def create_workflow_for_test( *, team_id: int, diff --git a/products/workflows/backend/services/message_assets.py b/products/workflows/backend/services/message_assets.py index 944a5166b259..eb7a1914e970 100644 --- a/products/workflows/backend/services/message_assets.py +++ b/products/workflows/backend/services/message_assets.py @@ -2,6 +2,7 @@ from datetime import UTC, datetime from typing import Any, Optional, cast +from posthog.clickhouse.client.connection import Workload from posthog.clickhouse.client.execute import sync_execute from products.workflows.backend.facade.contracts import MessageAsset @@ -129,6 +130,43 @@ def fetch_message_assets( return [_build_asset(row) for row in results] +def fetch_sent_email_asset( + team_id: int, + function_ids: Sequence[str], + invocation_id: str, + sent_after: datetime, +) -> Optional[MessageAsset]: + """The email one run sent, or None. Only searches the given workflows, so the read stays on the sort key.""" + if not function_ids: + return None + kwargs: dict[str, Any] = { + "team_id": team_id, + "function_ids": list(function_ids), + "invocation_id": invocation_id, + "sent_after": sent_after.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%S"), + } + query = f""" + SELECT {_OUTER_COLUMNS} + FROM ( + SELECT {_COLLAPSED_AGGREGATES} + FROM message_assets + WHERE team_id = %(team_id)s + AND function_kind = 'hog_flow' + AND function_id IN %(function_ids)s + AND invocation_id = %(invocation_id)s + AND kind = 'email' + AND sent_at >= toDateTime64(%(sent_after)s, 6) + GROUP BY invocation_id, action_id + ) + WHERE latest_is_deleted = 0 + ORDER BY latest_sent_at + LIMIT 1 + """ + # A server holding a project secret API key polls this. No person waits on screen. + results = cast(list, sync_execute(query, kwargs, workload=Workload.OFFLINE, team_id=team_id)) + return _build_asset(results[0]) if results else None + + def fetch_message_assets_for_person( team_id: int, person_id: str, diff --git a/products/workflows/backend/tests/api/test_message_assets.py b/products/workflows/backend/tests/api/test_message_assets.py index 57aff93f1c60..a82be1df87ea 100644 --- a/products/workflows/backend/tests/api/test_message_assets.py +++ b/products/workflows/backend/tests/api/test_message_assets.py @@ -15,6 +15,7 @@ from products.access_control.backend.models.property_access_control import PropertyAccessControl from products.access_control.backend.property_access_control import PropertyAccessLevel +from products.workflows.backend.facade.message_assets import fetch_sent_email_asset from products.workflows.backend.models.hog_flow.hog_flow import HogFlow from products.workflows.backend.models.message_assets_sql import INSERT_MESSAGE_ASSET_SQL @@ -227,6 +228,47 @@ def test_personal_api_key_requires_person_read_scope(self, path: str): assert "person:read" in res.json().get("detail", "") +class TestFetchSentEmailAsset(ClickhouseTestMixin, APIBaseTest): + @parameterized.expand( + [ + ("found", {}, {}, True), + ("deleted", {"is_deleted": 1}, {}, False), + ("another_workflow", {"function_id": "flow-other"}, {}, False), + ("sent_before_the_window", {"sent_at": datetime(2026, 1, 1, tzinfo=UTC)}, {}, False), + ("no_transactional_workflows", {}, {"function_ids": []}, False), + ] + ) + def test_finds_the_email_one_run_sent(self, _name, asset: dict, query: dict, found: bool): + sent_after = datetime.now(tz=UTC) - timedelta(hours=1) + fields = {"function_id": "flow-1", "sent_at": sent_after + timedelta(minutes=1), **asset} + create_message_asset(self.team.id, invocation_id="run-1", **fields) + + result = fetch_sent_email_asset( + team_id=self.team.id, + function_ids=query.get("function_ids", ["flow-1"]), + invocation_id="run-1", + sent_after=sent_after, + ) + + assert (result is not None) == found + if result is not None: + assert result.invocation_id == "run-1" + + def test_is_isolated_from_other_teams(self): + other = self.organization.teams.create(name="Other") + create_message_asset(other.id, "flow-1", "run-1") + + assert ( + fetch_sent_email_asset( + team_id=self.team.id, + function_ids=["flow-1"], + invocation_id="run-1", + sent_after=datetime.now(tz=UTC) - timedelta(hours=1), + ) + is None + ) + + class TestPersonEmails(ClickhouseTestMixin, APIBaseTest): def setUp(self): super().setUp() diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index f30f2bbe3363..e386c3770222 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -41103,6 +41103,28 @@ export namespace Schemas { name: string; } + /** + * * `queued` - Queued + * * `skipped` - Skipped + * * `sent` - Sent + * * `delivered` - Delivered + * * `bounced` - Bounced + * * `complained` - Complained + * * `failed` - Failed + */ + export type EmailDeliveryStatusEnum = typeof EmailDeliveryStatusEnum[keyof typeof EmailDeliveryStatusEnum]; + + + export const EmailDeliveryStatusEnum = { + Queued: 'queued', + Skipped: 'skipped', + Sent: 'sent', + Delivered: 'delivered', + Bounced: 'bounced', + Complained: 'complained', + Failed: 'failed', + } as const; + export interface EmailDomainConnectProvider { /** Domain Connect endpoint of the provider. */ endpoint: string; @@ -41477,6 +41499,30 @@ export namespace Schemas { readonly email_sending_suspension_reason: string; } + export interface EmailStatus { + /** The `message_id` that the send returned. */ + id: string; + /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means it was not sent, for example because the recipient was suppressed after the send. `failed` means PostHog could not send it. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. + * + * * `queued` - Queued + * * `skipped` - Skipped + * * `sent` - Sent + * * `delivered` - Delivered + * * `bounced` - Bounced + * * `complained` - Complained + * * `failed` - Failed */ + status: EmailDeliveryStatusEnum; + /** When the send was queued. */ + created_at: string; + /** When the email reached its current status. */ + status_at: string; + /** + * The bounce type for a bounced email, such as `hard` or `soft`, or the error kind of a failed email. Null for any other status. + * @nullable + */ + reason: string | null; + } + /** * Highest htmlID suffix per element type, e.g. {"u_row": 1, "u_content_text": 2}. */ diff --git a/tach.toml b/tach.toml index c3b1464303ac..ac3d5960862f 100644 --- a/tach.toml +++ b/tach.toml @@ -680,10 +680,11 @@ depends_on = [ # products.notifications.backend.facade.api. "products.notifications", # Facade-only: the recipient preferences page reads EmailTrackingConsentMode through - # products.workflows.backend.facade.enums. + # products.workflows.backend.facade.enums, and the transactional email status reads the send + # record through products.workflows.backend.facade.message_assets. "products.workflows", - # Presentation-only: the template design action wraps HTML through - # products.cdp.backend.presentation.hog_functions. + # The template design action wraps HTML through products.cdp.backend.presentation.hog_functions, + # and the transactional email status reads the run through products.cdp.backend.facade.api. "products.cdp", ] layer = "modules" From cd9935353044b1810c489f699e6c339b2d404e0b Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:24:52 +0000 Subject: [PATCH 2/5] fix(messaging): check workflow access and report skips from the worker's log - A session or personal API key reader needs viewer access to the email's workflow, as the send route needs editor access. - Tag the ClickHouse queries once before any read. The run lookup ran untagged and the endpoint answered 500 on a local stack. - A confirmed send wins over a run still marked running, and the send record and events still answer when an instance has the run records turned off. - skipped now comes from the worker's "Skipping send:" log, with its reason. A finished run with neither a send record nor a skip log is unknown. - Engagement events are aggregated per event before the result is read, so a long run of opens can't hide a later bounce or complaint, and the query filters on the send's distinct_id, which is in the events sort key. - Workflows an API call triggers count as transactional too, so a send from a workflow a template was relinked away from still resolves. - The 404 window right after a send is a few seconds, not one or two. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: e72a2c10-f145-4f7c-8754-a4fcc06a18be --- .../backend/services/invocation_lifecycle.py | 20 ++- .../messaging/backend/facade/contracts.py | 5 +- .../views/transactional_email_status.py | 39 ++++- .../services/transactional_email_status.py | 152 ++++++++++++------ .../api/test_transactional_email_status.py | 75 ++++++++- .../frontend/generated/api.schemas.ts | 7 +- products/workflows/backend/facade/api.py | 8 + services/mcp/src/api/generated.ts | 7 +- 8 files changed, 239 insertions(+), 74 deletions(-) diff --git a/products/cdp/backend/services/invocation_lifecycle.py b/products/cdp/backend/services/invocation_lifecycle.py index 06a7eee0a772..8ebbc340f734 100644 --- a/products/cdp/backend/services/invocation_lifecycle.py +++ b/products/cdp/backend/services/invocation_lifecycle.py @@ -14,6 +14,8 @@ @frozen class InvocationLifecycle: invocation_id: str + function_id: str + distinct_id: str status: InvocationLifecycleStatus scheduled_at: datetime # None while the run waits on the queue. The producer writes a running row with no start when it queues a run. @@ -42,10 +44,14 @@ def fetch_invocation_lifecycle( # Lifecycle rows collapse to the latest version. The aliases differ from the column names, because ClickHouse # would resolve a same-named alias in WHERE to the aggregate. query = """ - SELECT invocation_id, latest_status, latest_scheduled_at, latest_started_at, latest_finished_at, latest_error_kind + SELECT + invocation_id, latest_function_id, latest_distinct_id, latest_status, latest_scheduled_at, + latest_started_at, latest_finished_at, latest_error_kind FROM ( SELECT invocation_id, + argMax(function_id, version) AS latest_function_id, + argMax(distinct_id, version) AS latest_distinct_id, argMax(status, version) AS latest_status, min(scheduled_at) AS latest_scheduled_at, argMax(started_at, version) AS latest_started_at, @@ -70,9 +76,11 @@ def fetch_invocation_lifecycle( row = results[0] return InvocationLifecycle( invocation_id=row[0], - status=row[1], - scheduled_at=row[2], - started_at=row[3], - finished_at=row[4], - error_kind=row[5], + function_id=row[1], + distinct_id=row[2], + status=row[3], + scheduled_at=row[4], + started_at=row[5], + finished_at=row[6], + error_kind=row[7], ) diff --git a/products/messaging/backend/facade/contracts.py b/products/messaging/backend/facade/contracts.py index 999bff31092e..83bcee1bf9c7 100644 --- a/products/messaging/backend/facade/contracts.py +++ b/products/messaging/backend/facade/contracts.py @@ -112,6 +112,7 @@ class QueuedEmail: class EmailDeliveryStatus(LabeledStrEnum): QUEUED = "queued", "Queued" SKIPPED = "skipped", "Skipped" + UNKNOWN = "unknown", "Unknown" SENT = "sent", "Sent" DELIVERED = "delivered", "Delivered" BOUNCED = "bounced", "Bounced" @@ -125,5 +126,7 @@ class EmailStatus: status: EmailDeliveryStatus created_at: datetime status_at: datetime - # The bounce type for a bounced email, such as hard or soft, or the error kind of a failed run. + # The bounce type for a bounced email, why a send was skipped, or the error kind of a failed run. reason: str | None = None + # The workflow that sent the email, so a reader's access to it can be checked. Not part of the API response. + workflow_id: str | None = None diff --git a/products/messaging/backend/presentation/views/transactional_email_status.py b/products/messaging/backend/presentation/views/transactional_email_status.py index ec8c618fefbd..cbf0d41b6735 100644 --- a/products/messaging/backend/presentation/views/transactional_email_status.py +++ b/products/messaging/backend/presentation/views/transactional_email_status.py @@ -3,7 +3,7 @@ from drf_spectacular.types import OpenApiTypes from drf_spectacular.utils import OpenApiParameter, OpenApiResponse, extend_schema -from rest_framework import serializers, viewsets +from rest_framework import exceptions, serializers, viewsets from rest_framework.exceptions import NotFound from rest_framework.permissions import IsAuthenticated from rest_framework.request import Request @@ -12,10 +12,13 @@ from posthog.api.routing import TeamAndOrgViewSetMixin from posthog.auth import ProjectSecretAPIKeyAuthentication +from posthog.permissions import is_authenticated_via_project_secret_api_key from posthog.rate_limit import PersonalOrProjectSecretApiKeyRateThrottle, ProjectSecretApiKeyTeamRateThrottle from products.messaging.backend.facade.transactional_email import EmailDeliveryStatus, get_email_status from products.messaging.backend.presentation.views.message_templates import require_transactional_email_api +from products.workflows.backend.facade.api import WorkflowAccessDenied, WorkflowNotFound +from products.workflows.backend.facade.workflows import get_workflow MESSAGE_ID_PARAMETER = OpenApiParameter( name="id", @@ -30,16 +33,17 @@ class EmailStatusSerializer(serializers.Serializer): status = serializers.ChoiceField( choices=EmailDeliveryStatus.choices, help_text="What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. " - "`skipped` means it was not sent, for example because the recipient was suppressed after the send. " - "`failed` means PostHog could not send it. `delivered` means the recipient's mail server accepted it. " - "`bounced` and `complained` come later from the recipient's mail provider.", + "`skipped` means PostHog decided not to send it, and `reason` says why. `failed` means PostHog could not " + "send it. `unknown` means the run finished but PostHog has no record of the outcome. `delivered` means the " + "recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail " + "provider.", ) created_at = serializers.DateTimeField(help_text="When the send was queued.") status_at = serializers.DateTimeField(help_text="When the email reached its current status.") reason = serializers.CharField( allow_null=True, - help_text="The bounce type for a bounced email, such as `hard` or `soft`, or the error kind of a failed " - "email. Null for any other status.", + help_text="Why the email is skipped, the bounce type of a bounced email (such as `hard`), or the error kind " + "of a failed email. Null for any other status.", ) @@ -84,7 +88,7 @@ def get_throttles(self) -> list[BaseThrottle]: responses={ 200: EmailStatusSerializer, 404: OpenApiResponse( - description="No email of this project has this ID. A send can also answer 404 for a second or two " + description="No email of this project has this ID. A send can also answer 404 for a few seconds " "right after it was queued." ), }, @@ -104,4 +108,25 @@ def retrieve(self, request: Request, *args: Any, pk: str | None = None, **kwargs email_status = get_email_status(self.team, message_id) if email_status is None: raise NotFound() + self._require_workflow_viewer(request, email_status.workflow_id) return Response(EmailStatusSerializer(email_status).data) + + def _require_workflow_viewer(self, request: Request, workflow_id: str | None) -> None: + # The send route needs editor access to the email's workflow, so a read needs viewer access. A project + # secret API key has no user to check, and its scope already limits it to transactional email. + if is_authenticated_via_project_secret_api_key(request): + return + if workflow_id is None: + raise NotFound() + try: + get_workflow( + team_id=self.team_id, + workflow_id=workflow_id, + user_access_control=self.user_access_control, + required_level="viewer", + ) + except WorkflowAccessDenied: + raise exceptions.PermissionDenied("You need viewer access to this email's workflow.") from None + except WorkflowNotFound: + # The workflow was deleted since. Its sends keep the project's own access rules. + return diff --git a/products/messaging/backend/services/transactional_email_status.py b/products/messaging/backend/services/transactional_email_status.py index d1fa3e7a7cbd..8791f41fb003 100644 --- a/products/messaging/backend/services/transactional_email_status.py +++ b/products/messaging/backend/services/transactional_email_status.py @@ -1,7 +1,7 @@ """The status of one transactional email. -The send API records the run before it answers, and the email worker records every email the provider accepts, so -`queued`, `sent`, `skipped` and a failed run need nothing else. Delivery, bounces and complaints arrive as email +The send API records the run before it answers, and the email worker records every email the provider accepts and +logs every send it skips, so `queued`, `sent`, `skipped` and a failed run need nothing else. Delivery, bounces and complaints arrive as email events, which a project only records when it turns on engagement capture. """ @@ -19,6 +19,7 @@ from products.cdp.backend.facade.api import fetch_invocation_lifecycle from products.messaging.backend.facade.contracts import EmailDeliveryStatus, EmailStatus from products.messaging.backend.models.message_template import MessageTemplate +from products.workflows.backend.facade.api import api_triggered_workflow_ids from products.workflows.backend.facade.message_assets import fetch_sent_email_asset _STATUS_BY_EVENT: dict[str, EmailDeliveryStatus] = { @@ -34,19 +35,20 @@ } # Highest first. A later open does not undo a bounce or a complaint. -_PRECEDENCE = [ +_ENGAGEMENT_PRECEDENCE = [ EmailDeliveryStatus.COMPLAINED, EmailDeliveryStatus.BOUNCED, EmailDeliveryStatus.FAILED, EmailDeliveryStatus.DELIVERED, - EmailDeliveryStatus.SENT, ] +# The email worker logs this, followed by the reason, when it decides not to send. See EmailService in nodejs. +_SKIP_LOG_PREFIX = "Skipping send:" # The provider reports an email's delivery, bounce and complaint within days of the send. _EVENT_WINDOW = timedelta(days=30) # The Node API stamps the send's ID with its own clock, which can run ahead of the event timestamps. _CLOCK_SKEW = timedelta(minutes=5) -# How long after a run finishes its send record can still be on its way to ClickHouse. +# How long after a run finishes its send record and logs can still be on their way to ClickHouse. _SEND_RECORD_GRACE = timedelta(minutes=1) @@ -65,74 +67,126 @@ def get_email_status(team: Team, message_id: UUID) -> EmailStatus | None: created_at = queued_at(message_id) if created_at is None: return None - # Every send this API makes runs one of the project's transactional workflows, including ones whose template - # was deleted since. - workflow_ids = [ - str(flow_id) - for flow_id in MessageTemplate.objects.filter(team_id=team.id) - .exclude(transactional_hog_flow_id=None) - .values_list("transactional_hog_flow_id", flat=True) - ] - # The send API records the run before it answers 202, so a send that has no run is not this project's. - run = fetch_invocation_lifecycle(team.id, "hog_flow", workflow_ids, str(message_id), created_at - _CLOCK_SKEW) - if run is None: + # Tagged once here, because every read below, through the cdp and workflows facades too, runs a ClickHouse query. + tag_queries(product=Product.MESSAGING, feature=Feature.QUERY) + workflow_ids = _transactional_workflow_ids(team.id) + if not workflow_ids: return None + since = created_at - _CLOCK_SKEW + run = fetch_invocation_lifecycle(team.id, "hog_flow", workflow_ids, str(message_id), since) + sent = fetch_sent_email_asset(team.id, workflow_ids, str(message_id), since) + workflow_id = (run.function_id if run else None) or (sent.function_id if sent else None) + distinct_id = (run.distinct_id if run else "") or (sent.distinct_id if sent else "") + events = _engagement(team, message_id, created_at, distinct_id) def status_of(status: EmailDeliveryStatus, status_at: datetime, reason: str | None = None) -> EmailStatus: return EmailStatus( - message_id=str(message_id), status=status, created_at=created_at, status_at=status_at, reason=reason + message_id=str(message_id), + status=status, + created_at=created_at, + status_at=status_at, + reason=reason, + workflow_id=workflow_id, ) + # What the provider reported wins over the run, and a confirmed send wins over a run still marked running. + for status in _ENGAGEMENT_PRECEDENCE: + if status in events: + at, bounce_type = events[status] + return status_of(status, at, bounce_type if status == EmailDeliveryStatus.BOUNCED else None) + if sent is not None: + return status_of(EmailDeliveryStatus.SENT, sent.sent_at) + if EmailDeliveryStatus.SENT in events: + return status_of(EmailDeliveryStatus.SENT, events[EmailDeliveryStatus.SENT][0]) + # The run records are monitoring data that an instance can turn off, so without one the email is unknown. + if run is None: + return None + + if run.status in ("failed", "canceled"): + return status_of(EmailDeliveryStatus.FAILED, run.finished_at or run.scheduled_at, run.error_kind or None) if run.status == "running": return status_of(EmailDeliveryStatus.QUEUED, run.started_at or run.scheduled_at) - - sent = fetch_sent_email_asset(team.id, workflow_ids, str(message_id), created_at - _CLOCK_SKEW) - latest, bounce_type = _engagement(team, message_id, created_at) - if sent is not None and EmailDeliveryStatus.SENT not in latest: - latest[EmailDeliveryStatus.SENT] = sent.sent_at - for status in _PRECEDENCE: - if status in latest: - return status_of(status, latest[status], bounce_type if status == EmailDeliveryStatus.BOUNCED else None) - finished_at = run.finished_at or run.scheduled_at - if run.status in ("failed", "canceled"): - return status_of(EmailDeliveryStatus.FAILED, finished_at, run.error_kind or None) - # The worker writes the send record next to the finished run, so give it a moment to arrive. + skip_reason = _skip_reason(team, run.function_id, message_id, since) + if skip_reason is not None: + return status_of(EmailDeliveryStatus.SKIPPED, finished_at, skip_reason) + # The worker writes the send record and its logs next to the finished run, so give them a moment to arrive. if datetime.now(UTC) - finished_at < _SEND_RECORD_GRACE: return status_of(EmailDeliveryStatus.QUEUED, finished_at) - # The run finished without sending, for example because the recipient was suppressed after the send queued. - return status_of(EmailDeliveryStatus.SKIPPED, finished_at) + return status_of(EmailDeliveryStatus.UNKNOWN, finished_at) + + +def _transactional_workflow_ids(team_id: int) -> list[str]: + # A template's current workflow, plus every workflow an API call triggers, so a send from a workflow that a + # template was later relinked away from still resolves. + linked = ( + MessageTemplate.objects.filter(team_id=team_id) + .exclude(transactional_hog_flow_id=None) + .values_list("transactional_hog_flow_id", flat=True) + ) + return sorted({str(flow_id) for flow_id in linked} | set(api_triggered_workflow_ids(team_id=team_id))) def _engagement( - team: Team, message_id: UUID, created_at: datetime -) -> tuple[dict[EmailDeliveryStatus, datetime], str | None]: + team: Team, message_id: UUID, created_at: datetime, distinct_id: str +) -> dict[EmailDeliveryStatus, tuple[datetime, str | None]]: + """The latest time of each status the email's events report, with the bounce type of the latest bounce.""" + where = [ + "event IN {events}", + "timestamp >= {start}", + "timestamp < {end}", + "properties.$workflow_invocation_id = {message_id}", + ] + if distinct_id: + # distinct_id is in the events sort key, so this keeps a poll from scanning the project's email events. + where.append("distinct_id = {distinct_id}") query = parse_select( - """ - SELECT event, timestamp, properties.$bounce_type + f""" + SELECT event, max(timestamp), argMax(properties.$bounce_type, timestamp) FROM events - WHERE event IN {events} - AND timestamp >= {start} - AND timestamp < {end} - AND properties.$workflow_invocation_id = {message_id} - ORDER BY timestamp - LIMIT 100 + WHERE {" AND ".join(where)} + GROUP BY event """, placeholders={ "events": ast.Constant(value=list(_STATUS_BY_EVENT)), "start": ast.Constant(value=created_at - _CLOCK_SKEW), "end": ast.Constant(value=created_at + _EVENT_WINDOW), "message_id": ast.Constant(value=str(message_id)), + "distinct_id": ast.Constant(value=distinct_id), }, ) - tag_queries(product=Product.MESSAGING, feature=Feature.QUERY) # Callers hold a project secret API key and poll from a server. No person waits on screen. response = execute_hogql_query(query=query, team=team, workload=Workload.OFFLINE) - latest: dict[EmailDeliveryStatus, datetime] = {} - bounce_type: str | None = None - for event, timestamp, event_bounce_type in response.results: + latest: dict[EmailDeliveryStatus, tuple[datetime, str | None]] = {} + for event, timestamp, bounce_type in response.results: status = _STATUS_BY_EVENT[event] - latest[status] = timestamp - if status == EmailDeliveryStatus.BOUNCED: - bounce_type = event_bounce_type or None - return latest, bounce_type + if status not in latest or timestamp > latest[status][0]: + latest[status] = (timestamp, bounce_type or None) + return latest + + +def _skip_reason(team: Team, workflow_id: str, message_id: UUID, since: datetime) -> str | None: + """Why the email worker skipped the send, from the run's log, or None when it did not log a skip.""" + query = parse_select( + """ + SELECT message + FROM log_entries + WHERE log_source = 'hog_flow' + AND log_source_id = {workflow_id} + AND instance_id = {message_id} + AND timestamp >= {since} + AND message LIKE {pattern} + ORDER BY timestamp DESC + LIMIT 1 + """, + placeholders={ + "workflow_id": ast.Constant(value=workflow_id), + "message_id": ast.Constant(value=str(message_id)), + "since": ast.Constant(value=since), + "pattern": ast.Constant(value=f"%{_SKIP_LOG_PREFIX}%"), + }, + ) + response = execute_hogql_query(query=query, team=team, workload=Workload.OFFLINE) + if not response.results: + return None + return response.results[0][0].split(_SKIP_LOG_PREFIX, 1)[1].strip() diff --git a/products/messaging/backend/tests/api/test_transactional_email_status.py b/products/messaging/backend/tests/api/test_transactional_email_status.py index 866e3977b52d..dfa04423fe1a 100644 --- a/products/messaging/backend/tests/api/test_transactional_email_status.py +++ b/products/messaging/backend/tests/api/test_transactional_email_status.py @@ -1,6 +1,5 @@ from datetime import UTC, datetime, timedelta from typing import Any -from uuid import uuid4 from posthog.test.base import APIBaseTest, ClickhouseTestMixin, _create_event, flush_persons_and_events from unittest.mock import patch @@ -10,14 +9,17 @@ from rest_framework.test import APIClient from posthog.clickhouse.client.execute import sync_execute -from posthog.models import ProjectSecretAPIKey, Team +from posthog.clickhouse.log_entries import INSERT_LOG_ENTRY_SQL +from posthog.constants import AvailableFeature +from posthog.models import OrganizationMembership, ProjectSecretAPIKey, Team, User from posthog.models.personal_api_key import hash_key_value from posthog.models.utils import generate_random_token_secret from posthog.uuidt import UUIDT +from products.access_control.backend.facade.testing import create_access_control from products.cdp.backend.facade.clickhouse import INSERT_HOG_INVOCATION_RESULT_SQL from products.messaging.backend.models.message_template import MessageTemplate -from products.workflows.backend.facade.testing import create_sent_email_asset_for_test +from products.workflows.backend.facade.testing import create_sent_email_asset_for_test, create_workflow_for_test SENT = "$workflows_email_sent" DELIVERED = "$workflows_email_delivered" @@ -25,6 +27,7 @@ BOUNCED = "$workflows_email_bounced" BLOCKED = "$workflows_email_blocked" FAILED = "$workflows_email_failed" +SKIPPED_BY_MX = 'the domain "example.com" has no reachable mail servers, so this message would hard bounce.' @patch( @@ -39,7 +42,7 @@ def setUp(self): self.workflow_id = self._transactional_workflow(self.team) def _transactional_workflow(self, team: Team) -> str: - workflow_id = str(uuid4()) + workflow_id = create_workflow_for_test(team_id=team.id, trigger={"type": "api"}).id MessageTemplate.objects.create( team=team, name="Password reset", type="email", transactional_hog_flow_id=workflow_id ) @@ -85,6 +88,20 @@ def _run( } sync_execute(INSERT_HOG_INVOCATION_RESULT_SQL, params) + def _skip_log(self) -> None: + sync_execute( + INSERT_LOG_ENTRY_SQL, + { + "team_id": self.team.id, + "log_source": "hog_flow", + "log_source_id": self.workflow_id, + "instance_id": self.message_id, + "timestamp": self._at(1), + "level": "INFO", + "message": f"[Action:transactional_email] Skipping send: {SKIPPED_BY_MX}", + }, + ) + def _sent_record(self) -> None: create_sent_email_asset_for_test( team_id=self.team.id, function_id=self.workflow_id, invocation_id=self.message_id, sent_at=self._at(2) @@ -113,11 +130,15 @@ def _expected(self, email_status: str, status_offset: int, reason: str | None) - @parameterized.expand( [ ("queued", "running", "", False, [], "queued", 0, None), - ("skipped", "succeeded", "", False, [], "skipped", 1, None), + ("skipped", "succeeded", "", False, [], "skipped", 1, SKIPPED_BY_MX), + ("finished_without_a_record", "succeeded", "", False, [], "unknown", 1, None), ("failed_run", "failed", "timeout", False, [], "failed", 1, "timeout"), # Engagement capture off: the worker's record of the send is all there is. ("sent_without_engagement_capture", "succeeded", "", True, [], "sent", 2, None), - ("sent", "succeeded", "", True, [(SENT, {})], "sent", 3, None), + ("sent_while_the_run_still_reads_running", "running", "", True, [], "sent", 2, None), + # An instance can turn the run records off. The send record still answers. + ("sent_without_a_run_record", None, "", True, [], "sent", 2, None), + ("sent", "succeeded", "", True, [(SENT, {})], "sent", 2, None), ("delivered", "succeeded", "", True, [(SENT, {}), (DELIVERED, {})], "delivered", 4, None), ("an_open_proves_delivery", "succeeded", "", True, [(SENT, {}), (OPENED, {})], "delivered", 4, None), ("bounced", "succeeded", "", True, [(SENT, {}), (BOUNCED, {"$bounce_type": "hard"})], "bounced", 4, "hard"), @@ -137,7 +158,10 @@ def _expected(self, email_status: str, status_offset: int, reason: str | None) - def test_reports_the_status_of_the_send( self, _flag, _name, run_status, error_kind, sent, events, expected_status, status_offset, reason ): - self._run(run_status, error_kind=error_kind) + if run_status is not None: + self._run(run_status, error_kind=error_kind) + if expected_status == "skipped": + self._skip_log() if sent: self._sent_record() self._email_events(self.team, events) @@ -147,6 +171,43 @@ def test_reports_the_status_of_the_send( assert response.status_code == status.HTTP_200_OK, response.json() assert response.json() == self._expected(expected_status, status_offset, reason) + def test_a_bounce_after_many_opens_is_still_reported(self, _flag): + self._run("succeeded") + self._sent_record() + self._email_events(self.team, [(OPENED, {})] * 101 + [(BOUNCED, {"$bounce_type": "hard"})]) + + response = self.client.get(self._url(self.message_id)) + + assert response.json()["status"] == "bounced" + + def test_reports_a_send_from_a_workflow_no_template_links_to_any_more(self, _flag): + relinked_away = create_workflow_for_test(team_id=self.team.id, trigger={"type": "api"}).id + self._run("running", workflow_id=relinked_away) + + response = self.client.get(self._url(self.message_id)) + + assert response.json()["status"] == "queued" + + def test_a_member_without_access_to_the_workflow_cannot_read_its_status(self, _flag): + self.organization.available_product_features = [ + {"key": AvailableFeature.ACCESS_CONTROL, "name": AvailableFeature.ACCESS_CONTROL}, + ] + self.organization.save() + self._run("running") + member = User.objects.create_and_join(self.organization, "member@example.com", "testtest") + create_access_control( + team_id=self.team.id, + resource="hog_flow", + resource_id=self.workflow_id, + access_level="none", + organization_member_id=OrganizationMembership.objects.get(user=member, organization=self.organization).id, + ) + self.client.force_login(member) + + response = self.client.get(self._url(self.message_id)) + + assert response.status_code == status.HTTP_403_FORBIDDEN + def test_stays_queued_while_the_send_record_of_a_finished_run_is_on_its_way(self, _flag): finished_at = datetime.now(UTC).replace(microsecond=0) self._run("succeeded", finished_at=finished_at) diff --git a/products/messaging/frontend/generated/api.schemas.ts b/products/messaging/frontend/generated/api.schemas.ts index f4462532267d..cfedf005f263 100644 --- a/products/messaging/frontend/generated/api.schemas.ts +++ b/products/messaging/frontend/generated/api.schemas.ts @@ -10,6 +10,7 @@ /** * * `queued` - Queued * * `skipped` - Skipped + * * `unknown` - Unknown * * `sent` - Sent * * `delivered` - Delivered * * `bounced` - Bounced @@ -21,6 +22,7 @@ export type EmailDeliveryStatusEnumApi = (typeof EmailDeliveryStatusEnumApi)[key export const EmailDeliveryStatusEnumApi = { Queued: 'queued', Skipped: 'skipped', + Unknown: 'unknown', Sent: 'sent', Delivered: 'delivered', Bounced: 'bounced', @@ -31,10 +33,11 @@ export const EmailDeliveryStatusEnumApi = { export interface EmailStatusApi { /** The `message_id` that the send returned. */ id: string - /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means it was not sent, for example because the recipient was suppressed after the send. `failed` means PostHog could not send it. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. + /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means PostHog decided not to send it, and `reason` says why. `failed` means PostHog could not send it. `unknown` means the run finished but PostHog has no record of the outcome. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. * * * `queued` - Queued * * `skipped` - Skipped + * * `unknown` - Unknown * * `sent` - Sent * * `delivered` - Delivered * * `bounced` - Bounced @@ -46,7 +49,7 @@ export interface EmailStatusApi { /** When the email reached its current status. */ status_at: string /** - * The bounce type for a bounced email, such as `hard` or `soft`, or the error kind of a failed email. Null for any other status. + * Why the email is skipped, the bounce type of a bounced email (such as `hard`), or the error kind of a failed email. Null for any other status. * @nullable */ reason: string | null diff --git a/products/workflows/backend/facade/api.py b/products/workflows/backend/facade/api.py index 060614e43bb8..991ee07d09bc 100644 --- a/products/workflows/backend/facade/api.py +++ b/products/workflows/backend/facade/api.py @@ -229,6 +229,14 @@ 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 api_triggered_workflow_ids(*, team_id: int) -> list[str]: + """Every workflow of the team that an API call triggers, in any status, such as each transactional email's.""" + return [ + str(flow_id) + for flow_id in HogFlow.objects.filter(team_id=team_id, trigger__type="api").values_list("id", flat=True) + ] + + def has_active_workflows(*, team_id: int) -> bool: return HogFlow.objects.filter(team_id=team_id, status=HogFlow.State.ACTIVE).exists() diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index e386c3770222..4951fd8d14c2 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -41106,6 +41106,7 @@ export namespace Schemas { /** * * `queued` - Queued * * `skipped` - Skipped + * * `unknown` - Unknown * * `sent` - Sent * * `delivered` - Delivered * * `bounced` - Bounced @@ -41118,6 +41119,7 @@ export namespace Schemas { export const EmailDeliveryStatusEnum = { Queued: 'queued', Skipped: 'skipped', + Unknown: 'unknown', Sent: 'sent', Delivered: 'delivered', Bounced: 'bounced', @@ -41502,10 +41504,11 @@ export namespace Schemas { export interface EmailStatus { /** The `message_id` that the send returned. */ id: string; - /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means it was not sent, for example because the recipient was suppressed after the send. `failed` means PostHog could not send it. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. + /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means PostHog decided not to send it, and `reason` says why. `failed` means PostHog could not send it. `unknown` means the run finished but PostHog has no record of the outcome. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. * * * `queued` - Queued * * `skipped` - Skipped + * * `unknown` - Unknown * * `sent` - Sent * * `delivered` - Delivered * * `bounced` - Bounced @@ -41517,7 +41520,7 @@ export namespace Schemas { /** When the email reached its current status. */ status_at: string; /** - * The bounce type for a bounced email, such as `hard` or `soft`, or the error kind of a failed email. Null for any other status. + * Why the email is skipped, the bounce type of a bounced email (such as `hard`), or the error kind of a failed email. Null for any other status. * @nullable */ reason: string | null; From 29353cfa21f9c4f9003f9ef00fcd493147237aba Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:24:59 +0000 Subject: [PATCH 3/5] fix(messaging): only let email events refine a send the worker recorded Anyone with a project's public token can capture a $workflows_email_* event, so events no longer decide failed or sent. failed comes from the run record alone, which keeps its error kind and follows a successful rerun. Events are read only when the worker's send record exists, and then only to refine it to delivered, bounced or complained, filtered on the send record's distinct_id. An unknown ID no longer scans the project's email events. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: e72a2c10-f145-4f7c-8754-a4fcc06a18be --- .../services/transactional_email_status.py | 42 ++++++++----------- .../api/test_transactional_email_status.py | 13 +++++- 2 files changed, 29 insertions(+), 26 deletions(-) diff --git a/products/messaging/backend/services/transactional_email_status.py b/products/messaging/backend/services/transactional_email_status.py index 8791f41fb003..03787bff425e 100644 --- a/products/messaging/backend/services/transactional_email_status.py +++ b/products/messaging/backend/services/transactional_email_status.py @@ -1,8 +1,8 @@ """The status of one transactional email. The send API records the run before it answers, and the email worker records every email the provider accepts and -logs every send it skips, so `queued`, `sent`, `skipped` and a failed run need nothing else. Delivery, bounces and complaints arrive as email -events, which a project only records when it turns on engagement capture. +logs every send it skips, so `queued`, `sent`, `skipped` and `failed` need nothing else. Delivery, bounces and complaints arrive as email events, +which a project only records when it turns on engagement capture, and which only refine an email recorded as sent. """ from datetime import UTC, datetime, timedelta @@ -22,9 +22,9 @@ from products.workflows.backend.facade.api import api_triggered_workflow_ids from products.workflows.backend.facade.message_assets import fetch_sent_email_asset +# Anyone with the project's public token can capture an event, so events only refine an email the worker's own record +# shows as sent. Sent and failed come from the worker's records alone. _STATUS_BY_EVENT: dict[str, EmailDeliveryStatus] = { - "$workflows_email_sent": EmailDeliveryStatus.SENT, - "$workflows_email_failed": EmailDeliveryStatus.FAILED, "$workflows_email_delivered": EmailDeliveryStatus.DELIVERED, # An open or a click proves delivery, also when the delivery event arrives later or not at all. "$workflows_email_opened": EmailDeliveryStatus.DELIVERED, @@ -38,7 +38,6 @@ _ENGAGEMENT_PRECEDENCE = [ EmailDeliveryStatus.COMPLAINED, EmailDeliveryStatus.BOUNCED, - EmailDeliveryStatus.FAILED, EmailDeliveryStatus.DELIVERED, ] # The email worker logs this, followed by the reason, when it decides not to send. See EmailService in nodejs. @@ -76,8 +75,6 @@ def get_email_status(team: Team, message_id: UUID) -> EmailStatus | None: run = fetch_invocation_lifecycle(team.id, "hog_flow", workflow_ids, str(message_id), since) sent = fetch_sent_email_asset(team.id, workflow_ids, str(message_id), since) workflow_id = (run.function_id if run else None) or (sent.function_id if sent else None) - distinct_id = (run.distinct_id if run else "") or (sent.distinct_id if sent else "") - events = _engagement(team, message_id, created_at, distinct_id) def status_of(status: EmailDeliveryStatus, status_at: datetime, reason: str | None = None) -> EmailStatus: return EmailStatus( @@ -89,15 +86,14 @@ def status_of(status: EmailDeliveryStatus, status_at: datetime, reason: str | No workflow_id=workflow_id, ) - # What the provider reported wins over the run, and a confirmed send wins over a run still marked running. - for status in _ENGAGEMENT_PRECEDENCE: - if status in events: - at, bounce_type = events[status] - return status_of(status, at, bounce_type if status == EmailDeliveryStatus.BOUNCED else None) + # A confirmed send wins over a run still marked running, and what the provider reported refines it. if sent is not None: + events = _engagement(team, message_id, created_at, sent.distinct_id) + for status in _ENGAGEMENT_PRECEDENCE: + if status in events: + at, bounce_type = events[status] + return status_of(status, at, bounce_type if status == EmailDeliveryStatus.BOUNCED else None) return status_of(EmailDeliveryStatus.SENT, sent.sent_at) - if EmailDeliveryStatus.SENT in events: - return status_of(EmailDeliveryStatus.SENT, events[EmailDeliveryStatus.SENT][0]) # The run records are monitoring data that an instance can turn off, so without one the email is unknown. if run is None: return None @@ -131,20 +127,16 @@ def _engagement( team: Team, message_id: UUID, created_at: datetime, distinct_id: str ) -> dict[EmailDeliveryStatus, tuple[datetime, str | None]]: """The latest time of each status the email's events report, with the bounce type of the latest bounce.""" - where = [ - "event IN {events}", - "timestamp >= {start}", - "timestamp < {end}", - "properties.$workflow_invocation_id = {message_id}", - ] - if distinct_id: - # distinct_id is in the events sort key, so this keeps a poll from scanning the project's email events. - where.append("distinct_id = {distinct_id}") + # distinct_id is in the events sort key after the event name, so a poll reads only this recipient's email events. query = parse_select( - f""" + """ SELECT event, max(timestamp), argMax(properties.$bounce_type, timestamp) FROM events - WHERE {" AND ".join(where)} + WHERE event IN {events} + AND timestamp >= {start} + AND timestamp < {end} + AND distinct_id = {distinct_id} + AND properties.$workflow_invocation_id = {message_id} GROUP BY event """, placeholders={ diff --git a/products/messaging/backend/tests/api/test_transactional_email_status.py b/products/messaging/backend/tests/api/test_transactional_email_status.py index dfa04423fe1a..b8fd2e362580 100644 --- a/products/messaging/backend/tests/api/test_transactional_email_status.py +++ b/products/messaging/backend/tests/api/test_transactional_email_status.py @@ -139,6 +139,18 @@ def _expected(self, email_status: str, status_offset: int, reason: str | None) - # An instance can turn the run records off. The send record still answers. ("sent_without_a_run_record", None, "", True, [], "sent", 2, None), ("sent", "succeeded", "", True, [(SENT, {})], "sent", 2, None), + # Anyone with the project token can capture events, so they never override or stand in for the worker. + ( + "a_failure_event_does_not_override_a_confirmed_send", + "succeeded", + "", + True, + [(FAILED, {})], + "sent", + 2, + None, + ), + ("events_without_a_send_record_are_ignored", "succeeded", "", False, [(BOUNCED, {})], "unknown", 1, None), ("delivered", "succeeded", "", True, [(SENT, {}), (DELIVERED, {})], "delivered", 4, None), ("an_open_proves_delivery", "succeeded", "", True, [(SENT, {}), (OPENED, {})], "delivered", 4, None), ("bounced", "succeeded", "", True, [(SENT, {}), (BOUNCED, {"$bounce_type": "hard"})], "bounced", 4, "hard"), @@ -152,7 +164,6 @@ def _expected(self, email_status: str, status_offset: int, reason: str | None) - 4, None, ), - ("failed_send", "succeeded", "", False, [(FAILED, {})], "failed", 3, None), ] ) def test_reports_the_status_of_the_send( From 41f4cbe628540ab88a99835c294e395570d04d28 Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:25:04 +0000 Subject: [PATCH 4/5] fix(messaging): allow read-only keys for email status and throttle session reads - Project secret keys can now hold hog_flow:read, so a service that polls a transactional email's status doesn't need a key that can also send. No other hog_flow viewset accepts project keys, so the scope opens nothing else. - Session and personal API key callers of the status route keep the default API throttles. The email read buckets only count project keys. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: e72a2c10-f145-4f7c-8754-a4fcc06a18be --- frontend/src/lib/scopes.tsx | 1 + posthog/api/test/test_project_secret_api_keys.py | 15 +++++++++++---- posthog/scopes.py | 6 ++++-- .../views/transactional_email_status.py | 3 +++ 4 files changed, 19 insertions(+), 6 deletions(-) diff --git a/frontend/src/lib/scopes.tsx b/frontend/src/lib/scopes.tsx index 510580043bef..f522dcb17411 100644 --- a/frontend/src/lib/scopes.tsx +++ b/frontend/src/lib/scopes.tsx @@ -380,6 +380,7 @@ export const PROJECT_SECRET_API_KEY_ALLOWED_API_SCOPE_ACTION = [ 'billing:read', 'replay_scanner:write', 'session_recording:read', + 'hog_flow:read', 'hog_flow:write', ] as const diff --git a/posthog/api/test/test_project_secret_api_keys.py b/posthog/api/test/test_project_secret_api_keys.py index b24ad90598cc..7f5ac581e9a0 100644 --- a/posthog/api/test/test_project_secret_api_keys.py +++ b/posthog/api/test/test_project_secret_api_keys.py @@ -81,13 +81,20 @@ def test_create_project_secret_api_key(self): assert data["last_used_at"] is None assert data["value"].startswith("phs_") - def test_create_feature_flag_read_scope(self): + @parameterized.expand( + [ + ("feature_flag_local_evaluation", "feature_flag:read"), + # A service that polls a transactional email's status must not need a key that can also send. + ("transactional_email_status", "hog_flow:read"), + ] + ) + def test_create_read_only_scope(self, _name, scope): response = self.client.post( f"/api/projects/{self.team.id}/project_secret_api_keys", - {"label": "local eval key", "scopes": ["feature_flag:read"]}, + {"label": "read only key", "scopes": [scope]}, ) - assert response.status_code == 201 - assert response.json()["scopes"] == ["feature_flag:read"] + assert response.status_code == 201, response.json() + assert response.json()["scopes"] == [scope] def test_create_too_many_api_keys(self): for i in range(0, MAX_PROJECT_SECRET_API_KEYS_PER_TEAM): diff --git a/posthog/scopes.py b/posthog/scopes.py index ced1cc5e00b1..b4625d238f70 100644 --- a/posthog/scopes.py +++ b/posthog/scopes.py @@ -272,8 +272,10 @@ # Both are needed together; every other replay_scanner and session_recording route stays PSAK-denied. ("replay_scanner", "write"), ("session_recording", "read"), - # Lets a service send transactional email through `messaging_templates/:id/send/`. Every other - # hog_flow route stays PSAK-denied, because no other hog_flow viewset lists PSAK actions. + # Lets a service send transactional email (`messaging_templates/:id/send/`, `messaging/send/email/`) and read a + # send's status (`messaging/emails/:id/`). Every other hog_flow route stays PSAK-denied, because no other hog_flow + # viewset lists PSAK actions. + ("hog_flow", "read"), ("hog_flow", "write"), ] diff --git a/products/messaging/backend/presentation/views/transactional_email_status.py b/products/messaging/backend/presentation/views/transactional_email_status.py index cbf0d41b6735..714ba5dc5e93 100644 --- a/products/messaging/backend/presentation/views/transactional_email_status.py +++ b/products/messaging/backend/presentation/views/transactional_email_status.py @@ -76,6 +76,9 @@ class MessagingEmailsViewSet(TeamAndOrgViewSetMixin, viewsets.ViewSet): psak_allowed_actions = ["retrieve"] def get_throttles(self) -> list[BaseThrottle]: + # The email read buckets only count API keys, so every other caller keeps the default API throttles. + if not is_authenticated_via_project_secret_api_key(self.request): + return super().get_throttles() return [ TransactionalEmailReadBurstThrottle(), TransactionalEmailReadSustainedThrottle(), From 6451324a558659170e3bcd68e79255a053a72060 Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Tue, 27 Oct 2026 17:25:07 +0000 Subject: [PATCH 5/5] refactor(messaging): report a transactional email as queued, sent or failed The status now comes from two records PostHog writes itself: the run the send API records before it answers, and the worker's record of an email the provider accepted. queued while the run waits or its send record is on the way, sent once the send record exists, failed when the run errored (reason is the error kind) or finished without sending (reason not_sent). The checks before the send API answers already refuse every email the worker would skip, so a separate skipped status and its log read are dropped. Delivered, bounced and complained are dropped too: they need engagement capture, depend on #123902, and come from events anyone with the project token can capture. They can come back once provider outcomes have trusted storage. Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: e72a2c10-f145-4f7c-8754-a4fcc06a18be --- .github/new-events-schema-targets.txt | 1 - .../messaging/backend/facade/contracts.py | 7 +- .../views/transactional_email_status.py | 16 +-- .../services/transactional_email_status.py | 119 ++---------------- .../api/test_transactional_email_status.py | 91 ++------------ .../frontend/generated/api.schemas.ts | 19 +-- products/messaging/frontend/generated/api.ts | 2 +- services/mcp/src/api/generated.ts | 19 +-- 8 files changed, 31 insertions(+), 243 deletions(-) diff --git a/.github/new-events-schema-targets.txt b/.github/new-events-schema-targets.txt index 1b878ed6cee7..cdbdda95b3a7 100644 --- a/.github/new-events-schema-targets.txt +++ b/.github/new-events-schema-targets.txt @@ -43,7 +43,6 @@ products/mcp_analytics/backend/tests/test_model_breakdown.py products/mcp_analytics/backend/tests/test_protocol_version_breakdown.py products/mcp_analytics/backend/tests/test_tool_quality_tables.py products/mcp_analytics/backend/tests/test_tool_tables.py -products/messaging/backend/tests/api/test_transactional_email_status.py products/product_analytics/backend/hogql_queries products/product_analytics/backend/tests/test_insight_data_checks.py products/replay_vision/backend/tests/test_session_identity.py diff --git a/products/messaging/backend/facade/contracts.py b/products/messaging/backend/facade/contracts.py index 83bcee1bf9c7..6ebdf5fa0383 100644 --- a/products/messaging/backend/facade/contracts.py +++ b/products/messaging/backend/facade/contracts.py @@ -111,12 +111,7 @@ class QueuedEmail: class EmailDeliveryStatus(LabeledStrEnum): QUEUED = "queued", "Queued" - SKIPPED = "skipped", "Skipped" - UNKNOWN = "unknown", "Unknown" SENT = "sent", "Sent" - DELIVERED = "delivered", "Delivered" - BOUNCED = "bounced", "Bounced" - COMPLAINED = "complained", "Complained" FAILED = "failed", "Failed" @@ -126,7 +121,7 @@ class EmailStatus: status: EmailDeliveryStatus created_at: datetime status_at: datetime - # The bounce type for a bounced email, why a send was skipped, or the error kind of a failed run. + # Why a send failed: the run's error kind, or not_sent when the run finished without sending. reason: str | None = None # The workflow that sent the email, so a reader's access to it can be checked. Not part of the API response. workflow_id: str | None = None diff --git a/products/messaging/backend/presentation/views/transactional_email_status.py b/products/messaging/backend/presentation/views/transactional_email_status.py index 714ba5dc5e93..93e8fbe46980 100644 --- a/products/messaging/backend/presentation/views/transactional_email_status.py +++ b/products/messaging/backend/presentation/views/transactional_email_status.py @@ -32,18 +32,15 @@ class EmailStatusSerializer(serializers.Serializer): id = serializers.CharField(source="message_id", help_text="The `message_id` that the send returned.") status = serializers.ChoiceField( choices=EmailDeliveryStatus.choices, - help_text="What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. " - "`skipped` means PostHog decided not to send it, and `reason` says why. `failed` means PostHog could not " - "send it. `unknown` means the run finished but PostHog has no record of the outcome. `delivered` means the " - "recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail " - "provider.", + help_text="What happened to the email. `queued` means it waits to go out. `sent` means PostHog handed it to " + "the email provider. `failed` means it was not sent, and `reason` says why.", ) created_at = serializers.DateTimeField(help_text="When the send was queued.") status_at = serializers.DateTimeField(help_text="When the email reached its current status.") reason = serializers.CharField( allow_null=True, - help_text="Why the email is skipped, the bounce type of a bounced email (such as `hard`), or the error kind " - "of a failed email. Null for any other status.", + help_text="Why a failed email was not sent: the error kind of a run that errored, or `not_sent` when the run " + "finished without sending, for example because the recipient was suppressed after the send. Null otherwise.", ) @@ -97,9 +94,8 @@ def get_throttles(self) -> list[BaseThrottle]: }, summary="Get a transactional email's status", description=( - "Get the status of one transactional email by the `message_id` that the send returned. Every project sees " - "`queued`, `sent`, `skipped` and `failed`. `delivered`, `bounced` and `complained` come from the email's " - "events, so they need engagement capture turned on for the project." + "Get the status of one transactional email by the `message_id` that the send returned: `queued` while " + "it waits to go out, `sent` once PostHog hands it to the email provider, or `failed` when it was not sent." ), ) def retrieve(self, request: Request, *args: Any, pk: str | None = None, **kwargs: Any) -> Response: diff --git a/products/messaging/backend/services/transactional_email_status.py b/products/messaging/backend/services/transactional_email_status.py index 03787bff425e..254cf51329aa 100644 --- a/products/messaging/backend/services/transactional_email_status.py +++ b/products/messaging/backend/services/transactional_email_status.py @@ -1,18 +1,13 @@ """The status of one transactional email. -The send API records the run before it answers, and the email worker records every email the provider accepts and -logs every send it skips, so `queued`, `sent`, `skipped` and `failed` need nothing else. Delivery, bounces and complaints arrive as email events, -which a project only records when it turns on engagement capture, and which only refine an email recorded as sent. +The send API records the run before it answers, and the email worker records every email the provider accepts, so the +two records say whether an email is queued, sent or failed. The checks before the send API answers already refuse an +email the worker would skip, so a finished run without a send record is rare and reads as failed. """ from datetime import UTC, datetime, timedelta from uuid import UUID -from posthog.hogql import ast -from posthog.hogql.parser import parse_select -from posthog.hogql.query import execute_hogql_query - -from posthog.clickhouse.client.connection import Workload from posthog.clickhouse.query_tagging import Feature, Product, tag_queries from posthog.models.team import Team @@ -22,33 +17,13 @@ from products.workflows.backend.facade.api import api_triggered_workflow_ids from products.workflows.backend.facade.message_assets import fetch_sent_email_asset -# Anyone with the project's public token can capture an event, so events only refine an email the worker's own record -# shows as sent. Sent and failed come from the worker's records alone. -_STATUS_BY_EVENT: dict[str, EmailDeliveryStatus] = { - "$workflows_email_delivered": EmailDeliveryStatus.DELIVERED, - # An open or a click proves delivery, also when the delivery event arrives later or not at all. - "$workflows_email_opened": EmailDeliveryStatus.DELIVERED, - "$workflows_email_link_clicked": EmailDeliveryStatus.DELIVERED, - "$workflows_email_bounced": EmailDeliveryStatus.BOUNCED, - # The mailbox provider reported a spam complaint. - "$workflows_email_blocked": EmailDeliveryStatus.COMPLAINED, -} - -# Highest first. A later open does not undo a bounce or a complaint. -_ENGAGEMENT_PRECEDENCE = [ - EmailDeliveryStatus.COMPLAINED, - EmailDeliveryStatus.BOUNCED, - EmailDeliveryStatus.DELIVERED, -] -# The email worker logs this, followed by the reason, when it decides not to send. See EmailService in nodejs. -_SKIP_LOG_PREFIX = "Skipping send:" - -# The provider reports an email's delivery, bounce and complaint within days of the send. -_EVENT_WINDOW = timedelta(days=30) -# The Node API stamps the send's ID with its own clock, which can run ahead of the event timestamps. +# The Node API stamps the send's ID with its own clock, which can run ahead of the records' timestamps. _CLOCK_SKEW = timedelta(minutes=5) -# How long after a run finishes its send record and logs can still be on their way to ClickHouse. +# How long after a run finishes its send record can still be on its way to ClickHouse. _SEND_RECORD_GRACE = timedelta(minutes=1) +# The reason of a run that finished without sending, for example because the recipient was suppressed in the +# moments between the send API's checks and the worker. +NOT_SENT_REASON = "not_sent" def queued_at(message_id: UUID) -> datetime | None: @@ -66,7 +41,7 @@ def get_email_status(team: Team, message_id: UUID) -> EmailStatus | None: created_at = queued_at(message_id) if created_at is None: return None - # Tagged once here, because every read below, through the cdp and workflows facades too, runs a ClickHouse query. + # Tagged once here, because both reads below run a ClickHouse query through another product's facade. tag_queries(product=Product.MESSAGING, feature=Feature.QUERY) workflow_ids = _transactional_workflow_ids(team.id) if not workflow_ids: @@ -86,30 +61,19 @@ def status_of(status: EmailDeliveryStatus, status_at: datetime, reason: str | No workflow_id=workflow_id, ) - # A confirmed send wins over a run still marked running, and what the provider reported refines it. + # A confirmed send wins over a run still marked running, and still answers when an instance turns run records off. if sent is not None: - events = _engagement(team, message_id, created_at, sent.distinct_id) - for status in _ENGAGEMENT_PRECEDENCE: - if status in events: - at, bounce_type = events[status] - return status_of(status, at, bounce_type if status == EmailDeliveryStatus.BOUNCED else None) return status_of(EmailDeliveryStatus.SENT, sent.sent_at) - # The run records are monitoring data that an instance can turn off, so without one the email is unknown. if run is None: return None - if run.status in ("failed", "canceled"): return status_of(EmailDeliveryStatus.FAILED, run.finished_at or run.scheduled_at, run.error_kind or None) if run.status == "running": return status_of(EmailDeliveryStatus.QUEUED, run.started_at or run.scheduled_at) finished_at = run.finished_at or run.scheduled_at - skip_reason = _skip_reason(team, run.function_id, message_id, since) - if skip_reason is not None: - return status_of(EmailDeliveryStatus.SKIPPED, finished_at, skip_reason) - # The worker writes the send record and its logs next to the finished run, so give them a moment to arrive. if datetime.now(UTC) - finished_at < _SEND_RECORD_GRACE: return status_of(EmailDeliveryStatus.QUEUED, finished_at) - return status_of(EmailDeliveryStatus.UNKNOWN, finished_at) + return status_of(EmailDeliveryStatus.FAILED, finished_at, NOT_SENT_REASON) def _transactional_workflow_ids(team_id: int) -> list[str]: @@ -121,64 +85,3 @@ def _transactional_workflow_ids(team_id: int) -> list[str]: .values_list("transactional_hog_flow_id", flat=True) ) return sorted({str(flow_id) for flow_id in linked} | set(api_triggered_workflow_ids(team_id=team_id))) - - -def _engagement( - team: Team, message_id: UUID, created_at: datetime, distinct_id: str -) -> dict[EmailDeliveryStatus, tuple[datetime, str | None]]: - """The latest time of each status the email's events report, with the bounce type of the latest bounce.""" - # distinct_id is in the events sort key after the event name, so a poll reads only this recipient's email events. - query = parse_select( - """ - SELECT event, max(timestamp), argMax(properties.$bounce_type, timestamp) - FROM events - WHERE event IN {events} - AND timestamp >= {start} - AND timestamp < {end} - AND distinct_id = {distinct_id} - AND properties.$workflow_invocation_id = {message_id} - GROUP BY event - """, - placeholders={ - "events": ast.Constant(value=list(_STATUS_BY_EVENT)), - "start": ast.Constant(value=created_at - _CLOCK_SKEW), - "end": ast.Constant(value=created_at + _EVENT_WINDOW), - "message_id": ast.Constant(value=str(message_id)), - "distinct_id": ast.Constant(value=distinct_id), - }, - ) - # Callers hold a project secret API key and poll from a server. No person waits on screen. - response = execute_hogql_query(query=query, team=team, workload=Workload.OFFLINE) - latest: dict[EmailDeliveryStatus, tuple[datetime, str | None]] = {} - for event, timestamp, bounce_type in response.results: - status = _STATUS_BY_EVENT[event] - if status not in latest or timestamp > latest[status][0]: - latest[status] = (timestamp, bounce_type or None) - return latest - - -def _skip_reason(team: Team, workflow_id: str, message_id: UUID, since: datetime) -> str | None: - """Why the email worker skipped the send, from the run's log, or None when it did not log a skip.""" - query = parse_select( - """ - SELECT message - FROM log_entries - WHERE log_source = 'hog_flow' - AND log_source_id = {workflow_id} - AND instance_id = {message_id} - AND timestamp >= {since} - AND message LIKE {pattern} - ORDER BY timestamp DESC - LIMIT 1 - """, - placeholders={ - "workflow_id": ast.Constant(value=workflow_id), - "message_id": ast.Constant(value=str(message_id)), - "since": ast.Constant(value=since), - "pattern": ast.Constant(value=f"%{_SKIP_LOG_PREFIX}%"), - }, - ) - response = execute_hogql_query(query=query, team=team, workload=Workload.OFFLINE) - if not response.results: - return None - return response.results[0][0].split(_SKIP_LOG_PREFIX, 1)[1].strip() diff --git a/products/messaging/backend/tests/api/test_transactional_email_status.py b/products/messaging/backend/tests/api/test_transactional_email_status.py index b8fd2e362580..2b4cf8219e45 100644 --- a/products/messaging/backend/tests/api/test_transactional_email_status.py +++ b/products/messaging/backend/tests/api/test_transactional_email_status.py @@ -1,7 +1,7 @@ from datetime import UTC, datetime, timedelta from typing import Any -from posthog.test.base import APIBaseTest, ClickhouseTestMixin, _create_event, flush_persons_and_events +from posthog.test.base import APIBaseTest, ClickhouseTestMixin from unittest.mock import patch from parameterized import parameterized @@ -9,7 +9,6 @@ from rest_framework.test import APIClient from posthog.clickhouse.client.execute import sync_execute -from posthog.clickhouse.log_entries import INSERT_LOG_ENTRY_SQL from posthog.constants import AvailableFeature from posthog.models import OrganizationMembership, ProjectSecretAPIKey, Team, User from posthog.models.personal_api_key import hash_key_value @@ -21,14 +20,6 @@ from products.messaging.backend.models.message_template import MessageTemplate from products.workflows.backend.facade.testing import create_sent_email_asset_for_test, create_workflow_for_test -SENT = "$workflows_email_sent" -DELIVERED = "$workflows_email_delivered" -OPENED = "$workflows_email_opened" -BOUNCED = "$workflows_email_bounced" -BLOCKED = "$workflows_email_blocked" -FAILED = "$workflows_email_failed" -SKIPPED_BY_MX = 'the domain "example.com" has no reachable mail servers, so this message would hard bounce.' - @patch( "products.messaging.backend.presentation.views.message_templates.posthoganalytics.feature_enabled", @@ -88,36 +79,11 @@ def _run( } sync_execute(INSERT_HOG_INVOCATION_RESULT_SQL, params) - def _skip_log(self) -> None: - sync_execute( - INSERT_LOG_ENTRY_SQL, - { - "team_id": self.team.id, - "log_source": "hog_flow", - "log_source_id": self.workflow_id, - "instance_id": self.message_id, - "timestamp": self._at(1), - "level": "INFO", - "message": f"[Action:transactional_email] Skipping send: {SKIPPED_BY_MX}", - }, - ) - def _sent_record(self) -> None: create_sent_email_asset_for_test( team_id=self.team.id, function_id=self.workflow_id, invocation_id=self.message_id, sent_at=self._at(2) ) - def _email_events(self, team: Team, events: list[tuple[str, dict]]) -> None: - for offset, (event, properties) in enumerate(events, start=3): - _create_event( - team=team, - event=event, - distinct_id="user@example.com", - properties={"$workflow_invocation_id": self.message_id, **properties}, - timestamp=self._at(offset), - ) - flush_persons_and_events() - def _expected(self, email_status: str, status_offset: int, reason: str | None) -> dict: return { "id": self.message_id, @@ -129,68 +95,28 @@ def _expected(self, email_status: str, status_offset: int, reason: str | None) - @parameterized.expand( [ - ("queued", "running", "", False, [], "queued", 0, None), - ("skipped", "succeeded", "", False, [], "skipped", 1, SKIPPED_BY_MX), - ("finished_without_a_record", "succeeded", "", False, [], "unknown", 1, None), - ("failed_run", "failed", "timeout", False, [], "failed", 1, "timeout"), - # Engagement capture off: the worker's record of the send is all there is. - ("sent_without_engagement_capture", "succeeded", "", True, [], "sent", 2, None), - ("sent_while_the_run_still_reads_running", "running", "", True, [], "sent", 2, None), + ("queued", "running", "", False, "queued", 0, None), + ("sent", "succeeded", "", True, "sent", 2, None), + ("sent_while_the_run_still_reads_running", "running", "", True, "sent", 2, None), # An instance can turn the run records off. The send record still answers. - ("sent_without_a_run_record", None, "", True, [], "sent", 2, None), - ("sent", "succeeded", "", True, [(SENT, {})], "sent", 2, None), - # Anyone with the project token can capture events, so they never override or stand in for the worker. - ( - "a_failure_event_does_not_override_a_confirmed_send", - "succeeded", - "", - True, - [(FAILED, {})], - "sent", - 2, - None, - ), - ("events_without_a_send_record_are_ignored", "succeeded", "", False, [(BOUNCED, {})], "unknown", 1, None), - ("delivered", "succeeded", "", True, [(SENT, {}), (DELIVERED, {})], "delivered", 4, None), - ("an_open_proves_delivery", "succeeded", "", True, [(SENT, {}), (OPENED, {})], "delivered", 4, None), - ("bounced", "succeeded", "", True, [(SENT, {}), (BOUNCED, {"$bounce_type": "hard"})], "bounced", 4, "hard"), - ( - "a_later_open_keeps_the_complaint", - "succeeded", - "", - True, - [(SENT, {}), (BLOCKED, {}), (OPENED, {})], - "complained", - 4, - None, - ), + ("sent_without_a_run_record", None, "", True, "sent", 2, None), + ("failed_run", "failed", "timeout", False, "failed", 1, "timeout"), + ("finished_without_sending", "succeeded", "", False, "failed", 1, "not_sent"), ] ) def test_reports_the_status_of_the_send( - self, _flag, _name, run_status, error_kind, sent, events, expected_status, status_offset, reason + self, _flag, _name, run_status, error_kind, sent, expected_status, status_offset, reason ): if run_status is not None: self._run(run_status, error_kind=error_kind) - if expected_status == "skipped": - self._skip_log() if sent: self._sent_record() - self._email_events(self.team, events) response = self.client.get(self._url(self.message_id)) assert response.status_code == status.HTTP_200_OK, response.json() assert response.json() == self._expected(expected_status, status_offset, reason) - def test_a_bounce_after_many_opens_is_still_reported(self, _flag): - self._run("succeeded") - self._sent_record() - self._email_events(self.team, [(OPENED, {})] * 101 + [(BOUNCED, {"$bounce_type": "hard"})]) - - response = self.client.get(self._url(self.message_id)) - - assert response.json()["status"] == "bounced" - def test_reports_a_send_from_a_workflow_no_template_links_to_any_more(self, _flag): relinked_away = create_workflow_for_test(team_id=self.team.id, trigger={"type": "api"}).id self._run("running", workflow_id=relinked_away) @@ -239,7 +165,6 @@ def test_answers_404_for_an_email_it_cannot_report(self, _flag, _name, message_i if other_project: other = Team.objects.create(organization=self.organization, name="Other") self._run("succeeded", team=other, workflow_id=self._transactional_workflow(other)) - self._email_events(other, [(SENT, {})]) response = self.client.get(self._url(self.message_id if message_id == "message_id" else message_id)) diff --git a/products/messaging/frontend/generated/api.schemas.ts b/products/messaging/frontend/generated/api.schemas.ts index cfedf005f263..ff16860088ed 100644 --- a/products/messaging/frontend/generated/api.schemas.ts +++ b/products/messaging/frontend/generated/api.schemas.ts @@ -9,39 +9,24 @@ */ /** * * `queued` - Queued - * * `skipped` - Skipped - * * `unknown` - Unknown * * `sent` - Sent - * * `delivered` - Delivered - * * `bounced` - Bounced - * * `complained` - Complained * * `failed` - Failed */ export type EmailDeliveryStatusEnumApi = (typeof EmailDeliveryStatusEnumApi)[keyof typeof EmailDeliveryStatusEnumApi] export const EmailDeliveryStatusEnumApi = { Queued: 'queued', - Skipped: 'skipped', - Unknown: 'unknown', Sent: 'sent', - Delivered: 'delivered', - Bounced: 'bounced', - Complained: 'complained', Failed: 'failed', } as const export interface EmailStatusApi { /** The `message_id` that the send returned. */ id: string - /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means PostHog decided not to send it, and `reason` says why. `failed` means PostHog could not send it. `unknown` means the run finished but PostHog has no record of the outcome. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. + /** What happened to the email. `queued` means it waits to go out. `sent` means PostHog handed it to the email provider. `failed` means it was not sent, and `reason` says why. * * * `queued` - Queued - * * `skipped` - Skipped - * * `unknown` - Unknown * * `sent` - Sent - * * `delivered` - Delivered - * * `bounced` - Bounced - * * `complained` - Complained * * `failed` - Failed */ status: EmailDeliveryStatusEnumApi /** When the send was queued. */ @@ -49,7 +34,7 @@ export interface EmailStatusApi { /** When the email reached its current status. */ status_at: string /** - * Why the email is skipped, the bounce type of a bounced email (such as `hard`), or the error kind of a failed email. Null for any other status. + * Why a failed email was not sent: the error kind of a run that errored, or `not_sent` when the run finished without sending, for example because the recipient was suppressed after the send. Null otherwise. * @nullable */ reason: string | null diff --git a/products/messaging/frontend/generated/api.ts b/products/messaging/frontend/generated/api.ts index 6e331959a172..3ccc18b9c0ec 100644 --- a/products/messaging/frontend/generated/api.ts +++ b/products/messaging/frontend/generated/api.ts @@ -62,7 +62,7 @@ export const getMessagingEmailsRetrieveUrl = (projectId: string, id: string) => } /** - * Get the status of one transactional email by the `message_id` that the send returned. Every project sees `queued`, `sent`, `skipped` and `failed`. `delivered`, `bounced` and `complained` come from the email's events, so they need engagement capture turned on for the project. + * Get the status of one transactional email by the `message_id` that the send returned: `queued` while it waits to go out, `sent` once PostHog hands it to the email provider, or `failed` when it was not sent. * @summary Get a transactional email's status */ export const messagingEmailsRetrieve = async ( diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 4951fd8d14c2..f40d7a6f3a90 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -41105,12 +41105,7 @@ export namespace Schemas { /** * * `queued` - Queued - * * `skipped` - Skipped - * * `unknown` - Unknown * * `sent` - Sent - * * `delivered` - Delivered - * * `bounced` - Bounced - * * `complained` - Complained * * `failed` - Failed */ export type EmailDeliveryStatusEnum = typeof EmailDeliveryStatusEnum[keyof typeof EmailDeliveryStatusEnum]; @@ -41118,12 +41113,7 @@ export namespace Schemas { export const EmailDeliveryStatusEnum = { Queued: 'queued', - Skipped: 'skipped', - Unknown: 'unknown', Sent: 'sent', - Delivered: 'delivered', - Bounced: 'bounced', - Complained: 'complained', Failed: 'failed', } as const; @@ -41504,15 +41494,10 @@ export namespace Schemas { export interface EmailStatus { /** The `message_id` that the send returned. */ id: string; - /** What happened to the email. `queued` means it waits to go out. `sent` means it left PostHog. `skipped` means PostHog decided not to send it, and `reason` says why. `failed` means PostHog could not send it. `unknown` means the run finished but PostHog has no record of the outcome. `delivered` means the recipient's mail server accepted it. `bounced` and `complained` come later from the recipient's mail provider. + /** What happened to the email. `queued` means it waits to go out. `sent` means PostHog handed it to the email provider. `failed` means it was not sent, and `reason` says why. * * * `queued` - Queued - * * `skipped` - Skipped - * * `unknown` - Unknown * * `sent` - Sent - * * `delivered` - Delivered - * * `bounced` - Bounced - * * `complained` - Complained * * `failed` - Failed */ status: EmailDeliveryStatusEnum; /** When the send was queued. */ @@ -41520,7 +41505,7 @@ export namespace Schemas { /** When the email reached its current status. */ status_at: string; /** - * Why the email is skipped, the bounce type of a bounced email (such as `hard`), or the error kind of a failed email. Null for any other status. + * Why a failed email was not sent: the error kind of a run that errored, or `not_sent` when the run finished without sending, for example because the recipient was suppressed after the send. Null otherwise. * @nullable */ reason: string | null;