diff --git a/products/workflows/backend/facade/contracts.py b/products/workflows/backend/facade/contracts.py index 1be9ba6e03e4..0fd16302f888 100644 --- a/products/workflows/backend/facade/contracts.py +++ b/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 diff --git a/products/workflows/backend/presentation/views/hog_flow.py b/products/workflows/backend/presentation/views/hog_flow.py index 4c36a9122e9f..1fb4206d74d2 100644 --- a/products/workflows/backend/presentation/views/hog_flow.py +++ b/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", ] diff --git a/products/workflows/backend/services/hog_flow_reads.py b/products/workflows/backend/services/hog_flow_reads.py index 7d41c627823f..68370e29506e 100644 --- a/products/workflows/backend/services/hog_flow_reads.py +++ b/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 diff --git a/products/workflows/backend/services/integration_usage.py b/products/workflows/backend/services/integration_usage.py index 007993176e32..dd6d380f19ba 100644 --- a/products/workflows/backend/services/integration_usage.py +++ b/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: diff --git a/products/workflows/backend/tests/api/test_hog_flow.py b/products/workflows/backend/tests/api/test_hog_flow.py index 9575654dc5ac..69b076bcd43e 100644 --- a/products/workflows/backend/tests/api/test_hog_flow.py +++ b/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): diff --git a/products/workflows/frontend/Broadcasts/broadcastWizardLogic.test.ts b/products/workflows/frontend/Broadcasts/broadcastWizardLogic.test.ts index 71fee2b3e5f7..ec126a165b97 100644 --- a/products/workflows/frontend/Broadcasts/broadcastWizardLogic.test.ts +++ b/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: [], } } diff --git a/products/workflows/frontend/Workflows/hogflows/types.ts b/products/workflows/frontend/Workflows/hogflows/types.ts index 2fa6f7dcf182..353466e8a805 100644 --- a/products/workflows/frontend/Workflows/hogflows/types.ts +++ b/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 { 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 diff --git a/products/workflows/frontend/generated/api.schemas.ts b/products/workflows/frontend/generated/api.schemas.ts index 594776b2e704..54d09bbaf081 100644 --- a/products/workflows/frontend/generated/api.schemas.ts +++ b/products/workflows/frontend/generated/api.schemas.ts @@ -659,6 +659,13 @@ export interface HogFlowScheduleApi { readonly updated_at: string } +export interface HogFlowUnverifiedEmailSenderApi { + /** The email sender's integration id. */ + readonly id: number + /** The sender's domain, which is not verified yet. */ + readonly domain: string +} + /** * Mixin for serializers to add user access control fields */ @@ -770,6 +777,8 @@ export interface HogFlowApi { * @nullable */ readonly went_live_at: string | null + /** 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. */ + readonly unverified_email_senders: readonly HogFlowUnverifiedEmailSenderApi[] /** Newest task this loop workflow created, as its last run. Null when the workflow is not a loop or has not run. */ readonly last_run: HogFlowLastRunApi | null } @@ -889,6 +898,8 @@ export interface HogFlowUpdateApi { * @nullable */ readonly went_live_at: string | null + /** 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. */ + readonly unverified_email_senders: readonly HogFlowUnverifiedEmailSenderApi[] /** Newest task this loop workflow created, as its last run. Null when the workflow is not a loop or has not run. */ readonly last_run: HogFlowLastRunApi | null } @@ -1008,6 +1019,8 @@ export interface PatchedHogFlowUpdateApi { * @nullable */ readonly went_live_at?: string | null + /** 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. */ + readonly unverified_email_senders?: readonly HogFlowUnverifiedEmailSenderApi[] /** Newest task this loop workflow created, as its last run. Null when the workflow is not a loop or has not run. */ readonly last_run?: HogFlowLastRunApi | null } diff --git a/products/workflows/frontend/generated/api.zod.ts b/products/workflows/frontend/generated/api.zod.ts index a272f008dc1e..f85c156e0697 100644 --- a/products/workflows/frontend/generated/api.zod.ts +++ b/products/workflows/frontend/generated/api.zod.ts @@ -2549,6 +2549,16 @@ export const HogFlowsInvocationsCreateBody = /* @__PURE__ */ zod.object({ .describe( "When the workflow last went live. Earlier than enable_requested_at, or null, while the workflow waits for its email sender's domain." ), + unverified_email_senders: zod + .array( + zod.object({ + id: zod.number().describe("The email sender's integration id."), + domain: zod.string().describe("The sender's domain, which is not verified yet."), + }) + ) + .describe( + '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.' + ), last_run: zod .union([ zod.object({ diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index 83da412effb3..e3cc554a698c 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -58682,6 +58682,13 @@ export namespace Schemas { readonly updated_at: string; } + export interface HogFlowUnverifiedEmailSender { + /** The email sender's integration id. */ + readonly id: number; + /** The sender's domain, which is not verified yet. */ + readonly domain: string; + } + export interface HogFlowLastRun { /** The task this run belongs to. */ readonly task_id: string; @@ -58802,6 +58809,8 @@ export namespace Schemas { * @nullable */ readonly went_live_at: string | null; + /** 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. */ + readonly unverified_email_senders: readonly HogFlowUnverifiedEmailSender[]; /** Newest task this loop workflow created, as its last run. Null when the workflow is not a loop or has not run. */ readonly last_run: HogFlowLastRun | null; } @@ -59426,6 +59435,8 @@ export namespace Schemas { * @nullable */ readonly went_live_at: string | null; + /** 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. */ + readonly unverified_email_senders: readonly HogFlowUnverifiedEmailSender[]; /** Newest task this loop workflow created, as its last run. Null when the workflow is not a loop or has not run. */ readonly last_run: HogFlowLastRun | null; } @@ -87504,6 +87515,8 @@ export namespace Schemas { * @nullable */ readonly went_live_at?: string | null; + /** 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. */ + readonly unverified_email_senders?: readonly HogFlowUnverifiedEmailSender[]; /** Newest task this loop workflow created, as its last run. Null when the workflow is not a loop or has not run. */ readonly last_run?: HogFlowLastRun | null; }