feat(cdp): skip sandbox sends of a blocked organization

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

@@ -29,7 +29,7 @@ Only workflows that send email are subject to the tiers; SMS, push, and webhook
 The `workflows-sandbox-sender` flag enables a PostHog-managed email integration per project through `POST /api/projects/:project_id/integrations/email_sandbox_sender/`.
 The endpoint requires `integration:write` and project membership. Repeated requests keep the same integration and refresh its display name from the organization name.
 Staff turn the sandbox sender off for one organization by adding a `SandboxSenderOrganizationBlock` row in the Django admin, and lift the block by deleting the row.
-While the block exists, the endpoint returns 404 for the organization's projects. Existing sandbox senders stay, and the email worker does not read the block.
+While the block exists, the endpoint returns 404 for the organization's projects. Existing sandbox senders stay, but their sends skip.
 Set `WORKFLOWS_SANDBOX_SENDER_DOMAIN`, `SES_SANDBOX_FROM_ADDRESS`, and `SES_SANDBOX_TENANT_NAME` in Django; any empty value leaves the endpoint unavailable.
 The worker sends from the same `SES_SANDBOX_FROM_ADDRESS`, so set it once for Django and the worker. The address the UI shows is then the address the worker sends from.
 Workflow email steps and test sends offer it as a sender.
@@ -117,6 +117,8 @@ Every sandbox send, test sends included, skips while the shared sandbox SES tena
 The worker checks the pause again after the daily allowance claim and gives the claim back when the pause blocks delivery.
 Django mirrors the tenant named by `SES_SANDBOX_TENANT_NAME` into `SandboxSenderTenantState` from the SES tenant-state webhook and the daily reconciliation, without emailing project admins. Set the same `SES_SANDBOX_TENANT_NAME` on Django web, Celery, and the worker.
 A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state.
+A blocked organization's sandbox sends skip with the reason `organization_blocked`.
+The worker reads the block with the organization members, so a block and its removal reach the workers within one minute. Own-sender sends do not read it.
 Each delivered send emits a `workflows sandbox email sent` event.
 Test sends never record `email_sent`, so they do not count toward the sending tier.
 A successful live sandbox send records `email_sandbox_sent` next to `email_sent`, so the workflow's metrics still show it.

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

@@ -3,39 +3,47 @@ import { PostgresRouter } from '~/common/utils/db/postgres'
 import { OrganizationMembersService } from './organization-members.service'
 
 describe('OrganizationMembersService', () => {
+    const queriesPerLookup = 2
     const memberRow = (email: string): { team_id: number; email: string; is_email_verified: boolean } => ({
         team_id: 1,
         email,
         is_email_verified: true,
     })
+    const rowsFor = (name: string, member: string): { rows: unknown[] } => ({
+        rows: name === 'fetch-sandbox-team-organization-members' ? [memberRow(member)] : [],
+    })
+    const allowed = { organizationBlocked: false, blockedRecipients: [] }
 
     it('fails closed when every member lookup outlasts its one-minute decision', async () => {
         let now = 0
-        const query = jest.fn(() => {
+        const query = jest.fn((_use: unknown, _sql: string, _params: unknown, name: string) => {
             now += 60_000
-            return Promise.resolve({ rows: [memberRow('member@example.com')] })
+            return Promise.resolve(rowsFor(name, 'member@example.com'))
         })
         const service = new OrganizationMembersService({ query } as unknown as PostgresRouter, () => now)
 
-        await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired')
-        expect(query).toHaveBeenCalledTimes(2)
+        await expect(service.checkRecipients(1, ['member@example.com'])).rejects.toThrow('expired')
+        expect(query).toHaveBeenCalledTimes(2 * queriesPerLookup)
     })
 
     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 query = jest.fn((_use: unknown, _sql: string, _params: unknown, name: string) =>
+            Promise.resolve(rowsFor(name, member))
+        )
         const service = new OrganizationMembersService({ query } as unknown as PostgresRouter, () => now)
 
-        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([])
-        expect(await service.getBlockedRecipients(1, ['member@example.com'])).toEqual([])
-        expect(query).toHaveBeenCalledTimes(1)
+        expect(await service.checkRecipients(1, ['member@example.com'])).toEqual(allowed)
+        expect(await service.checkRecipients(1, ['member@example.com'])).toEqual(allowed)
+        expect(query).toHaveBeenCalledTimes(queriesPerLookup)
 
         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)
+        expect(await service.checkRecipients(1, ['member@example.com'])).toEqual({
+            organizationBlocked: false,
+            blockedRecipients: [{ address: 'member@example.com', unverifiedMember: false }],
+        })
+        expect(query).toHaveBeenCalledTimes(2 * queriesPerLookup)
     })
 })

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

@@ -4,11 +4,16 @@ import { LazyLoader } from '~/common/utils/lazy-loader'
 type OrganizationMemberSnapshot = {
     verifiedEmails: ReadonlySet<string>
     unverifiedEmails: ReadonlySet<string>
+    sandboxSenderBlocked: boolean
     expiresAt: number
 }
 
 export type BlockedRecipient = { readonly address: string; readonly unverifiedMember: boolean }
 
