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

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

products/cdp/backend/facade/api.py

@@ -21,14 +21,22 @@
 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."
 
 _LAZY = {"HogFunctionSerializer": "api.hog_function"}
 
 __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",

products/cdp/backend/services/invocation_lifecycle.py

@@ -0,0 +1,86 @@
+"""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
+    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.
+    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_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,
+                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],
+        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],
+    )

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,21 @@ 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"
+    SENT = "sent", "Sent"
+    FAILED = "failed", "Failed"
+
+
+@frozen
+class EmailStatus:
+    message_id: str
+    status: EmailDeliveryStatus
+    created_at: datetime
+    status_at: datetime
+    # 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

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,131 @@
+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
+from products.workflows.backend.facade.api import WorkflowAccessDenied, WorkflowNotFound
+from products.workflows.backend.facade.workflows import get_workflow
+
+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.",
+    )
+    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: 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.",
+    )
+
+
+# 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]:
+        # 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(),
+            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 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: `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:
+        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()
+        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

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"])