fix(workflows): guard sandbox sends on empty settings and SES throttle

.github/scripts/check-idor-model-coverage.py

@@ -302,6 +302,7 @@ def get_scoped_models() -> tuple[dict[str, set[str]], set[str], set[str], set[st
         "AsyncMigrationError",
         "InstanceSetting",
         "License",
+        "SandboxSenderTenantState",  # instance-global mirror of the shared sandbox SES tenant state
         # --- Deprecated ---
         "Prompt",
         "PromptSequence",

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.
@@ -57,6 +58,15 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
 
 - The batch audience cap is decided when the batch is dispatched. Adding an email step to the workflow while a batch is queued does not re-cap it; the send-time buckets still cap every email at execution. This is why enforcement requires the worker caps to be deployed (see the rollout order).
 - Test-panel sends bypass the team buckets on purpose, matching the per-workflow rate limit.
+- Sandbox sender sends bypass the team buckets without spending their tokens. Sandbox limits run independently.
+  The worker holds every sandbox send, test sends included, to a daily cap per project (`WORKFLOWS_SANDBOX_DAILY_TEAM_CAP`, one token per recipient copy) and a daily cap per recipient address (`WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP`).
+  A reached cap skips the send instead of delaying it. A limiter error or an empty or invalid cap value skips every sandbox send.
+  The SES tenant-state webhook and daily reconciliation mirror `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` without emailing project admins.
+  Every sandbox send, including test sends, skips when that tenant's mirrored sending status is `DISABLED` or the pause lookup fails. A missing row allows sending, with SES enforcing its own state.
+  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.
+  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.
+  Keep `WORKFLOWS_SANDBOX_SENDER_ENABLED` off until the recipient membership checks, sandbox caps and sandbox pause gate are deployed.
 - 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.
   The computed wait is capped at one hour and then jittered 1x to 2x, so a parked send can wait just under two hours between attempts.

nodejs/package.json

@@ -133,6 +133,7 @@
         "node-schedule": "^2.1.0",
         "nodemailer": "^8.0.8",
         "p-limit": "3.1.0",
+        "parse5": "7.1.2",
         "pg": "^8.6.0",
         "pino": "^8.6.0",
         "posthog-node": "5.54.1",

nodejs/src/cdp/cdp-api.ts

@@ -195,7 +195,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

@@ -30,7 +30,9 @@ import { InvocationResultsService } from './services/invocation-results.service'
 import { HogFunctionManagerService } from './services/managers/hog-function-manager.service'
 import { HogFunctionTemplateManagerService } from './services/managers/hog-function-template-manager.service'
 import { IntegrationManagerService } from './services/managers/integration-manager.service'
+import { OrganizationMembersService } from './services/managers/organization-members.service'
 import { RecipientsManagerService } from './services/managers/recipients-manager.service'
+import { SandboxSenderStateService } from './services/managers/sandbox-sender-state.service'
 import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service'
 import { EmailSuppressionService } from './services/messaging/email-suppression.service'
 import { EmailValidationService } from './services/messaging/email-validation.service'
@@ -40,6 +42,7 @@ 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 { SandboxEmailSender } from './services/messaging/sandbox-email-sender'
 import { HogFunctionMonitoringService } from './services/monitoring/hog-function-monitoring.service'
 import { HogInvocationResultsService } from './services/monitoring/hog-invocation-results.service'
 import { HogWatcherService } from './services/monitoring/hog-watcher.service'
@@ -107,6 +110,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
 }
@@ -159,6 +163,12 @@ export type CdpCoreServicesConfig = Pick<
         | 'SES_ENDPOINT'
         | 'SES_TRACKED_CONFIGURATION_SET'
         | 'SES_UNTRACKED_CONFIGURATION_SET'
+        | 'WORKFLOWS_SANDBOX_SENDER_ENABLED'
+        | 'WORKFLOWS_SANDBOX_DAILY_TEAM_CAP'
+        | 'WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP'
+        | 'SES_SANDBOX_TENANT_NAME'
+        | 'SES_SANDBOX_CONFIGURATION_SET'
+        | 'SES_SANDBOX_FROM_ADDRESS'
         | 'EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD'
         | 'EMAIL_TEAM_SENDING_CAP_MODE'
         | 'EMAIL_TEAM_SENDING_CAP_HOURLY_BY_TIER'
@@ -433,6 +443,22 @@ export function createCdpCoreServices(
     const teamEmailRateLimiter = deps.emailValidationValkey
         ? new RateLimiterService(deps.emailValidationValkey, { name: 'team-email' })
         : null
+    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,
+        new SandboxSenderStateService(deps.postgres, deps.pubSub)
+    )
     const emailService = new EmailService(
         {
             sesAccessKeyId: config.SES_ACCESS_KEY_ID,
@@ -454,7 +480,9 @@ export function createCdpCoreServices(
         recipientsManager,
         messageAssetsService,
         workflowEmailRateLimiter,
-        teamEmailRateLimiter
+        teamEmailRateLimiter,
+        sandboxEmailSender,
+        new OrganizationMembersService(deps.postgres)
     )
     const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL)
     const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields)
@@ -578,6 +606,7 @@ export function createCdpCoreServices(
         recipientTokensService,
         outputs,
         emailService,
+        sandboxEmailSender,
         messageAssetsService,
     }
 }

