feat(workflows): separate sandbox delivery history

docs/internal/workflows-email-sending-tiers.md

@@ -52,6 +52,7 @@ None of them are wired there yet; the charts change is part of turning the rollo
 
 In order: staff suspension or an AWS-paused tenant drops to tier 0; a dirty 7-day window, a workflow auto-pause, or a HIGH tenant reputation impact demotes one tier (at most once per cooldown, anchored on the last rate demotion); a tier above 0 with zero sends for the decay period drops one step per period; promotion needs the dwell served, real use of the tier (at least the use ratio of the daily cap on the minimum number of separate days, counted only after the tier anchor), a clean 30-day window, and a clean tenant reputation.
 Rates only count on a meaningful denominator; below the complaint floor, the absolute complaint backstop still applies, including in windows with no sends.
+Sandbox sender sends never count as sends: the sweep subtracts `email_sandbox_sent` from `email_sent` per day, floored at zero. Their bounces and complaints still count.
 Pinned teams never move automatically.
 
 While the tier is enforced, the sweep notifies teams it moved: a rate demotion sends an in-app notification plus an email to project admins, and an earned promotion sends an in-app notification only.
@@ -81,6 +82,7 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
   A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state.
   Test sends run inline in the CDP API, so the API reads the same `SES_RATE_LIMITER_VALKEY_*` settings as the email worker. Without them, every sandbox test send is skipped.
   CDP API opens the SES Valkey pool only when sandbox delivery is enabled; otherwise its MX validation uses the existing local cache and DNS.
+  The CDP API's SES webhook captures `workflows sandbox email delivered` for a delivery on `SES_SANDBOX_CONFIGURATION_SET`, test sends included. Without that setting on the CDP API, the event never fires.
   Deploy membership checks, daily caps, and the shared pause gate first. Configure the SES identity and Valkey pool, enable `WORKFLOWS_SANDBOX_SENDER_ENABLED`, then enable the UI flag.
 - The buckets are token buckets: a full idle bucket plus refill allows up to roughly twice the stated cap in the very first period. The bucket TTLs exceed the refill periods so this does not recur from idling.
 - A denied send parks until every short bucket has refilled enough to cover it, instead of retrying on a fixed few-minute cadence.

nodejs/src/cdp/cdp-api.ts

