feat(cdp): refuse a sandbox code send before it is queued

nodejs/src/cdp/cdp-api.ts

@@ -34,7 +34,7 @@ import {
 import { HogTransformerService, createHogTransformerService } from './hog-transformations/hog-transformer.service'
 import { RerunJobManager } from './rerun/rerun-job.manager'
 import { RerunRequest } from './rerun/rerun-job.types'
-import { HogFlow, HogFlowAction } from './schema/hogflow'
+import { HogFlowAction } from './schema/hogflow'
 import { BatchExportHogFunctionService, NotFoundError, ParseError } from './services/batch-export-hog-function.service'
 import type { CyclotronV2JobProducer } from './services/cyclotron-v2'
 import { HogExecutorAsyncService, HogExecutorExecuteAsyncOptions } from './services/hog-executor-async.service'
@@ -75,6 +75,12 @@ import { EmailTrackingService } from './services/messaging/email-tracking.servic
 import { EmailValidationService } from './services/messaging/email-validation.service'
 import { EmailTrackingCodeSigner } from './services/messaging/helpers/tracking-code'
 import { RecipientTokensService } from './services/messaging/recipient-tokens.service'
+import {
+    DeliveryMode,
+    SandboxSendGate,
+    SandboxSendRefusal,
+    SandboxSendRequest,
+} from './services/messaging/sandbox-send-gate'
 import {
     HogWatcherService,
     HogWatcherState,
@@ -173,6 +179,28 @@ function parseRealtimeCohort(value: unknown): BatchResolverState['realtimeCohort
 
 const ACCEPTED_SEND_FINGERPRINT_MARKER = 'accepted_send'
 
+const SANDBOX_REFUSAL_STATUS: Record<SandboxSendRefusal, number> = {
+    sandbox_recipient_not_allowed: 400,
+    sandbox_limit_reached: 400,
+    sending_suspended: 422,
+    sandbox_check_failed: 503,
+}
+
+type TransactionalEmailAnswer = { deliveryMode: DeliveryMode }
+
+type TransactionalEmailCheck =
+    | { type: 'refused'; outcome: TransactionalEmailOutcome; status: number }
+    | { type: 'admitted'; deliveryMode: DeliveryMode }
+
+const refused = (outcome: TransactionalEmailOutcome, status: number): TransactionalEmailCheck => ({
+    type: 'refused',
+    outcome,
+    status,
+})
+
+const deliveryModeBody = (answer: TransactionalEmailAnswer | undefined): { delivery_mode?: DeliveryMode } =>
+    answer ? { delivery_mode: answer.deliveryMode } : {}
+
 const transactionalEmailFingerprint = (
     hogFlowId: string,
     { to, distinct_id, variables, send }: TransactionalEmailRequest
@@ -232,11 +260,12 @@ export class CdpApi {
     private cancelBatchJwt: ScopedServiceJwt
     private stepResumeJwt: ScopedServiceJwt
     private transactionalEmailJwt: ScopedServiceJwt
-    private transactionalEmailIdempotencyKeys: IdempotencyKeys
+    private transactionalEmailIdempotencyKeys: IdempotencyKeys<TransactionalEmailAnswer>
     private emailSuppressionService: EmailSuppressionService
     private emailValidationService: EmailValidationService
     private teamWorkflowsConfigService: TeamWorkflowsConfigService
     private emailStatusService: EmailStatusService
+    private sandboxSendGate: SandboxSendGate
     private resumeBatchJwt: ScopedServiceJwt
 
     constructor(
@@ -262,6 +291,11 @@ export class CdpApi {
         this.hogWatcherMirror = services.hogWatcherMirror
         this.invocationResultsService = services.invocationResultsService
         this.emailStatusService = services.emailStatusService
+        this.sandboxSendGate = new SandboxSendGate(
+            services.sandboxEmailSender,
+            services.organizationMembers,
+            deps.integrationManager
+        )
 
         // API-only services. The hog-transformer's monitoring service reuses the same
         // resolved outputs registry as the core CDP services — no separate construction.
@@ -320,7 +354,7 @@ export class CdpApi {
             PosthogJwtAudience.WORKFLOWS_SEND_TRANSACTIONAL_EMAIL,
             config.WORKFLOWS_TRANSACTIONAL_EMAIL_JWT_SECRET || ''
         )
-        this.transactionalEmailIdempotencyKeys = new IdempotencyKeys(
+        this.transactionalEmailIdempotencyKeys = new IdempotencyKeys<TransactionalEmailAnswer>(
             deps.idempotencyValkey ?? null,
             'transactional-email'
         )
@@ -1311,6 +1345,7 @@ export class CdpApi {
                                 status: 'queued',
                                 invocation_id: reservation.invocationId,
                                 replayed: true,
+                                ...deliveryModeBody(reservation.answer),
                             })
                         case 'in_flight':
                             return reply('idempotency_key_in_flight', 409, { error: 'idempotency_key_in_flight' })
@@ -1320,16 +1355,24 @@ export class CdpApi {
                             return reply('idempotency_unavailable', 503, { error: 'idempotency_unavailable' })
                     }
                 }
+                let deliveryMode: DeliveryMode
                 try {
                     stages.start('checks')
-                    const refusal = await this.refuseTransactionalEmail(team.id, hogFlow, to)
-                    if (refusal) {
+                    const check = await this.checkTransactionalEmail({
+                        teamId: team.id,
+                        invocationId: invocation.id,
+                        hogFlow,
+                        send,
+                        to,
+                    })
+                    if (check.type === 'refused') {
                         if (idempotencyKey) {
                             stages.start('idempotency')
                             await idempotencyKeys.release(team.id, idempotencyKey, invocation.id)
                         }
-                        return reply(refusal.outcome, refusal.status, { error: refusal.outcome })
+                        return reply(check.outcome, check.status, { error: check.outcome })
                     }
+                    deliveryMode = check.deliveryMode
                     stages.start('queue_insert')
                     this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running')
                     try {
@@ -1349,7 +1392,9 @@ export class CdpApi {
                 }
                 if (idempotencyKey) {
                     stages.start('idempotency')
-                    await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id)
+                    await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id, {
+                        deliveryMode,
+                    })
                 }
                 stages.start('metrics_flush')
                 this.emailStatusService.queue({ teamId: team.id, messageId: invocation.id, status: 'queued' })
@@ -1364,7 +1409,12 @@ export class CdpApi {
                     await this.invocationResultsService.invocationResultsRowsService.handOffQueuedRows()
                 void produced.catch(warnTelemetryFlushFailed)
 
-                return reply('queued', 202, { status: 'queued', invocation_id: invocation.id, replayed: false })
+                return reply('queued', 202, {
+                    status: 'queued',
+                    invocation_id: invocation.id,
+                    replayed: false,
+                    delivery_mode: deliveryMode,
+                })
             } catch (e) {
                 logger.error('Error queueing transactional email', {
                     error: e instanceof Error ? e.message : String(e),
@@ -1378,32 +1428,33 @@ export class CdpApi {
     // The email worker skips a send for these reasons too, but only after the caller has its 202. Checking them
     // here tells the caller that the email will not go out. The worker keeps its checks, because the state can
     // change while the job waits on the queue.
-    private async refuseTransactionalEmail(
-        teamId: number,
-        hogFlow: HogFlow,
-        to: string
-    ): Promise<{ outcome: TransactionalEmailOutcome; status: number } | null> {
+    private async checkTransactionalEmail(request: SandboxSendRequest): Promise<TransactionalEmailCheck> {
+        const { teamId, hogFlow, to } = request
         if (hogFlow.status !== 'active') {
-            return { outcome: 'not_sendable', status: 400 }
+            return refused('not_sendable', 400)
         }
         if (await this.teamWorkflowsConfigService.getEmailSendingSuspension(teamId)) {
-            return { outcome: 'sending_suspended', status: 422 }
+            return refused('sending_suspended', 422)
         }
         if (hogFlow.email_sending_paused_at) {
-            return { outcome: 'sending_paused', status: 422 }
+            return refused('sending_paused', 422)
         }
         if (await this.emailSuppressionService.isSuppressed(teamId, to, { transactional: true })) {
-            return { outcome: 'recipient_suppressed', status: 422 }
+            return refused('recipient_suppressed', 422)
         }
         const { isLimited } = await checkHogFlowQuotaLimits(hogFlow, teamId, this.deps.quotaLimiting)
         if (isLimited) {
-            return { outcome: 'quota_limited', status: 429 }
+            return refused('quota_limited', 429)
+        }
+        const admission = await this.sandboxSendGate.admit(request)
+        if (admission.deliveryMode === 'sandbox' && admission.refusal) {
+            return refused(admission.refusal, SANDBOX_REFUSAL_STATUS[admission.refusal])
         }
         // Last, because a cache miss costs a DNS lookup. A DNS failure lets the send through.
         if (await this.emailValidationService.getRecipientSkipReason(teamId, to)) {
-            return { outcome: 'recipient_unreachable', status: 422 }
+            return refused('recipient_unreachable', 422)
         }
-        return null
+        return { type: 'admitted', deliveryMode: admission.deliveryMode }
     }
 
     private verifyScopedWorkflowJwt(

nodejs/src/cdp/cdp-services.ts

@@ -121,6 +121,7 @@ export interface CdpCoreServices {
     outputs: CdpOutputs
     emailService: EmailService
     sandboxEmailSender: SandboxEmailSender
+    organizationMembers: OrganizationMembersService
     /** Buffers rendered-email asset rows and bulk-flushes them at the batch boundary. */
     messageAssetsService: MessageAssetsService
     /** Buffers transactional email status rows. `invocationResultsService.flush()` produces them. */
@@ -489,6 +490,7 @@ export function createCdpCoreServices(
         sandboxEmailRateLimiter,
         new SandboxSenderStateService(deps.postgres, deps.pubSub)
     )
+    const organizationMembers = new OrganizationMembersService(deps.postgres)
     const emailService = new EmailService(
         {
             sesAccessKeyId: config.SES_ACCESS_KEY_ID,
@@ -513,7 +515,7 @@ export function createCdpCoreServices(
         workflowEmailRateLimiter,
         teamEmailRateLimiter,
         sandboxEmailSender,
-        new OrganizationMembersService(deps.postgres)
+        organizationMembers
     )
     const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL)
     const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields)
@@ -653,6 +655,7 @@ export function createCdpCoreServices(
         outputs,
         emailService,
         sandboxEmailSender,
+        organizationMembers,
         messageAssetsService,
         emailStatusService,
     }

nodejs/src/cdp/services/hogflows/idempotency-keys.ts

@@ -111,32 +111,33 @@ const sortKeys = (value: unknown): unknown => {
     return value
 }
 
-export type IdempotencyReservation =
+export type IdempotencyReservation<Answer extends object = object> =
     | { status: 'reserved' }
     // The key already queued a run for the same request.
-    | { status: 'replayed'; invocationId: string }
+    | { status: 'replayed'; invocationId: string; answer?: Answer }
     // The first request with the key has not finished queueing yet.
     | { status: 'in_flight' }
     // The key was used for a different request.
     | { status: 'conflict' }
     | { status: 'unavailable' }
 
-type StoredKey = { fingerprint: string; invocationId: string; queued: boolean }
+type StoredKey<Answer extends object> = { fingerprint: string; invocationId: string; queued: boolean; answer?: Answer }
 
-type PendingCompletion = { redisKey: string; value: StoredKey; expiresAt: number }
+type PendingCompletion<Answer extends object> = { redisKey: string; value: StoredKey<Answer>; expiresAt: number }
 
 // Keyed by run, not by Redis key. Several runs of one key can wait at once, for example a webhook
 // retry that queued without a reservation during the outage. Only the store knows which run owns
 // the key, so every run keeps its entry until its own write succeeds, loses ownership, or expires.
-const pendingCompletionKey = ({ redisKey, value }: PendingCompletion): string => `${redisKey}/${value.invocationId}`
+const pendingCompletionKey = <Answer extends object>({ redisKey, value }: PendingCompletion<Answer>): string =>
+    `${redisKey}/${value.invocationId}`
 
 /**
  * Caller-supplied idempotency keys, scoped to a team. A key reserves a run before it is queued, and
  * holds the run's id for 24 hours once it is. A queue failure releases the key, so the caller's
  * retry can queue the run.
  */
-export class IdempotencyKeys {
-    private pendingCompletions = new Map<string, PendingCompletion>()
+export class IdempotencyKeys<Answer extends object = object> {
+    private pendingCompletions = new Map<string, PendingCompletion<Answer>>()
     private draining = false
 
     constructor(
@@ -156,7 +157,7 @@ export class IdempotencyKeys {
         key: string,
         fingerprint: string,
         invocationId: string
-    ): Promise<IdempotencyReservation> {
+    ): Promise<IdempotencyReservation<Answer>> {
         const reservation = await this.reserveInStore(teamId, key, fingerprint, invocationId)
         this.count('reserve', reservation.status)
         return reservation
@@ -167,12 +168,12 @@ export class IdempotencyKeys {
         key: string,
         fingerprint: string,
         invocationId: string
-    ): Promise<IdempotencyReservation> {
+    ): Promise<IdempotencyReservation<Answer>> {
         const valkey = this.valkey
         if (!valkey) {
             return { status: 'unavailable' }
         }
-        const value: StoredKey = { fingerprint, invocationId, queued: false }
+        const value: StoredKey<Answer> = { fingerprint, invocationId, queued: false }
         let existing: string | null
         try {
             // SET ... NX GET sets the key when absent and returns the existing value when present.
@@ -196,17 +197,25 @@ export class IdempotencyKeys {
         if (existing === null) {
             return { status: 'reserved' }
         }
-        const stored = parseJSON(existing) as StoredKey
+        const stored = parseJSON(existing) as StoredKey<Answer>
         if (stored.fingerprint !== fingerprint) {
             return { status: 'conflict' }
         }
-        return stored.queued ? { status: 'replayed', invocationId: stored.invocationId } : { status: 'in_flight' }
+        return stored.queued
+            ? { status: 'replayed', invocationId: stored.invocationId, answer: stored.answer }
+            : { status: 'in_flight' }
     }
 
-    async markQueued(teamId: number, key: string, fingerprint: string, invocationId: string): Promise<void> {
-        const completion: PendingCompletion = {
+    async markQueued(
+        teamId: number,
+        key: string,
+        fingerprint: string,
+        invocationId: string,
+        answer?: Answer
+    ): Promise<void> {
+        const completion: PendingCompletion<Answer> = {
             redisKey: this.redisKey(teamId, key),
-            value: { fingerprint, invocationId, queued: true },
+            value: { fingerprint, invocationId, queued: true, answer },
             expiresAt: Date.now() + QUEUED_TTL_SECONDS * 1000,
         }
         for (let attempt = 1; attempt <= MARK_QUEUED_INLINE_ATTEMPTS; attempt++) {
@@ -225,7 +234,7 @@ export class IdempotencyKeys {
         this.retryInBackground(completion)
     }
 
-    private retryInBackground(completion: PendingCompletion): void {
+    private retryInBackground(completion: PendingCompletion<Answer>): void {
         if (this.pendingCompletions.size >= MAX_PENDING_COMPLETIONS) {
             logger.error('🔑', 'idempotency key completion dropped, too many waiting for the store', {
                 scope: this.scope,
@@ -270,7 +279,7 @@ export class IdempotencyKeys {
         return true
     }
 
-    private async writeQueued({ redisKey, value, expiresAt }: PendingCompletion): Promise<void> {
+    private async writeQueued({ redisKey, value, expiresAt }: PendingCompletion<Answer>): Promise<void> {
         const ttlSeconds = Math.ceil((expiresAt - Date.now()) / 1000)
         const written = await this.timed(
             'mark_queued',

nodejs/src/cdp/services/hogflows/transactional-email.ts

@@ -66,6 +66,9 @@ export type TransactionalEmailOutcome =
     | 'sending_paused'
     | 'recipient_suppressed'
     | 'recipient_unreachable'
+    | 'sandbox_recipient_not_allowed'
+    | 'sandbox_limit_reached'
+    | 'sandbox_check_failed'
     | 'error'
 
 export const transactionalEmailRequestsTotal = new Counter({
@@ -88,7 +91,7 @@ export type TransactionalEmailStage = (typeof TRANSACTIONAL_EMAIL_STAGES)[number
 
 export const transactionalEmailStageDurationMs = new Histogram({
     name: 'cdp_transactional_email_stage_duration_ms',
-    help: 'Milliseconds one transactional email request spends in each stage: auth (the JWT), validate (the body), load (the team and the workflow), build (the run and its idempotency fingerprint), idempotency (every Idempotency-Key call to Valkey), checks (suspension, pause, suppression, quota and MX), queue_insert (the email queue insert), metrics_flush (the hand-off of the result and status rows to the Kafka producer, without the wait for the broker), and total. A request records only the stages it reached, so a refusal shows in the stage that refused it.',
+    help: 'Milliseconds one transactional email request spends in each stage: auth (the JWT), validate (the body), load (the team and the workflow), build (the run and its idempotency fingerprint), idempotency (every Idempotency-Key call to Valkey), checks (suspension, pause, suppression, quota, the sandbox member, pause and allowance checks, and MX), queue_insert (the email queue insert), metrics_flush (the hand-off of the result and status rows to the Kafka producer, without the wait for the broker), and total. A request records only the stages it reached, so a refusal shows in the stage that refused it.',
     labelNames: ['stage'],
     buckets: [1, 2.5, 5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000],
 })

nodejs/src/cdp/services/managers/integration-manager.service.test.ts

@@ -70,6 +70,15 @@ describe('IntegrationManager', () => {
         expect(item?.config).toEqual({ team: 'foobar' })
     })
 
+    it('finds an integration the read replica does not have yet on the primary', async () => {
+        const query = jest.spyOn(hub.postgres, 'query').mockResolvedValueOnce({ rows: [] } as any)
+
+        const item = await manager.get(integrations[0].id)
+
+        expect(item?.config).toEqual({ team: 'foobar' })
+        expect(query.mock.calls.map(([use]) => use)).toEqual([PostgresUse.COMMON_READ, PostgresUse.COMMON_WRITE])
+    })
+
     it('updates cached integration data when integration changes', async () => {
         // First check - initial state
         const item = await manager.get(integrations[0].id)