+export type RecipientCheck =
+    | { readonly organizationBlocked: true }
+    | { readonly organizationBlocked: false; readonly blockedRecipients: BlockedRecipient[] }
+
 const MAX_SNAPSHOT_AGE_MS = 60_000
 const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - 1_000
 
@@ -28,20 +33,33 @@ export class OrganizationMembersService {
             bufferMs: 0,
             loader: async (teamIds) => {
                 const expiresAt = this.now() + MAX_SNAPSHOT_AGE_MS
-                // Only an explicit false deactivates an organization, matching the web middleware.
-                const { rows } = await postgres.query<{ team_id: number; email: string; is_email_verified: boolean }>(
-                    PostgresUse.COMMON_WRITE,
-                    `SELECT team.id AS team_id, users.email, users.is_email_verified
-                     FROM posthog_team AS team
-                     JOIN posthog_organization AS organization ON organization.id = team.organization_id
-                     JOIN posthog_organizationmembership AS membership ON membership.organization_id = organization.id
-                     JOIN posthog_user AS users ON users.id = membership.user_id
-                     WHERE team.id = ANY($1::bigint[])
-                       AND organization.is_active IS NOT FALSE
-                       AND users.is_active = true`,
-                    [teamIds.map(Number)],
-                    'fetch-sandbox-team-organization-members'
-                )
+                const ids = [teamIds.map(Number)]
+                const [{ rows }, blocks] = await Promise.all([
+                    // Only an explicit false deactivates an organization, matching the web middleware.
+                    postgres.query<{ team_id: number; email: string; is_email_verified: boolean }>(
+                        PostgresUse.COMMON_WRITE,
+                        `SELECT team.id AS team_id, users.email, users.is_email_verified
+                         FROM posthog_team AS team
+                         JOIN posthog_organization AS organization ON organization.id = team.organization_id
+                         JOIN posthog_organizationmembership AS membership ON membership.organization_id = organization.id
+                         JOIN posthog_user AS users ON users.id = membership.user_id
+                         WHERE team.id = ANY($1::bigint[])
+                           AND organization.is_active IS NOT FALSE
+                           AND users.is_active = true`,
+                        ids,
+                        'fetch-sandbox-team-organization-members'
+                    ),
+                    postgres.query<{ team_id: number }>(
+                        PostgresUse.COMMON_WRITE,
+                        `SELECT team.id AS team_id
+                         FROM posthog_team AS team
+                         JOIN workflows_sandboxsenderorganizationblock AS block ON block.organization_id = team.organization_id
+                         WHERE team.id = ANY($1::bigint[])`,
+                        ids,
+                        'fetch-sandbox-team-organization-blocks'
+                    ),
+                ])
+                const blockedTeamIds = new Set(blocks.rows.map((row) => String(row.team_id)))
                 const members = new Map(
                     teamIds.map((id) => [id, { verified: new Set<string>(), unverified: new Set<string>() }])
                 )
@@ -52,18 +70,27 @@ export class OrganizationMembersService {
                 return Object.fromEntries(
                     [...members].map(([id, { verified, unverified }]) => [
                         id,
-                        { verifiedEmails: verified, unverifiedEmails: unverified, expiresAt },
+                        {
+                            verifiedEmails: verified,
+                            unverifiedEmails: unverified,
+                            sandboxSenderBlocked: blockedTeamIds.has(id),
+                            expiresAt,
+                        },
                     ])
                 )
             },
         })
     }
 
