feat(cdp): send sandbox email from the sandbox worker

nodejs/src/cdp/cdp-services.ts

@@ -46,6 +46,10 @@ import { MessageAssetsService } from './services/messaging/message-assets.servic
 import { PushNotificationService } from './services/messaging/push-notification.service'
 import { RecipientPreferencesService } from './services/messaging/recipient-preferences.service'
 import { RecipientTokensService } from './services/messaging/recipient-tokens.service'
+import { SandboxEmailRuntime } from './services/messaging/sandbox/sandbox-email-runtime'
+import { SandboxEmailSendStep } from './services/messaging/sandbox/sandbox-email-step'
+import { SandboxEmailTester } from './services/messaging/sandbox/sandbox-email-tester'
+import { deliveredAddressReader } from './services/messaging/sandbox/sandbox-render'
 import {
     HogFunctionMonitoringService,
     parseInternalLogsSources,
@@ -177,6 +181,8 @@ export type CdpCoreServicesConfig = Pick<
         | 'CDP_WATCHER_OBSERVE_RESULTS_BUFFER_MAX_RESULTS'
         | 'CDP_HOGFLOW_AWAITED_STEPS_ENABLED'
         | 'WORKFLOWS_SANDBOX_SENDER_ENABLED'
+        | 'WORKFLOWS_SANDBOX_EMAIL_TEST_URL'
+        | 'WORKFLOWS_SANDBOX_EMAIL_TEST_JWT_SECRET'
         | 'CDP_INTERNAL_LOGS_SOURCES'
         | 'SES_ACCESS_KEY_ID'
         | 'SES_SECRET_ACCESS_KEY'
@@ -253,6 +259,7 @@ export interface CdpCoreServicesDeps {
      * `null` elsewhere, and when no host is configured, which turns the cap off.
      */
     frequencyCapValkey: RedisV2 | null
+    sandboxEmail?: SandboxEmailRuntime | null
 }
 
 /**
@@ -598,6 +605,7 @@ export function createCdpCoreServices(
         createUsageIngestionClient(config, 'cdp'),
         usageReportTeamMatcher(config)
     )
+    const sandboxDeliveredAddress = deliveredAddressReader(deps.postgres)
     const hogFlowExecutor = new HogFlowExecutorService(
         hogFlowFunctionsService,
         recipientPreferencesService,
@@ -609,7 +617,30 @@ export function createCdpCoreServices(
         cdpUsageReporter,
         {
             awaitedStepsEnabled: config.CDP_HOGFLOW_AWAITED_STEPS_ENABLED,
-            sandboxEmail: { enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED },
+            sandboxEmail: {
+                enabled: config.WORKFLOWS_SANDBOX_SENDER_ENABLED,
+                send: deps.sandboxEmail
+                    ? new SandboxEmailSendStep({
+                          service: deps.sandboxEmail.service,
+                          events: deps.sandboxEmail.events,
+                          hogInputsService,
+                          deliveredAddress: sandboxDeliveredAddress,
+                          siteUrl: config.SITE_URL,
+                          recipientPreferences: recipientPreferencesService,
+                          sesBucket: deps.sandboxEmail.sesBucket,
+                      })
+                    : null,
+                testClient: new SandboxEmailTester({
+                    hogInputsService,
+                    deliveredAddress: sandboxDeliveredAddress,
+                    siteUrl: config.SITE_URL,
+                    endpoint: config.WORKFLOWS_SANDBOX_EMAIL_TEST_URL,
+                    jwt: new ScopedServiceJwt(
+                        PosthogJwtAudience.WORKFLOWS_SANDBOX_EMAIL_TEST,
+                        config.WORKFLOWS_SANDBOX_EMAIL_TEST_JWT_SECRET
+                    ),
+                }),
+            },
         }
     )
 

nodejs/src/cdp/config.ts

@@ -75,6 +75,16 @@ export type CdpConfig = ClickhouseConfig & {
     // Off until Django emits `$workflow_step_resume`, or parked steps never wake.
     CDP_HOGFLOW_AWAITED_STEPS_ENABLED: boolean
     WORKFLOWS_SANDBOX_SENDER_ENABLED: boolean
+    SES_SANDBOX_CONFIGURATION_SET: string
+    SES_SANDBOX_TENANT_NAME: string
+    WORKFLOWS_SANDBOX_DAILY_TEAM_CAP: number
+    WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP: number
+    WORKFLOWS_SANDBOX_CAPS_VALKEY_HOST: string
+    WORKFLOWS_SANDBOX_CAPS_VALKEY_PORT: number
+    WORKFLOWS_SANDBOX_CAPS_VALKEY_PASSWORD: string
+    WORKFLOWS_SANDBOX_CAPS_VALKEY_TLS: boolean
+    WORKFLOWS_SANDBOX_EMAIL_TEST_URL: string
+    WORKFLOWS_SANDBOX_EMAIL_TEST_JWT_SECRET: string
     // Comma-separated log sources, such as `hog_flow`, that also send their logs to the logs product. Empty sends none.
     CDP_INTERNAL_LOGS_SOURCES: string
     // Django produces cdp_internal_events to the cyclotron cluster; empty inherits the consumer default.
@@ -362,6 +372,16 @@ export function getDefaultCdpConfig(): CdpConfig {
         CDP_CYCLOTRON_COMPRESS_KAFKA_DATA: true,
         CDP_HOGFLOW_AWAITED_STEPS_ENABLED: isProdEnv() ? false : true,
         WORKFLOWS_SANDBOX_SENDER_ENABLED: false,
+        SES_SANDBOX_CONFIGURATION_SET: '',
+        SES_SANDBOX_TENANT_NAME: '',
+        WORKFLOWS_SANDBOX_DAILY_TEAM_CAP: 0,
+        WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP: 0,
+        WORKFLOWS_SANDBOX_CAPS_VALKEY_HOST: '',
+        WORKFLOWS_SANDBOX_CAPS_VALKEY_PORT: 6379,
+        WORKFLOWS_SANDBOX_CAPS_VALKEY_PASSWORD: '',
+        WORKFLOWS_SANDBOX_CAPS_VALKEY_TLS: false,
+        WORKFLOWS_SANDBOX_EMAIL_TEST_URL: '',
+        WORKFLOWS_SANDBOX_EMAIL_TEST_JWT_SECRET: '',
         CDP_INTERNAL_LOGS_SOURCES: '',
         CDP_INTERNAL_EVENTS_CONSUMER_METADATA_BROKER_LIST: '',
         CDP_REDIS_HOST: '127.0.0.1',

nodejs/src/cdp/services/hog-inputs.service.ts

@@ -52,6 +52,8 @@ const accessTokenPlaceholderBase = (integrationId: number): string => `${ACCESS_
 
 const signedAccessTokenPlaceholder = (base: string, signature: string): string => `${base}_${signature.slice(0, 32)}`
 
+export type BuildInputsOptions = { recipientIdentifier?: (email: string) => string | Promise<string> }
+
 export class HogInputsService {
     constructor(
         private integrationManager: IntegrationManagerService,
@@ -64,7 +66,8 @@ export class HogInputsService {
         globals: HogFunctionInvocationGlobals,
         additionalInputs?: Record<string, any>,
         preferencesTokenSource?: PreferencesTokenSource,
-        signal?: AbortSignal
+        signal?: AbortSignal,
+        { recipientIdentifier = (email: string) => email }: BuildInputsOptions = {}
     ): Promise<Record<string, any>> {
         // TODO: Load the values from the integrationManager
         const newGlobals: HogFunctionInvocationGlobalsWithInputs = {
@@ -145,7 +148,11 @@ export class HogInputsService {
             // If we have an email value then we template it out to get the email address
             const emailValue = await _formatInput(emailInput, emailInputSchema.key)
             if (emailValue?.to?.email) {
-                const recipient = { team_id: hogFunction.team_id, identifier: emailValue.to.email }
+                const identifier = recipientIdentifier(emailValue.to.email)
+                const recipient = {
+                    team_id: hogFunction.team_id,
+                    identifier: typeof identifier === 'string' ? identifier : await identifier,
+                }
                 newGlobals.unsubscribe_url = this.recipientTokensService.generatePreferencesUrl(
                     recipient,
                     preferencesTokenSource
@@ -186,11 +193,19 @@ export class HogInputsService {
         globals: HogFunctionInvocationGlobals,
         additionalInputs?: Record<string, any>,
         preferencesTokenSource?: PreferencesTokenSource,
-        signal?: AbortSignal
+        signal?: AbortSignal,
+        options?: BuildInputsOptions
     ): Promise<HogFunctionInvocationGlobalsWithInputs> {
         return {
             ...globals,
-            inputs: await this.buildInputs(hogFunction, globals, additionalInputs, preferencesTokenSource, signal),
+            inputs: await this.buildInputs(
+                hogFunction,
+                globals,
+                additionalInputs,
+                preferencesTokenSource,
+                signal,
+                options
+            ),
         }
     }
 

nodejs/src/cdp/services/hogflows/actions/email_sandbox.test.ts

@@ -236,6 +236,13 @@ describe('SandboxEmailActionHandler', () => {
             { status: 'pending', reason: 'pause_state_unavailable' },
             undefined,
         ],
+        [
+            'a real test over the shared send rate asks the editor to try again',
+            { isTest: true },
+            { send: () => Promise.resolve({ status: 'failed', reason: 'send_rate_limited' }) },
+            { status: 'failed', reason: 'send_rate_limited' },
+            'Too many emails are being sent right now. Send the test again in a moment.',
+        ],
     ])('runs a test inline: %s', async (_name, hogExecutorOptions, testClient, execResult, error) => {
         const mailbox = new FakeMailbox()
 

nodejs/src/cdp/services/hogflows/actions/email_sandbox.ts

@@ -58,14 +58,15 @@ export class SandboxEmailActionHandler implements ActionHandler {
     }
 
     private async runTest(options: ActionHandlerOptions<SandboxEmailAction>): Promise<ActionHandlerResult> {
-        if (options.hogExecutorOptions?.functions) {
-            this.log(options, 'Mocked the sandbox email send')
-            return this.finish(options, { status: 'accepted' })
-        }
+        const mocked = !!options.hogExecutorOptions?.functions
         if (!this.testClient) {
-            return this.finish(options, toStepResult(WORKER_UNAVAILABLE))
+            return this.finish(options, mocked ? { status: 'accepted' } : toStepResult(WORKER_UNAVAILABLE))
+        }
+        const stepResult = await this.testClient.send(options)
+        if (mocked && stepResult.status === 'accepted') {
+            this.log(options, 'Mocked the sandbox email send')
         }
-        return this.finish(options, await this.testClient.send(options))
+        return this.finish(options, stepResult)
     }
 
     private routeToSandboxQueue(
@@ -111,7 +112,7 @@ export class SandboxEmailActionHandler implements ActionHandler {
         stepResult: SandboxEmailStepResult
     ): ActionHandlerResult {
         if (stepResult.status === 'failed') {
-            return { result: stepResult, error: new Error(`Sandbox email was not sent: ${stepResult.reason}`) }
+            return { result: stepResult, error: new Error(failureMessage(stepResult.reason)) }
         }
         this.log(
             options,
@@ -131,6 +132,14 @@ export class SandboxEmailActionHandler implements ActionHandler {
     }
 }
 
+const FAILURE_MESSAGES: Record<string, string> = {
+    send_rate_limited: 'Too many emails are being sent right now. Send the test again in a moment.',
+}
+
+function failureMessage(reason: string | undefined): string {
+    return (reason && FAILURE_MESSAGES[reason]) ?? `Sandbox email was not sent: ${reason}`
+}
+
 function toStepResult({ outcome, reason, messageId }: SandboxEmailOutcome): SandboxEmailStepResult {
     return { status: outcome, ...(reason ? { reason } : {}), ...(messageId ? { message_id: messageId } : {}) }
 }

nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts

@@ -19,6 +19,7 @@ import { template as setVariableTemplate } from '~/cdp/templates/_destinations/p
 import { compileHog } from '~/cdp/templates/compiler'
 import { convertToHogFunctionFilterGlobal } from '~/cdp/utils/hog-function-filtering'
 import { createRedisV2PoolFromConfig } from '~/common/redis/redis-v2'
+import { FakeUsageIngestion } from '~/common/usage-ingestion/_tests/fake-usage-ingestion'
 import { createHub } from '~/common/utils/db/hub'
 import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres'
 import { parseJSON } from '~/common/utils/json-parse'
@@ -50,7 +51,11 @@ import { EmailTrackingCodeSigner } from '../messaging/helpers/tracking-code'
 import { PushNotificationFetchUtils, PushNotificationService } from '../messaging/push-notification.service'
 import { RecipientPreferencesService } from '../messaging/recipient-preferences.service'
 import { RecipientTokensService } from '../messaging/recipient-tokens.service'
+import { SandboxEmailService } from '../messaging/sandbox/sandbox-email.service'
+import { SandboxEmailSendStep } from '../messaging/sandbox/sandbox-email-step'
+import { SandboxMessage } from '../messaging/sandbox/sandbox-message'
 import { RateLimiterService } from '../rate-limiter/rate-limiter.service'
+import { CdpUsageReporterService } from '../usage/cdp-usage-reporter.service'
 import { HogFlowExecutorService, createHogFlowInvocation } from './hogflow-executor.service'
 import { HogFlowFunctionsService } from './hogflow-functions.service'
 import { TRANSACTIONAL_EMAIL_EVENT } from './hogflow-sensitivity'
@@ -183,6 +188,8 @@ describe('Hogflow Executor', () => {
     let executor: HogFlowExecutorService
     let cohortMembershipRepository: CohortMembershipRepository
     let hogFlowFunctionsService: HogFlowFunctionsService
+    let recipientPreferencesService: RecipientPreferencesService
+    let emailValidationService: EmailValidationService
     let hub: Hub
     let team: Team
     let integrationId: number
@@ -300,14 +307,14 @@ describe('Hogflow Executor', () => {
             hogExecutor
         )
         const recipientsManager = new RecipientsManagerService(hub.postgres)
-        const recipientPreferencesService = new RecipientPreferencesService(
+        recipientPreferencesService = new RecipientPreferencesService(
             recipientsManager,
             new EmailSuppressionService(hub.postgres, emailSuppressionConfigFromEnv())
         )
         // Stubbed to always allow: this suite covers executor routing and flow control,
         // not MX validation (email-validation.service.test.ts does), and the real
         // service would fire live DNS lookups for the fixture recipients here.
-        const emailValidationService = {
+        emailValidationService = {
             getSkipReason: () => Promise.resolve(null),
         } as unknown as EmailValidationService
         return {
@@ -3290,7 +3297,74 @@ describe('Hogflow Executor', () => {
                 .build()
         }
 
-        it('should record billing metrics for both regular hog functions and email functions', async () => {
+        it('bills hog functions and own-domain email, and never a sandbox email in the same run', async () => {
+            const usageIngestion = new FakeUsageIngestion()
+            const usageReporter = new CdpUsageReporterService(await usageIngestion.start(), () => true)
+            const sandboxMailbox: SandboxMessage[] = []
+            const sandboxSes = {
+                send: (message: SandboxMessage) => {
+                    sandboxMailbox.push(message)
+                    return Promise.resolve({ status: 'accepted' as const, messageId: 'ses-1' })
+                },
+            }
+            executor = new HogFlowExecutorService(
+                hogFlowFunctionsService,
+                recipientPreferencesService,
+                emailValidationService,
+                cohortMembershipRepository,
+                hub.integrationManager,
+                null,
+                undefined,
+                usageReporter,
+                {
+                    sandboxEmail: {
+                        enabled: true,
+                        send: new SandboxEmailSendStep({
+                            service: new SandboxEmailService({
+                                enabled: true,
+                                fromAddress: 'sandbox@send.example.com',
+                                tenantName: 'sandbox-tenant',
+                                isProvisioned: () => Promise.resolve(true),
+                                checkRecipient: () => Promise.resolve({ ok: true }),
+                                authorize: () =>
+                                    Promise.resolve({
+                                        ok: true,
+                                        organizationId: 'org-1',
+                                        organizationName: 'Acme',
+                                        storedAddress: 'member@example.com',
+                                        tenant: { sendingStatus: 'ENABLED', syncedAt: new Date(Date.now()) },
+                                    }),
+                                claimCaps: () =>
+                                    Promise.resolve({
+                                        ok: true,
+                                        markSending: () => Promise.resolve(true),
+                                        releaseUnsent: () => Promise.resolve(true),
+                                        refundUnsent: () => Promise.resolve(true),
+                                    }),
+                                teamWorkflowsConfig: new TeamWorkflowsConfigService(hub.postgres, hub.pubSub),
+                                ses: sandboxSes,
+                                now: () => new Date(Date.now()),
+                            }),
+                            hogInputsService: new HogInputsService(
+                                hub.integrationManager,
+                                new RecipientTokensService(hub.ENCRYPTION_SALT_KEYS, hub.SITE_URL),
+                                hub.encryptedFields
+                            ),
+                            deliveredAddress: () => Promise.resolve('member@example.com'),
+                            siteUrl: hub.SITE_URL,
+                            recipientPreferences: recipientPreferencesService,
+                            events: { capture: () => Promise.resolve() },
+                            sesBucket: {
+                                limiter: { claimUpTo: () => Promise.resolve(1) } as unknown as RateLimiterService,
+                                key: '@posthog/ses/billing-test',
+                                capacity: 1,
+                                refillPerSecond: 1,
+                            },
+                        }),
+                    },
+                }
+            )
+
             await insertIntegration(hub.postgres, team.id, {
                 id: integrationId,
                 kind: 'email',
@@ -3398,13 +3472,24 @@ describe('Hogflow Executor', () => {
                             },
                         },
                     },
+                    sandbox_email: {
+                        type: 'email_sandbox',
+                        config: {
+                            inputs: {
+                                email: {
+                                    value: { to: { email: 'member@example.com' }, subject: 'Sandbox', text: 'Hi' },
+                                },
+                            },
+                        },
+                    },
                 },
                 edges: [
                     { from: 'trigger', to: 'function_1', type: 'continue' },
                     { from: 'function_1', to: 'function_2', type: 'continue' },
                     { from: 'function_2', to: 'email_1', type: 'continue' },
                     { from: 'email_1', to: 'email_2', type: 'continue' },
-                    { from: 'email_2', to: 'exit', type: 'continue' },
+                    { from: 'email_2', to: 'sandbox_email', type: 'continue' },
+                    { from: 'sandbox_email', to: 'exit', type: 'continue' },
                 ],
             })
 
@@ -3440,6 +3525,24 @@ describe('Hogflow Executor', () => {
                 (m) => m.metric_kind === 'email' && m.metric_name === 'billable_invocation'
             )
             expect(emailBilling).toHaveLength(2)
+
+            expect(sandboxMailbox.map((message) => message.to.address)).toEqual(['member@example.com'])
+            await usageReporter.shutdown()
+            await usageIngestion.stop()
+            expect(
+                usageIngestion.accepted.map(({ teamId, usageKey, quantity }) => [teamId, usageKey, quantity]).sort()
+            ).toEqual([
+                [invocation.teamId, 'workflow_billable_invocations', 1],
+                [invocation.teamId, 'workflow_billable_invocations', 1],
+                [invocation.teamId, 'workflow_emails_sent', 1],
+                [invocation.teamId, 'workflow_emails_sent', 1],
+            ])
+            expect(
+                metrics.filter((m) => m.instance_id === 'sandbox_email').map((m) => [m.metric_kind, m.metric_name])
+            ).toEqual([
+                ['email', 'email_sandbox_sent'],
+                ['success', 'succeeded'],
+            ])
         })
 
         const suspendEmailSending = async (): Promise<void> => {