From 35ce76039dd7a521c43905bdfc0b0eb58b7de163 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Tue, 27 Oct 2026 08:07:08 +0000 Subject: [PATCH 1/2] feat(workflows): skip sandbox sends while the shared ses tenant is paused The email worker reads the sandbox SES tenant state that Django mirrors, skips every sandbox send while the tenant is paused or the state cannot be read, and reloads the state on the reload-sandbox-sender-state message. After the daily allowance claim it checks the pause again and gives the claim back when the pause blocks delivery. Refs Silthus/personal#199 Co-Authored-By: Claude Opus 5.5 --- .../internal/workflows-email-sending-tiers.md | 4 +- nodejs/src/cdp/cdp-services.ts | 4 +- .../sandbox-sender-state.service.test.ts | 24 ++ .../managers/sandbox-sender-state.service.ts | 52 +++++ .../services/messaging/email.service.test.ts | 211 +++++++++++++++++- .../cdp/services/messaging/email.service.ts | 29 ++- .../messaging/sandbox-email-sender.test.ts | 7 +- .../messaging/sandbox-email-sender.ts | 14 +- .../rate-limiter/rate-limiter.service.test.ts | 3 +- 9 files changed, 339 insertions(+), 9 deletions(-) create mode 100644 nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts create mode 100644 nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts diff --git a/docs/internal/workflows-email-sending-tiers.md b/docs/internal/workflows-email-sending-tiers.md index 37d323ec0f5c..1b7eb595becc 100644 --- a/docs/internal/workflows-email-sending-tiers.md +++ b/docs/internal/workflows-email-sending-tiers.md @@ -110,8 +110,10 @@ A reached allowance skips the send instead of delaying it. A limiter error, or a The worker checks members after the claim and gives the claim back when that check skips the send. An SES throttle response also gives the claim back before the delivery retry. Returns are deduplicated, and Valkey ignores a return that arrives more than five minutes after it was issued. Test sends run inline in the CDP API, so a sandbox-enabled CDP API opens the SES Valkey pool (`SES_RATE_LIMITER_VALKEY_*`) for the allowances. Without it, every sandbox test send skips. +Every sandbox send, test sends included, skips while the shared sandbox SES tenant is paused (`DISABLED`), or when the pause state cannot be read. A tenant Django has never synced counts as open, and SES enforces its own state. +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. +A committed state change reaches the workers through the `reload-sandbox-sender-state` message. Own-sender sends do not read this state. 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. Live workflow sends still count as `email_sent`, which the sending tier reads, until sandbox sends get their own metric. diff --git a/nodejs/src/cdp/cdp-services.ts b/nodejs/src/cdp/cdp-services.ts index cf9ca2accf6d..fbe4641e098b 100644 --- a/nodejs/src/cdp/cdp-services.ts +++ b/nodejs/src/cdp/cdp-services.ts @@ -34,6 +34,7 @@ import { IntegrationManagerService } from './services/managers/integration-manag 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 { 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' @@ -499,7 +500,8 @@ export function createCdpCoreServices( }, deps.teamManager, redis, - sandboxEmailRateLimiter + sandboxEmailRateLimiter, + new SandboxSenderStateService(deps.postgres, deps.pubSub) ), new OrganizationMembersService(deps.postgres) ) diff --git a/nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts new file mode 100644 index 000000000000..6bdde7cfa59e --- /dev/null +++ b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.test.ts @@ -0,0 +1,24 @@ +import { DependencyUnavailableError } from '~/common/utils/db/error' +import { PostgresRouter } from '~/common/utils/db/postgres' +import { PubSub } from '~/common/utils/pubsub' + +import { SandboxSenderStateService } from './sandbox-sender-state.service' + +describe('SandboxSenderStateService', () => { + it('rides out a transient database error instead of failing the sandbox send', async () => { + const query = jest + .fn() + .mockRejectedValueOnce( + new DependencyUnavailableError('server closed the connection unexpectedly', 'Postgres', new Error()) + ) + .mockResolvedValueOnce({ rows: [{ tenant_name: 'sandbox-tenant', sending_status: 'DISABLED' }] }) + const service = new SandboxSenderStateService( + { query } as unknown as PostgresRouter, + { + on: jest.fn(), + } as unknown as PubSub + ) + + await expect(service.isPaused('sandbox-tenant')).resolves.toBe(true) + }) +}) diff --git a/nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts new file mode 100644 index 000000000000..3a82caf91e31 --- /dev/null +++ b/nodejs/src/cdp/services/managers/sandbox-sender-state.service.ts @@ -0,0 +1,52 @@ +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' +import { DEFAULT_LOADER_RETRY, LazyLoader } from '~/common/utils/lazy-loader' +import { PubSub } from '~/common/utils/pubsub' + +type SandboxSenderState = { sendingStatus: string | undefined; generation: number } + +export class SandboxSenderStateService { + private lazyLoader: LazyLoader + private generation = 0 + + 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), + loaderRetry: DEFAULT_LOADER_RETRY, + }) + pubSub.on<{ tenantName: string }>('reload-sandbox-sender-state', ({ tenantName }): void => { + this.generation++ + this.lazyLoader.markForRefresh(tenantName) + }) + } + + public async isPaused(tenantName: string): Promise { + for (let attempt = 0; attempt < 2; attempt++) { + const state = await this.lazyLoader.get(tenantName) + if (state?.generation === this.generation) { + // No row means never synced; SES stays the final word. + return state.sendingStatus === 'DISABLED' + } + this.lazyLoader.markForRefresh(tenantName) + } + throw new Error('Sandbox sender state changed during lookup') + } + + private async fetchStates(tenantNames: string[]): Promise> { + const generation = this.generation + 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' + ) + const statuses = new Map(rows.map((row) => [row.tenant_name, row.sending_status])) + return Object.fromEntries(tenantNames.map((name) => [name, { sendingStatus: statuses.get(name), generation }])) + } +} diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index d84b5280a370..b1792202d392 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -31,6 +31,7 @@ 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 { SandboxSenderStateService } from '../managers/sandbox-sender-state.service' import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { RateLimiterService } from '../rate-limiter/rate-limiter.service' import { selectEmailSenderIntegrationId } from './email-sender-selection' @@ -280,12 +281,14 @@ describe('EmailService', () => { teamEmailRateLimiter = null, messageAssetsService, dailyCapLimiter = new RateLimiterService(sandboxRedis, { name: 'sandbox-email' }), + senderState = new SandboxSenderStateService(hub.postgres, hub.pubSub), }: { membersPostgres?: PostgresRouter senderSettings?: Partial teamEmailRateLimiter?: RateLimiterService | null messageAssetsService?: MessageAssetsService dailyCapLimiter?: RateLimiterService | null + senderState?: SandboxSenderStateService } = {} ): EmailService => { const workflowsConfig = new TeamWorkflowsConfigService(hub.postgres, hub.pubSub) @@ -324,7 +327,8 @@ describe('EmailService', () => { }, hub.teamManager, sandboxRedis, - dailyCapLimiter + dailyCapLimiter, + senderState ), new OrganizationMembersService(membersPostgres) ) @@ -757,6 +761,211 @@ describe('EmailService', () => { expectBlockedEvent('switch_off', false, 0) }) + describe('pause gate', () => { + const pausedMessage = + 'Skipping send: the sandbox sender is paused right now. Verify your own domain to keep sending.' + const tenantName = (): string => `sandbox-tenant-${team.id}` + const setTenantStatus = async (status: string): Promise => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + `INSERT INTO workflows_sandboxsendertenantstate (tenant_name, sending_status, synced_at) + VALUES ($1, $2, now()) + ON CONFLICT (tenant_name) DO UPDATE SET sending_status = $2`, + [tenantName(), status], + 'test:set-sandbox-tenant-status' + ) + } + + afterEach(async () => { + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + 'DELETE FROM workflows_sandboxsendertenantstate WHERE tenant_name = $1', + [tenantName()], + 'test:delete-sandbox-tenant-status' + ) + }) + + it.each([false, true])('skips while the sandbox tenant is paused (isTest=%s)', async (isTest) => { + await setTenantStatus('DISABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + + const result = await service.executeSendEmail(invocation, isTest) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.error).toBeUndefined() + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toEqual( + expect.arrayContaining([expect.objectContaining({ level: 'info', message: pausedMessage })]) + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'paused', is_test: isTest, blocked_recipient_count: 0 } + ) + }) + + it.each(['ENABLED', 'REINSTATED'])('sends while the sandbox tenant is %s', async (status) => { + await setTenantStatus(status) + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(result.skipped).toBeFalsy() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + }) + + it('skips when the pause state cannot be read', async () => { + const brokenPostgres = new PostgresRouter(hub) + await brokenPostgres.end() + service = createSandboxService(true, { + senderSettings: { tenantName: tenantName() }, + senderState: new SandboxSenderStateService(brokenPostgres, hub.pubSub), + }) + + const result = await service.executeSendEmail(invocation) + + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.invocation.state.vmState?.stack).toEqual([{ success: false }]) + expect(result.logs).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + level: 'info', + message: + 'Skipping send: could not check whether the sandbox sender is paused. Try again later, or verify your own domain to keep sending.', + }), + ]) + ) + expect(capture).toHaveBeenCalledWith( + expect.objectContaining({ id: team.id }), + 'workflows sandbox email blocked', + { reason: 'check_failed', is_test: false, blocked_recipient_count: 0 } + ) + }) + + it('leaves own-sender sends alone while the sandbox tenant is paused', async () => { + await setTenantStatus('DISABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + invocation.queueParameters = createEmailParams({ from: { integrationId: 1 } }) + + const result = await service.executeSendEmail(invocation) + + expect(result.error).toBeUndefined() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + }) + + it('rechecks a pause after a delayed cap claim and returns the grant', async () => { + await setTenantStatus('ENABLED') + const now = Date.now() + const clock = jest.spyOn(Date, 'now').mockReturnValue(now) + const dailyCapLimiter = new RateLimiterService(sandboxRedis, { name: 'sandbox-email' }) + service = createSandboxService(true, { + dailyCapLimiter, + senderSettings: { tenantName: tenantName(), dailyTeamCap: '1', dailyRecipientCap: '1' }, + }) + const claim = dailyCapLimiter.claimAllOrNothing.bind(dailyCapLimiter) + const delayedClaim = jest + .spyOn(dailyCapLimiter, 'claimAllOrNothing') + .mockImplementationOnce(async (buckets) => { + const grant = await claim(buckets) + await setTenantStatus('DISABLED') + clock.mockReturnValue(now + 40_000) + return grant + }) + try { + const blocked = await service.executeSendEmail(invocation) + expect(blocked).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(blocked.error).toBeUndefined() + expect(sendEmailSpy).not.toHaveBeenCalled() + expect(blocked.logs).toContainEqual(expect.objectContaining({ message: pausedMessage })) + + await setTenantStatus('ENABLED') + clock.mockReturnValue(now + 80_000) + invocation.state.vmState = { stack: [] } as any + const resumed = await service.executeSendEmail(invocation) + expect(resumed.error).toBeUndefined() + expect(resumed.skipped).toBeFalsy() + expect(sendEmailSpy).toHaveBeenCalledTimes(1) + } finally { + delayedClaim.mockRestore() + clock.mockRestore() + } + }) + + it('skips when a pause invalidates a pending state read', async () => { + await setTenantStatus('ENABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + let snapshotLoaded!: () => void + let releaseRead!: () => void + const snapshot = new Promise((resolve) => { + snapshotLoaded = resolve + }) + const heldRead = new Promise((resolve) => { + releaseRead = resolve + }) + let acknowledgeReload!: () => void + const reload = new Promise((resolve) => { + acknowledgeReload = resolve + }) + hub.pubSub.on<{ tenantName: string }>('reload-sandbox-sender-state', ({ tenantName: name }) => { + if (name === tenantName()) { + acknowledgeReload() + } + }) + const query = hub.postgres.query.bind(hub.postgres) + let held = false + const delayedRead = jest.spyOn(hub.postgres, 'query').mockImplementation(async (...args) => { + const result = await query(...args) + if (args[3] === 'fetch-sandbox-sender-state' && !held) { + held = true + snapshotLoaded() + await heldRead + } + return result + }) + try { + const pendingSend = service.executeSendEmail(invocation) + await snapshot + await setTenantStatus('DISABLED') + await hub.pubSub.publish( + 'reload-sandbox-sender-state', + JSON.stringify({ tenantName: tenantName() }) + ) + await reload + releaseRead() + const result = await pendingSend + expect(result).toMatchObject({ finished: true, skipped: true, metrics: [] }) + expect(result.error).toBeUndefined() + expect(result.logs).toContainEqual(expect.objectContaining({ message: pausedMessage })) + expect(sendEmailSpy).not.toHaveBeenCalled() + } finally { + releaseRead() + delayedRead.mockRestore() + } + }) + + it('picks up a pause announced while the state is cached', async () => { + await setTenantStatus('ENABLED') + service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) + expect((await service.executeSendEmail(invocation)).skipped).toBeFalsy() + + await setTenantStatus('DISABLED') + + await waitForExpect(async () => { + await hub.pubSub.publish( + 'reload-sandbox-sender-state', + JSON.stringify({ tenantName: tenantName() }) + ) + const result = await service.executeSendEmail(invocation) + expect(result.logs).toEqual( + expect.arrayContaining([expect.objectContaining({ message: pausedMessage })]) + ) + }, 3000) + }) + }) + describe('daily caps', () => { const TEAM_CAP_REACHED = "Skipping send: this email would go over the sandbox sender's daily limit for this project. Verify your own domain to send more." diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index 10efa7f9b92b..24302dcb7121 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/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 +type SandboxRecipientBlockReason = Exclude type SesRoute = { tenantName: string @@ -445,6 +445,12 @@ const SANDBOX_RECIPIENT_BLOCK_COPY: Record< }, } +const SANDBOX_PAUSE_SKIP_MESSAGES = { + paused: 'Skipping send: the sandbox sender is paused right now. Verify your own domain to keep sending.', + check_failed: + 'Skipping send: could not check whether the sandbox sender is paused. Try again later, or verify your own domain to keep sending.', +} as const + const SANDBOX_CAP_SKIP_MESSAGES = { project_cap_reached: "Skipping send: this email would go over the sandbox sender's daily limit for this project. Verify your own domain to send more.", @@ -612,6 +618,9 @@ export class EmailService { await this.sandboxSender?.captureBlocked(invocation.teamId, isTest, 'switch_off', 0) return result } + if (isSandbox && !(await this.passesSandboxPauseGate(result, isTest))) { + return result + } const from = this.resolveFromSender(integration, params.from, addLog) @@ -829,7 +838,10 @@ export class EmailService { if (!capGrant) { return result } - if (!(await this.checkSandboxRecipients(result, deliveryRecipients, isTest))) { + if ( + !(await this.passesSandboxPauseGate(result, isTest)) || + !(await this.checkSandboxRecipients(result, deliveryRecipients, isTest)) + ) { await this.sandboxSender!.returnDailyCaps(capGrant) return result } @@ -1096,6 +1108,19 @@ export class EmailService { return false } + private async passesSandboxPauseGate( + result: CyclotronJobInvocationResult, + isTest: boolean + ): Promise { + const gate = await this.sandboxSender!.pauseGate() + if (gate === 'open') { + return true + } + this.skipSandboxSend(result, SANDBOX_PAUSE_SKIP_MESSAGES[gate]) + await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, gate, 0) + return false + } + private async blockSandboxRecipients( result: CyclotronJobInvocationResult, blockedRecipients: string[], diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts index 08ace91b93af..08a78499abd0 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.test.ts @@ -11,6 +11,7 @@ const teamManager = { getTeam: jest.fn().mockResolvedValue({ organization_id: '00000000-0000-0000-0000-000000000001' }), } as unknown as TeamManager const blockedEventRedis = { useClient: jest.fn() } +const senderState = { isPaused: jest.fn().mockResolvedValue(false) } const sender = new SandboxEmailSender( { enabled: true, @@ -22,7 +23,8 @@ const sender = new SandboxEmailSender( }, teamManager, blockedEventRedis, - null + null, + senderState ) const params = (html: string, text?: string, preheader?: string): CyclotronInvocationQueueParametersEmailType => ({ type: 'email', @@ -72,7 +74,8 @@ describe('sandbox email content', () => { }, teamManager, blockedEventRedis, - null + null, + senderState ) const result = await paddedSender.withIdentificationFooter( params('

