fix(workflows): refresh pending email sender verification

posthog/api/test/test_integration.py

@@ -666,6 +666,7 @@ def test_integration_from_domain(
             "reply_to": "support@posthog.com",
             "verified": False,
             "provider": "ses",
+            "verification_requested_at": ANY,
         }
         assert integration.sensitive_config == {}
         assert integration.created_by == self.user
@@ -724,7 +725,9 @@ def test_email_verify_returns_ses_result(self, mock_create_email_domain, mock_ve
             created_by=self.user,
         )
         email_integration = EmailIntegration(integration)
-        verification_result = email_integration.verify()
+        clicked_at = timezone.now() + timedelta(hours=2)
+        with time_machine.travel(clicked_at, tick=False):
+            verification_result = email_integration.verify()
 
         assert verification_result == expected_result
 
@@ -741,6 +744,7 @@ def test_email_verify_returns_ses_result(self, mock_create_email_domain, mock_ve
             "reply_to": "",
             "verified": False,
             "provider": "ses",
+            "verification_requested_at": int(clicked_at.timestamp()),
         }
 
     @patch("products.workflows.backend.facade.api.verify_ses_email_domain")
@@ -778,6 +782,7 @@ def test_email_verify_updates_integration(self, mock_create_email_domain, mock_v
             "reply_to": "",
             "verified": True,
             "provider": "ses",
+            "verification_requested_at": ANY,
         }
 
     @patch("products.workflows.backend.facade.api.verify_ses_email_domain")
@@ -802,7 +807,13 @@ def test_email_verify_updates_all_other_integrations_with_same_domain(
             created_by=self.user,
         )
         integration2 = EmailIntegration.create_native_integration(
-            {**self.valid_config, "provider": "ses"},
+            {**self.valid_config, "email": "other@posthog.com", "provider": "ses"},
+            team_id=self.team.id,
+            organization_id=self.organization.id,
+            created_by=self.user,
+        )
+        integrationOtherMailFrom = EmailIntegration.create_native_integration(
+            {**self.valid_config, "email": "bounces@posthog.com", "mail_from_subdomain": "mail", "provider": "ses"},
             team_id=self.team.id,
             organization_id=self.organization.id,
             created_by=self.user,
@@ -829,10 +840,12 @@ def test_email_verify_updates_all_other_integrations_with_same_domain(
 
         integration1.refresh_from_db()
         integration2.refresh_from_db()
+        integrationOtherMailFrom.refresh_from_db()
         integrationOtherDomain.refresh_from_db()
 
         assert integration1.config["verified"]
         assert integration2.config["verified"]
+        assert not integrationOtherMailFrom.config["verified"]
         assert not integrationOtherDomain.config["verified"]
 
 

posthog/models/integration/__init__.py

@@ -38,7 +38,7 @@
     dot_get,
 )
 from .databricks import DatabricksIntegration, DatabricksIntegrationError
-from .email import EmailIntegration, cleanup_ses_identity_on_integration_delete
+from .email import EmailIntegration, cleanup_ses_identity_on_integration_delete, mark_verification_requested
 from .external_issues import (
     SUPPORTED_EXTERNAL_ISSUE_PROVIDERS,
     external_issue_url,
@@ -276,4 +276,5 @@
     "RedshiftIntegration",
     "SnowflakeIntegrationError",
     "SnowflakeIntegration",
+    "mark_verification_requested",
 ]

posthog/models/integration/email.py

@@ -8,6 +8,7 @@
 from django.conf import settings
 from django.core.cache import cache
 from django.db import models, transaction
+from django.db.models import F, Func, JSONField, Q, Value
 from django.dispatch import receiver
 from django.utils import timezone
 
@@ -39,6 +40,23 @@ class EmailDomainStatusReport:
     checked_at: datetime
 
 
+VERIFICATION_REQUESTED_AT = "verification_requested_at"
+
+
+def _with_config_key(key: str, value: object) -> Func:
+    return Func(
+        F("config"),
+        Value([key]),
+        Value(value, output_field=JSONField()),
+        function="jsonb_set",
+        output_field=JSONField(),
+    )
+
+
+def mark_verification_requested(senders: "models.QuerySet[model.Integration]") -> int:
+    return senders.update(config=_with_config_key(VERIFICATION_REQUESTED_AT, int(timezone.now().timestamp())))
+
+
 class EmailIntegration:
     integration: model.Integration
 
@@ -100,6 +118,7 @@ def create_native_integration(
                     "name": name,
                     "provider": provider,
                     "verified": True if provider == "maildev" else False,
+                    VERIFICATION_REQUESTED_AT: int(timezone.now().timestamp()),
                 },
                 "created_by": created_by,
             },
@@ -151,6 +170,26 @@ def update_native_integration(self, config: dict, team_id: int) -> model.Integra
 
         return self.integration
 
+    def apply_verification(self, verification_result: "EmailDomainVerification") -> "EmailDomainVerification":
+        if verification_result.get("status") == "success":
+            self._mark_domain_verified(on_domain_verified=None)
+        return verification_result
+
+    def _domain_senders(self) -> "models.QuerySet[model.Integration]":
+        return model.Integration.objects.filter(
+            team_id=self.integration.team_id,
+            kind="email",
+            config__domain=self.integration.config.get("domain"),
+            config__provider=self.integration.config.get("provider", "ses"),
+        )
+
+    def _matching_senders(self) -> "models.QuerySet[model.Integration]":
+        mail_from_subdomain = self.integration.config.get("mail_from_subdomain", "feedback")
+        matching_mail_from = Q(config__mail_from_subdomain=mail_from_subdomain)
+        if mail_from_subdomain == "feedback":
+            matching_mail_from |= Q(config__mail_from_subdomain__isnull=True)
+        return self._domain_senders().filter(matching_mail_from)
+
     def verify(self, *, on_domain_verified: Callable[[], None] | None = None) -> "EmailDomainVerification":
         domain = self.integration.config.get("domain")
         provider = self.integration.config.get("provider", "ses")