@@ -240,7 +240,8 @@ export class CdpApi {
             services.capturedEventsService,
             services.teamWorkflowsConfigService,
             new EmailTrackingCodeSigner(config.ENCRYPTION_SALT_KEYS, config.CDP_EMAIL_TRACKING_URL),
-            services.emailSuppressionService
+            services.emailSuppressionService,
+            services.sandboxEmailSender
         )
         this.groupsManager = new GroupsManagerService(deps.teamManager, deps.groupRepository)
         this.batchExportHogFunctionService = new BatchExportHogFunctionService(

nodejs/src/cdp/cdp-services.ts

@@ -118,6 +118,7 @@ export interface CdpCoreServices {
     /** Resolved outputs shared across every CDP service/consumer. */
     outputs: CdpOutputs
     emailService: EmailService
+    sandboxEmailSender: SandboxEmailSender
     /** Buffers rendered-email asset rows and bulk-flushes them at the batch boundary. */
     messageAssetsService: MessageAssetsService
 }
@@ -466,6 +467,20 @@ export function createCdpCoreServices(
     const sandboxEmailRateLimiter = deps.emailValidationValkey
         ? new RateLimiterService(deps.emailValidationValkey, { name: 'sandbox-email' })
         : null
+    const sandboxEmailSender = new SandboxEmailSender(
+        {
+            enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED,
+            tenantName: config.SES_SANDBOX_TENANT_NAME,
+            configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET,
+            fromAddress: config.SES_SANDBOX_FROM_ADDRESS,
+            dailyTeamCap: config.WORKFLOWS_SANDBOX_DAILY_TEAM_CAP,
+            dailyRecipientCap: config.WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP,
+        },
+        deps.teamManager,
+        sandboxEmailRateLimiter,
+        redis,
+        new SandboxSenderStateService(deps.postgres, deps.pubSub)
+    )
     const emailService = new EmailService(
         {
             sesAccessKeyId: config.SES_ACCESS_KEY_ID,
@@ -488,20 +503,7 @@ export function createCdpCoreServices(
         messageAssetsService,
         workflowEmailRateLimiter,
         teamEmailRateLimiter,
-        new SandboxEmailSender(
-            {
-                enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED,
-                tenantName: config.SES_SANDBOX_TENANT_NAME,
-                configurationSetName: config.SES_SANDBOX_CONFIGURATION_SET,
-                fromAddress: config.SES_SANDBOX_FROM_ADDRESS,
-                dailyTeamCap: config.WORKFLOWS_SANDBOX_DAILY_TEAM_CAP,
-                dailyRecipientCap: config.WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP,
-            },
-            deps.teamManager,
-            sandboxEmailRateLimiter,
-            redis,
-            new SandboxSenderStateService(deps.postgres, deps.pubSub)
-        ),
+        sandboxEmailSender,
         new OrganizationMembersService(deps.postgres)
     )
     const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL)
@@ -638,6 +640,7 @@ export function createCdpCoreServices(
         recipientTokensService,
         outputs,
         emailService,
+        sandboxEmailSender,
         messageAssetsService,
     }
 }

nodejs/src/cdp/services/messaging/email-tracking.service.test.ts

@@ -17,6 +17,7 @@ import { KAFKA_APP_METRICS_2, KAFKA_LOG_ENTRIES } from '~/common/config/kafka-to
 import { closeHub, createHub } from '~/common/utils/db/hub'
 import { PostgresUse } from '~/common/utils/db/postgres'
 import * as envUtils from '~/common/utils/env-utils'
+import * as posthog from '~/common/utils/posthog'
 import { createCdpConsumerDeps } from '~/tests/helpers/cdp'
 import { waitForExpect } from '~/tests/helpers/expectations'
 import { createTestTeamFixture } from '~/tests/helpers/sql'
@@ -819,6 +820,119 @@ describe('EmailTrackingService', () => {
         })
     })
 
