feat(workflows): show email sending health per mailbox provider

frontend/src/lib/components/Sparkline.tsx

@@ -2,7 +2,7 @@ import clsx from 'clsx'
 import { useCallback, useMemo } from 'react'
 
 import { DefaultTooltip, Sparkline as QuillSparklineChart, useChartTheme } from '@posthog/quill-charts'
-import type { Series, TooltipContext } from '@posthog/quill-charts'
+import type { Series, TooltipContext, ValueDomain } from '@posthog/quill-charts'
 
 import { getColorVar } from 'lib/colors'
 import { humanFriendlyNumber } from 'lib/utils/numbers'
@@ -41,6 +41,30 @@ export interface SparklineProps {
     sortTooltipByCount?: boolean
     /** Format the per-series tooltip value. Defaults to `humanFriendlyNumber`. */
     renderTooltipValue?: (value: number) => string
+    /**
+     * X-axis value range to highlight as a translucent box behind the bars. Values are
+     * in the x-axis's own units: epoch ms for a time scale (positioned with sub-bar
+     * precision), or a label for a category scale. Used to mirror an external selection
+     * (e.g. the rows currently visible in a paired virtualized list) onto the chart.
+     * Callers pass an already-ordered `xMin <= xMax`; pass `null`/`undefined` to clear.
+     */
+    highlightedRange?: { xMin: number | string; xMax: number | string } | null
+    /**
+     * Bar indices that are still being ingested (incomplete). Those bars render with a faded
+     * diagonal-hatch fill, and hovering one adds `tooltip` to the hover tooltip. Used to flag the
+     * most recent bucket(s) when ingestion hasn't caught up. Pass `null`/`undefined` or an empty
+     * `indices` array to clear.
+     */
+    incompleteBars?: { indices: number[]; tooltip?: string } | null
+    /**
+     * Let the pointer move onto the tooltip without dismissing it, so a tooltip taller than its
+     * max height can be scrolled. Off by default: an interactive tooltip sits over the canvas and
+     * would swallow clicks meant for the chart (e.g. clickable markers or drag-to-select).
+     */
+    interactiveTooltip?: boolean
+    /** Pin the value axis instead of auto-scaling to the data, so sparklines in a column stay
+     *  comparable to each other. */
+    valueDomain?: ValueDomain
 }
 
 function normalizeSparklineData(
@@ -112,6 +136,7 @@ export function Sparkline({
     hideZerosInTooltip = false,
     sortTooltipByCount = false,
     renderTooltipValue,
+    valueDomain,
 }: SparklineProps): JSX.Element {
     const theme = useChartTheme()
 
@@ -156,6 +181,7 @@ export function Sparkline({
                 labels={chartLabels}
                 theme={theme}
                 type={type}
+                valueDomain={valueDomain}
                 fill
                 className="h-full"
                 tooltip={renderTooltip}

packages/quill/packages/charts/src/charts/Sparkline/Sparkline.tsx

@@ -3,7 +3,7 @@ import React, { useEffect, useMemo } from 'react'
 import { useChartHover } from '../../core/chart-context'
 import { ChartErrorBoundary } from '../../core/ChartErrorBoundary'
 import { useLatest } from '../../core/hooks/useLatest'
-import type { BarChartConfig, ChartTheme, LineChartConfig, Series, TooltipContext } from '../../core/types'
+import type { BarChartConfig, ChartTheme, LineChartConfig, Series, TooltipContext, ValueDomain } from '../../core/types'
 import { BarChart } from '../BarChart/BarChart'
 import { LineChart } from '../LineChart/LineChart'
 
@@ -28,6 +28,9 @@ export interface SparklineProps {
     fillOpacity?: number
     /** Dash the line from this index onward (e.g. an in-progress trailing period). Omit for a fully solid line. */
     dashedFromIndex?: number
+    /** Value-axis domain control — omit for data-derived auto-scaling. Pin both ends to keep
+     *  separate sparklines comparable (e.g. a column of per-provider rates all read against 0–100). */
+    valueDomain?: ValueDomain
     /** Fires the hovered index, or -1 when not hovering. */
     onHoverIndexChange?: (index: number) => void
     /** Tooltip content renderer. Sparkline tooltips are off by default; supplying this enables them. */
@@ -72,6 +75,7 @@ function SparklineInner({
     fill = false,
     fillOpacity = 0.35,
     dashedFromIndex,
+    valueDomain,
     onHoverIndexChange,
     tooltip,
     className,
@@ -111,8 +115,9 @@ function SparklineInner({
                   }
                 : { showCrosshair: true, margins: LINE_MARGINS }),
             ...(hasTooltip ? {} : { tooltip: { enabled: false } }),
+            ...(valueDomain ? { valueDomain } : {}),
         }),
-        [type, hasTooltip]
+        [type, hasTooltip, valueDomain]
     )
     const wrapperStyle = useMemo<React.CSSProperties | undefined>(() => (fill ? undefined : { height }), [fill, height])
 

posthog/settings/ses.py

@@ -33,3 +33,12 @@
 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()
 ]
+
+# Mailbox providers that sending health is broken down by, as SES ISP dimension values.
+# SES exposes no API to enumerate them and AWS documents the vocabulary only as "e.g. Gmail,
+# Yahoo", so this stays configurable: an unrecognized value returns zeros rather than an error,
+# and correcting one must not need a deploy. Each provider costs one BatchGetMetricData query
+# per metric, batched ten at a time.
+SES_ISP_DIMENSIONS: list[str] = [
+    isp.strip() for isp in os.getenv("SES_ISP_DIMENSIONS", "Gmail,Yahoo,Outlook,Apple").split(",") if isp.strip()
+]

products/workflows/backend/api/hog_flow.py

@@ -73,6 +73,7 @@
 from posthog.event_usage import AGENT_EVENT_SOURCES, EventSource, get_event_source, report_user_action
 from posthog.models import Team
 from posthog.models.filters import Filter
+from posthog.models.integration import Integration
 from posthog.plugins.plugin_server_api import (
     cancel_hog_flow_batch_job,
     cancel_hog_flow_invocations,
@@ -1977,6 +1978,71 @@ def _fetch_aws_tenant_reputation(team_id: int) -> dict[str, Any] | None:
     return value
 
 
+ISP_METRICS_CACHE_SECONDS = 5 * 60
+ISP_METRICS_ERROR_CACHE_SECONDS = 60
+# Bounds the BatchGetMetricData fan-out: every extra domain costs one query per provider per
+# metric. Projects with more sending domains than this get a breakdown over their first few.
+ISP_METRICS_MAX_DOMAINS = 5
+
+
+def _fetch_isp_metrics(team_id: int, window_days: int) -> list[dict[str, Any]]:
+    """
+    Per-mailbox-provider sending health for the project's verified domains, cached like the tenant
+    reputation above and for the same reason: the endpoint reloads on every search keystroke.
+
+    Returns an empty list rather than raising when SES is unreachable or VDM is not collecting yet
+    — the provider breakdown is an addition to the rates display, never a reason it fails to load.
+    """
+    cache_key = f"workflows_ses_isp_metrics_{team_id}_{window_days}"
+    cached = cache.get(cache_key)
+    if cached is not None:
+        return cached["value"]
+
+    # Dedupe before the cap: several senders commonly share one domain, and letting duplicates
+    # consume the budget would drop real domains from the breakdown.
+    domains = list(
+        dict.fromkeys(
+            domain
+            for domain in Integration.objects.filter(team_id=team_id, kind="email", config__verified=True)
+            .order_by("id")
+            .values_list("config__domain", flat=True)
+            if domain
+        )
+    )[:ISP_METRICS_MAX_DOMAINS]
+    if not domains:
+        cache.set(cache_key, {"value": []}, ISP_METRICS_CACHE_SECONDS)
+        return []
+
+    try:
+        rows = SESProvider().get_identity_isp_metrics(domains, window_days=window_days)
+    except Exception:
+        logger.exception("Failed to fetch SES per-ISP metrics", team_id=team_id)
+        cache.set(cache_key, {"value": []}, ISP_METRICS_ERROR_CACHE_SECONDS)
+        return []
+
+    value = [
+        {
+            "isp": row.isp,
+            "emails_sent": row.emails_sent,
+            "delivery_rate": row.delivery_rate,
+            "bounce_rate": row.bounce_rate,
+            "complaint_rate": row.complaint_rate,
+            "daily": [
+                {
+                    "date": point.date,
+                    "emails_sent": point.emails_sent,
+                    "delivery_rate": point.delivery_rate,
+                    "bounce_rate": point.bounce_rate,
+                }
+                for point in row.daily
+            ],
+        }
+        for row in rows
+    ]
+    cache.set(cache_key, {"value": value}, ISP_METRICS_CACHE_SECONDS)
+    return value
+
+
 class EmailSendingRatesSerializer(serializers.Serializer):
     """Bounce/complaint rates over the last 30 days of workflow email, computed on the fly from app metrics."""
 
@@ -2062,6 +2128,57 @@ class AwsTenantReputationSerializer(serializers.Serializer):
     )
 
 
+class IspDailyPointSerializer(serializers.Serializer):
+    """One bucket of a provider's sending history."""
+
+    date = serializers.CharField(read_only=True, help_text="Bucket date, as an ISO 8601 calendar date.")
+    emails_sent = serializers.IntegerField(read_only=True, help_text="Emails sent to this provider on this date.")
+    delivery_rate = serializers.FloatField(
+        read_only=True, help_text="Emails this provider accepted on this date, divided by emails sent to it (0-1)."
+    )
+    bounce_rate = serializers.FloatField(
+        read_only=True, help_text="Hard bounces at this provider on this date, divided by emails sent to it (0-1)."
+    )
+
+
+class IspSendingHealthSerializer(serializers.Serializer):
+    """How one mailbox provider treated this project's email, from AWS SES's own delivery data."""
+
+    isp = serializers.CharField(
+        read_only=True,
+        help_text="The recipient mailbox provider, as AWS names it — for example Gmail or Yahoo.",
+    )
+    emails_sent = serializers.IntegerField(read_only=True, help_text="Emails sent to this provider during the window.")
+    delivery_rate = serializers.FloatField(
+        read_only=True,
+        help_text=(
+            "Emails this provider accepted, divided by emails sent to it (0-1). Acceptance is not "
+            "inbox placement: a provider can accept a message and still file it as spam."
+        ),
+    )
+    bounce_rate = serializers.FloatField(
+        read_only=True,
+        help_text="Hard (permanent) bounces at this provider, divided by emails sent to it (0-1).",
+    )
+    complaint_rate = serializers.FloatField(
+        read_only=True,
+        allow_null=True,
+        help_text=(
+            "Spam complaints from this provider, divided by the deliveries it reports complaints "
+            "for (0-1). Null when the provider runs no feedback loop, so complaints are "
+            "unmeasurable here rather than zero."
+        ),
+    )
+    daily = IspDailyPointSerializer(
+        many=True,
+        read_only=True,
+        help_text=(
+            "Sending history for this provider, oldest first, so a drop can be dated rather than "
+            "averaged into the window. Dates this provider received nothing are omitted."
+        ),
+    )
+
+
 class TeamEmailReputationResponseSerializer(serializers.Serializer):
     aws = AwsTenantReputationSerializer(
         allow_null=True,
@@ -2085,6 +2202,14 @@ class TeamEmailReputationResponseSerializer(serializers.Serializer):
         read_only=True,
         help_text="Rates per workflow, worst first (complaint rate, then bounce rate), capped at the worst 50.",
     )
+    isps = IspSendingHealthSerializer(
+        many=True,
+        read_only=True,
+        help_text=(
+            "Sending health per mailbox provider, busiest first. Empty when the caller lacks "
+            "project-wide workflow access, no sending domain is verified, or AWS has no data yet."
+        ),
+    )
     email_sending_suspended = serializers.BooleanField(
         read_only=True,
         help_text="True while workflow email sending is suspended for this project to protect deliverability.",
@@ -4660,6 +4785,11 @@ def team_reputation(self, request: Request, **kwargs) -> Response:
                     "aws": _fetch_aws_tenant_reputation(self.team_id) if can_read_all_workflows else None,
                     "reputation": reputation,
                     "workflows": workflow_rows,
+                    # Same project-wide gate as `reputation`: the breakdown pools every workflow's
+                    # email for a sending domain, so object-level grants alone don't earn it.
+                    "isps": (
+                        _fetch_isp_metrics(self.team_id, self.REPUTATION_WINDOW_DAYS) if can_read_all_workflows else []
+                    ),
                     "email_sending_suspended": suspended_at is not None,
                     "email_sending_suspended_at": suspended_at,
                     "email_sending_suspension_reason": suspension_reason if suspended_at is not None else "",