feat(cdp): send sandbox email through its ses identity to org members

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

@@ -75,3 +75,21 @@ Decay, suspension drops, admin recomputes, and the backfill stay silent.
   A send denied because the limiter itself failed keeps the shorter token bucket cadence, at most 5 to 10 minutes.
 - Gmail provides no per-message feedback loop, so complaint rates (ours and AWS's) cannot see Gmail complaints at all.
 - The `HOGFLOW_BATCH_TRIGGER_ELEVATED_TEAM_IDS` allowlist elevates a team to the top tier in Django only: the UI, the admin, and the batch audience cap show top-tier numbers, but the worker's send-time buckets read the stored tier. An allowlisted team below the top tier is therefore throttled at its stored tier while being told otherwise, and its shadow-mode delay counts overstate real impact. Before enforcing, pin each allowlisted team at the top tier from the Django admin and empty the allowlist, rather than plumbing the list into the worker.
+
+### Sandbox delivery
+
+The worker sends through a `sandbox` email integration only when `WORKFLOWS_SANDBOX_SENDER_ENABLED` is on and `SES_SANDBOX_TENANT_NAME`, `SES_SANDBOX_CONFIGURATION_SET`, and `SES_SANDBOX_FROM_ADDRESS` are set.
+Otherwise every send through it skips with a reason in the run log and a `workflows sandbox email blocked` event with reason `switch_off`.
+Every To, CC, and BCC address must belong to an active member of the team's active organization with a verified email address.
+Membership snapshots expire within one minute, and the worker checks recipients again right before delivery.
+A failed or stale membership check skips the send and lists the blocked addresses in the run log.
+The blocked event fires at most once an hour per team, reason, and send kind (workflow or test).
+A send that passes every check goes out through the fixed SES tenant, configuration set, and From address, whatever the step's custom sender and Reply-To say.
+Sandbox sends skip the team's tier budget, click and open tracking, and unsubscribe headers, because they reach only members of the sending organization.
+The worker parses and normalizes the HTML, then appends a footer that names the sender and the organization ID.
+It overrides root clipping and visibility styles, drops any root style declaration it cannot parse cleanly, and keeps the template's typography, so malformed markup may render differently from an own-domain send.
+A template whose HTML cannot keep the footer fails the send.
+The visible footer is best effort across email clients and template CSS; the signed tracking header still attributes the message.
+Each delivered send emits a `workflows sandbox email sent` event.
+Live workflow sends record `email_sent`, so workflow reports count them; test sends do not.
+The worker has no daily sandbox caps yet, so only switch it on where Django provisions sandbox senders for a small set of projects.

nodejs/package.json

@@ -137,6 +137,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.61.2",

nodejs/src/cdp/cdp-services.ts

@@ -5,7 +5,7 @@ import { PersonReadRepository } from '~/common/persons/repositories/person-repos
 import { RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2'
 import { UsageIngestionConfig, createUsageIngestionClient, usageReportTeamMatcher } from '~/common/usage-ingestion'
 import { PostgresRouter } from '~/common/utils/db/postgres'
-import { getRedisHost } from '~/common/utils/db/redis'
+import { RedisPoolConfig, getRedisHost } from '~/common/utils/db/redis'
 import { logger } from '~/common/utils/logger'
 import { PubSub } from '~/common/utils/pubsub'
 import { TeamManager } from '~/common/utils/team-manager'
@@ -32,6 +32,7 @@ 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 { PersonsManagerService } from './services/managers/persons-manager.service'
 import { RecipientsManagerService } from './services/managers/recipients-manager.service'
 import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service'
@@ -44,6 +45,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, sandboxBlockedEventRedisConfig } 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'
@@ -175,6 +177,10 @@ export type CdpCoreServicesConfig = Pick<
         | 'SES_ENDPOINT'
         | 'SES_TRACKED_CONFIGURATION_SET'
         | 'SES_UNTRACKED_CONFIGURATION_SET'
+        | 'WORKFLOWS_SANDBOX_SENDER_ENABLED'
+        | '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'
@@ -390,17 +396,24 @@ export function createCdpCoreServices(
     deps: CdpCoreServicesDeps,
     redisName = 'cdp-redis'
 ): CdpCoreServices {
-    const redis = createRedisV2PoolFromConfig({
-        connection: config.CDP_REDIS_HOST
+    const cdpRedisConnection = (name: string): RedisPoolConfig['connection'] =>
+        config.CDP_REDIS_HOST
             ? {
                   url: config.CDP_REDIS_HOST,
                   options: { port: config.CDP_REDIS_PORT, password: config.CDP_REDIS_PASSWORD },
-                  name: redisName,
+                  name,
               }
-            : { url: config.REDIS_URL, name: `${redisName}-fallback` },
+            : { url: config.REDIS_URL, name: `${name}-fallback` }
+    const redis = createRedisV2PoolFromConfig({
+        connection: cdpRedisConnection(redisName),
         poolMinSize: config.REDIS_POOL_MIN_SIZE,
         poolMaxSize: config.REDIS_POOL_MAX_SIZE,
     })
+    // A stalled blocked-event call keeps its connection until Redis answers, so these calls get their own
+    // connections: on the shared pool they took the ones hog watcher and the duplicate observer wait for.
+    const sandboxBlockedEventRedis = createRedisV2PoolFromConfig(
+        sandboxBlockedEventRedisConfig(cdpRedisConnection(`${redisName}-sandbox-blocked-event`))
+    )
 
     const redisReader = createCdpReaderRedisPool(config, redis, redisName)
     const valkeyShadow = createCdpValkeyShadowPools(config, redisName)
@@ -491,7 +504,18 @@ export function createCdpCoreServices(
         workflowsActivationReporter,
         messageAssetsService,
         workflowEmailRateLimiter,
-        teamEmailRateLimiter
+        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,
+            },
+            deps.teamManager,
+            sandboxBlockedEventRedis
+        ),
+        new OrganizationMembersService(deps.postgres)
     )
     const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL)
     const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields)

