feat(workflows): list a workflow's unverified email senders on its read

products/workflows/backend/facade/contracts.py

@@ -599,6 +599,14 @@ class WorkflowHasNoDraft(Exception):
     """Publish found no staged draft."""
 
 
+@frozen
+class UnverifiedEmailSender:
+    """An email sender of a workflow whose domain is not verified yet."""
+
+    id: int
+    domain: str
+
+
 @frozen
 class Workflow:
     """A workflow as the API reads it.
@@ -649,6 +657,8 @@ class Workflow:
     # Read on the list only. A single-workflow read leaves both None and ``schedules`` filled.
     pending_suggestions: int | None = None
     suggestions_enabled: bool | None = None
+    # Read for a single workflow only. The list leaves it empty.
+    unverified_email_senders: tuple[UnverifiedEmailSender, ...] = ()
 
 
 @frozen

products/workflows/backend/presentation/views/hog_flow.py

@@ -3281,6 +3281,11 @@ class Meta(HogFlowMinimalSerializer.Meta):
         read_only_fields = fields
 
 
+class HogFlowUnverifiedEmailSenderSerializer(serializers.Serializer):
+    id = serializers.IntegerField(read_only=True, help_text="The email sender's integration id.")
+    domain = serializers.CharField(read_only=True, help_text="The sender's domain, which is not verified yet.")
+
+
 class HogFlowSerializer(HogFlowMinimalSerializer):
     origin_product = serializers.ChoiceField(
         choices=HogFlowOriginProduct.choices,
@@ -3470,6 +3475,14 @@ def get_email_sending_pause_requires_support(self, hog_flow: Workflow) -> bool:
             "waits for its email sender's domain."
         ),
     )
+    unverified_email_senders = HogFlowUnverifiedEmailSenderSerializer(
+        many=True,
+        read_only=True,
+        help_text=(
+            "The email senders of this workflow whose domain is not verified yet. Enabling the workflow while "
+            "this list is not empty makes it wait for those domains. Not returned by the list endpoint."
+        ),
+    )
 
     def validate_status(self, value: str) -> str:
         instance = cast(Optional[WorkflowEditState], self.instance)
@@ -3665,6 +3678,7 @@ class Meta:
             "email_sending_resumed_at",
             "enable_requested_at",
             "went_live_at",
+            "unverified_email_senders",
             "last_run",
             "tags",
         ]
@@ -3692,6 +3706,7 @@ class Meta:
             "email_sending_resumed_at",
             "enable_requested_at",
             "went_live_at",
+            "unverified_email_senders",
             "last_run",
         ]
 

products/workflows/backend/services/hog_flow_reads.py

@@ -18,6 +18,7 @@
 from posthog.models import TaggedItem
 
 from products.workflows.backend.facade.contracts import (
+    UnverifiedEmailSender,
     Workflow,
     WorkflowAccessDenied,
     WorkflowArchived,
@@ -36,6 +37,7 @@
 from products.workflows.backend.services.batch_jobs import hog_flow_ids_with_broadcast_status
 from products.workflows.backend.services.hog_flow_schedules import list_schedules_oldest_first
 from products.workflows.backend.services.hog_flow_secrets import TemplateCache, mask_workflow_fields
+from products.workflows.backend.services.integration_usage import email_senders_of, unverified_sender_domains
 from products.workflows.backend.services.publish_impact import build_publish_impact
 
 if TYPE_CHECKING:
@@ -53,7 +55,7 @@ def get_workflow(
     it is below `required_level`. Pass None for both access arguments to skip the check, as a
     service credential does."""
     flow = _checked_flow(team_id, workflow_id, user_access_control, required_level, created_by=True)