@@ -180,6 +219,8 @@ def verify(self, *, on_domain_verified: Callable[[], None] | None = None) -> "Em
 
         if verification_result.get("status") == "success":
             self._mark_domain_verified(on_domain_verified)
+        else:
+            mark_verification_requested(model.Integration.objects.filter(id=self.integration.id))
 
         return verification_result
 
@@ -248,39 +289,34 @@ def _check_domain(self) -> "EmailDomainCheck":
 
     def _mark_domain_verified(self, on_domain_verified: Callable[[], None] | None) -> None:
         with transaction.atomic():
-            senders = list(
-                model.Integration.objects.select_for_update()
-                .filter(
-                    team_id=self.integration.team_id,
-                    kind="email",
-                    config__domain=self.integration.config.get("domain"),
-                    config__provider=self.integration.config.get("provider", "ses"),
+            senders = list(self._matching_senders().select_for_update().order_by("id"))
+            sender_ids = [sender.id for sender in senders]
+            unverified_ids = [sender.id for sender in senders if not sender.config.get("verified")]
+            domain_was_verified = self._domain_senders().filter(config__verified=True).exists()
+            model.Integration.objects.filter(id__in=unverified_ids).update(config=_with_config_key("verified", True))
+            # A concurrent edit can move this sender off the checked MAIL FROM subdomain, so it may not be among them.
+            verifies_this_sender = self.integration.id in sender_ids
+            if verifies_this_sender:
+                from products.workflows.backend.facade.api import (
+                    set_default_email_sender_if_unset,  # noqa: PLC0415 — keeps the workflows facade off the model import path
                 )
-                .order_by("id")
-            )
-            unverified_senders = [sender for sender in senders if not sender.config.get("verified")]
-            for sender in unverified_senders:
-                sender.config["verified"] = True
-                sender.save()
 
-        self.integration.config["verified"] = True
-
-        from products.workflows.backend.facade.api import (
-            set_default_email_sender_if_unset,  # noqa: PLC0415 — keeps the workflows facade off the model import path
-        )
-
-        set_default_email_sender_if_unset(self.integration.team_id, self.integration.id)
+                set_default_email_sender_if_unset(self.integration.team_id, self.integration.id)
 
+        if sender_ids:
+            self._reload_senders_on_workers(sender_ids)
+        if not verifies_this_sender:
+            return
+        self.integration.config["verified"] = True
         # A sender added to a domain that is already verified joins it, so only the first verification counts.
-        if on_domain_verified and len(unverified_senders) == len(senders):
+        if on_domain_verified and not domain_was_verified:
             on_domain_verified()
-        self._reload_senders_on_workers(senders)
 
     # Workers also refresh integrations on their own schedule, so a failed reload only delays sending.
     # Failing the request instead would hide a verification that is already saved.
-    def _reload_senders_on_workers(self, senders: list[model.Integration]) -> None:
+    def _reload_senders_on_workers(self, sender_ids: list[int]) -> None:
         try:
-            reload_integrations_on_workers(self.integration.team_id, [sender.id for sender in senders])
+            reload_integrations_on_workers(self.integration.team_id, sender_ids)
         except Exception as error:
             capture_exception(error, {"team_id": self.integration.team_id})
 

posthog/tasks/scheduled.py

@@ -159,6 +159,8 @@
     poll_ses_reputation_findings,
     recompute_workflows_email_sending_tiers,
     reconcile_ses_tenant_states,
+    refresh_new_email_senders,
+    refresh_settled_email_senders,
     sweep_workflow_email_deliverability,
 )
 
@@ -339,6 +341,22 @@ def setup_periodic_tasks(sender: Celery, **kwargs: Any) -> None:
         expires_seconds=12 * 3600,
     )
 
+    add_periodic_task_with_expiry(
+        sender,
+        crontab(minute="*"),
+        refresh_new_email_senders.s(),
+        name="refresh new workflow email senders",
+        expires_seconds=60,
+    )
+
+    add_periodic_task_with_expiry(
+        sender,
+        crontab(minute="*/5"),
+        refresh_settled_email_senders.s(),
+        name="refresh settled workflow email senders",
+        expires_seconds=300,
+    )
+
     # SES tenant reputation reconciliation - daily at 6:30 AM UTC. EventBridge events are the
     # real-time path; this sweep catches missed deliveries. Sequential SES API calls per team
     # with an SES email integration, so kept daily to stay well inside SES API rate limits.

products/workflows/backend/facade/tasks.py

@@ -1,4 +1,8 @@
 from products.workflows.backend.tasks.batch_jobs import resume_batch_jobs_waiting_for_realtime_cohort
+from products.workflows.backend.tasks.email_sender_verification import (
+    refresh_new_email_senders,
+    refresh_settled_email_senders,
+)
 from products.workflows.backend.tasks.email_sending_tiers import recompute_workflows_email_sending_tiers
 from products.workflows.backend.tasks.hog_flows import refresh_affected_hog_flows
 from products.workflows.backend.tasks.ses_account_reputation import (
@@ -15,5 +19,7 @@
     "recompute_workflows_email_sending_tiers",
     "reconcile_ses_tenant_states",
     "refresh_affected_hog_flows",
+    "refresh_new_email_senders",
+    "refresh_settled_email_senders",
     "sweep_workflow_email_deliverability",
 ]