feat(workflows): email admins on AWS SES tenant reputation changes

posthog/settings/ses.py

@@ -13,3 +13,10 @@
     SES_SECRET_ACCESS_KEY = os.getenv("SES_SECRET_ACCESS_KEY", "") or None
 
 SES_REGION = os.getenv("SES_REGION", "us-east-1")
+
+# SNS topics allowed to deliver SES tenant reputation events (EventBridge -> SNS -> webhook).
+# Empty (the default) leaves the webhook inert: the SNS signature proves a message came from AWS,
+# but only the allowlist proves it came from *our* topic.
+WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS: list[str] = [
+    arn.strip() for arn in os.getenv("WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS", "").split(",") if arn.strip()
+]

posthog/tasks/email.py

@@ -631,6 +631,41 @@ def send_email_sending_suspended(team_id: int, reason: str, suspended_at: str) -
     message.send()
 
 
+@shared_task(**EMAIL_TASK_KWARGS)
+@with_team_scope()
+def send_email_sending_reputation_finding(team_id: int, impact: str, found_at: str) -> None:
+    """
+    Warn a team's admins that our email provider raised reputation findings against their
+    project's sending. LOW impact is a fix-this warning; HIGH impact means sending can be
+    paused automatically. Not gated by notification settings — inaction escalates to a pause.
+    """
+    if not is_email_available(with_absolute_urls=True):
+        return
+    team = Team.objects.get(id=team_id)
+    memberships_to_email = _get_project_admins_to_notify_of_email_sending_suspension(team)
+    if not memberships_to_email:
+        return
+    high_impact = impact == "HIGH"
+    subject = (
+        f"[Action required] Email sending for project '{team}' is at risk of being paused"
+        if high_impact
+        else f"Reputation warning for email sending in project '{team}'"
+    )
+    message = EmailMessage(
+        campaign_key=f"email_sending_reputation_finding_{team_id}_{impact}_{found_at}",
+        subject=subject,
+        template_name="email_sending_reputation_finding",
+        template_context={
+            "team": team,
+            "high_impact": high_impact,
+            "reputation_path": f"/project/{team.id}/workflows/reputation",
+        },
+    )
+    for membership in memberships_to_email:
+        message.add_user_recipient(membership.user)
+    message.send()
+
+
 @shared_task(**EMAIL_TASK_KWARGS)
 @with_team_scope()
 def send_email_sending_unsuspended(team_id: int, unsuspended_at: str) -> None:

posthog/tasks/scheduled.py

@@ -116,6 +116,7 @@
     reap_stale_prewarm_heatmaps,
     report_stuck_heatmap_screenshots,
 )
+from products.workflows.backend.tasks.ses_tenant_state import reconcile_ses_tenant_states
 
 TWENTY_FOUR_HOURS = 24 * 60 * 60
 
@@ -236,6 +237,15 @@ def setup_periodic_tasks(sender: Celery, **kwargs: Any) -> None:
         name="team metadata expiry tracking cleanup",
     )
 
+    # 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.
+    sender.add_periodic_task(
+        crontab(hour="6", minute="30"),
+        reconcile_ses_tenant_states.s(),
+        name="ses tenant reputation reconciliation",
+    )
+
     # LLM gateway policy cache sync - hourly at :05 to stagger from team_metadata at :00
     sender.add_periodic_task(
         crontab(hour="*", minute="5"),

posthog/templates/email/email_sending_reputation_finding.html

@@ -0,0 +1,38 @@
+{% extends "email/base.html" %} {% load posthog_assets %}
+{% block heading %}{% if high_impact %}Email sending is at risk of being paused{% else %}Reputation warning for your email sending{% endif %}{% endblock %}
+{% block section %}
+<!-- prettier-ignore -->
+<p>
+    Our email provider has raised {% if high_impact %}<b>high-impact</b>{% else %}low-impact{% endif %} reputation findings against workflow email sending in project <b>{{ team }}</b>.
+</p>
+{% if high_impact %}
+<p>
+    At this level, email sending for your project can be <b>paused automatically</b> to protect deliverability. Act now to avoid an interruption.
+</p>
+{% else %}
+<p>
+    Sending still works, but if these findings are not fixed they can escalate and email sending for your project can be paused.
+</p>
+{% endif %}
+<p>
+    Open the Reputation tab to see each finding with the provider's recommended fix, and which workflows have the highest bounce or spam complaint rates. Typical fixes: remove old or purchased addresses from your lists, and stop sending to recipients who never engage.
+</p>
+<div class="mb mt text-center">
+    <a class="button" href="{% absolute_uri reputation_path %}">View findings</a>
+</div>
+<div class="mt">
+    <p>
+        If the button above doesn't work, paste this link into your browser:<br />
+        <a href="{% absolute_uri reputation_path %}">{% absolute_uri reputation_path %}</a>
+    </p>
+</div>
+{% endblock %}
+
+{% block footer %}
+Need help?
+<a href="https://posthog.com/questions?{{ utm_tags }}"
+    target="_blank"><b>Contact support</b></a>
+or
+<a href="https://posthog.com/docs?{{ utm_tags }}"
+    target="_blank"><b>read our documentation</b></a>.
+{% endblock %}

posthog/urls.py

@@ -82,6 +82,7 @@
 )
 from products.warehouse_sources.backend.presentation.views.public_source_configs import PublicSourceConfigViewSet
 from products.workflows.backend.api import hog_flow, hog_flow_template
