feat(cdp): deliver and charge a repeated sandbox address once

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

@@ -369,7 +369,12 @@ describe('EmailService', () => {
                 expect(capture).toHaveBeenLastCalledWith(
                     expect.objectContaining({ id: team.id }),
                     'workflows sandbox email blocked',
-                    { reason, is_test: isTest, blocked_recipient_count: blockedRecipientCount }
+                    {
+                        reason,
+                        is_test: isTest,
+                        blocked_recipient_count: blockedRecipientCount,
+                        source: isTest ? 'test' : 'workflow',
+                    }
                 )
             }
 
@@ -610,6 +615,39 @@ describe('EmailService', () => {
                 )
             })
 
+            it('delivers to each member once when an address repeats across to, cc and bcc', async () => {
+                service = createSandboxService(true)
+                invocation.queueParameters = createSandboxParams({
+                    to: { email: memberEmail() },
+                    cc: `${memberEmail().toUpperCase()}, ${memberEmail('cc')}, ${memberEmail('cc')}`,
+                    bcc: `Example colleague <${memberEmail('cc')}>, ${memberEmail('bcc')}, ${memberEmail()}`,
+                })
+
+                const result = await service.executeSendEmail(invocation)
+
+                expect(result.error).toBeUndefined()
+                expect((sendEmailSpy.mock.calls[0][0] as SendEmailCommand).input.Destination).toEqual({
+                    ToAddresses: [memberEmail()],
+                    CcAddresses: [memberEmail('cc')],
+                    BccAddresses: [memberEmail('bcc')],
+                })
+            })
+
+            it('records a send from the transactional email API as an api send', async () => {
+                service = createSandboxService(true)
+                invocation.queueParameters = createSandboxParams({ to: { email: memberEmail() } })
+                invocation.state.globals.event.event = TRANSACTIONAL_EMAIL_EVENT
+
+                const result = await service.executeSendEmail(invocation)
+
+                expect(result.error).toBeUndefined()
+                expect(capture).toHaveBeenCalledWith(
+                    expect.objectContaining({ id: team.id }),
+                    'workflows sandbox email sent',
+                    { recipient_count: 1, source: 'api', is_test: false }
+                )
+            })
+
             it.each(['cc', 'bcc'] as const)('delivers a quoted %s mailbox whole', async (field) => {
                 const email = `"example,colleague-${team.id}"@example.com`
                 await createMember(email)
@@ -876,7 +914,12 @@ describe('EmailService', () => {
                     expect(capture).toHaveBeenCalledWith(
                         expect.objectContaining({ id: team.id }),
                         'workflows sandbox email blocked',
-                        { reason: 'paused', is_test: isTest, blocked_recipient_count: 0 }
+                        {
+                            reason: 'paused',
+                            is_test: isTest,
+                            blocked_recipient_count: 0,
+                            source: isTest ? 'test' : 'workflow',
+                        }
                     )
                 })
 
@@ -916,7 +959,7 @@ describe('EmailService', () => {
                     expect(capture).toHaveBeenCalledWith(
                         expect.objectContaining({ id: team.id }),
                         'workflows sandbox email blocked',
-                        { reason: 'check_failed', is_test: false, blocked_recipient_count: 0 }
+                        { reason: 'check_failed', is_test: false, blocked_recipient_count: 0, source: 'workflow' }
                     )
                 })
 
@@ -1164,19 +1207,20 @@ describe('EmailService', () => {
                     }
                 )
 
-                it('charges a repeated address once per address and once per copy for the project', async () => {
+                it('charges a repeated address once, as it is delivered once', async () => {
                     service = createSandboxService(true, {
-                        senderSettings: { dailyTeamCap: '3', dailyRecipientCap: '1' },
+                        senderSettings: { dailyTeamCap: '2', dailyRecipientCap: '1' },
                     })
 
                     expectSent(await send(false, { cc: memberEmail().toUpperCase(), bcc: ` ${memberEmail()} ` }))
+                    expectSent(await send(false, { to: { email: memberEmail('cc') } }))
                     expectCapSkip(
-                        await send(false, { to: { email: memberEmail('cc') } }),
+                        await send(false, { to: { email: memberEmail('bcc') } }),
                         false,
                         TEAM_CAP_REACHED,
                         'cap_reached'
                     )
-                    expect(sendEmailSpy).toHaveBeenCalledTimes(1)
+                    expect(sendEmailSpy).toHaveBeenCalledTimes(2)
                 })
 
                 it('refills a spent allowance over a day rather than seconds', async () => {

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

@@ -44,7 +44,13 @@ import {
     RecipientTokensService,
     preferencesTokenSourceForInvocation,
 } from './recipient-tokens.service'
-import { SandboxDailyCapGrant, SandboxEmailBlockReason, SandboxEmailSender } from './sandbox-email-sender'
+import {
+    SandboxDailyCapGrant,
+    SandboxEmailBlockReason,
+    SandboxEmailSender,
+    sandboxSendSource,
+} from './sandbox-email-sender'
+import { sandboxEnvelope, sandboxEnvelopeRecipients } from './sandbox-envelope'
 
 const sesThrottleResponsesTotal = new Counter({
     name: 'cdp_ses_throttle_responses_total',
@@ -408,60 +414,8 @@ 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)]
+    return sandboxEnvelopeRecipients(sandboxEnvelope(params.to.email, params.cc, params.bcc))
 }
 
 type SandboxRecipientBlockReason = Exclude<SandboxEmailBlockReason, 'switch_off' | 'paused' | 'organization_blocked'>
@@ -687,7 +641,12 @@ export class EmailService {
                     'Skipping send: the sandbox sender is unavailable right now. Verify your own domain to keep sending.',
                     'sending_suspended'
                 )
-                await this.sandboxSender?.captureBlocked(invocation.teamId, isTest, 'switch_off', 0)
+                await this.sandboxSender?.captureBlocked(
+                    invocation.teamId,
+                    sandboxSendSource(invocation, isTest),
+                    'switch_off',
+                    0
+                )
                 return result
             }
             if (isSandbox && !(await this.passesSandboxPauseGate(result, isTest))) {
@@ -920,7 +879,11 @@ export class EmailService {
                         return result
                     }
                     await this.sendSandboxEmail(result, deliveryParams, from, isTest, queueWait, route, capGrant)
-                    await this.sandboxSender!.captureSent(invocation.teamId, isTest, deliveryRecipients.length)
+                    await this.sandboxSender!.captureSent(
+                        invocation.teamId,
+                        sandboxSendSource(invocation, isTest),
+                        deliveryRecipients.length
+                    )
                     break
                 }
 
@@ -1206,7 +1169,12 @@ export class EmailService {
         }
         if (organizationBlocked) {
             this.skipSandboxSend(result, isTest, SANDBOX_ORGANIZATION_BLOCKED_MESSAGE, 'sending_suspended')
-            await this.sandboxSender?.captureBlocked(teamId, isTest, 'organization_blocked', 0)
+            await this.sandboxSender?.captureBlocked(
+                teamId,
+                sandboxSendSource(result.invocation, isTest),
+                'organization_blocked',
+                0
+            )
             return false
         }
         if (!blockedRecipients.length) {
@@ -1225,7 +1193,12 @@ export class EmailService {
             return true
         }
         this.skipSandboxSend(result, isTest, SANDBOX_PAUSE_SKIP_MESSAGES[gate], 'sending_suspended')
-        await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, gate, 0)
+        await this.sandboxSender!.captureBlocked(
+            result.invocation.teamId,
+            sandboxSendSource(result.invocation, isTest),
+            gate,
+            0
+        )
         return false
     }
 
@@ -1242,7 +1215,12 @@ export class EmailService {
             `${explanation}${label}${blockedRecipients.join(', ')}. ${guidance}`,
             reason === 'cap_reached' ? 'frequency_capped' : 'recipient_unreachable'
         )
-        await this.sandboxSender?.captureBlocked(result.invocation.teamId, isTest, reason, blockedRecipients.length)
+        await this.sandboxSender?.captureBlocked(
+            result.invocation.teamId,
+            sandboxSendSource(result.invocation, isTest),
+            reason,
+            blockedRecipients.length
+        )
     }
 
     private async claimSandboxDailyCaps(
@@ -1265,7 +1243,12 @@ export class EmailService {
             claim.type === 'check_failed' ? 'sending_suspended' : 'frequency_capped'
         )
         const reason = claim.type === 'check_failed' ? 'check_failed' : 'cap_reached'
-        await this.sandboxSender!.captureBlocked(result.invocation.teamId, isTest, reason, recipients.length)
+        await this.sandboxSender!.captureBlocked(
+            result.invocation.teamId,
+            sandboxSendSource(result.invocation, isTest),
+            reason,
+            recipients.length
+        )
         return null
     }
 
@@ -1366,12 +1349,13 @@ export class EmailService {
     // set, and no recipient gets an unsubscribe link: only members of the sending organization
     // receive it. CC and BCC are parsed once so the final member check covers exactly this list.
     private sandboxSesRoute(params: CyclotronInvocationQueueParametersEmailType): SesRoute {
+        const envelope = sandboxEnvelope(params.to.email, params.cc, params.bcc)
         return {
             tenantName: this.sandboxSender!.config.tenantName,
             configurationSetName: this.sandboxSender!.config.configurationSetName,
             unsubscribeHeaders: false,
-            ccAddresses: sandboxAddressList(params.cc),
-            bccAddresses: sandboxAddressList(params.bcc),
+            ccAddresses: envelope.cc,
+            bccAddresses: envelope.bcc,
         }
     }
 

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

@@ -2,13 +2,15 @@ import { createHash } from 'crypto'
 import { type DefaultTreeAdapterMap, defaultTreeAdapter, parse, parseFragment, serialize } from 'parse5'
 
 import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron'
+import { CyclotronJobInvocationHogFunction } from '~/cdp/types'
 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'
 
 import { SandboxSenderStateService } from '../managers/sandbox-sender-state.service'
 import { ClaimRequest, IssuedClaim, RateLimiterService } from '../rate-limiter/rate-limiter.service'
+import { isTransactionalEmailSend } from './email-status.service'
 import { maybeAddPreheaderToEmail } from './helpers/preheader'
 
 const SECONDS_PER_DAY = 86400
@@ -83,6 +85,17 @@ function distinctAddresses(recipients: string[]): string[] {
 
 const BLOCKED_EVENT_WINDOW_SECONDS = 3600
 
+export type SandboxSendSource = 'test' | 'workflow' | 'api'
+
+const BLOCKED_EVENT_WINDOW_NAMES: Record<SandboxSendSource, string> = { test: 'test', workflow: 'live', api: 'api' }
+
+export function sandboxSendSource(invocation: CyclotronJobInvocationHogFunction, isTest: boolean): SandboxSendSource {
+    if (isTest) {
+        return 'test'
+    }
+    return isTransactionalEmailSend(invocation) ? 'api' : 'workflow'
+}
+
 function htmlBody(document: DefaultTreeAdapterMap['document']): DefaultTreeAdapterMap['element'] | undefined {
     const root = document.childNodes.find(defaultTreeAdapter.isElementNode)
     return root?.childNodes.find(
@@ -259,11 +272,11 @@ export class SandboxEmailSender {
         }
     }
 
-    public async captureSent(teamId: number, isTest: boolean, recipientCount: number): Promise<void> {
+    public async captureSent(teamId: number, source: SandboxSendSource, recipientCount: number): Promise<void> {
         await this.captureEvent(teamId, 'workflows sandbox email sent', {
             recipient_count: recipientCount,
-            source: isTest ? 'test' : 'workflow',
-            is_test: isTest,
+            source,
+            is_test: source === 'test',
         })
     }
 
@@ -276,15 +289,15 @@ export class SandboxEmailSender {
 
     public async captureBlocked(
         teamId: number,
-        isTest: boolean,
+        source: SandboxSendSource,
         reason: SandboxEmailBlockReason,
         blockedRecipientCount: number
     ): Promise<void> {
         await this.captureEvent(
             teamId,
             'workflows sandbox email blocked',
-            { reason, blocked_recipient_count: blockedRecipientCount, is_test: isTest },
-            () => this.isFirstBlockInWindow(teamId, isTest, reason)
+            { reason, blocked_recipient_count: blockedRecipientCount, is_test: source === 'test', source },
+            () => this.isFirstBlockInWindow(teamId, source, reason)
         )
     }
 
@@ -309,14 +322,14 @@ export class SandboxEmailSender {
 
     private async isFirstBlockInWindow(
         teamId: number,
-        isTest: boolean,
+        source: SandboxSendSource,
         reason: SandboxEmailBlockReason
     ): Promise<boolean> {
         const firstBlock = await this.blockedEventRedis.useClient(
             { name: 'sandbox-blocked-event', timeout: 1000 },
             (client) =>
                 client.set(
-                    `@posthog/workflows-sandbox-blocked/${teamId}/${reason}/${isTest ? 'test' : 'live'}`,
+                    `@posthog/workflows-sandbox-blocked/${teamId}/${reason}/${BLOCKED_EVENT_WINDOW_NAMES[source]}`,
                     '1',
                     'EX',
                     BLOCKED_EVENT_WINDOW_SECONDS,