-    public async getBlockedRecipients(teamId: number, recipients: string[]): Promise<BlockedRecipient[]> {
+    public async checkRecipients(teamId: number, recipients: string[]): Promise<RecipientCheck> {
         const snapshot = await this.getFreshSnapshot(String(teamId))
-        return recipients
+        if (snapshot.sandboxSenderBlocked) {
+            return { organizationBlocked: true }
+        }
+        const blockedRecipients = recipients
             .filter((email) => !snapshot.verifiedEmails.has(normalizeEmail(email)))
             .map((address) => ({ address, unverifiedMember: snapshot.unverifiedEmails.has(normalizeEmail(address)) }))
+        return { organizationBlocked: false, blockedRecipients }
     }
 
     // A snapshot older than a minute never decides a send: it is reloaded once, and a reload that

nodejs/src/cdp/services/messaging/email.service.test.ts

@@ -601,6 +601,59 @@ describe('EmailService', () => {
                 expect(field === 'cc' ? destination?.CcAddresses : destination?.BccAddresses).toEqual([email])
             })
 
+            it.each([false, true])(
+                'a running worker stops sandbox sends within a minute of a block and resumes within a minute of its removal (isTest=%s)',
+                async (isTest) => {
+                    const setBlocked = (blocked: boolean): Promise<unknown> =>
+                        hub.postgres.query(
+                            PostgresUse.COMMON_WRITE,
+                            blocked
+                                ? `INSERT INTO workflows_sandboxsenderorganizationblock (organization_id, reason, created_at)
+                                   VALUES ($1, 'Abuse report', now())`
+                                : 'DELETE FROM workflows_sandboxsenderorganizationblock WHERE organization_id = $1',
+                            [team.organization_id],
+                            'test:set-sandbox-organization-block'
+                        )
+                    const sendAt = (time: number) => {
+                        clock.mockReturnValue(time)
+                        invocation.state.vmState = { stack: [] } as any
+                        return service.executeSendEmail(invocation, isTest)
+                    }
+                    const now = Date.now()
+                    const clock = jest.spyOn(Date, 'now').mockReturnValue(now)
+                    try {
+                        service = createSandboxService(true)
+                        await sendAt(now)
+                        expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+
+                        await setBlocked(true)
+                        const blocked = await sendAt(now + 61_000)
+
+                        expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+                        expect(blocked).toMatchObject({ finished: true, skipped: true, metrics: [] })
+                        expect(blocked.invocation.state.vmState?.stack).toEqual([{ success: false }])
+                        expect(blocked.logs).toContainEqual(
+                            expect.objectContaining({
+                                level: 'info',
+                                message:
+                                    'Skipping send: the sandbox sender is turned off for this organization. Contact support to find out why.',
+                            })
+                        )
+                        expectBlockedEvent('organization_blocked', isTest, 0)
+
+                        await setBlocked(false)
+                        const unblocked = await sendAt(now + 122_000)
+
+                        expect(unblocked.error).toBeUndefined()
+                        expect(unblocked.skipped).not.toBe(true)
+                        expect(sendEmailSpy).toHaveBeenCalledTimes(2)
+                    } finally {
+                        clock.mockRestore()
+                        await setBlocked(false)
+                    }
+                }
+            )
+
             it.each([false, true])('blocks a non-member To address (isTest=%s)', async (isTest) => {
                 service = createSandboxService(true)
                 invocation.queueParameters = createSandboxParams({ to: { email: 'outside@example.com' } })

nodejs/src/cdp/services/messaging/email.service.ts

@@ -407,7 +407,7 @@ function sandboxRecipientsOf(params: CyclotronInvocationQueueParametersEmailType
     return [params.to.email.trim(), ...sandboxAddressList(params.cc), ...sandboxAddressList(params.bcc)]
 }
 
-type SandboxRecipientBlockReason = Exclude<SandboxEmailBlockReason, 'switch_off' | 'paused'>
+type SandboxRecipientBlockReason = Exclude<SandboxEmailBlockReason, 'switch_off' | 'paused' | 'organization_blocked'>
 
 type SesRoute = {
     tenantName: string
@@ -445,6 +445,9 @@ const SANDBOX_RECIPIENT_BLOCK_COPY: Record<
     },
 }
 
+const SANDBOX_ORGANIZATION_BLOCKED_MESSAGE =
+    'Skipping send: the sandbox sender is turned off for this organization. Contact support to find out why.'
+
 const SANDBOX_PAUSE_SKIP_MESSAGES = {
     paused: 'Skipping send: the sandbox sender is paused right now. Verify your own domain to keep sending.',
     check_failed:
@@ -1093,13 +1096,16 @@ export class EmailService {
         isTest: boolean
     ): Promise<boolean> {
         const { teamId } = result.invocation
+        let organizationBlocked = false
         let blockedRecipients: string[]
         let reason: SandboxRecipientBlockReason
         try {
             if (!this.organizationMembers) {
                 throw new Error('Organization member checks are not configured')
             }
-            const blocked = await this.organizationMembers.getBlockedRecipients(teamId, recipients)
+            const check = await this.organizationMembers.checkRecipients(teamId, recipients)
+            organizationBlocked = check.organizationBlocked
+            const blocked = check.organizationBlocked ? [] : check.blockedRecipients
             blockedRecipients = blocked.map(({ address }) => address)
             reason = blocked.every(({ unverifiedMember }) => unverifiedMember)
                 ? 'recipient_unverified'
@@ -1112,6 +1118,11 @@ export class EmailService {
                 error: error instanceof Error ? error.message : 'Unknown error',
             })
         }
+        if (organizationBlocked) {
+            this.skipSandboxSend(result, SANDBOX_ORGANIZATION_BLOCKED_MESSAGE)
+            await this.sandboxSender?.captureBlocked(teamId, isTest, 'organization_blocked', 0)
+            return false
+        }
         if (!blockedRecipients.length) {
             return true
         }

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

@@ -42,6 +42,7 @@ export type SandboxEmailBlockReason =
     | 'check_failed'
     | 'cap_reached'
     | 'paused'
+    | 'organization_blocked'
 
 export type SandboxDailyCapGrant = { type: 'granted'; claim: IssuedClaim; buckets: ClaimRequest[] }