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,7 @@ 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 { SandboxSendGate, SandboxSendRefusal, SandboxSendRequest } from './services/messaging/sandbox-send-gate'
 import {
     HogWatcherService,
     HogWatcherState,
@@ -173,6 +174,23 @@ 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 TransactionalEmailCheck =
+    | { type: 'refused'; outcome: TransactionalEmailOutcome; status: number }
+    | { type: 'admitted' }
+
+const refused = (outcome: TransactionalEmailOutcome, status: number): TransactionalEmailCheck => ({
+    type: 'refused',
+    outcome,
+    status,
+})
+
 const transactionalEmailFingerprint = (
     hogFlowId: string,
     { to, distinct_id, variables, send }: TransactionalEmailRequest
@@ -237,6 +255,7 @@ export class CdpApi {
     private emailValidationService: EmailValidationService
     private teamWorkflowsConfigService: TeamWorkflowsConfigService
     private emailStatusService: EmailStatusService
+    private sandboxSendGate: SandboxSendGate
     private resumeBatchJwt: ScopedServiceJwt
 
     constructor(
@@ -262,6 +281,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.
@@ -1322,13 +1346,19 @@ export class CdpApi {
                 }
                 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 })
                     }
                     stages.start('queue_insert')
                     this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running')
@@ -1378,32 +1408,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' }
     }
 
     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/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)

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

@@ -39,16 +39,26 @@ export class IntegrationManagerService {
     private async fetchIntegrations(ids: string[]): Promise<Record<string, IntegrationType | undefined>> {
         logger.info('[IntegrationManager]', 'Fetching integrations', { ids })
 
+        const fromReplica = await this.queryIntegrations(PostgresUse.COMMON_READ, ids)
+        const missingFromReplica = ids.filter((id) => !fromReplica[id])
+        if (!missingFromReplica.length) {
+            return fromReplica
+        }
+        return { ...fromReplica, ...(await this.queryIntegrations(PostgresUse.COMMON_WRITE, missingFromReplica)) }
+    }
+
+    private async queryIntegrations(
+        use: PostgresUse,
+        ids: string[]
+    ): Promise<Record<string, IntegrationType | undefined>> {
         const response = await this.postgres.query<IntegrationType>(
-            PostgresUse.COMMON_READ,
+            use,
             `SELECT id, team_id, kind, config, sensitive_config FROM posthog_integration WHERE id = ANY($1)`,
             [ids],
             'fetchIntegrations'
         )
 
-        const items = response.rows
-
-        return items.reduce<Record<string, IntegrationType | undefined>>((acc, item) => {
+        return response.rows.reduce<Record<string, IntegrationType | undefined>>((acc, item) => {
             const sensitiveConfig = this.encryptedFields.decryptObject(item.sensitive_config || {}, {
                 ignoreDecryptionErrors: true,
             })

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

@@ -54,6 +54,19 @@ export type SandboxDailyCapClaim =
     | { type: 'recipient_cap_reached'; addresses: string[] }
     | { type: 'check_failed' }
 
+type DailyCapBuckets = { limiter: RateLimiterService; buckets: ClaimRequest[]; addresses: string[] }
+
+type SandboxDailyCapDenial = { type: 'project_cap_reached' } | { type: 'recipient_cap_reached'; addresses: string[] }
+
+export type SandboxDailyCapPeek = { type: 'available' } | SandboxDailyCapDenial | { type: 'check_failed' }
+
+function dailyCapDenial({ addresses }: DailyCapBuckets, deniedIndexes: number[]): SandboxDailyCapDenial {
+    if (deniedIndexes.includes(0)) {
+        return { type: 'project_cap_reached' }
+    }
+    return { type: 'recipient_cap_reached', addresses: deniedIndexes.map((index) => addresses[index - 1]) }
+}
+
 function dailyBucket(key: string, requested: number, capacity: number): ClaimRequest {
     return { key, requested, capacity, refillPerSecond: capacity / SECONDS_PER_DAY, ttlSeconds: DAILY_CAP_TTL_SECONDS }
 }
@@ -212,15 +225,39 @@ export class SandboxEmailSender {
     }
 
     public async claimDailyCaps(teamId: number, recipients: string[]): Promise<SandboxDailyCapClaim> {
+        const caps = await this.dailyCapBuckets(teamId, recipients)
+        if (!caps) {
+            return { type: 'check_failed' }
+        }
+        const claim = await caps.limiter.claimAllOrNothing(caps.buckets)
+        if (claim.granted) {
+            return { type: 'granted', claim, buckets: caps.buckets }
+        }
+        return claim.deniedIndexes === null ? { type: 'check_failed' } : dailyCapDenial(caps, claim.deniedIndexes)
+    }
+
+    public async peekDailyCaps(teamId: number, recipients: string[]): Promise<SandboxDailyCapPeek> {
+        const caps = await this.dailyCapBuckets(teamId, recipients)
+        if (!caps) {
+            return { type: 'check_failed' }
+        }
+        const peek = await caps.limiter.peekAllOrNothing(caps.buckets)
+        if (!peek) {
+            return { type: 'check_failed' }
+        }
+        return peek.deniedIndexes.length ? dailyCapDenial(caps, peek.deniedIndexes) : { type: 'available' }
+    }
+
+    private async dailyCapBuckets(teamId: number, recipients: string[]): Promise<DailyCapBuckets | null> {
         const dailyTeamCap = parseDailyCap(this.config.dailyTeamCap)
         const dailyRecipientCap = parseDailyCap(this.config.dailyRecipientCap)
         if (!this.dailyCapLimiter || dailyTeamCap === null || dailyRecipientCap === null) {
             logger.error('Sandbox email daily caps are not configured correctly', { teamId })
-            return { type: 'check_failed' }
+            return null
         }
         const team = await this.teamManager.getTeam(teamId)
         if (!team) {
-            return { type: 'check_failed' }
+            return null
         }
         const prefix = dailyCapKeyPrefix(team.organization_id)
         const addresses = distinctAddresses(recipients)
@@ -230,17 +267,7 @@ export class SandboxEmailSender {
                 dailyBucket(`${prefix}/recipient/${addressDigest(address)}`, 1, dailyRecipientCap)
             ),
         ]
-        const claim = await this.dailyCapLimiter.claimAllOrNothing(buckets)
-        if (claim.granted) {
-            return { type: 'granted', claim, buckets }
-        }
-        if (claim.deniedIndexes === null) {
-            return { type: 'check_failed' }
-        }
-        if (claim.deniedIndexes.includes(0)) {
-            return { type: 'project_cap_reached' }
-        }
-        return { type: 'recipient_cap_reached', addresses: claim.deniedIndexes.map((index) => addresses[index - 1]) }
+        return { limiter: this.dailyCapLimiter, buckets, addresses }
     }
 
     public async returnDailyCaps(grant: SandboxDailyCapGrant): Promise<void> {