feat(workflows): keep sandbox sends out of the sending tier

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

@@ -39,6 +39,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.
@@ -92,5 +93,7 @@ The worker checks the pause again after the daily allowance claim and gives the
 Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker.
 A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state.
 Each delivered send emits a `workflows sandbox email sent` event.
-Successful sandbox sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric.
-Keep the switch off in production until that metric is deployed.
+A successful sandbox send records `email_sandbox_sent` next to `email_sent`, so the workflow's metrics still show it.
+The sending tier subtracts `email_sandbox_sent` from `email_sent` per day, and the team's sending allowance per workflow, both floored at zero. Sandbox bounces and complaints still count.
+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 this metric to the worker and Django before you enable the switch in production.

nodejs/src/cdp/cdp-api.ts

@@ -246,7 +246,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,
+        redis,
+        sandboxEmailRateLimiter,
+        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,
-            redis,
-            sandboxEmailRateLimiter,
-            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.captureDelivered(parsedTeamId, isTest, recipientCount)
+            }
+        }
+    }
+
     private parseTrackingParams(query: Record<string, any>): {
         functionId?: string
         invocationId?: string

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

@@ -509,7 +509,7 @@ describe('EmailService', () => {
                         }),
                     ])
                     expect(result.metrics.map((metric) => metric.metric_name)).toEqual(
-                        isTest ? [] : ['email_sent', 'email_untracked']
+                        isTest ? [] : ['email_sent', 'email_untracked', 'email_sandbox_sent']
                     )
                     expect(capture).toHaveBeenCalledWith(
                         expect.objectContaining({ id: team.id }),

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

@@ -897,6 +897,17 @@ export class EmailService {
                 })
             }
 
+            if (success && isSandbox) {
+                result.metrics.push({
+                    team_id: invocation.teamId,
+                    app_source_id: invocation.parentRunId ?? invocation.functionId,
+                    instance_id: invocation.state.actionId || invocation.id,
+                    metric_kind: 'email',
+                    metric_name: 'email_sandbox_sent',
+                    count: 1,
+                })
+            }
+
             if (success && assetRow) {
                 result.messageAssets.push(assetRow)
             }

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

@@ -266,6 +266,13 @@ export class SandboxEmailSender {
         })
     }
 
+    public async captureDelivered(teamId: number, isTest: boolean, recipientCount: number): Promise<void> {
+        await this.captureEvent(teamId, 'workflows sandbox email delivered', {
+            recipient_count: recipientCount,
+            is_test: isTest,
+        })
+    }
+
     public async captureBlocked(
         teamId: number,
         isTest: boolean,

nodejs/src/cdp/types.ts

@@ -269,6 +269,7 @@ export type MinimalAppMetric = {
         | 'email_blocked'
         | 'email_unsubscribed'
         | 'email_untracked'
+        | 'email_sandbox_sent'
         | 'push_sent'
         | 'push_failed'
         | 'push_skipped'

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