-    return _to_workflow(flow, user_access_control, template_cache={}, with_schedules=True)
+    return _to_workflow(flow, user_access_control, template_cache={}, single=True)
 
 
 def get_workflow_ref(
@@ -142,7 +144,7 @@ def to_edit_state(flow: HogFlow) -> WorkflowEditState:
 def workflow_from_fields(*, fields: Mapping[str, object], user_access_control: "UserAccessControl | None") -> Workflow:
     """The workflow a write left behind, built from the field values the write returned rather than
     read again, so a write another request commits afterwards does not show in this response."""
-    return _to_workflow(HogFlow(**fields), user_access_control, template_cache={}, with_schedules=True)
+    return _to_workflow(HogFlow(**fields), user_access_control, template_cache={}, single=True)
 
 
 def workflow_publish_impact(
@@ -229,7 +231,7 @@ def list_workflows(
     template_cache: TemplateCache = {}
     return WorkflowPage(
         count=count,
-        results=[_to_workflow(flow, user_access_control, template_cache, with_schedules=False) for flow in flows],
+        results=[_to_workflow(flow, user_access_control, template_cache, single=False) for flow in flows],
     )
 
 
@@ -357,8 +359,9 @@ def _to_workflow(
     user_access_control: "UserAccessControl | None",
     template_cache: TemplateCache,
     *,
-    with_schedules: bool,
+    single: bool,
 ) -> Workflow:
+    """``single`` reads what only a single-workflow response carries: the schedules and the unverified senders."""
     masked: dict[str, object] = {"actions": flow.actions, "trigger": flow.trigger, "draft": flow.draft}
     mask_workflow_fields(
         masked,
@@ -392,9 +395,7 @@ def _to_workflow(
         abort_action=flow.abort_action,
         variables=flow.variables,
         billable_action_types=flow.billable_action_types,
-        schedules=tuple(list_schedules_oldest_first(team_id=flow.team_id, hog_flow_id=flow.id))
-        if with_schedules
-        else (),
+        schedules=tuple(list_schedules_oldest_first(team_id=flow.team_id, hog_flow_id=flow.id)) if single else (),
         draft=cast("dict | None", masked["draft"]),
         draft_updated_at=flow.draft_updated_at,
         action_redirects=flow.action_redirects,
@@ -408,9 +409,18 @@ def _to_workflow(
         tags=tuple(sorted(current_tag_names(flow))),
         pending_suggestions=getattr(flow, "pending_suggestions", None),
         suggestions_enabled=getattr(flow, "suggestions_enabled", None),
+        unverified_email_senders=_unverified_email_senders(flow) if single else (),
     )
 
 
+def _unverified_email_senders(flow: HogFlow) -> tuple[UnverifiedEmailSender, ...]:
+    # Transactional email workflows refuse an unverified sender on their own, so they never wait for one.
+    if flow.origin_product == HogFlow.OriginProduct.TRANSACTIONAL_EMAILS:
+        return ()
+    domains = unverified_sender_domains(team_id=flow.team_id, sender_ids=email_senders_of(flow))
+    return tuple(UnverifiedEmailSender(id=sender_id, domain=domain) for sender_id, domain in sorted(domains.items()))
+
+
 # A workflow's type is what owns it, else what it does. `loop` and `broadcast` name the surfaces that
 # have their own page, and the behavioural values exclude them: a flow those surfaces own is tagged
 # by surface in the UI (see WorkflowTypeTag), so returning it under `messaging` would contradict the

products/workflows/backend/services/integration_usage.py

@@ -1,6 +1,8 @@
 from typing import Any
 from uuid import UUID
 
+from django.db.models import QuerySet
+
 from posthog.models.integration import Integration
 
 from products.cdp.backend.models.hog_function_template import HogFunctionTemplate
@@ -108,18 +110,42 @@ def email_senders_of(flow: HogFlow) -> frozenset[int]:
     )
 
 
+def _email_integrations(team_id: int, sender_ids: frozenset[int]) -> "QuerySet[Integration]":
+    return Integration.objects.filter(team_id=team_id, kind="email", id__in=sender_ids).only("id", "config")
+
+
+def _is_unverified(sender: Integration) -> bool:
+    return not (sender.config or {}).get("verified")
+
+
 def lock_unverified_senders(*, team_id: int, sender_ids: frozenset[int]) -> frozenset[int]:
     """The senders among ``sender_ids`` that exist and are not verified yet. A sender that no longer exists
     can never verify, so it does not hold a workflow back. Locks the senders until the caller's transaction
     ends, so a verification of them waits for the caller's status write and then finds it."""
     if not sender_ids:
         return frozenset()
-    senders = Integration.objects.filter(team_id=team_id, kind="email", id__in=sender_ids).only("id", "config")
     return frozenset(
-        sender.id for sender in senders.order_by("id").select_for_update() if not (sender.config or {}).get("verified")
+        sender.id
+        for sender in _email_integrations(team_id, sender_ids).order_by("id").select_for_update()
+        if _is_unverified(sender)
     )
 
 
+def unverified_sender_domains(*, team_id: int, sender_ids: frozenset[int]) -> dict[int, str]:
+    """The domain of each sender among ``sender_ids`` that exists and is not verified yet, keyed by sender id."""
+    if not sender_ids:
+        return {}
+    return {
+        sender.id: _sender_domain(sender.config or {})
+        for sender in _email_integrations(team_id, sender_ids)
+        if _is_unverified(sender)
+    }
+
+
+def _sender_domain(config: dict) -> str:
+    return config.get("domain") or str(config.get("email") or "").rpartition("@")[2]
+
+
 def waiting_flows_using(*, team_id: int, sender_ids: frozenset[int]) -> list[UUID]:
     """The waiting workflows of the team that send email from any of ``sender_ids``."""
     if not sender_ids:

products/workflows/backend/tests/api/test_hog_flow.py

@@ -6614,6 +6614,37 @@ def test_enable_saves_the_status_its_senders_allow(self, _name, verified, flag_o
             assert body["went_live_at"] == body["enable_requested_at"]
             assert [event["activated_by"] for event in self._activated_events()] == ["user"]
 
+    def test_the_detail_lists_unverified_senders_until_they_verify(self):
+        sender = self._sender(verified=False)
+        created = self._create(self._email_flow(sender))
+        assert created.status_code == 201, created.json()
+        detail_url = f"/api/projects/{self.team.id}/hog_flows/{created.json()['id']}"
+
+        assert created.json()["unverified_email_senders"] == [{"id": sender.id, "domain": "example.com"}]
+        assert self.client.get(detail_url).json()["unverified_email_senders"] == [
+            {"id": sender.id, "domain": "example.com"}
+        ]
+        listed = self.client.get(f"/api/projects/{self.team.id}/hog_flows").json()["results"]
+        assert "unverified_email_senders" not in listed[0]
+
+        sender.config = {**sender.config, "verified": True}
+        sender.save()
+
+        assert self.client.get(detail_url).json()["unverified_email_senders"] == []
+
+    @parameterized.expand([("a_string", "not an email object"), ("a_list", ["to", "from"])])
+    def test_a_malformed_email_step_reads_as_having_no_unverified_sender(self, _name, malformed_value):
+        flow_id = self._create_draft(self._sender(verified=False))
+        flow = HogFlow.objects.get(pk=flow_id)
+        actions = deepcopy(flow.actions)
+        actions[1]["config"]["inputs"]["email"]["value"] = malformed_value
+        HogFlow.objects.filter(pk=flow_id).update(actions=actions)
+
+        response = self.client.get(f"/api/projects/{self.team.id}/hog_flows/{flow_id}")
+
+        assert response.status_code == 200, response.json()
+        assert response.json()["unverified_email_senders"] == []
+
     def test_a_sender_rotation_waits_for_its_unverified_sender(self):
         verified, unverified = self._sender(verified=True), self._sender(verified=False)
         payload = self._email_flow(verified)
@@ -6627,6 +6658,7 @@ def test_a_sender_rotation_waits_for_its_unverified_sender(self):
         with self._flag(True):
             response = self._set_status(created.json()["id"], "active")
 
+        assert created.json()["unverified_email_senders"] == [{"id": unverified.id, "domain": "example.com"}]
         assert response.json()["status"] == "waiting_for_domain"
 
     def test_enabling_a_waiting_workflow_again_once_its_sender_verified_activates_it_once(self):

products/workflows/frontend/Broadcasts/broadcastWizardLogic.test.ts

@@ -67,6 +67,7 @@ function savedBroadcast(overrides: { name: string; subject: string; updatedAt: s
         email_sending_resumed_at: null,
         enable_requested_at: null,
         went_live_at: null,
+        unverified_email_senders: [],
     }
 }
 

products/workflows/frontend/Workflows/hogflows/types.ts

@@ -6,7 +6,7 @@ import { EmailFieldErrors } from 'scenes/hog-functions/email-templater/types'
 
 import { AccessControlLevel, UserBasicType } from '~/types'
 
-import { HogFlowOriginProductEnumApi } from '../../generated/api.schemas'
+import { HogFlowOriginProductEnumApi, HogFlowUnverifiedEmailSenderApi } from '../../generated/api.schemas'
 import { CyclotronJobInputSchemaTypeSchema, HogFlowActionSchema, HogFlowTriggerSchema } from './steps/types'
 
 const HogFlowEdgeSchema = z.object({
@@ -112,6 +112,8 @@ export interface HogFlow extends z.infer<typeof HogFlowSchema> {
     enable_requested_at?: string | null
     // Later than enable_requested_at when the workflow went live after waiting for its email domain.
     went_live_at?: string | null
+    // Senders whose domain is not verified yet. Only a single-workflow read carries it, not the list.
+    unverified_email_senders?: HogFlowUnverifiedEmailSenderApi[]
     // Suggested changes waiting for a person. Only the list endpoint counts it.
     pending_suggestions?: number | null
     suggestions_enabled?: boolean | null