+    describe('SES webhook records sandbox deliveries', () => {
+        const sandboxConfigurationSet = 'posthog-messaging-sandbox'
+        let app: express.Application
+        let server: Server
+        let verifySignatureSpy: jest.SpyInstance
+        let capture: jest.SpyInstance
+
+        beforeEach(() => {
+            hub.SES_SANDBOX_CONFIGURATION_SET = sandboxConfigurationSet
+            const api = new CdpApi(hub, createCdpConsumerDeps(hub), {
+                hogQueue: createMockJobQueue(),
+                hogflowQueue: createMockJobQueue(),
+            })
+            app = setupExpressApp()
+            app.use('/', api.router())
+            server = app.listen(0, () => {})
+
+            verifySignatureSpy = jest
+                .spyOn(SesWebhookHandler.prototype as any, 'verifySnsSignature')
+                .mockResolvedValue(true)
+            capture = jest.spyOn(posthog, 'captureTeamEvent').mockImplementation(() => {})
+        })
+
+        afterEach(() => {
+            server.close()
+            verifySignatureSpy.mockRestore()
+            capture.mockRestore()
+        })
+
+        const postDelivery = async (
+            configurationSet: string,
+            isTest: boolean,
+            signedCode = true,
+            recipients: string[] | null = ['member@example.com']
+        ): Promise<supertest.Response> => {
+            const signedTrackingCode = signer.generate(
+                { functionId: 'function-id', id: 'invocation-id', teamId: team.id },
+                isTest
+            )
+            const [unsignedTrackingCode] = signedTrackingCode.split('.')
+            const sesRecord = {
+                eventType: 'Delivery',
+                mail: {
+                    timestamp: '2024-01-01T00:00:00.000Z',
+                    source: 'sandbox@example.com',
+                    messageId: 'ses-message-id',
+                    destination: ['member@example.com'],
+                    headers: signedCode ? [{ name: TRACKING_CODE_HEADER_NAME, value: signedTrackingCode }] : [],
+                    tags: {
+                        'ses:configuration-set': [configurationSet],
+                        ...(signedCode ? {} : { ph_id: [unsignedTrackingCode] }),
+                    },
+                },
+                delivery: {
+                    timestamp: '2024-01-01T00:00:01.000Z',
+                    ...(recipients ? { recipients } : {}),
+                },
+            }
+            const envelope = {
+                Type: 'Notification',
+                MessageId: 'sns-message-id',
+                TopicArn: 'arn:aws:sns:us-east-1:123456789012:ses-events',
+                Message: JSON.stringify(sesRecord),
+                Timestamp: '2024-01-01T00:00:01.000Z',
+                SignatureVersion: '1',
+                Signature: 'stubbed',
+                SigningCertURL: 'https://sns.us-east-1.amazonaws.com/cert.pem',
+            }
+            return await supertest(app)
+                .post('/public/m/ses_webhook')
+                .set('Content-Type', 'text/plain')
+                .send(JSON.stringify(envelope))
+        }
+
+        it.each([false, true])(
+            'captures one delivered event for a delivery on the sandbox set (test send: %s)',
+            async (isTest) => {
+                const res = await postDelivery(sandboxConfigurationSet, isTest)
+
+                expect(res.status).toBe(200)
+                expect(capture).toHaveBeenCalledTimes(1)
+                expect(capture).toHaveBeenCalledWith(
+                    expect.objectContaining({ id: team.id, organization_id: team.organization_id }),
+                    'workflows sandbox email delivered',
+                    { is_test: isTest, recipient_count: 1 }
+                )
+            }
+        )
+
+        it.each([
+            ['a delivery on another configuration set', 'posthog-messaging-tracked', true],
+            ['a sandbox delivery whose tracking code is not signed', sandboxConfigurationSet, false],
+        ])('captures no delivered event for %s', async (_name, configurationSet, signedCode) => {
+            const res = await postDelivery(configurationSet, false, signedCode)
+
+            expect(res.status).toBe(200)
+            expect(capture).not.toHaveBeenCalled()
+        })
+
+        it.each([
+            { recipients: ['member@example.com', 'another@example.com'], count: 2 },
+            { recipients: null, count: 0 },
+        ])('counts recipient copies in a delivery notification ($count)', async ({ recipients, count }) => {
+            const res = await postDelivery(sandboxConfigurationSet, false, true, recipients)
+            expect(res.status).toBe(200)
+            expect(capture).toHaveBeenCalledWith(
+                expect.objectContaining({ id: team.id }),
+                'workflows sandbox email delivered',
+                { is_test: false, recipient_count: count }
+            )
+        })
+    })
+
     describe('resolveEmailEngagementDistinctId', () => {
         const buildInvocation = (
             globals: Partial<NonNullable<CyclotronJobInvocationHogFunction['state']['globals']>>

nodejs/src/cdp/services/messaging/email-tracking.service.ts

@@ -24,6 +24,7 @@ import { EmailSuppressionService } from './email-suppression.service'
 import { escapeHtml } from './helpers/escape-html'
 import { SES_LINK_INDEX_TAG, SesWebhookHandler } from './helpers/ses'
 import { EmailTrackingCodeSigner, trackingCodeFormatCounter } from './helpers/tracking-code'
+import { SandboxEmailSender } from './sandbox-email-sender'
 
 export const PIXEL_GIF = Buffer.from('R0lGODlhAQABAAAAACH5BAEKAAEALAAAAAABAAEAAAICTAEAOw==', 'base64')
 const LINK_REGEX =
@@ -280,10 +281,15 @@ export class EmailTrackingService {
         private capturedEventsService: CapturedEventsService,
         private teamWorkflowsConfigService: TeamWorkflowsConfigService,
         private trackingCodeSigner: EmailTrackingCodeSigner,
-        private emailSuppressionService: EmailSuppressionService
+        private emailSuppressionService: EmailSuppressionService,
+        private sandboxEmailSender: SandboxEmailSender
     ) {
         const allowedTopicArns = (process.env.SES_ALLOWED_SNS_TOPIC_ARNS ?? '').split(',')
-        this.sesWebhookHandler = new SesWebhookHandler(this.trackingCodeSigner, allowedTopicArns)
+        this.sesWebhookHandler = new SesWebhookHandler(
+            this.trackingCodeSigner,
+            allowedTopicArns,
+            this.sandboxEmailSender.config.configurationSetName
+        )
     }
 
     public async trackMetric({
@@ -493,6 +499,7 @@ export class EmailTrackingService {
                 hardBounceRecipients,
                 complainedRecipients,
                 deliveredRecipients,
+                sandboxDeliveries,
             } = await this.sesWebhookHandler.handleWebhook({
                 body: parseJSON(req.body),
                 headers: req.headers,
@@ -577,6 +584,8 @@ export class EmailTrackingService {
                 emailTrackingErrorsCounter.inc({ error_type: 'suppression_update_failed', source: 'ses' })
             }
 
+            await this.captureSandboxDeliveries(sandboxDeliveries || [])
+
             return { status, message: body as string }
         } catch (error) {
             emailTrackingErrorsCounter.inc({ error_type: error.name || 'unknown' })
@@ -585,6 +594,17 @@ export class EmailTrackingService {
         }
     }
 
+    private async captureSandboxDeliveries(
+        deliveries: { teamId: string; isTest: boolean; recipientCount: number }[]
+    ): Promise<void> {
+        for (const { teamId, isTest, recipientCount } of deliveries) {
+            const parsedTeamId = Number(teamId)
+            if (Number.isSafeInteger(parsedTeamId) && parsedTeamId > 0) {
+                await this.sandboxEmailSender.capture(parsedTeamId, isTest, { type: 'delivered', recipientCount })
+            }
+        }
+    }
+
     private parseTrackingParams(query: Record<string, any>): {
         functionId?: string
         invocationId?: string

nodejs/src/cdp/services/messaging/helpers/ses.ts

@@ -396,7 +396,8 @@ export class SesWebhookHandler {
         private trackingCodeSigner: EmailTrackingCodeSigner,
         // Empty set means no restriction (dev/test); prod is expected to configure the workflow SES
         // topic ARN.
-        allowedTopicArns: string[] = []
+        allowedTopicArns: string[] = [],
+        private sandboxConfigurationSetName = ''
     ) {
         this.allowedTopicArns = new Set(allowedTopicArns.map((arn) => arn.trim()).filter(Boolean))
     }
@@ -612,6 +613,7 @@ export class SesWebhookHandler {
             emailAddresses: string[]
             timestamp?: string
         }[]
+        sandboxDeliveries?: { teamId: string; isTest: boolean; recipientCount: number }[]
     }> {
         // Payloads carry recipients and clicked links, including reset links from sensitive emails,
         // so the logs keep identifiers only.
@@ -714,6 +716,7 @@ export class SesWebhookHandler {
             emailAddresses: string[]
             timestamp?: string
         }[] = []
+        const sandboxDeliveries: { teamId: string; isTest: boolean; recipientCount: number }[] = []
 
         for (const rec of records) {
             logger.info('[SesWebhookHandler] processing record', {
@@ -877,6 +880,20 @@ export class SesWebhookHandler {
             // carrier is only used for engagement metrics/log entries above.
             const codeIsTrusted = parsedCode?.format === 'signed'
 
+            if (
+                rec.eventType === 'Delivery' &&
+                codeIsTrusted &&
+                teamId &&
+                this.sandboxConfigurationSetName &&
+                rec.mail.tags?.['ses:configuration-set']?.[0] === this.sandboxConfigurationSetName
+            ) {
+                sandboxDeliveries.push({
+                    teamId,
+                    isTest: isTest ?? false,
+                    recipientCount: rec.delivery.recipients?.length ?? 0,
+                })
+            }
+
             // Suppression writes (below) skip test sends — editor "Run test" traffic must not be
             // able to perturb production suppression state by targeting a bad recipient.
             const suppressionAllowed = teamId && codeIsTrusted && !isTest
@@ -937,6 +954,7 @@ export class SesWebhookHandler {
             hardBounceRecipients,
             complainedRecipients,
             deliveredRecipients,
+            sandboxDeliveries,
         }
     }
 }

nodejs/src/cdp/services/messaging/sandbox-email-sender.ts

@@ -51,6 +51,7 @@ export type SandboxDailyCapClaim =
 
 type SandboxEmailOutcome =
     | { type: 'sent'; recipientCount: number }
+    | { type: 'delivered'; recipientCount: number }
     | {
           type: 'blocked'
           reason:
@@ -63,6 +64,17 @@ type SandboxEmailOutcome =
           blockedRecipientCount: number
       }
 
+function outcomeProperties(outcome: SandboxEmailOutcome, isTest: boolean): Record<string, unknown> {
+    switch (outcome.type) {
+        case 'sent':
+            return { recipient_count: outcome.recipientCount, source: isTest ? 'test' : 'workflow' }
+        case 'blocked':
+            return { reason: outcome.reason, blocked_recipient_count: outcome.blockedRecipientCount }
+        case 'delivered':
+            return { recipient_count: outcome.recipientCount }
+    }
+}
+
 function dailyBucket(key: string, requested: number, capacity: number): ClaimRequest {
     return { key, requested, capacity, refillPerSecond: capacity / SECONDS_PER_DAY, ttlSeconds: DAILY_CAP_TTL_SECONDS }
 }
@@ -292,11 +304,10 @@ export class SandboxEmailSender {
             }
             const team = await this.teamManager.getTeam(teamId)
             if (team) {
-                const properties =
-                    outcome.type === 'sent'
-                        ? { recipient_count: outcome.recipientCount, source: isTest ? 'test' : 'workflow' }
-                        : { reason: outcome.reason, blocked_recipient_count: outcome.blockedRecipientCount }
-                captureTeamEvent(team, `workflows sandbox email ${outcome.type}`, { ...properties, is_test: isTest })
+                captureTeamEvent(team, `workflows sandbox email ${outcome.type}`, {
+                    ...outcomeProperties(outcome, isTest),
+                    is_test: isTest,
+                })
             }
         } catch (error) {
             logger.warn('Could not capture sandbox email event', { teamId, error })

products/workflows/backend/services/email_health.py

@@ -22,6 +22,7 @@
 from products.workflows.backend.models.hog_flow.hog_flow import HogFlow
 from products.workflows.backend.providers.ses import SESProvider
 from products.workflows.backend.services.email_sending_attribution import fold_email_totals_by_flow
+from products.workflows.backend.services.email_sending_tier import SANDBOX_SENT_METRIC, SENT_METRIC, own_sends
 from products.workflows.backend.utils.email_sending_tiers import max_email_sending_tier, resolve_team_email_sending_tier
 
 logger = structlog.get_logger(__name__)
@@ -89,9 +90,9 @@ def team_email_sending_allowance(team_id: int) -> EmailSendingAllowance:
 
 def _team_email_sends_since(team_id: int, after: datetime) -> int:
     totals = fetch_app_metric_totals_by_team_and_source(
-        app_source="hog_flow", name=["email_sent"], after=after, team_ids=[team_id]
+        app_source="hog_flow", name=[SENT_METRIC, SANDBOX_SENT_METRIC], after=after, team_ids=[team_id]
     )
-    return sum(counts.get("email_sent", 0) for counts in totals.get(team_id, {}).values())
+    return sum(own_sends(counts) for counts in totals.get(team_id, {}).values())
 
 
 AWS_TENANT_REPUTATION_CACHE_SECONDS = 5 * 60