+from products.workflows.backend.api.ses_events_webhook import ses_tenant_events_webhook
 
 from .utils import opt_slash_path, render_template
 from .views import (
@@ -675,6 +676,8 @@ def authorize_and_redirect(request: HttpRequest) -> HttpResponse:
     opt_slash_path("webhooks/github", github_webhook),
     # Stamphog runs as its own GitHub App with a dedicated inbound endpoint (not the fan-out above)
     opt_slash_path("webhooks/stamphog/github", stamphog_github_webhook),
+    # AWS SES tenant reputation events (EventBridge -> SNS HTTPS subscription)
+    opt_slash_path("webhooks/workflows/ses-events", ses_tenant_events_webhook),
     # Message preferences
     path("messaging-preferences/<str:token>/", preferences_page, name="message_preferences"),
     opt_slash_path("messaging-preferences/update", update_preferences, name="message_preferences_update"),

products/workflows/backend/api/ses_events_webhook.py

@@ -0,0 +1,106 @@
+"""
+Inbound webhook for AWS SES tenant reputation events, delivered via EventBridge → SNS HTTPS
+subscription. Events are treated as change signals only — the handler never trusts the payload's
+state; it enqueues a sync that reads the authoritative tenant state from the SES API. That makes
+duplicate, reordered, or partially-shaped events harmless.
+
+Authentication is the SNS message signature (proves "from AWS") plus a topic-ARN allowlist
+(proves "from our topic"). AWS-side wiring: an EventBridge rule on `source = aws.ses` for the
+tenant detail-types, targeting an SNS topic with an HTTPS subscription to this endpoint.
+"""
+
+import re
+import json
+from typing import Any
+
+from django.conf import settings
+from django.http import HttpRequest, HttpResponse
+from django.views.decorators.csrf import csrf_exempt
+
+import requests
+import structlog
+
+from products.workflows.backend.services.sns_verification import is_valid_sns_url, verify_sns_message
+from products.workflows.backend.tasks.ses_tenant_state import sync_ses_tenant_state_task
+
+logger = structlog.get_logger(__name__)
+
+_TENANT_TEAM_RE = re.compile(r"\bteam-(\d+)\b")
+
+
+def _extract_team_id(event: dict[str, Any]) -> int | None:
+    """Pull the `team-<id>` tenant name out of an EventBridge event, wherever AWS put it."""
+    detail = event.get("detail") or {}
+    candidates: list[Any] = [
+        detail.get("tenantName"),
+        detail.get("tenant-name"),
+        detail.get("reputationEntityReference"),
+    ]
+    resources = event.get("resources")
+    if isinstance(resources, list):
+        candidates.extend(resources)
+    for candidate in candidates:
+        if isinstance(candidate, str) and (match := _TENANT_TEAM_RE.search(candidate)):
+            return int(match.group(1))
+    return None
+
+
+@csrf_exempt
+def ses_tenant_events_webhook(request: HttpRequest) -> HttpResponse:
+    """Verify an SNS delivery and enqueue a tenant-state sync for the affected team."""
+    if request.method != "POST":
+        return HttpResponse(status=405)
+
+    allowed_topics = {arn for arn in settings.WORKFLOWS_SES_EVENTS_SNS_TOPIC_ARNS if arn}
+    if not allowed_topics:
+        logger.error("ses_tenant_events_webhook_not_configured")
+        return HttpResponse("Webhook not configured", status=500)
+
+    try:
+        # RecursionError: deeply nested JSON from an unauthenticated caller must 400, not 500.
+        message = json.loads(request.body)
+    except (json.JSONDecodeError, UnicodeDecodeError, RecursionError):
+        return HttpResponse("Invalid payload", status=400)
+    if not isinstance(message, dict):
+        return HttpResponse("Invalid payload", status=400)
+
+    if message.get("TopicArn") not in allowed_topics:
+        logger.warning("ses_tenant_events_webhook_unknown_topic", topic=message.get("TopicArn"))
+        return HttpResponse("Unknown topic", status=403)
+    if not verify_sns_message(message):
+        logger.warning("ses_tenant_events_webhook_invalid_signature", message_id=message.get("MessageId"))
+        return HttpResponse("Invalid signature", status=403)
+
+    message_type = message.get("Type")
+    if message_type == "SubscriptionConfirmation":
+        subscribe_url = message.get("SubscribeURL")
+        if not isinstance(subscribe_url, str) or not is_valid_sns_url(subscribe_url):
+            return HttpResponse("Invalid subscribe URL", status=400)
+        try:
+            requests.get(subscribe_url, timeout=5).raise_for_status()
+        except requests.RequestException:
+            # Non-2xx makes SNS retry the confirmation later.
+            logger.exception("ses_tenant_events_webhook_subscription_confirm_failed", topic=message.get("TopicArn"))
+            return HttpResponse("Subscription confirmation failed", status=502)
+        logger.info("ses_tenant_events_webhook_subscription_confirmed", topic=message.get("TopicArn"))
+        return HttpResponse(status=200)
+
+    if message_type != "Notification":
+        # UnsubscribeConfirmation and anything unknown: ack so SNS stops retrying.
+        return HttpResponse(status=200)
+
+    try:
+        event = json.loads(message.get("Message", ""))
+    except (json.JSONDecodeError, TypeError):
+        return HttpResponse(status=200)
+    if not isinstance(event, dict) or event.get("source") != "aws.ses":
+        return HttpResponse(status=200)
+
+    team_id = _extract_team_id(event)
+    if team_id is None:
+        logger.warning("ses_tenant_events_webhook_no_tenant", detail_type=event.get("detail-type"))
+        return HttpResponse(status=200)
+
+    # Ack fast; the sync fetches authoritative state and sends any transition emails.
+    sync_ses_tenant_state_task.delay(team_id)
+    return HttpResponse(status=202)

products/workflows/backend/migrations/0017_teamworkflowsconfig_ses_tenant_state.py

@@ -0,0 +1,27 @@
+# Generated by Django 5.2.14 on 2026-07-30 08:27
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("workflows", "0016_drop_emailreputationsnapshot_fk"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="teamworkflowsconfig",
+            name="ses_tenant_reputation_impact",
+            field=models.CharField(blank=True, db_default="", default="", max_length=8),
+        ),
+        migrations.AddField(
+            model_name="teamworkflowsconfig",
+            name="ses_tenant_sending_status",
+            field=models.CharField(blank=True, db_default="", default="", max_length=16),
+        ),
+        migrations.AddField(
+            model_name="teamworkflowsconfig",
+            name="ses_tenant_state_synced_at",
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+    ]

products/workflows/backend/migrations/max_migration.txt

@@ -1 +1 @@
-0016_drop_emailreputationsnapshot_fk
+0017_teamworkflowsconfig_ses_tenant_state

products/workflows/backend/models/team_workflows_config.py

@@ -35,5 +35,13 @@ class TeamWorkflowsConfig(models.Model):
     email_sending_suspended_at = models.DateTimeField(null=True, blank=True)
     email_sending_suspension_reason = models.TextField(blank=True, default="")
 
+    # Last-known state of the team's AWS SES tenant, mirrored so state *changes* can trigger
+    # customer emails exactly once (EventBridge events are best-effort; a periodic sweep
+    # reconciles). Empty string = never synced; the first sync sets a baseline without notifying.
+    # db_default so raw INSERTs from non-Django writers keep working.
+    ses_tenant_sending_status = models.CharField(max_length=16, blank=True, default="", db_default="")
+    ses_tenant_reputation_impact = models.CharField(max_length=8, blank=True, default="", db_default="")
+    ses_tenant_state_synced_at = models.DateTimeField(null=True, blank=True)
+
 
 register_team_extension_signal(TeamWorkflowsConfig, logger=logger)

products/workflows/backend/services/ses_tenant_state.py

@@ -0,0 +1,108 @@
+from collections.abc import Callable
+
+from django.db import transaction
+from django.utils import timezone
+
+import structlog
+
+from posthog.tasks.email import (
+    send_email_sending_reputation_finding,
+    send_email_sending_suspended,
+    send_email_sending_unsuspended,
+)
+
+from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig
+from products.workflows.backend.providers.ses import SESProvider
+
+logger = structlog.get_logger(__name__)
+
+PROVIDER_PAUSE_REASON = "Our email provider paused sending for this project because of high-impact reputation findings"
+
+_IMPACT_SEVERITY = {"": 0, "LOW": 1, "HIGH": 2}
+
+_STATE_FIELDS = ["ses_tenant_sending_status", "ses_tenant_reputation_impact", "ses_tenant_state_synced_at"]
+
+
+def sync_ses_tenant_state(team_id: int) -> None:
+    """
+    Fetch the authoritative AWS SES tenant state for a team and apply it. Called from the
+    EventBridge webhook (events only say "something changed" — the API is the source of truth)
+    and from the periodic reconciliation sweep (event delivery is best-effort).
+    """
+    tenant = SESProvider().get_tenant_reputation(team_id)
+    if tenant is None:
+        return
+    apply_ses_tenant_state(
+        team_id, sending_status=tenant["sending_status"], reputation_impact=tenant["reputation_impact"]
+    )
+
+
+def apply_ses_tenant_state(team_id: int, *, sending_status: str, reputation_impact: str | None) -> None:
+    """
+    Persist the tenant state and email the project's admins on meaningful transitions:
+    sending paused, sending re-enabled, or reputation findings escalating. The stored state is
+    the dedupe mechanism — the row is locked for the read-compare-write so overlapping syncs
+    (webhook + sweep, or duplicate events) serialize, and the loser sees an unchanged state.
+    """
+    impact = reputation_impact or ""
+    TeamWorkflowsConfig.objects.get_or_create(team_id=team_id)
+
+    with transaction.atomic():
+        config = TeamWorkflowsConfig.objects.select_for_update().get(team_id=team_id)
+        previous_status = config.ses_tenant_sending_status
+        previous_impact = config.ses_tenant_reputation_impact
+        if previous_status == sending_status and previous_impact == impact:
+            config.ses_tenant_state_synced_at = timezone.now()
+            config.save(update_fields=["ses_tenant_state_synced_at"])
+            return
+
+        now = timezone.now()
+        config.ses_tenant_sending_status = sending_status
+        config.ses_tenant_reputation_impact = impact
+        config.ses_tenant_state_synced_at = now
+        config.save(update_fields=_STATE_FIELDS)
+        logger.info(
+            "SES tenant state changed",
+            team_id=team_id,
+            from_status=previous_status,
+            to_status=sending_status,
+            from_impact=previous_impact,
+            to_impact=impact,
+        )
+
+        notify = _pick_notification(
+            previous_status=previous_status,
+            previous_impact=previous_impact,
+            sending_status=sending_status,
+            impact=impact,
+            team_id=team_id,
+            now_iso=now.isoformat(),
+        )
+        if notify is not None:
+            # Dispatch after commit so a rollback can't leave an email claiming a state
+            # that was never persisted.
+            transaction.on_commit(notify)
+
+
+def _pick_notification(
+    *, previous_status: str, previous_impact: str, sending_status: str, impact: str, team_id: int, now_iso: str
+) -> Callable[[], None] | None:
+    if previous_status == "":
+        # First sync is baseline adoption, not a transition — healthy teams stay silent so
+        # rollout doesn't mass-email. But a tenant that is ALREADY paused or already carries
+        # high-impact findings is exactly who this feature exists to tell, so those do notify.
+        if sending_status == "DISABLED":
+            return lambda: send_email_sending_suspended.delay(team_id, PROVIDER_PAUSE_REASON, now_iso)
+        if impact == "HIGH":
+            return lambda: send_email_sending_reputation_finding.delay(team_id, impact, now_iso)
+        return None
+
+    if sending_status == "DISABLED" and previous_status != "DISABLED":
+        return lambda: send_email_sending_suspended.delay(team_id, PROVIDER_PAUSE_REASON, now_iso)
+    if previous_status == "DISABLED" and sending_status in ("ENABLED", "REINSTATED"):
+        return lambda: send_email_sending_unsuspended.delay(team_id, now_iso)
+    if sending_status != "DISABLED" and _IMPACT_SEVERITY.get(impact, 0) > _IMPACT_SEVERITY.get(previous_impact, 0):
+        # Escalation only, and only while still sending — a paused tenant already got the
+        # suspension email, and de-escalations don't need action.
+        return lambda: send_email_sending_reputation_finding.delay(team_id, impact, now_iso)
+    return None