Hello

'), diff --git a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts index 4df643e09d7c..0bb188858895 100644 --- a/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts +++ b/nodejs/src/cdp/services/messaging/sandbox-email-sender.ts @@ -7,6 +7,7 @@ import { logger } from '~/common/utils/logger' import { captureTeamEvent } from '~/common/utils/posthog' import { TeamManager } from '~/common/utils/team-manager' +import { SandboxSenderStateService } from '../managers/sandbox-sender-state.service' import { ClaimRequest, IssuedClaim, RateLimiterService } from '../rate-limiter/rate-limiter.service' import { maybeAddPreheaderToEmail } from './helpers/preheader' @@ -40,6 +41,7 @@ export type SandboxEmailBlockReason = | 'recipient_unverified' | 'check_failed' | 'cap_reached' + | 'paused' export type SandboxDailyCapGrant = { type: 'granted'; claim: IssuedClaim; buckets: ClaimRequest[] } @@ -168,7 +170,8 @@ export class SandboxEmailSender { public readonly config: SandboxEmailSenderConfig, private teamManager: TeamManager, private blockedEventRedis: Pick, - private dailyCapLimiter: RateLimiterService | null + private dailyCapLimiter: RateLimiterService | null, + private senderState: Pick ) { this.config = { ...config, @@ -185,6 +188,15 @@ export class SandboxEmailSender { this.canSend = config.enabled && !emptySettings.length } + public async pauseGate(): Promise<'open' | 'paused' | 'check_failed'> { + try { + return (await this.senderState.isPaused(this.config.tenantName)) ? 'paused' : 'open' + } catch (error) { + logger.warn('Could not check the sandbox sender pause state', { error }) + return 'check_failed' + } + } + public async claimDailyCaps(teamId: number, recipients: string[]): Promise { const dailyTeamCap = parseDailyCap(this.config.dailyTeamCap) const dailyRecipientCap = parseDailyCap(this.config.dailyRecipientCap) diff --git a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts index 6ef291ebd57b..15095c35f6ab 100644 --- a/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts +++ b/nodejs/src/cdp/services/rate-limiter/rate-limiter.service.test.ts @@ -434,7 +434,8 @@ describe('RateLimiterService', () => { }, teamManager, redis, - limiter + limiter, + { isPaused: () => Promise.resolve(false) } ) try { From 2179168e42195b3832e9ef4ee5995427bb1623ff Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:35:36 +0000 Subject: [PATCH 2/2] fix(workflows): record a sandbox pause skip as sending_suspended A paused or unreadable sandbox tenant now writes the same email status row as master's own paused SES tenant, so a transactional caller sees the skip. Co-Authored-By: Claude Opus 5.5 --- .../services/messaging/email.service.test.ts | 43 +++++++++++++++++++ .../cdp/services/messaging/email.service.ts | 2 +- 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index b752ef9c7de1..9c0741b2e56c 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -867,6 +867,49 @@ describe('EmailService', () => { ) }) + it.each([ + [ + 'the sandbox tenant is paused', + async (): Promise<{ senderState?: SandboxSenderStateService }> => { + await setTenantStatus('DISABLED') + return {} + }, + ], + [ + 'the pause state cannot be read', + async (): Promise<{ senderState?: SandboxSenderStateService }> => { + const brokenPostgres = new PostgresRouter(hub) + await brokenPostgres.end() + return { senderState: new SandboxSenderStateService(brokenPostgres, hub.pubSub) } + }, + ], + ])('records a skipped transactional email as sending_suspended when %s', async (_case, setUp) => { + const produce = jest.fn().mockResolvedValue(undefined) + const emailStatusService = new EmailStatusService( + { produce } as unknown as IngestionOutputs, + { EMAIL_STATUS_ENABLED: true } + ) + service = createSandboxService(true, { + senderSettings: { tenantName: tenantName() }, + ...(await setUp()), + }) + service['emailStatusService'] = emailStatusService + invocation.id = new UUIDT().toString() + invocation.state.globals.event.event = TRANSACTIONAL_EMAIL_EVENT + + await service.executeSendEmail(invocation) + await emailStatusService.flush() + + const rows = produce.mock.calls.map(([, message]) => parseJSON(message.value.toString())) + expect(rows).toEqual([ + expect.objectContaining({ + message_id: invocation.id, + status: 'failed', + reason: 'sending_suspended', + }), + ]) + }) + it('leaves own-sender sends alone while the sandbox tenant is paused', async () => { await setTenantStatus('DISABLED') service = createSandboxService(true, { senderSettings: { tenantName: tenantName() } }) diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index 523fd227f506..2657fa03585a 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -1200,7 +1200,7 @@ export class EmailService { if (gate === 'open') { return true } - this.skipSandboxSend(result, SANDBOX_PAUSE_SKIP_MESSAGES[gate]) + this.skipSandboxSend(result, isTest, SANDBOX_PAUSE_SKIP_MESSAGES[gate], 'sending_suspended') await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, gate, 0) return false }