nodejs/src/cdp/config.ts

@@ -173,6 +173,11 @@ 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
+    // The sandbox sender delivers only with the switch on and all three route settings set.
+    WORKFLOWS_SANDBOX_SENDER_ENABLED: boolean
+    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
@@ -423,6 +428,10 @@ 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,
+        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.test.ts

@@ -0,0 +1,158 @@
+import { PostgresRouter } from '~/common/utils/db/postgres'
+
+import { OrganizationMembersService } from './organization-members.service'
+
+const routerFor = (select: jest.Mock): PostgresRouter =>
+    ({
+        transaction: (_usage: unknown, _tag: string, run: (tx: unknown) => Promise<unknown>) => run({}),
+        query: (_tx: unknown, sql: string) => (sql.startsWith('SET LOCAL') ? Promise.resolve({ rows: [] }) : select()),
+    }) as unknown as PostgresRouter
+
+describe('OrganizationMembersService', () => {
+    const memberRow = (email: string): { team_id: number; email: string; is_email_verified: boolean } => ({
+        team_id: 1,
+        email,
+        is_email_verified: true,
+    })
+
+    it('fails closed when every member lookup outlasts its one-minute decision', async () => {
+        let now = 0
+        const query = jest.fn(() => {
+            now += 60_000
+            return Promise.resolve({ rows: [memberRow('member@example.com')] })
+        })
+        const service = new OrganizationMembersService(routerFor(query), () => now)
+
+        await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired')
+        expect(query).toHaveBeenCalledTimes(2)
+    })
+
+    it('fails closed when the fresh lookup outlasts its one-minute decision', async () => {
+        let now = 0
+        const query = jest.fn(() => {
+            now += 60_000
+            return Promise.resolve({ rows: [memberRow('member@example.com')] })
+        })
+        const service = new OrganizationMembersService(routerFor(query), () => now)
+
+        await expect(service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).rejects.toThrow(
+            'expired'
+        )
+        expect(query).toHaveBeenCalledTimes(1)
+    })
+
+    it('fails closed after three seconds when the lookup cannot get a connection', async () => {
+        jest.useFakeTimers()
+        try {
+            const query = jest.fn()
+            const stalledPool = {
+                transaction: () => new Promise<never>(() => {}),
+                query,
+            } as unknown as PostgresRouter
+            const service = new OrganizationMembersService(stalledPool)
+
+            const check = service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })
+            const outcome = expect(check).rejects.toThrow('timed out')
+            await jest.advanceTimersByTimeAsync(3_000)
+
+            await outcome
+            expect(query).not.toHaveBeenCalled()
+        } finally {
+            jest.useRealTimers()
+        }
+    })
+
+    it('keeps at most five lookups open while the pool has no connection, and runs no query for them once it recovers', async () => {
+        jest.useFakeTimers()
+        try {
+            let recovered = false
+            const waitingForConnection: (() => void)[] = []
+            const query = jest.fn((_tx: unknown, sql: string) =>
+                Promise.resolve({ rows: sql.startsWith('SET LOCAL') ? [] : [memberRow('member@example.com')] })
+            )
+            const pool = {
+                transaction: (_usage: unknown, _tag: string, run: (tx: unknown) => Promise<unknown>) =>
+                    recovered
+                        ? run({})
+                        : new Promise((resolve, reject) =>
+                              waitingForConnection.push(() => void run({}).then(resolve, reject))
+                          ),
+                query,
+            } as unknown as PostgresRouter
+            const service = new OrganizationMembersService(pool)
+
+            const checks = Array.from({ length: 20 }, () =>
+                service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })
+            )
+            const outcomes = Promise.allSettled(checks)
+            await jest.advanceTimersByTimeAsync(3_000)
+
+            const settled = await outcomes
+            expect(settled.every((outcome) => outcome.status === 'rejected')).toBe(true)
+            expect(waitingForConnection).toHaveLength(5)
+
+            recovered = true
+            waitingForConnection.forEach((connect) => connect())
+            await jest.advanceTimersByTimeAsync(0)
+            expect(query).not.toHaveBeenCalled()
+
+            await expect(service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).resolves.toEqual([])
+            expect(query).toHaveBeenCalledTimes(2)
+        } finally {
+            jest.useRealTimers()
+        }
+    })
+
+    it('shares one member lookup across sends and reloads it after a minute', async () => {
+        let now = 0
+        let member = 'member@example.com'
+        const query = jest.fn(() => Promise.resolve({ rows: [memberRow(member)] }))
+        const service = new OrganizationMembersService(routerFor(query), () => now)
+
+        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([])
+        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([])
+        expect(query).toHaveBeenCalledTimes(1)
+
+        now = 60_001
+        member = 'new-member@example.com'
+        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([
+            { address: 'member@example.com', unverifiedMember: false },
+        ])
+        expect(query).toHaveBeenCalledTimes(2)
+    })
+
+    it('reloads members for a fresh check inside the minute', async () => {
+        let member = 'member@example.com'
+        const query = jest.fn(() => Promise.resolve({ rows: [memberRow(member)] }))
+        const service = new OrganizationMembersService(routerFor(query), () => 0)
+
+        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([])
+        member = 'new-member@example.com'
+
+        expect(await service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })).toEqual([
+            { address: 'member@example.com', unverifiedMember: false },
+        ])
+        expect(query).toHaveBeenCalledTimes(2)
+    })
+
+    it('does not let a fresh check join a lookup that started before it', async () => {
+        let finishEarlierLookup: (rows: unknown) => void = () => {}
+        const query = jest
+            .fn()
+            .mockReturnValueOnce(new Promise((resolve) => (finishEarlierLookup = resolve)))
+            .mockResolvedValue({ rows: [memberRow('new-member@example.com')] })
+        const service = new OrganizationMembersService(routerFor(query), () => 0)
+
+        // Another send starts loading members, then the member is removed before the final check.
+        const earlierCheck = service.getBlockedRecipients(1, ['member@example.com'])
+        while (query.mock.calls.length === 0) {
+            await new Promise((resolve) => setImmediate(resolve))
+        }
+        const freshCheck = service.getBlockedRecipients(1, ['member@example.com'], { fresh: true })
+        finishEarlierLookup({ rows: [memberRow('member@example.com')] })
+
+        expect(await earlierCheck).toEqual([])
+        expect(await freshCheck).toEqual([{ address: 'member@example.com', unverifiedMember: false }])
+        expect(query).toHaveBeenCalledTimes(2)
+    })
+})