feat(messaging): read a transactional email's status by message id

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
 

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):

posthog/scopes.py

@@ -275,8 +275,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"),
 ]
 

products/messaging/backend/facade/contracts.py

@@ -1,4 +1,5 @@
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any
 from uuid import UUID
 
@@ -158,3 +159,22 @@ class SupabaseHookResult:
     error_code: str = ""
     # How many emails went out. Email change can send two.
     sent_count: int = 0
+
+
+class EmailDeliveryStatus(LabeledStrEnum):
+    QUEUED = "queued", "Queued"
+    FAILED = "failed", "Failed"
+    SENT = "sent", "Sent"
+    DELIVERED = "delivered", "Delivered"
+    BOUNCED = "bounced", "Bounced"
+    COMPLAINED = "complained", "Complained"
+
+
+@frozen
+class EmailStatus:
+    message_id: str
+    status: EmailDeliveryStatus
+    created_at: datetime
+    status_at: datetime
+    # Why a send failed, or the bounce type of a bounced email.
+    reason: str | None = None

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",

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:

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

@@ -0,0 +1,118 @@
+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 exceptions, 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.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
+
+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 PostHog handed it to "
+        "the email provider. `failed` means it was not sent, and `reason` says why. A rerun of the send can still move it to `sent`. `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="Why a failed email was not sent, or the bounce type of a bounced email. A failed send has "
+        "`provider_error`, or `provider_rejected` when the provider refused the email after it accepted the send. A skipped send has `recipient_suppressed`, `recipient_unreachable`, `recipient_opted_out`, "
+        "`missing_recipient`, `sending_paused`, `sending_suspended` or `frequency_capped`. A run that stopped before "
+        "the send has `canceled`, `workflow_unavailable` or an error kind such as `timeout`, and more error kinds can "
+        "be added. A bounce has `hard`, `soft` or `unknown`. Null otherwise.",
+    )
+
+
+# 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]:
+        # Only a project secret API key can read, and the email read buckets count those. Any other caller gets the
+        # default API throttles on its way to a 401.
+        if not is_authenticated_via_project_secret_api_key(self.request):
+            return super().get_throttles()
+        return [
+            TransactionalEmailReadBurstThrottle(),
+            TransactionalEmailReadSustainedThrottle(),
+            TransactionalEmailReadTeamBurstThrottle(),
+            TransactionalEmailReadTeamSustainedThrottle(),
+        ]
+
+    @extend_schema(
+        parameters=[MESSAGE_ID_PARAMETER],
+        responses={
+            200: EmailStatusSerializer,
+            401: OpenApiResponse(description="The request has no project secret API key."),
+            404: OpenApiResponse(
+                description="No email of this project has this ID, or it is older than 30 days. A send can also "
+                "answer 404 for a few seconds 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, for 30 days after "
+            "the send. Needs a project secret API key with the `hog_flow:read` scope."
+        ),
+    )
+    def retrieve(self, request: Request, *args: Any, pk: str | None = None, **kwargs: Any) -> Response:
+        # The routing mixin adds session and personal API key authentication to every route. A status is read by the
+        # server that sent the email, so only its project secret API key can read it.
+        if not is_authenticated_via_project_secret_api_key(request):
+            raise exceptions.NotAuthenticated()
+        require_transactional_email_api(self.team)
+        try:
+            message_id = UUID(str(pk))
+        except ValueError:
+            raise NotFound()
+        email_status = get_email_status(self.team_id, message_id)
+        if email_status is None:
+            raise NotFound()
+        return Response(EmailStatusSerializer(email_status).data)

products/messaging/backend/routes.py

@@ -14,6 +14,7 @@
     SupabaseAuthEmailsViewSet,
     SupabaseSendEmailHookView,
 )
+from products.messaging.backend.presentation.views.transactional_email_status import MessagingEmailsViewSet
 
 webhook_urlpatterns: list[URLPattern] = [
     re_path(
@@ -48,3 +49,4 @@ def register_routes(routers: RouterRegistry) -> None:
         "project_messaging_supabase_auth_emails",
         ["team_id"],
     )
+    routers.projects.register(r"messaging/emails", MessagingEmailsViewSet, "project_messaging_emails", ["team_id"])

products/messaging/backend/services/transactional_email_status.py

@@ -0,0 +1,76 @@
+"""The status of one transactional email, read from the email_status table that the email path writes.
+
+The send API writes `queued`, the email worker `sent` or `failed`, and the SES webhook `delivered`, `bounced` or
+`complained`. A row's version only grows as an email moves forward, so the row with the highest version is the status.
+"""
+
+from datetime import UTC, datetime, timedelta
+from typing import cast
+from uuid import UUID
+
+from posthog.clickhouse.client.connection import Workload
+from posthog.clickhouse.client.execute import sync_execute
+from posthog.clickhouse.query_tagging import Feature, Product, tag_queries
+
+from products.messaging.backend.facade.contracts import EmailDeliveryStatus, EmailStatus
+from products.messaging.backend.models.email_status_sql import EMAIL_STATUS_RETENTION_DAYS, EMAIL_STATUS_TABLE
+
+# The Node API stamps the send's ID with its own clock, which can run ahead of this one.
+_CLOCK_SKEW = timedelta(minutes=5)
+_RETENTION = timedelta(days=EMAIL_STATUS_RETENTION_DAYS)
+# Every row of a message carries the message's own created_at, so the window prunes to its partition.
+_EMAIL_STATUS_QUERY = f"""
+SELECT argMax(status, version), argMax(status_at, version), argMax(reason, version)
+FROM {EMAIL_STATUS_TABLE}
+WHERE team_id = %(team_id)s
+  AND message_id = %(message_id)s
+  AND created_at >= %(start)s
+  AND created_at < %(end)s
+GROUP BY message_id
+"""
+
+
+def queued_at(message_id: UUID) -> datetime | None:
+    """When the send was queued, or None when no send the table still keeps can have this ID."""
+    # The first 6 bytes of a UUIDT are the Unix time in milliseconds. They are compared as a number first, because a
+    # made-up ID can name a time far past the last year a datetime holds.
+    queued_ms = int.from_bytes(message_id.bytes[:6], "big")
+    now = datetime.now(UTC)
+    if not (now - _RETENTION).timestamp() * 1000 <= queued_ms <= (now + _CLOCK_SKEW).timestamp() * 1000:
+        return None
+    return datetime.fromtimestamp(queued_ms / 1000, tz=UTC)
+
+
+def get_email_status(team_id: int, message_id: UUID) -> EmailStatus | None:
+    """The send's status, or None when no send of this team has this ID or its first status has not arrived."""
+    created_at = queued_at(message_id)
+    if created_at is None:
+        return None
+    tag_queries(product=Product.MESSAGING, feature=Feature.QUERY)
+    rows = cast(
+        list,
+        sync_execute(
+            _EMAIL_STATUS_QUERY,
+            {
+                "team_id": team_id,
+                "message_id": str(message_id),
+                "start": created_at - _CLOCK_SKEW,
+                "end": created_at + _CLOCK_SKEW,
+            },
+            # A server holding a project secret API key polls this. No person waits on screen.
+            workload=Workload.OFFLINE,
+            team_id=team_id,
+        ),
+    )
+    if not rows:
+        return None
+    status, status_at, reason = rows[0]
+    if status not in EmailDeliveryStatus.values:
+        return None
+    return EmailStatus(
+        message_id=str(message_id),
+        status=EmailDeliveryStatus(status),
+        created_at=created_at,
+        status_at=status_at,
+        reason=reason or None,
+    )