From b3f3e6ced5167187d01172da8db409baaeca681e Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 05:21:39 +0000 Subject: [PATCH 1/3] feat(workflows): restrict sandbox email to verified org members A send through a `sandbox` email integration now skips unless the worker switch and its three SES route settings are set, and unless every To, CC and BCC address belongs to an active, verified member of the team's active organization. The check runs again right before delivery. Each skip logs a reason and emits a deduplicated blocked event. The sandbox route does not exist yet, so a send that passes every check still skips as unavailable instead of reporting success. Co-Authored-By: Claude Opus 5.5 --- .../internal/workflows-email-sending-tiers.md | 11 + nodejs/src/cdp/cdp-services.ts | 19 +- nodejs/src/cdp/config.ts | 9 + .../organization-members.service.test.ts | 41 ++ .../managers/organization-members.service.ts | 86 ++++ .../services/messaging/email.service.test.ts | 406 +++++++++++++++++- .../cdp/services/messaging/email.service.ts | 160 ++++++- .../messaging/sandbox-email-sender.ts | 93 ++++ 8 files changed, 819 insertions(+), 6 deletions(-) create mode 100644 nodejs/src/cdp/services/managers/organization-members.service.test.ts create mode 100644 nodejs/src/cdp/services/managers/organization-members.service.ts create mode 100644 nodejs/src/cdp/services/messaging/sandbox-email-sender.ts diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index b511cea77ec8..c5a718d4db77 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -74,3 +74,14 @@ 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). +The sandbox route is not built yet, so a send that passes every check still skips as unavailable. +Keep the switch off in production until the sandbox caps are deployed. diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index 47643b470597..91ea747695ef 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -31,6 +31,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' @@ -42,6 +43,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' @@ -168,6 +170,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' @@ -474,7 +480,18 @@ export function createCdpCoreServices( recipientsManager, 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, + redis + ), + new OrganizationMembersService(deps.postgres) ) const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL) const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields) diff --git a/nodejs/src/cdp/config.ts b/nodejs/src/cdp/config.ts index dd7198cb4a09..33d291e3b187 100644 --- a/nodejs/src/cdp/config.ts +++ b/nodejs/src/cdp/config.ts @@ -166,6 +166,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 @@ -407,6 +412,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, diff --git a/nodejs/src/cdp/services/managers/organization-members.service.test.ts b/nodejs/src/cdp/services/managers/organization-members.service.test.ts new file mode 100644 index 000000000000..1f9440626105 --- /dev/null +++ b/nodejs/src/cdp/services/managers/organization-members.service.test.ts @@ -0,0 +1,41 @@ +import { PostgresRouter } from '~/common/utils/db/postgres' + +import { OrganizationMembersService } from './organization-members.service' + +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({ query } as unknown as PostgresRouter, () => now) + + await expect(service.getBlockedRecipients(1, ['member@example.com'])).rejects.toThrow('expired') + expect(query).toHaveBeenCalledTimes(2) + }) + + 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({ 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) + + 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) + }) +}) diff --git a/nodejs/src/cdp/services/managers/organization-members.service.ts b/nodejs/src/cdp/services/managers/organization-members.service.ts new file mode 100644 index 000000000000..1362b35123fa --- /dev/null +++ b/nodejs/src/cdp/services/managers/organization-members.service.ts @@ -0,0 +1,86 @@ +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' +import { LazyLoader } from '~/common/utils/lazy-loader' + +type OrganizationMemberSnapshot = { + verifiedEmails: ReadonlySet + unverifiedEmails: ReadonlySet + expiresAt: number +} + +export type BlockedRecipient = { readonly address: string; readonly unverifiedMember: boolean } + +const MAX_SNAPSHOT_AGE_MS = 60_000 +const REFRESH_AGE_MS = MAX_SNAPSHOT_AGE_MS - 1_000 + +const normalizeEmail = (email: string): string => email.trim().toLowerCase() + +export class OrganizationMembersService { + private snapshots: LazyLoader + + constructor( + postgres: PostgresRouter, + private now: () => number = Date.now + ) { + this.snapshots = new LazyLoader({ + name: 'sandbox_team_organization_members', + refreshAgeMs: REFRESH_AGE_MS, + refreshJitterMs: 0, + 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 members = new Map( + teamIds.map((id) => [id, { verified: new Set(), unverified: new Set() }]) + ) + for (const row of rows) { + const { verified, unverified } = members.get(String(row.team_id))! + ;(row.is_email_verified === true ? verified : unverified).add(normalizeEmail(row.email)) + } + return Object.fromEntries( + [...members].map(([id, { verified, unverified }]) => [ + id, + { verifiedEmails: verified, unverifiedEmails: unverified, expiresAt }, + ]) + ) + }, + }) + } + + public async getBlockedRecipients(teamId: number, recipients: string[]): Promise { + const snapshot = await this.getFreshSnapshot(String(teamId)) + return recipients + .filter((email) => !snapshot.verifiedEmails.has(normalizeEmail(email))) + .map((address) => ({ address, unverifiedMember: snapshot.unverifiedEmails.has(normalizeEmail(address)) })) + } + + // A snapshot older than a minute never decides a send: it is reloaded once, and a reload that + // itself outlasts the minute fails the check rather than trusting stale membership. + private async getFreshSnapshot(teamId: string): Promise { + let snapshot = await this.snapshots.get(teamId) + if (snapshot && this.now() >= snapshot.expiresAt) { + this.snapshots.markForRefresh(teamId) + snapshot = await this.snapshots.get(teamId) + } + if (!snapshot) { + throw new Error('Could not check organization members') + } + if (this.now() >= snapshot.expiresAt) { + this.snapshots.markForRefresh(teamId) + throw new Error('The organization member check expired') + } + return snapshot + } +} diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index 459f664a7da7..b74db59a9d4c 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -7,16 +7,23 @@ import { CyclotronInvocationQueueParametersEmailSchema, CyclotronInvocationQueueParametersEmailType, } from '~/cdp/schema/cyclotron' -import { CyclotronJobInvocationHogFunction } from '~/cdp/types' +import { CyclotronJobInvocationHogFunction, CyclotronJobInvocationResult } from '~/cdp/types' import { IngestionOutputs } from '~/common/outputs/ingestion-outputs' -import { createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' +import { RedisClient, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { closeHub, createHub } from '~/common/utils/db/hub' -import { PostgresUse } from '~/common/utils/db/postgres' +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' +import * as posthog from '~/common/utils/posthog' import { waitForExpect } from '~/tests/helpers/expectations' -import { createTestTeamFixture } from '~/tests/helpers/sql' +import { + createOrganization, + createOrganizationMembership, + createTestTeamFixture, + createUser, +} from '~/tests/helpers/sql' import { INVALID_TEAM_ID_KEYS } from '~/tests/helpers/team' import { Hub, Team } from '../../../types' +import { OrganizationMembersService } from '../managers/organization-members.service' import { RecipientsManagerService } from '../managers/recipients-manager.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' @@ -33,6 +40,7 @@ import { MailDevAPI } from './helpers/maildev' import { EmailTrackingCodeSigner } from './helpers/tracking-code' import { MessageAssetsService } from './message-assets.service' import { RecipientTokensService } from './recipient-tokens.service' +import { SandboxEmailSender, SandboxEmailSenderConfig } from './sandbox-email-sender' class ThrottlingException extends Error { constructor(message: string) { @@ -219,6 +227,396 @@ describe('EmailService', () => { sendEmailSpy = jest.spyOn(service.sesV2Client!, 'send') as any sendEmailSpy.mockResolvedValue({ MessageId: 'test-message-id' }) }) + describe('sandbox sender', () => { + const UNAVAILABLE_LOG = + 'Skipping send: the sandbox sender is unavailable right now. Verify your own domain to keep sending.' + let capture: jest.SpyInstance + let memberId: number + let memberIds: number[] + let extraOrganizationIds: string[] + const memberEmail = (name = 'test'): string => `${name}-${team.id}@example.com` + const createSandboxParams = ( + params: Partial = {} + ): CyclotronInvocationQueueParametersEmailType => + createEmailParams({ + from: { integrationId: 4 }, + to: { email: memberEmail(), name: 'Example member' }, + ...params, + }) + const createMember = async (email: string, organizationId = team.organization_id): Promise => { + const id = await createUser(hub.postgres, email) + memberIds.push(id) + await createOrganizationMembership(hub.postgres, organizationId, id) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET email = $1, is_active = true, is_email_verified = true WHERE id = $2', + [email, id], + 'test:verify-sandbox-member' + ) + return id + } + const createSandboxService = ( + enabled: boolean, + membersPostgres: PostgresRouter = hub.postgres, + routeSettings: Partial = {} + ): EmailService => { + const sandboxService = new EmailService( + { + sesAccessKeyId: hub.SES_ACCESS_KEY_ID, + sesSecretAccessKey: hub.SES_SECRET_ACCESS_KEY, + sesRegion: hub.SES_REGION, + sesEndpoint: hub.SES_ENDPOINT, + sesTrackedConfigurationSet: hub.SES_TRACKED_CONFIGURATION_SET, + sesUntrackedConfigurationSet: hub.SES_UNTRACKED_CONFIGURATION_SET, + }, + hub.integrationManager, + new TeamWorkflowsConfigService(hub.postgres, hub.pubSub), + hub.ENCRYPTION_SALT_KEYS, + hub.SITE_URL, + new EmailTrackingCodeSigner(hub.ENCRYPTION_SALT_KEYS, hub.CDP_EMAIL_TRACKING_URL), + new EmailSuppressionService(hub.postgres, emailSuppressionConfigFromEnv()), + new RecipientsManagerService(hub.postgres), + undefined, + null, + null, + new SandboxEmailSender( + { + enabled, + tenantName: 'sandbox-tenant', + configurationSetName: 'sandbox-email', + fromAddress: 'fixed-sandbox@example.com', + ...routeSettings, + }, + hub.teamManager, + { + useClient: async (_options, callback) => { + const client = await hub.redisPool.acquire() + try { + return await callback(client as unknown as RedisClient) + } finally { + await hub.redisPool.release(client) + } + }, + } + ), + new OrganizationMembersService(membersPostgres) + ) + sendEmailSpy = jest.spyOn(sandboxService.sesV2Client!, 'send') as jest.SpyInstance + sendEmailSpy.mockResolvedValue({ MessageId: 'sandbox-message-id' }) + return sandboxService + } + const expectSkipped = (result: CyclotronJobInvocationResult): void => { + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [], messageAssets: [] }) + expect(result.error).toBeUndefined() + expect(result.emailSendResult).toMatchObject({ status: 'skipped' }) + expect(result.invocation.state.vmState?.stack.at(-1)).toEqual({ success: false }) + } + const expectBlockedEvent = (reason: string, isTest: boolean, blockedRecipientCount: number): void => { + expect(capture).toHaveBeenLastCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason, is_test: isTest, blocked_recipient_count: blockedRecipientCount } + ) + } + + beforeEach(async () => { + capture = jest.spyOn(posthog, 'captureTeamEvent').mockImplementation(() => {}) + memberIds = [] + extraOrganizationIds = [] + memberId = await createMember(memberEmail()) + await createMember(memberEmail('cc')) + await createMember(memberEmail('bcc')) + await insertIntegration(hub.postgres, team.id, { + id: getIntegrationId(4), + kind: 'email', + config: { + provider: 'sandbox', + email: 'sandbox@example.com', + domain: 'example.com', + name: 'Example organization via PostHog', + verified: true, + }, + }) + invocation.queueParameters = createSandboxParams() + Object.assign(invocation, { hogFlow: { id: invocation.functionId, team_id: team.id } }) + }) + + afterEach(async () => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organizationmembership WHERE user_id = ANY($1::integer[])', + [memberIds], + 'test:delete-sandbox-memberships' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_user WHERE id = ANY($1::integer[])', + [memberIds], + 'test:delete-sandbox-members' + ) + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organization WHERE id = ANY($1::uuid[])', + [extraOrganizationIds], + 'test:delete-sandbox-organizations' + ) + const client = await hub.redisPool.acquire() + try { + const keys = await client.keys(`@posthog/workflows-sandbox-blocked/${team.id}/*`) + if (keys.length) { + await client.del(...keys) + } + } finally { + await hub.redisPool.release(client) + } + capture.mockRestore() + }) + + it.each([ + ['the switch is off', false, false, {}], + ['the switch is off', true, false, {}], + ['the tenant is empty', false, true, { tenantName: '' }], + ['the configuration set is blank', true, true, { configurationSetName: ' ' }], + ['the From address is empty', false, true, { fromAddress: '' }], + ] as const)( + 'skips when the sandbox sender is switched off because %s (isTest=%s)', + async (_reason, isTest, enabled, routeSettings: Partial) => { + service = createSandboxService(enabled, hub.postgres, routeSettings) + + const result = await service.executeSendEmail(invocation, isTest) + + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ level: 'info', message: UNAVAILABLE_LOG }) + ) + expectBlockedEvent('switch_off', isTest, 0) + } + ) + + it.each([false, true])( + 'permits sandbox destination invocations only as explicit tests (isTest=%s)', + async (isTest) => { + service = createSandboxService(true) + const destination = createExampleInvocation({ team_id: team.id, id: 'email-destination' }) + destination.state.vmState = { stack: [] } as any + destination.queueParameters = createSandboxParams() + + const result = await service.executeSendEmail(destination, isTest) + + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ + message: isTest + ? UNAVAILABLE_LOG + : 'Skipping send: the sandbox sender is only available for workflows and test sends.', + }) + ) + } + ) + + it.each([false, true])( + 'skips a member send as unavailable until the sandbox route exists (isTest=%s)', + async (isTest) => { + service = createSandboxService(true) + + const result = await service.executeSendEmail(invocation, isTest) + + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ level: 'info', message: UNAVAILABLE_LOG }) + ) + expect(capture).not.toHaveBeenCalled() + } + ) + + it.each([ + ['a display name', 'Example colleague', false], + ['a quoted comma', '"Example, colleague"', false], + ['an escaped quote', '"Example \\"colleague, teammate\\""', false], + ['mixed-case stored addresses', 'Example colleague', true], + ])('passes members through the gate whatever the case and %s', async (_case, name, storedMixedCase) => { + if (storedMixedCase) { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET email = initcap(email) WHERE id = ANY($1::integer[])', + [memberIds], + 'test:store-mixed-case-sandbox-members' + ) + } + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ + to: { email: memberEmail().toUpperCase() }, + cc: `${name} <${memberEmail('cc').toUpperCase()}>`, + bcc: ` ${name} <${memberEmail('bcc').toUpperCase()}> `, + }) + + const result = await service.executeSendEmail(invocation) + + expectSkipped(result) + expect(result.logs).toContainEqual(expect.objectContaining({ message: UNAVAILABLE_LOG })) + expect(capture).not.toHaveBeenCalled() + }) + + 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' } }) + + const result = await service.executeSendEmail(invocation, isTest) + + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ + level: 'info', + message: + 'Skipping send: the sandbox sender only sends to active organization members with verified email addresses. Blocked addresses: outside@example.com. Verify your own domain to send to anyone.', + }) + ) + expectBlockedEvent('recipient_not_member', isTest, 1) + }) + + it.each([ + 'cc', + 'bcc', + 'cc list', + 'multiple', + 'ambiguous cc mailbox', + 'unverified', + 'legacy', + 'inactive', + 'other organization', + 'deactivated organization', + ])('blocks all delivery for %s', async (recipient) => { + let blocked = ['outside@example.com'] + let params = createSandboxParams() + if (recipient === 'cc' || recipient === 'bcc') { + params = { ...params, [recipient]: 'Outside member ' } + blocked = ['OUTSIDE@example.com'] + } else if (recipient === 'cc list') { + params = { ...params, cc: `"Example, colleague" <${memberEmail('cc')}>, outside@example.com` } + } else if (recipient === 'multiple') { + params = { ...params, cc: 'outside@example.com', bcc: 'another@example.com' } + blocked = ['outside@example.com', 'another@example.com'] + } else if (recipient === 'ambiguous cc mailbox') { + params = { ...params, cc: `Example <${memberEmail('cc')}> ` } + blocked = [`Example <${memberEmail('cc')}> `] + } else if (recipient === 'other organization') { + const organizationId = await createOrganization(hub.postgres) + extraOrganizationIds.push(organizationId) + blocked = [memberEmail('outside')] + await createMember(blocked[0], organizationId) + params = { ...params, to: { email: blocked[0] } } + } else if (recipient === 'deactivated organization') { + blocked = [memberEmail()] + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_organization SET is_active = false WHERE id = $1', + [team.organization_id], + 'test:deactivate-sandbox-organization' + ) + } else { + blocked = [memberEmail()] + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'UPDATE posthog_user SET is_active = $1, is_email_verified = $2 WHERE id = $3', + [recipient !== 'inactive', recipient === 'legacy' ? null : recipient === 'inactive', memberId], + 'test:change-sandbox-member-status' + ) + } + service = createSandboxService(true) + invocation.queueParameters = params + + const result = await service.executeSendEmail(invocation) + + expectSkipped(result) + const unverifiedMember = recipient === 'unverified' || recipient === 'legacy' + expect(result.logs).toContainEqual( + expect.objectContaining({ + level: 'info', + message: unverifiedMember + ? `Skipping send: the sandbox sender only sends to organization members with verified email addresses. Unverified addresses: ${blocked.join(', ')}. Verify your email address in PostHog, or ask these members to verify theirs.` + : `Skipping send: the sandbox sender only sends to active organization members with verified email addresses. Blocked addresses: ${blocked.join(', ')}. Verify your own domain to send to anyone.`, + }) + ) + expect(capture).toHaveBeenCalledTimes(1) + expectBlockedEvent( + unverifiedMember ? 'recipient_unverified' : 'recipient_not_member', + false, + blocked.length + ) + }) + + it('fails closed when the member check fails', async () => { + const membersPostgres = new PostgresRouter(hub) + await membersPostgres.end() + service = createSandboxService(true, membersPostgres) + invocation.queueParameters = createSandboxParams({ cc: memberEmail('cc') }) + + const result = await service.executeSendEmail(invocation) + + expectSkipped(result) + expect(result.logs).toContainEqual( + expect.objectContaining({ + message: `Skipping send: could not check organization members for these addresses: ${memberEmail()}, ${memberEmail('cc')}. Try again, or verify your own domain to send to anyone.`, + }) + ) + expectBlockedEvent('check_failed', false, 2) + }) + + it('checks members again right before delivery', async () => { + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + const preparation = jest.spyOn(EmailSuppressionService.prototype, 'isSuppressed') + const isSuppressed = preparation.getMockImplementation()! + try { + service = createSandboxService(true) + await service.executeSendEmail(invocation) + expect(capture).not.toHaveBeenCalled() + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM posthog_organizationmembership WHERE user_id = $1', + [memberId], + 'test:revoke-sandbox-member-during-preparation' + ) + clock.mockReturnValue(now + 58_000) + preparation.mockImplementation(async function (this: EmailSuppressionService, ...args) { + const suppressed = await isSuppressed.apply(this, args) + clock.mockReturnValue(now + 63_000) + return suppressed + }) + invocation.state.vmState = { stack: [] } as any + + const result = await service.executeSendEmail(invocation) + + expectSkipped(result) + expectBlockedEvent('recipient_not_member', false, 1) + } finally { + clock.mockRestore() + preparation.mockRestore() + } + }) + + it('captures one blocked event per reason and send kind while logging every skip', async () => { + for (let attempt = 0; attempt < 2; attempt++) { + service = createSandboxService(true) + invocation.queueParameters = createSandboxParams({ to: { email: 'outside@example.com' } }) + const result = await service.executeSendEmail(invocation) + expect(result.logs).toContainEqual( + expect.objectContaining({ message: expect.stringContaining('outside@example.com') }) + ) + } + expect(capture).toHaveBeenCalledTimes(1) + + await createSandboxService(true).executeSendEmail(invocation, true) + expect(capture).toHaveBeenCalledTimes(2) + expectBlockedEvent('recipient_not_member', true, 1) + + invocation.queueParameters = createSandboxParams() + await createSandboxService(false).executeSendEmail(invocation) + expect(capture).toHaveBeenCalledTimes(3) + expectBlockedEvent('switch_off', false, 0) + }) + }) describe('integration validation', () => { beforeEach(async () => { await insertIntegration(hub.postgres, team.id, { diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index c0f4581a699c..c69b429cc36d 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -17,6 +17,7 @@ import { createInvocationResult } from '~/cdp/utils/invocation-utils' import { logger } from '~/common/utils/logger' import { IntegrationManagerService } from '../managers/integration-manager.service' +import { OrganizationMembersService } from '../managers/organization-members.service' import { RecipientManagerRecipient, RecipientsManagerService } from '../managers/recipients-manager.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' @@ -39,6 +40,7 @@ import { RecipientTokensService, preferencesTokenSourceForInvocation, } from './recipient-tokens.service' +import { SandboxEmailBlockReason, SandboxEmailSender } from './sandbox-email-sender' const sesThrottleResponsesTotal = new Counter({ name: 'cdp_ses_throttle_responses_total', @@ -349,6 +351,90 @@ export function parseAddressList(value?: string): string[] | undefined { return result.length > 0 ? result : undefined } +type QuoteScan = { unquotedIndexes: number[]; balanced: boolean } + +function scanQuotes(text: string): QuoteScan { + const unquotedIndexes: number[] = [] + let quoted = false + let escaped = false + for (let index = 0; index < text.length; index++) { + const character = text[index] + if (escaped) { + escaped = false + } else if (quoted && character === '\\') { + escaped = true + } else if (character === '"') { + quoted = !quoted + } else if (!quoted) { + unquotedIndexes.push(index) + } + } + return { unquotedIndexes, balanced: !quoted } +} + +function splitOutsideQuotes(text: string): string[] { + const commas = scanQuotes(text).unquotedIndexes.filter((index) => text[index] === ',') + return [-1, ...commas].map((comma, position) => text.slice(comma + 1, commas[position] ?? text.length)) +} + +function angleAddress(entry: string): string | undefined { + const { unquotedIndexes, balanced } = scanQuotes(entry) + const opening = unquotedIndexes.find((index) => entry[index] === '<') + const closing = entry.length - 1 + if (!balanced || opening === undefined || entry[closing] !== '>' || /[\r\n]/.test(entry)) { + return undefined + } + const unquotedBetween = (start: number, end: number): string => + unquotedIndexes + .filter((index) => index >= start && index < end) + .map((index) => entry[index]) + .join('') + const mailbox = entry.slice(opening + 1, closing).trim() + const isFramed = !/[>@;]/.test(unquotedBetween(0, opening)) && !/[<>]/.test(unquotedBetween(opening + 1, closing)) + return isFramed && mailbox ? mailbox : undefined +} + +// Unlike `extractEmailsFromAddressList`, this keeps quoted commas inside one entry and leaves an +// ambiguous mailbox whole, so the member check sees exactly the addresses SES would deliver to. +function sandboxAddressList(value?: string): string[] { + return splitOutsideQuotes(value ?? '') + .map((entry) => entry.trim()) + .filter(Boolean) + .map((entry) => angleAddress(entry) ?? entry) +} + +function sandboxRecipientsOf(params: CyclotronInvocationQueueParametersEmailType): string[] { + return [params.to.email.trim(), ...sandboxAddressList(params.cc), ...sandboxAddressList(params.bcc)] +} + +type SandboxRecipientBlockReason = Exclude + +const SANDBOX_UNAVAILABLE_LOG = + 'Skipping send: the sandbox sender is unavailable right now. Verify your own domain to keep sending.' + +const SANDBOX_RECIPIENT_BLOCK_COPY: Record< + SandboxRecipientBlockReason, + { explanation: string; label: string; guidance: string } +> = { + recipient_not_member: { + explanation: + 'Skipping send: the sandbox sender only sends to active organization members with verified email addresses', + label: '. Blocked addresses: ', + guidance: 'Verify your own domain to send to anyone.', + }, + recipient_unverified: { + explanation: + 'Skipping send: the sandbox sender only sends to organization members with verified email addresses', + label: '. Unverified addresses: ', + guidance: 'Verify your email address in PostHog, or ask these members to verify theirs.', + }, + check_failed: { + explanation: 'Skipping send: could not check organization members', + label: ' for these addresses: ', + guidance: 'Try again, or verify your own domain to send to anyone.', + }, +} + export class EmailService { sesV2Client: SESv2Client | null @@ -366,7 +452,9 @@ export class EmailService { private recipientsManager: RecipientsManagerService, private messageAssetsService?: MessageAssetsService, private workflowEmailRateLimiter: RateLimiterService | null = null, - private teamEmailRateLimiter: RateLimiterService | null = null + private teamEmailRateLimiter: RateLimiterService | null = null, + private sandboxSender: SandboxEmailSender | null = null, + private organizationMembers: OrganizationMembersService | null = null ) { this.sesV2Client = this.sesConfig.sesRegion ? new SESv2Client({ @@ -403,6 +491,7 @@ export class EmailService { const params = invocation.queueParameters const integrationId = selectEmailSenderIntegrationId(invocation.id, params.from) const integration = await this.integrationManager.get(integrationId) + const isSandbox = integration?.config.provider === 'sandbox' let success: boolean = false let throttled: boolean = false @@ -490,6 +579,19 @@ export class EmailService { ) } + if (isSandbox && !isTest && !('hogFlow' in invocation)) { + this.skipSandboxSend( + result, + 'Skipping send: the sandbox sender is only available for workflows and test sends.' + ) + return result + } + if (isSandbox && !this.sandboxSender?.canSend) { + this.skipSandboxSend(result, SANDBOX_UNAVAILABLE_LOG) + await this.sandboxSender?.captureBlocked(invocation.teamId, isTest, 'switch_off', 0) + return result + } + const from = this.resolveFromSender(integration, params.from, addLog) // An empty `to` comes from the customer's data (a person without an email property, or no @@ -516,6 +618,11 @@ export class EmailService { return result } + const recipients = isSandbox ? sandboxRecipientsOf(params) : [] + if (isSandbox && !(await this.checkSandboxRecipients(result, recipients, isTest))) { + return result + } + // Single choke point for the suppression check — every send path lands here regardless // of whether the invocation came from a workflow action or an email destination hog // function. Checking here means callers can't bypass it by taking a different upstream @@ -678,6 +785,11 @@ export class EmailService { case 'ses': await this.sendEmailWithSES(result, sendParams, from, trackingEnabled, isTest) break + case 'sandbox': + if (await this.checkSandboxRecipients(result, recipients, isTest)) { + this.skipSandboxSend(result, SANDBOX_UNAVAILABLE_LOG) + } + return result case 'unsupported': throw new Error('Email delivery mode not supported') @@ -893,6 +1005,52 @@ export class EmailService { return null } + private skipSandboxSend( + result: CyclotronJobInvocationResult, + reason: string + ): void { + createAddLogFunction(result.logs)('info', reason) + result.emailSendResult = { status: 'skipped', reason } + result.skipped = true + result.invocation.state.vmState?.stack.push({ success: false }) + } + + // Runs once before the send is prepared and again right before delivery, because a member can + // leave the organization while the send waits on suppression checks and rate limits. + private async checkSandboxRecipients( + result: CyclotronJobInvocationResult, + recipients: string[], + isTest: boolean + ): Promise { + const { teamId } = result.invocation + 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) + blockedRecipients = blocked.map(({ address }) => address) + reason = blocked.every(({ unverifiedMember }) => unverifiedMember) + ? 'recipient_unverified' + : 'recipient_not_member' + } catch (error) { + blockedRecipients = recipients + reason = 'check_failed' + logger.warn('Could not check sandbox email recipients', { + teamId, + error: error instanceof Error ? error.message : 'Unknown error', + }) + } + if (!blockedRecipients.length) { + return true + } + const { explanation, label, guidance } = SANDBOX_RECIPIENT_BLOCK_COPY[reason] + this.skipSandboxSend(result, `${explanation}${label}${blockedRecipients.join(', ')}. ${guidance}`) + await this.sandboxSender?.captureBlocked(teamId, isTest, reason, blockedRecipients.length) + return false + } + // Per-send tracking decision, combining two independent controls: // 1. Message-level: `tracking_enabled` on the email step's config (via `hogFunction.metadata`, // same as `message_category_type`). False is a hard off regardless of consent. Absent means diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts new file mode 100644 index 000000000000..37f4a9ada68c --- /dev/null +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -0,0 +1,93 @@ +import { RedisV2 } from '~/common/redis/redis-v2' +import { logger } from '~/common/utils/logger' +import { captureTeamEvent } from '~/common/utils/posthog' +import { TeamManager } from '~/common/utils/team-manager' + +export interface SandboxEmailSenderConfig { + enabled: boolean + tenantName: string + configurationSetName: string + fromAddress: string +} + +const ROUTE_SETTING_NAMES = { + tenantName: 'SES_SANDBOX_TENANT_NAME', + configurationSetName: 'SES_SANDBOX_CONFIGURATION_SET', + fromAddress: 'SES_SANDBOX_FROM_ADDRESS', +} as const + +function emptyRouteSettings(config: SandboxEmailSenderConfig): string[] { + return (Object.keys(ROUTE_SETTING_NAMES) as (keyof typeof ROUTE_SETTING_NAMES)[]) + .filter((setting) => !config[setting].trim()) + .map((setting) => ROUTE_SETTING_NAMES[setting]) +} + +export type SandboxEmailBlockReason = 'switch_off' | 'recipient_not_member' | 'recipient_unverified' | 'check_failed' + +const BLOCKED_EVENT_WINDOW_SECONDS = 3600 + +export class SandboxEmailSender { + public readonly canSend: boolean + + constructor( + public readonly config: SandboxEmailSenderConfig, + private teamManager: TeamManager, + private blockedEventRedis: Pick + ) { + this.config = { + ...config, + tenantName: config.tenantName.trim(), + configurationSetName: config.configurationSetName.trim(), + fromAddress: config.fromAddress.trim(), + } + const emptySettings = emptyRouteSettings(config) + if (config.enabled && emptySettings.length) { + logger.error('The sandbox sender is switched on with empty route settings, so it skips every send', { + emptySettings, + }) + } + this.canSend = config.enabled && !emptySettings.length + } + + public async captureBlocked( + teamId: number, + isTest: boolean, + reason: SandboxEmailBlockReason, + blockedRecipientCount: number + ): Promise { + try { + if (!(await this.isFirstBlockInWindow(teamId, isTest, reason))) { + return + } + const team = await this.teamManager.getTeam(teamId) + if (team) { + captureTeamEvent(team, 'workflows sandbox email blocked', { + reason, + blocked_recipient_count: blockedRecipientCount, + is_test: isTest, + }) + } + } catch (error) { + logger.warn('Could not capture sandbox email event', { teamId, error }) + } + } + + private async isFirstBlockInWindow( + teamId: number, + isTest: boolean, + reason: SandboxEmailBlockReason + ): Promise { + const firstBlock = await this.blockedEventRedis.useClient( + { name: 'sandbox-blocked-event', timeout: 1000 }, + (client) => + client.set( + `@posthog/workflows-sandbox-blocked/${teamId}/${reason}/${isTest ? 'test' : 'live'}`, + '1', + 'EX', + BLOCKED_EVENT_WINDOW_SECONDS, + 'NX' + ) + ) + return firstBlock === 'OK' + } +} From 4d5ab71d1dedbb85fbf2b449f0fe2174877d5ab2 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 05:54:17 +0000 Subject: [PATCH 2/3] fix(workflows): keep sandbox sends off the team sending budget A sandbox send reaches only the team's own members, so it must not spend or wait on the team's tier budget that its real sends share. Co-Authored-By: Claude Opus 5.5 --- .../internal/workflows-email-sending-tiers.md | 1 + .../services/messaging/email.service.test.ts | 27 ++++++++++++++++--- .../cdp/services/messaging/email.service.ts | 3 ++- 3 files changed, 27 insertions(+), 4 deletions(-) diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index c5a718d4db77..13a714ac1a32 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -83,5 +83,6 @@ Every To, CC, and BCC address must belong to an active member of the team's acti 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). +Sandbox sends never spend the team's tier budget, because they reach only its own members. The sandbox route is not built yet, so a send that passes every check still skips as unavailable. Keep the switch off in production until the sandbox caps are deployed. diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index b74db59a9d4c..e58ee8c3209c 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -258,8 +258,11 @@ describe('EmailService', () => { const createSandboxService = ( enabled: boolean, membersPostgres: PostgresRouter = hub.postgres, - routeSettings: Partial = {} + routeSettings: Partial = {}, + teamEmailRateLimiter: RateLimiterService | null = null ): EmailService => { + const workflowsConfig = new TeamWorkflowsConfigService(hub.postgres, hub.pubSub) + jest.spyOn(workflowsConfig, 'getEmailSendingTier').mockResolvedValue(0) const sandboxService = new EmailService( { sesAccessKeyId: hub.SES_ACCESS_KEY_ID, @@ -268,9 +271,12 @@ describe('EmailService', () => { sesEndpoint: hub.SES_ENDPOINT, sesTrackedConfigurationSet: hub.SES_TRACKED_CONFIGURATION_SET, sesUntrackedConfigurationSet: hub.SES_UNTRACKED_CONFIGURATION_SET, + teamEmailCapMode: 'enforce', + teamEmailTierHourlyCaps: [1], + teamEmailTierDailyCaps: [1], }, hub.integrationManager, - new TeamWorkflowsConfigService(hub.postgres, hub.pubSub), + workflowsConfig, hub.ENCRYPTION_SALT_KEYS, hub.SITE_URL, new EmailTrackingCodeSigner(hub.ENCRYPTION_SALT_KEYS, hub.CDP_EMAIL_TRACKING_URL), @@ -278,7 +284,7 @@ describe('EmailService', () => { new RecipientsManagerService(hub.postgres), undefined, null, - null, + teamEmailRateLimiter, new SandboxEmailSender( { enabled, @@ -430,6 +436,21 @@ describe('EmailService', () => { } ) + it('never spends the team sending budget on a sandbox send', async () => { + const claimAllOrNothingPair = jest + .fn() + .mockResolvedValue({ granted: false, deniedIndex: 0, retryAfterMs: 60_000, reserved: false }) + service = createSandboxService(true, hub.postgres, {}, { + claimAllOrNothingPair, + } as unknown as RateLimiterService) + + const result = await service.executeSendEmail(invocation) + + expectSkipped(result) + expect(result.logs).toContainEqual(expect.objectContaining({ level: 'info', message: UNAVAILABLE_LOG })) + expect(claimAllOrNothingPair).not.toHaveBeenCalled() + }) + it.each([ ['a display name', 'Example colleague', false], ['a quoted comma', '"Example, colleague"', false], diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index c69b429cc36d..26fc33357f68 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -727,7 +727,8 @@ export class EmailService { 1 + extractEmailsFromAddressList(deliverableParams.cc).length + extractEmailsFromAddressList(deliverableParams.bcc).length - const capDelay = await this.claimTeamSendingBudget(invocation, isTest, capRecipients) + // Sandbox sends never spend the team's tier budget: they reach only its own members. + const capDelay = isSandbox ? null : await this.claimTeamSendingBudget(invocation, isTest, capRecipients) if (capDelay) { result.finished = false // Re-attach the email payload before rescheduling, for the same reason as the From 9ffcb7a96ecaddce9cbc74acc684a76070c07fe0 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Sat, 31 Oct 2026 10:11:53 +0000 Subject: [PATCH 3/3] fix(workflows): record skipped sandbox sends as failed transactional emails A transactional email whose sandbox send is skipped stayed queued, because the sandbox skip paths never wrote the skipped status that master's other skip paths write. Sender-side skips record sending_suspended and recipient-side skips record recipient_unreachable. Co-Authored-By: Claude Opus 5.5 --- .../services/messaging/email.service.test.ts | 29 +++++++++++++++++++ .../cdp/services/messaging/email.service.ts | 22 ++++++++++---- 2 files changed, 45 insertions(+), 6 deletions(-) diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index e7f73c2a22b5..0fe4d8e67f68 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -659,6 +659,35 @@ describe('EmailService', () => { expect(capture).toHaveBeenCalledTimes(3) expectBlockedEvent('switch_off', false, 0) }) + + it.each([ + ['the sandbox sender is switched off', false, memberEmail, 'sending_suspended'], + ['the recipient is not a member', true, () => 'outside@example.com', 'recipient_unreachable'], + ])('records a skipped transactional email when %s', async (_case, enabled, recipient, expectedReason) => { + const produce = jest.fn().mockResolvedValue(undefined) + const emailStatusService = new EmailStatusService( + { produce } as unknown as IngestionOutputs, + { EMAIL_STATUS_ENABLED: true } + ) + service = createSandboxService(enabled) + service['emailStatusService'] = emailStatusService + invocation.id = new UUIDT().toString() + invocation.state.globals.event.event = TRANSACTIONAL_EMAIL_EVENT + invocation.queueParameters = createSandboxParams({ to: { email: recipient() } }) + + await service.executeSendEmail(invocation) + await emailStatusService.flush() + + const rows = produce.mock.calls.map(([, message]) => parseJSON(message.value.toString())) + expect(rows).toEqual([ + expect.objectContaining({ + team_id: team.id, + message_id: invocation.id, + status: 'failed', + reason: expectedReason, + }), + ]) + }) }) describe('integration validation', () => { beforeEach(async () => { diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index b9f5c1a06023..3e0b16601818 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -24,7 +24,7 @@ import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.se import { RateLimiterService } from '../rate-limiter/rate-limiter.service' import { getEmailQueuePriorityClass } from './email-priority' import { selectEmailSenderIntegrationId } from './email-sender-selection' -import { EmailStatusService, isTransactionalEmailSend } from './email-status.service' +import { EmailSkipReason, EmailStatusService, isTransactionalEmailSend } from './email-status.service' import { EmailSuppressionService, SuppressionCheckOptions } from './email-suppression.service' import { addTrackingToEmail, @@ -646,12 +646,14 @@ export class EmailService { if (isSandbox && !isTest && !('hogFlow' in invocation)) { this.skipSandboxSend( result, - 'Skipping send: the sandbox sender is only available for workflows and test sends.' + isTest, + 'Skipping send: the sandbox sender is only available for workflows and test sends.', + 'sending_suspended' ) return result } if (isSandbox && !this.sandboxSender?.canSend) { - this.skipSandboxSend(result, SANDBOX_UNAVAILABLE_LOG) + this.skipSandboxSend(result, isTest, SANDBOX_UNAVAILABLE_LOG, 'sending_suspended') await this.sandboxSender?.captureBlocked(invocation.teamId, isTest, 'switch_off', 0) return result } @@ -854,7 +856,7 @@ export class EmailService { break case 'sandbox': if (await this.checkSandboxRecipients(result, recipients, isTest)) { - this.skipSandboxSend(result, SANDBOX_UNAVAILABLE_LOG) + this.skipSandboxSend(result, isTest, SANDBOX_UNAVAILABLE_LOG, 'sending_suspended') } return result @@ -1085,10 +1087,13 @@ export class EmailService { private skipSandboxSend( result: CyclotronJobInvocationResult, - reason: string + isTest: boolean, + reason: string, + statusReason: EmailSkipReason ): void { createAddLogFunction(result.logs)('info', reason) result.emailSendResult = { status: 'skipped', reason } + this.emailStatusService?.queueSkippedSend(result.invocation, statusReason, isTest) result.skipped = true result.invocation.state.vmState?.stack.push({ success: false }) } @@ -1124,7 +1129,12 @@ export class EmailService { return true } const { explanation, label, guidance } = SANDBOX_RECIPIENT_BLOCK_COPY[reason] - this.skipSandboxSend(result, `${explanation}${label}${blockedRecipients.join(', ')}. ${guidance}`) + this.skipSandboxSend( + result, + isTest, + `${explanation}${label}${blockedRecipients.join(', ')}. ${guidance}`, + 'recipient_unreachable' + ) await this.sandboxSender?.captureBlocked(teamId, isTest, reason, blockedRecipients.length) return false }