diff --git a/frontend/src/lib/components/Sparkline.tsx b/frontend/src/lib/components/Sparkline.tsx index a3eae5514442..cc1b9287160d 100644 --- a/frontend/src/lib/components/Sparkline.tsx +++ b/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} diff --git a/packages/quill/packages/charts/src/charts/Sparkline/Sparkline.tsx b/packages/quill/packages/charts/src/charts/Sparkline/Sparkline.tsx index 06342cb5d714..344be4743713 100644 --- a/packages/quill/packages/charts/src/charts/Sparkline/Sparkline.tsx +++ b/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(() => (fill ? undefined : { height }), [fill, height]) diff --git a/posthog/settings/ses.py b/posthog/settings/ses.py index 659ff307e428..fd08c1f71139 100644 --- a/posthog/settings/ses.py +++ b/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() +] diff --git a/products/workflows/backend/api/hog_flow.py b/products/workflows/backend/api/hog_flow.py index 8d21cdb12333..d04aa70c3907 100644 --- a/products/workflows/backend/api/hog_flow.py +++ b/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 "", diff --git a/products/workflows/backend/api/test/test_email_reputation.py b/products/workflows/backend/api/test/test_email_reputation.py index 1be9f69c5b51..23ec9ac3980f 100644 --- a/products/workflows/backend/api/test/test_email_reputation.py +++ b/products/workflows/backend/api/test/test_email_reputation.py @@ -10,12 +10,14 @@ from posthog.constants import AvailableFeature from posthog.models import Team +from posthog.models.integration import Integration from posthog.models.organization import OrganizationMembership from posthog.models.user import User from products.access_control.backend.models.access_control import AccessControl from products.workflows.backend.models import HogFlow, HogFlowBatchJob from products.workflows.backend.models.team_workflows_config import TeamWorkflowsConfig +from products.workflows.backend.providers.ses import IspDailyPoint, IspSendingMetrics class TestEmailReputationAPI(APIBaseTest): @@ -37,13 +39,21 @@ def _create_flow(self, name: str) -> HogFlow: ) def _get_reputation( - self, totals_by_source: dict, query: str = "", aws_tenant: dict | None | Exception = None + self, + totals_by_source: dict, + query: str = "", + aws_tenant: dict | None | Exception = None, + isp_metrics: list | Exception | None = None, ) -> dict: provider = MagicMock() if isinstance(aws_tenant, Exception): provider.get_tenant_reputation.side_effect = aws_tenant else: provider.get_tenant_reputation.return_value = aws_tenant + if isinstance(isp_metrics, Exception): + provider.get_identity_isp_metrics.side_effect = isp_metrics + else: + provider.get_identity_isp_metrics.return_value = isp_metrics or [] with ( patch( "products.workflows.backend.api.hog_flow.fetch_app_metric_totals_by_source", @@ -243,6 +253,80 @@ def test_reputation_endpoint_reports_email_sending_suspension(self): assert data["email_sending_suspended_at"] == suspended_at.isoformat().replace("+00:00", "Z") assert data["email_sending_suspension_reason"] == "critical bounce rate" + def _verify_sending_domain(self, domain: str = "mail.example.com") -> None: + Integration.objects.create( + team=self.team, + kind="email", + integration_id=domain, + config={"domain": domain, "provider": "ses", "verified": True}, + ) + + def test_reputation_endpoint_returns_the_per_provider_breakdown(self): + self._verify_sending_domain() + + body = self._get_reputation( + {}, + isp_metrics=[ + IspSendingMetrics( + isp="Gmail", + emails_sent=900, + delivery_rate=0.97, + bounce_rate=0.01, + complaint_rate=None, + daily=(IspDailyPoint(date="2026-08-01", emails_sent=900, delivery_rate=0.97, bounce_rate=0.01),), + ), + IspSendingMetrics( + isp="Yahoo", + emails_sent=100, + delivery_rate=0.99, + bounce_rate=0.0, + complaint_rate=0.002, + daily=(), + ), + ], + ) + + assert body["isps"] == [ + { + "isp": "Gmail", + "emails_sent": 900, + "delivery_rate": 0.97, + "bounce_rate": 0.01, + # Null rather than 0 — Gmail runs no feedback loop, so a complaint rate would be + # a number we can't actually measure. + "complaint_rate": None, + "daily": [ + { + "date": "2026-08-01", + "emails_sent": 900, + "delivery_rate": 0.97, + "bounce_rate": 0.01, + } + ], + }, + { + "isp": "Yahoo", + "emails_sent": 100, + "delivery_rate": 0.99, + "bounce_rate": 0.0, + "complaint_rate": 0.002, + "daily": [], + }, + ] + + def test_reputation_endpoint_still_loads_when_the_provider_breakdown_fails(self): + # The breakdown is an addition to the rates display; SES being unreachable must not take + # the whole reputation page down with it. + self._verify_sending_domain() + + body = self._get_reputation( + {"src": {"email_sent": 100, "email_bounced_hard": 1}}, + isp_metrics=Exception("SES timeout"), + ) + + assert body["isps"] == [] + assert body["reputation"]["emails_sent"] == 100 + @pytest.mark.ee class TestEmailReputationAccessControl(APIBaseTest): @@ -285,6 +369,16 @@ def test_object_level_only_member_gets_rows_but_no_project_wide_state(self): "reputation_impact": "HIGH", "findings": [], } + provider.get_identity_isp_metrics.return_value = [ + IspSendingMetrics( + isp="Gmail", + emails_sent=100, + delivery_rate=0.9, + bounce_rate=0.05, + complaint_rate=None, + daily=(), + ) + ] with ( patch( "products.workflows.backend.api.hog_flow.fetch_app_metric_totals_by_source", @@ -298,4 +392,7 @@ def test_object_level_only_member_gets_rows_but_no_project_wide_state(self): data = response.json() assert data["aws"] is None assert data["reputation"] is None + # The per-provider breakdown pools every workflow's email for a sending domain, so it is + # project-wide state too — an object grant alone must not disclose it. + assert data["isps"] == [] assert [row["hog_flow_id"] for row in data["workflows"]] == [str(flow.id)] diff --git a/products/workflows/backend/providers/ses.py b/products/workflows/backend/providers/ses.py index 311e47c8a153..53382c2cd976 100644 --- a/products/workflows/backend/providers/ses.py +++ b/products/workflows/backend/providers/ses.py @@ -1,7 +1,10 @@ import re import logging -from collections.abc import Iterable, Iterator +from collections import defaultdict +from collections.abc import Iterable, Iterator, Sequence +from datetime import UTC, datetime, timedelta from functools import cached_property +from itertools import batched from typing import TYPE_CHECKING, Any from django.conf import settings @@ -12,6 +15,8 @@ from botocore.exceptions import BotoCoreError, ClientError from rest_framework import exceptions +from posthog.dataclasses import frozen + if TYPE_CHECKING: from types_boto3_ses.client import SESClient from types_boto3_sesv2.client import SESV2Client @@ -19,6 +24,51 @@ logger = logging.getLogger(__name__) +# Everything needed to express one provider's sending health as rates. DELIVERY_COMPLAINT is the +# complaint denominator rather than SEND: AWS defines it as deliveries excluding recipients at +# ISPs it has no feedback-loop agreement with, which is the difference between "nobody complained" +# and "this provider never tells us". +ISP_METRICS: tuple[str, ...] = ("SEND", "DELIVERY", "PERMANENT_BOUNCE", "COMPLAINT", "DELIVERY_COMPLAINT") + +# SES caps a BatchGetMetricData request at ten queries. +METRIC_QUERY_BATCH_SIZE = 10 + + +def _bucket_date(timestamp: Any) -> str: + """ + An ISO date key for one series bucket. boto3 hands back datetimes, but the key has to be + stable across sending domains so their series line up when summed, and JSON-safe on the way + out; anything unrecognized falls through as its own string rather than collapsing buckets + together. + """ + if isinstance(timestamp, datetime): + return timestamp.date().isoformat() + return str(timestamp) + + +@frozen +class IspDailyPoint: + """One bucket of a provider's series, so a drop can be dated rather than just averaged away.""" + + date: str + emails_sent: int + delivery_rate: float + bounce_rate: float + + +@frozen +class IspSendingMetrics: + """How one mailbox provider treated a project's mail over the requested window.""" + + isp: str + emails_sent: int + delivery_rate: float + bounce_rate: float + # None when the provider runs no feedback loop, so complaints are unmeasurable rather than zero. + complaint_rate: float | None + # Oldest bucket first. Buckets SES returned nothing for are absent rather than zero-filled. + daily: tuple[IspDailyPoint, ...] + class SESProvider: ses_client: "SESClient" @@ -471,3 +521,121 @@ def delete_identity(self, identity: str): except (ClientError, BotoCoreError) as e: logger.exception(f"SES API error deleting identity: {e}") raise + + def get_identity_isp_metrics( + self, + domains: Sequence[str], + window_days: int, + isps: Sequence[str] | None = None, + ) -> list[IspSendingMetrics]: + """ + Sending health per mailbox provider for the given sending domains, newest window first. + + This is the only view that separates "our mail is being accepted but filtered" from "our + mail is fine": delivery and bounce rates diverging at one provider is invisible in the + project-wide rates, which pool every provider together. + + Counts are summed across `domains` because a project's rates are project-wide; VDM's + EMAIL_IDENTITY dimension is per verified domain, so a project sending from several needs + one query set each. Providers that received nothing in the window are omitted rather than + shown as a row of zeros. + """ + isps = list(isps) if isps is not None else list(settings.SES_ISP_DIMENSIONS) + # A project can verify several senders on one domain, and each duplicate would query the + # same VDM series again and sum it in twice, inflating volume while leaving rates intact. + domains = list(dict.fromkeys(domains)) + if not domains or not isps: + return [] + + # VDM aggregates daily and rejects the whole batch if either bound is a partial day + # ("you must not specify partial-day timestamps"), so the window is whole UTC days ending + # at the last midnight. Today's partial day is therefore excluded. + end = datetime.now(tz=UTC).replace(hour=0, minute=0, second=0, microsecond=0) + start = end - timedelta(days=window_days) + + # Dimensions filter rather than group, and the response echoes only the query id — so a + # per-provider breakdown means one query per (domain, provider, metric) and a local index + # back to what each id asked for. + queries: list[dict[str, Any]] = [] + query_subjects: dict[str, tuple[str, str]] = {} + for domain in domains: + for isp in isps: + for metric in ISP_METRICS: + query_id = f"q{len(queries)}" + query_subjects[query_id] = (isp, metric) + queries.append( + { + "Id": query_id, + "Namespace": "VDM", + "Metric": metric, + "Dimensions": {"EMAIL_IDENTITY": domain, "ISP": isp}, + "StartDate": start, + "EndDate": end, + } + ) + + # Kept per bucket rather than collapsed on arrival: the series is what lets a customer + # date a drop, and the window totals fall out of it by summing. + series: dict[tuple[str, str], dict[str, int]] = defaultdict(lambda: defaultdict(int)) + # strict=False: the final batch is short whenever the query count isn't a multiple of ten. + for batch in batched(queries, METRIC_QUERY_BATCH_SIZE, strict=False): + response = self.ses_v2_client.batch_get_metric_data(Queries=list(batch)) # type: ignore[arg-type] + for result in response.get("Results", []): + isp, metric = query_subjects[result["Id"]] + buckets = series[(isp, metric)] + # AWS documents Values as "cumulative / sum" without saying which, so this reads + # them as per-bucket counts. If they turn out to be running totals the symptom is + # loud rather than subtle: summed deliveries would overshoot sends and every + # provider would pin to a 100% delivery rate against the clamp below. + for timestamp, value in zip(result.get("Timestamps", []), result.get("Values", []), strict=False): + buckets[_bucket_date(timestamp)] += value + for error in response.get("Errors", []): + # A failed query leaves its metric at zero, which understates a rate rather than + # breaking the panel. Log it so a systematic failure (ACCESS_DENIED on a fresh + # VDM subscription, say) is visible rather than silently reading as healthy. + logger.warning( + "SES metric query failed", + extra={ + "query": query_subjects.get(error.get("Id", "")), + "code": error.get("Code"), + "message": error.get("Message"), + }, + ) + + rows: list[IspSendingMetrics] = [] + for isp in isps: + sent_by_date = series[(isp, "SEND")] + emails_sent = sum(sent_by_date.values()) + if emails_sent == 0: + continue + delivered_by_date = series[(isp, "DELIVERY")] + bounced_by_date = series[(isp, "PERMANENT_BOUNCE")] + complaint_base = sum(series[(isp, "DELIVERY_COMPLAINT")].values()) + rows.append( + IspSendingMetrics( + isp=isp, + emails_sent=emails_sent, + # Feedback for a send can arrive after the window closes, so a rate can exceed + # its denominator at the boundary. Clamp, as the project-wide rates do. + delivery_rate=min(1.0, sum(delivered_by_date.values()) / emails_sent), + bounce_rate=min(1.0, sum(bounced_by_date.values()) / emails_sent), + complaint_rate=( + min(1.0, sum(series[(isp, "COMPLAINT")].values()) / complaint_base) if complaint_base else None + ), + daily=tuple( + IspDailyPoint( + date=date, + emails_sent=sent, + delivery_rate=min(1.0, delivered_by_date.get(date, 0) / sent), + bounce_rate=min(1.0, bounced_by_date.get(date, 0) / sent), + ) + # Only buckets that sent something: a rate over zero sends is undefined, + # and a zero-filled gap would draw as a cliff in the trend. + for date, sent in sorted(sent_by_date.items()) + if sent > 0 + ), + ) + ) + + rows.sort(key=lambda row: -row.emails_sent) + return rows diff --git a/products/workflows/backend/test/test_ses_provider.py b/products/workflows/backend/test/test_ses_provider.py index efc7a71b2ccf..9a7310463dfc 100644 --- a/products/workflows/backend/test/test_ses_provider.py +++ b/products/workflows/backend/test/test_ses_provider.py @@ -1,3 +1,4 @@ +from datetime import datetime from typing import Optional import pytest @@ -584,3 +585,148 @@ def test_missing_enforcement_status_fails_the_poll(self): with pytest.raises(KeyError): self.provider.get_account_reputation() + + +class TestGetIdentityIspMetrics(TestCase): + def setUp(self): + patcher = patch("products.workflows.backend.providers.ses.boto3.client") + mock_boto3_client = patcher.start() + self.addCleanup(patcher.stop) + self.mock_client = mock_boto3_client.return_value + self.provider = SESProvider() + + def _serve(self, values_by_subject: dict[tuple[str, str], int]) -> None: + """Answer each batch from a {(isp, metric): value} table, defaulting anything absent to 0.""" + + def respond(Queries): + return { + "Results": [ + { + "Id": query["Id"], + "Timestamps": ["2026-08-01T00:00:00Z"], + "Values": [values_by_subject.get((query["Dimensions"]["ISP"], query["Metric"]), 0)], + } + for query in Queries + ] + } + + self.mock_client.batch_get_metric_data.side_effect = lambda Queries: respond(Queries) + + def _serve_series(self, values_by_subject: dict[tuple[str, str], dict[str, int]]) -> None: + """Answer from a {(isp, metric): {date: value}} table, so a query spans several buckets.""" + + def respond(Queries): + results = [] + for query in Queries: + buckets = values_by_subject.get((query["Dimensions"]["ISP"], query["Metric"]), {}) + results.append( + { + "Id": query["Id"], + "Timestamps": [datetime.fromisoformat(date) for date in sorted(buckets)], + "Values": [buckets[date] for date in sorted(buckets)], + } + ) + return {"Results": results} + + self.mock_client.batch_get_metric_data.side_effect = lambda Queries: respond(Queries) + + def test_daily_series_is_ordered_and_merged_across_domains(self): + # Two domains reporting the same days have to land in one bucket per day, or the trend + # draws each domain as its own point and every rate is computed against half the sends. + self._serve_series( + { + ("Gmail", "SEND"): {"2026-08-02": 50, "2026-08-01": 100}, + ("Gmail", "DELIVERY"): {"2026-08-02": 10, "2026-08-01": 95}, + } + ) + + rows = self.provider.get_identity_isp_metrics( + [TEST_DOMAIN, "other.posthog.com"], window_days=30, isps=["Gmail"] + ) + + assert [(point.date, point.emails_sent, point.delivery_rate) for point in rows[0].daily] == [ + ("2026-08-01", 200, 0.95), + # The drop the whole feature exists to make visible; the 30-day average hides it. + ("2026-08-02", 100, 0.2), + ] + + def test_daily_series_skips_buckets_with_no_sends(self): + # A rate over zero sends is undefined, and zero-filling would draw a cliff that never + # happened. + self._serve_series( + { + ("Gmail", "SEND"): {"2026-08-01": 10, "2026-08-02": 0}, + ("Gmail", "DELIVERY"): {"2026-08-01": 9, "2026-08-02": 0}, + } + ) + + rows = self.provider.get_identity_isp_metrics([TEST_DOMAIN], window_days=30, isps=["Gmail"]) + + assert [point.date for point in rows[0].daily] == ["2026-08-01"] + + @parameterized.expand( + [ + # A provider that reports complaints: the rate is complaints over the deliveries it + # reports them for, NOT over everything we sent it. + ("reporting_provider", 40, 4, 0.1), + # SES excludes recipients at providers it has no feedback-loop agreement with from + # DELIVERY_COMPLAINT. A zero base means unmeasurable, which must not read as 0%. + ("provider_without_a_feedback_loop", 0, 0, None), + ] + ) + def test_complaint_rate_is_measured_against_delivery_complaint( + self, _name: str, delivery_complaint: int, complaints: int, expected: Optional[float] + ): + self._serve( + { + ("Gmail", "SEND"): 100, + ("Gmail", "DELIVERY"): 95, + ("Gmail", "DELIVERY_COMPLAINT"): delivery_complaint, + ("Gmail", "COMPLAINT"): complaints, + } + ) + + rows = self.provider.get_identity_isp_metrics([TEST_DOMAIN], window_days=30, isps=["Gmail"]) + + assert [row.complaint_rate for row in rows] == [expected] + + def test_counts_are_summed_across_the_projects_sending_domains(self): + # Each domain is queried separately because EMAIL_IDENTITY is per verified domain, but a + # project's rates are project-wide, so the two domains' counts have to add up. + self._serve({("Gmail", "SEND"): 50, ("Gmail", "DELIVERY"): 40, ("Gmail", "PERMANENT_BOUNCE"): 5}) + + rows = self.provider.get_identity_isp_metrics( + [TEST_DOMAIN, "other.posthog.com"], window_days=30, isps=["Gmail"] + ) + + assert len(rows) == 1 + assert rows[0].emails_sent == 100 + assert rows[0].delivery_rate == 0.8 + assert rows[0].bounce_rate == 0.1 + + def test_queries_are_batched_within_the_ses_ten_query_limit(self): + self._serve({("Gmail", "SEND"): 1, ("Yahoo", "SEND"): 1, ("Outlook", "SEND"): 1}) + + self.provider.get_identity_isp_metrics([TEST_DOMAIN], window_days=30, isps=["Gmail", "Yahoo", "Outlook"]) + + batches = [kwargs["Queries"] for _, kwargs in self.mock_client.batch_get_metric_data.call_args_list] + # 3 providers x 5 metrics = 15 queries, and SES rejects a batch of more than ten. + assert [len(batch) for batch in batches] == [10, 5] + sent = {(query["Dimensions"]["ISP"], query["Metric"]) for batch in batches for query in batch} + assert len(sent) == 15 + + def test_providers_that_received_nothing_are_omitted(self): + # Without this a silent provider divides by zero; a row of zeros would also read as a + # delivery problem rather than as "we never mailed anyone here". + self._serve({("Gmail", "SEND"): 10, ("Gmail", "DELIVERY"): 10}) + + rows = self.provider.get_identity_isp_metrics([TEST_DOMAIN], window_days=30, isps=["Gmail", "Yahoo"]) + + assert [row.isp for row in rows] == ["Gmail"] + + def test_busiest_provider_is_reported_first(self): + self._serve({("Gmail", "SEND"): 10, ("Yahoo", "SEND"): 90}) + + rows = self.provider.get_identity_isp_metrics([TEST_DOMAIN], window_days=30, isps=["Gmail", "Yahoo"]) + + assert [row.isp for row in rows] == ["Yahoo", "Gmail"] diff --git a/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.stories.tsx b/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.stories.tsx new file mode 100644 index 000000000000..2409d48d9449 --- /dev/null +++ b/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.stories.tsx @@ -0,0 +1,100 @@ +import { Meta, StoryFn } from '@storybook/react' + +import { useStorybookMocks } from '~/mocks/browser' + +import type { TeamEmailReputationResponseApi } from 'products/workflows/frontend/generated/api.schemas' + +import { WorkflowsReputation } from './WorkflowsReputation' + +const reputationEndpoint = '/api/projects/:team_id/hog_flows/reputation' + +// A fortnight where Gmail starts filtering us halfway through while everyone else stays fine. +// This is the shape the project-wide rates hide, so it's what the stories are built around. +function gmailSeries(): TeamEmailReputationResponseApi['isps'][number]['daily'] { + return Array.from({ length: 14 }, (_, day) => { + const filtered = day >= 7 + return { + date: `2026-08-${String(day + 1).padStart(2, '0')}`, + emails_sent: 6000, + delivery_rate: filtered ? 0.42 : 0.97, + bounce_rate: 0.008, + } + }) +} + +function steadySeries(deliveryRate: number): TeamEmailReputationResponseApi['isps'][number]['daily'] { + return Array.from({ length: 14 }, (_, day) => ({ + date: `2026-08-${String(day + 1).padStart(2, '0')}`, + emails_sent: 900, + delivery_rate: deliveryRate, + bounce_rate: 0.004, + })) +} + +const baseResponse: TeamEmailReputationResponseApi = { + aws: { health: 'healthy', sending_status: 'ENABLED', findings: [] }, + reputation: { bounce_rate: 0.0062, complaint_rate: 0.0001, emails_sent: 115025 }, + workflows: [], + isps: [ + { + isp: 'Gmail', + emails_sent: 84000, + delivery_rate: 0.69, + bounce_rate: 0.008, + // Gmail runs no feedback loop, so a complaint rate here would be unmeasurable. + complaint_rate: null, + daily: gmailSeries(), + }, + { + isp: 'Outlook', + emails_sent: 12600, + delivery_rate: 0.98, + bounce_rate: 0.004, + complaint_rate: 0.0004, + daily: steadySeries(0.98), + }, + { + isp: 'Apple', + emails_sent: 8100, + // Steady, but steadily poor: the case an auto-scaled axis would draw as a flat line + // indistinguishable from a healthy provider. + delivery_rate: 0.45, + bounce_rate: 0.012, + complaint_rate: null, + daily: steadySeries(0.45), + }, + { + isp: 'Yahoo', + emails_sent: 6300, + delivery_rate: 0.96, + bounce_rate: 0.006, + complaint_rate: 0.0011, + daily: steadySeries(0.96), + }, + ], + email_sending_suspended: false, + email_sending_suspended_at: null, + email_sending_suspension_reason: '', +} + +const meta: Meta = { + title: 'Products/Workflows/Reputation', + component: WorkflowsReputation, + parameters: { layout: 'padded', testOptions: { waitForLoadersToDisappear: true } }, +} +export default meta + +function mockReputation(response: TeamEmailReputationResponseApi): Record { + return { get: { [reputationEndpoint]: response } } +} + +export const OneProviderFiltering: StoryFn = () => { + useStorybookMocks(mockReputation(baseResponse)) + return +} + +export const NoProviderData: StoryFn = () => { + // What every project sees until Virtual Deliverability Manager is collecting. + useStorybookMocks(mockReputation({ ...baseResponse, isps: [] })) + return +} diff --git a/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.tsx b/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.tsx index f7be4ab7c40f..855c3a0cb0ff 100644 --- a/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.tsx +++ b/products/workflows/frontend/Workflows/Reputation/WorkflowsReputation.tsx @@ -2,6 +2,7 @@ import { useActions, useValues } from 'kea' import { LemonBanner, LemonInput, LemonTable, LemonTag, LemonTagType, Link, Tooltip } from '@posthog/lemon-ui' +import { Sparkline } from 'lib/components/Sparkline' import { humanFriendlyNumber, percentage } from 'lib/utils/numbers' import { urls } from 'scenes/urls' @@ -9,6 +10,7 @@ import type { AwsTenantReputationApi, AwsTenantReputationHealthEnumApi, EmailSendingRatesApi, + IspSendingHealthApi, WorkflowEmailSendingRatesApi, } from 'products/workflows/frontend/generated/api.schemas' @@ -134,12 +136,104 @@ function AwsFindings({ aws }: { aws: AwsTenantReputationApi }): JSX.Element | nu ) } +// A single point is a reading, not a trend, so the column stays empty until there are two. +const MIN_TREND_POINTS = 2 + +function DeliveryTrend({ isp }: { isp: IspSendingHealthApi }): JSX.Element | null { + if (isp.daily.length < MIN_TREND_POINTS) { + return null + } + return ( + point.delivery_rate * 100)} + labels={isp.daily.map((point) => point.date)} + name={`${isp.isp} delivery rate (%)`} + type="line" + renderTooltipValue={(value) => `${value.toFixed(1)}%`} + // Fixed 0-100 rather than auto-scaled per row: these are read against each other, and + // an auto-scaled axis draws a steady 40% provider identically to a steady 98% one. + valueDomain={{ min: 0, max: 100 }} + // Sizes Sparkline's own container: without a height it grows to fill the table cell + // and the chart bleeds across rows. + className="h-8 w-28" + /> + ) +} + +function IspBreakdown({ isps }: { isps: readonly IspSendingHealthApi[] }): JSX.Element | null { + if (isps.length === 0) { + return null + } + return ( +
+ + row.isp} + columns={[ + { + title: 'Provider', + key: 'isp', + render: (_, row: IspSendingHealthApi) => {row.isp}, + }, + { + title: 'Delivery rate', + key: 'delivery_rate', + align: 'right', + render: (_, row: IspSendingHealthApi) => ( + + {formatRate(row.delivery_rate)} + + ), + }, + { + title: 'Delivery trend', + key: 'delivery_trend', + tooltip: 'Every provider is drawn on the same 0-100% axis, so rows can be compared by height.', + render: (_, row: IspSendingHealthApi) => , + }, + { + title: 'Bounce rate', + key: 'bounce_rate', + align: 'right', + render: (_, row: IspSendingHealthApi) => , + }, + { + title: 'Complaint rate', + key: 'complaint_rate', + align: 'right', + render: (_, row: IspSendingHealthApi) => + row.complaint_rate === null ? ( + + Not reported + + ) : ( + + ), + }, + { + title: 'Emails sent', + key: 'emails_sent', + align: 'right', + render: (_, row: IspSendingHealthApi) => humanFriendlyNumber(row.emails_sent), + }, + ]} + /> +
+ ) +} + function TeamRatesCard({ reputation, aws, + isps, }: { reputation: EmailSendingRatesApi | null aws: AwsTenantReputationApi | null + isps: readonly IspSendingHealthApi[] }): JSX.Element { return (
@@ -180,12 +274,13 @@ function TeamRatesCard({
)} {aws && } + ) } export function WorkflowsReputation(): JSX.Element { - const { awsReputation, teamReputation, workflowSnapshots, reputationResponseLoading, search } = + const { awsReputation, teamReputation, ispSendingHealth, workflowSnapshots, reputationResponseLoading, search } = useValues(workflowsReputationLogic) const { setSearch } = useActions(workflowsReputationLogic) @@ -203,7 +298,7 @@ export function WorkflowsReputation(): JSX.Element { We judge and enforce reputation per project. {teamReputation || awsReputation ? ( - + ) : ( !reputationResponseLoading && (
diff --git a/products/workflows/frontend/Workflows/Reputation/workflowsReputationLogic.ts b/products/workflows/frontend/Workflows/Reputation/workflowsReputationLogic.ts index cb53b6b9edfa..cc702ac52f61 100644 --- a/products/workflows/frontend/Workflows/Reputation/workflowsReputationLogic.ts +++ b/products/workflows/frontend/Workflows/Reputation/workflowsReputationLogic.ts @@ -7,6 +7,7 @@ import { hogFlowsReputationRetrieve } from 'products/workflows/frontend/generate import type { AwsTenantReputationApi, EmailSendingRatesApi, + IspSendingHealthApi, TeamEmailReputationResponseApi, WorkflowEmailSendingRatesApi, } from 'products/workflows/frontend/generated/api.schemas' @@ -15,6 +16,7 @@ import type { export interface workflowsReputationLogicValues { currentProjectId: number | null // projectLogic awsReputation: AwsTenantReputationApi | null + ispSendingHealth: readonly IspSendingHealthApi[] reputationResponse: TeamEmailReputationResponseApi | null reputationResponseLoading: boolean search: string @@ -55,6 +57,7 @@ export interface workflowsReputationLogicMeta { __keaTypeGenInternalSelectorTypes: { awsReputation: (reputationResponse: TeamEmailReputationResponseApi | null) => AwsTenantReputationApi | null teamReputation: (reputationResponse: TeamEmailReputationResponseApi | null) => EmailSendingRatesApi | null + ispSendingHealth: (reputationResponse: TeamEmailReputationResponseApi | null) => readonly IspSendingHealthApi[] workflowSnapshots: ( reputationResponse: TeamEmailReputationResponseApi | null ) => readonly WorkflowEmailSendingRatesApi[] @@ -124,6 +127,10 @@ export const workflowsReputationLogic = kea([ (response: TeamEmailReputationResponseApi | null): EmailSendingRatesApi | null => response?.reputation ?? null, ], + ispSendingHealth: [ + (s) => [s.reputationResponse], + (response: TeamEmailReputationResponseApi | null): readonly IspSendingHealthApi[] => response?.isps ?? [], + ], workflowSnapshots: [ (s) => [s.reputationResponse], (response: TeamEmailReputationResponseApi | null): readonly WorkflowEmailSendingRatesApi[] => diff --git a/products/workflows/frontend/generated/api.schemas.ts b/products/workflows/frontend/generated/api.schemas.ts index b3ad15e5465a..a0505e210bb5 100644 --- a/products/workflows/frontend/generated/api.schemas.ts +++ b/products/workflows/frontend/generated/api.schemas.ts @@ -1400,6 +1400,41 @@ export interface WorkflowEmailSendingRatesApi { readonly hog_flow_name: string } +/** + * One bucket of a provider's sending history. + */ +export interface IspDailyPointApi { + /** Bucket date, as an ISO 8601 calendar date. */ + readonly date: string + /** Emails sent to this provider on this date. */ + readonly emails_sent: number + /** Emails this provider accepted on this date, divided by emails sent to it (0-1). */ + readonly delivery_rate: number + /** Hard bounces at this provider on this date, divided by emails sent to it (0-1). */ + readonly bounce_rate: number +} + +/** + * How one mailbox provider treated this project's email, from AWS SES's own delivery data. + */ +export interface IspSendingHealthApi { + /** The recipient mailbox provider, as AWS names it — for example Gmail or Yahoo. */ + readonly isp: string + /** Emails sent to this provider during the window. */ + readonly emails_sent: number + /** 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. */ + readonly delivery_rate: number + /** Hard (permanent) bounces at this provider, divided by emails sent to it (0-1). */ + readonly bounce_rate: number + /** + * 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. + * @nullable + */ + readonly complaint_rate: number | null + /** 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. */ + readonly daily: readonly IspDailyPointApi[] +} + export interface TeamEmailReputationResponseApi { /** Sending health as judged and enforced by AWS SES for this project's tenant; null when the caller lacks project-wide workflow access, no tenant is provisioned, or AWS is unreachable. */ readonly aws: AwsTenantReputationApi | null @@ -1407,6 +1442,8 @@ export interface TeamEmailReputationResponseApi { readonly reputation: EmailSendingRatesApi | null /** Rates per workflow, worst first (complaint rate, then bounce rate), capped at the worst 50. */ readonly workflows: readonly WorkflowEmailSendingRatesApi[] + /** 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. */ + readonly isps: readonly IspSendingHealthApi[] /** True while workflow email sending is suspended for this project to protect deliverability. */ readonly email_sending_suspended: boolean /** diff --git a/services/mcp/src/api/generated.ts b/services/mcp/src/api/generated.ts index bdd099002626..e644a145fa10 100644 --- a/services/mcp/src/api/generated.ts +++ b/services/mcp/src/api/generated.ts @@ -45452,6 +45452,41 @@ export namespace Schemas { identifier: string; } + /** + * One bucket of a provider's sending history. + */ + export interface IspDailyPoint { + /** Bucket date, as an ISO 8601 calendar date. */ + readonly date: string; + /** Emails sent to this provider on this date. */ + readonly emails_sent: number; + /** Emails this provider accepted on this date, divided by emails sent to it (0-1). */ + readonly delivery_rate: number; + /** Hard bounces at this provider on this date, divided by emails sent to it (0-1). */ + readonly bounce_rate: number; + } + + /** + * How one mailbox provider treated this project's email, from AWS SES's own delivery data. + */ + export interface IspSendingHealth { + /** The recipient mailbox provider, as AWS names it — for example Gmail or Yahoo. */ + readonly isp: string; + /** Emails sent to this provider during the window. */ + readonly emails_sent: number; + /** 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. */ + readonly delivery_rate: number; + /** Hard (permanent) bounces at this provider, divided by emails sent to it (0-1). */ + readonly bounce_rate: number; + /** + * 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. + * @nullable + */ + readonly complaint_rate: number | null; + /** 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. */ + readonly daily: readonly IspDailyPoint[]; + } + export interface JiraIssueSignalExtra { key: string; url: string | null; @@ -83359,6 +83394,8 @@ export namespace Schemas { readonly reputation: EmailSendingRates | null; /** Rates per workflow, worst first (complaint rate, then bounce rate), capped at the worst 50. */ readonly workflows: readonly WorkflowEmailSendingRates[]; + /** 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. */ + readonly isps: readonly IspSendingHealth[]; /** True while workflow email sending is suspended for this project to protect deliverability. */ readonly email_sending_suspended: boolean; /**