nodejs/src/cdp/config.ts

@@ -136,6 +136,13 @@ export type CdpConfig = ClickhouseConfig & {
     // Configuration set without open/click tracking (same delivery/bounce/complaint event destination).
     // Empty means not provisioned: tracking-off sends fall back to the tracked set with a warning.
     SES_UNTRACKED_CONFIGURATION_SET: string
+    WORKFLOWS_SANDBOX_SENDER_ENABLED: boolean
+    // An empty or invalid value blocks every sandbox send.
+    WORKFLOWS_SANDBOX_DAILY_TEAM_CAP: string
+    WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP: string
+    SES_SANDBOX_TENANT_NAME: string
+    SES_SANDBOX_CONFIGURATION_SET: string
+    SES_SANDBOX_FROM_ADDRESS: string
     // Comma-separated allowlist of SNS Topic ARNs the SES webhook accepts events from. Empty string
     // means no restriction (dev/test); production should set this to the workflow SES topic ARN(s).
     SES_ALLOWED_SNS_TOPIC_ARNS: string
@@ -344,6 +351,12 @@ export function getDefaultCdpConfig(): CdpConfig {
         SES_REGION: isTestEnv() || isDevEnv() ? 'us-east-1' : '',
         SES_TRACKED_CONFIGURATION_SET: 'posthog-messaging',
         SES_UNTRACKED_CONFIGURATION_SET: '',
+        WORKFLOWS_SANDBOX_SENDER_ENABLED: false,
+        WORKFLOWS_SANDBOX_DAILY_TEAM_CAP: '25',
+        WORKFLOWS_SANDBOX_DAILY_RECIPIENT_CAP: '5',
+        SES_SANDBOX_TENANT_NAME: '',
+        SES_SANDBOX_CONFIGURATION_SET: '',
+        SES_SANDBOX_FROM_ADDRESS: '',
         SES_ALLOWED_SNS_TOPIC_ARNS: '',
         EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD: 5,
 

nodejs/src/cdp/services/managers/organization-members.service.ts

@@ -0,0 +1,73 @@
+import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres'
+import { LazyLoader } from '~/common/utils/lazy-loader'
+
+type OrganizationMemberSnapshot = {
+    emails: ReadonlySet<string>
+    expiresAt: number
+}
+
+const MAX_SNAPSHOT_AGE_MS = 60_000
+const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - 1_000
+
+export class OrganizationMembersService {
+    private snapshots: LazyLoader<OrganizationMemberSnapshot>
+
+    constructor(private postgres: PostgresRouter) {
+        this.snapshots = new LazyLoader({
+            name: 'organization_email_members',
+            refreshAgeMs: REFRESH_AGE_MS,
+            refreshJitterMs: 0,
+            bufferMs: 0,
+            loader: async (organizationIds) => {
+                const expiresAt = Date.now() + MAX_SNAPSHOT_AGE_MS
+                const { rows } = await postgres.query<{ organization_id: string; email: string }>(
+                    PostgresUse.COMMON_WRITE,
+                    `SELECT membership.organization_id, users.email
+                     FROM posthog_organizationmembership AS membership
+                     JOIN posthog_user AS users ON users.id = membership.user_id
+                     WHERE membership.organization_id = ANY($1::uuid[])
+                       AND users.is_active = true AND users.is_email_verified = true`,
+                    [organizationIds],
+                    'fetch-organization-email-members'
+                )
+                const emailsByOrganization: Record<string, Set<string>> = Object.fromEntries(
+                    organizationIds.map((id) => [id, new Set<string>()])
+                )
+                for (const row of rows) {
+                    emailsByOrganization[row.organization_id].add(row.email.trim().toLowerCase())
+                }
+                return Object.fromEntries(
+                    Object.entries(emailsByOrganization).map(([id, emails]) => [id, { emails, expiresAt }])
+                )
+            },
+        })
+    }
+
+    public async getBlockedRecipients(teamId: number, recipients: string[]): Promise<string[]> {
+        const expiresAt = Date.now() + MAX_SNAPSHOT_AGE_MS
+        const {
+            rows: [team],
+        } = await this.postgres.query<{ organization_id: string }>(
+            PostgresUse.COMMON_WRITE,
+            'SELECT organization_id FROM posthog_team WHERE id = $1',
+            [teamId],
+            'fetch-sandbox-team-organization'
+        )
+        if (!team) {
+            throw new Error('Could not identify the organization for this team')
+        }
+        let snapshot = await this.snapshots.get(team.organization_id)
+        if (snapshot && Date.now() >= snapshot.expiresAt && Date.now() < expiresAt) {
+            this.snapshots.markForRefresh(team.organization_id)
+            snapshot = await this.snapshots.get(team.organization_id)
+        }
+        if (!snapshot) {
+            throw new Error('Could not check organization members')
+        }
+        if (Date.now() >= Math.min(expiresAt, snapshot.expiresAt)) {
+            this.snapshots.markForRefresh(team.organization_id)
+            throw new Error('The organization member check expired')
+        }
+        return recipients.filter((email) => !snapshot.emails.has(email.trim().toLowerCase()))
+    }
+}

nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts

@@ -0,0 +1,40 @@
+import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres'
+import { LazyLoader } from '~/common/utils/lazy-loader'
+import { PubSub } from '~/common/utils/pubsub'
+
+type SandboxSenderState = { sendingStatus: string }
+
+export class SandboxSenderStateService {
+    private lazyLoader: LazyLoader<SandboxSenderState>
+
+    constructor(
+        private postgres: PostgresRouter,
+        pubSub: PubSub
+    ) {
+        this.lazyLoader = new LazyLoader({
+            name: 'sandbox_sender_state',
+            refreshAgeMs: 30_000,
+            refreshJitterMs: 5_000,
+            loader: async (tenantNames) => await this.fetchStates(tenantNames),
+        })
+        pubSub.on<{ tenantName: string }>('reload-sandbox-sender-state', ({ tenantName }): void => {
+            this.lazyLoader.markForRefresh(tenantName)
+        })
+    }
+
+    public async isPaused(tenantName: string): Promise<boolean> {
+        // No row means never synced; SES stays the final word.
+        return (await this.lazyLoader.get(tenantName))?.sendingStatus === 'DISABLED'
+    }
+
+    private async fetchStates(tenantNames: string[]): Promise<Record<string, SandboxSenderState>> {
+        const { rows } = await this.postgres.query<{ tenant_name: string; sending_status: string }>(
+            // The reload message fires right after the commit, so a lagging replica would cache the old state again.
+            PostgresUse.COMMON_WRITE,
+            'SELECT tenant_name, sending_status FROM workflows_sandboxsendertenantstate WHERE tenant_name = ANY($1)',
+            [tenantNames],
+            'fetch-sandbox-sender-state'
+        )
+        return Object.fromEntries(rows.map((row) => [row.tenant_name, { sendingStatus: row.sending_status }]))
+    }
+}