feat(cdp): deliver transactional email attachments from a side table

nodejs/src/cdp/async-functions/send-email.ts

@@ -7,8 +7,9 @@ import { registerAsyncFunction } from '../async-function-registry'
 
 registerAsyncFunction('sendEmail', {
     execute: (args, context, result) => {
+        const { attachments: _onlyAnAcceptedSendHasFiles, ...stepEmail } = args[0] ?? {}
         result.invocation.queueParameters = CyclotronInvocationQueueParametersEmailSchema.parse({
-            ...args[0],
+            ...stepEmail,
             ...acceptedSendEmailFields(context.invocation.state.send),
             type: 'email',
         })

nodejs/src/cdp/cdp-api.serial.test.ts

@@ -2,8 +2,10 @@
 import { createMockJobQueue } from '../../tests/helpers/mocks/job-queue.mock'
 import { mockFetch } from '../../tests/helpers/mocks/request.mock'
 
+import { createHash } from 'crypto'
 import { Server } from 'http'
 import jwt from 'jsonwebtoken'
+import { Pool } from 'pg'
 import supertest from 'supertest'
 import express from 'ultimate-express'
 
@@ -41,6 +43,7 @@ import {
     transactionalEmailRequestsTotal,
     transactionalEmailStageDurationMs,
 } from './services/hogflows/transactional-email'
+import { decodeInvocationGlobals } from './services/monitoring/hog-invocation-results.service'
 import { BASE_REDIS_KEY, HogWatcherState } from './services/monitoring/hog-watcher.service'
 import { compileHog } from './templates/compiler'
 import { HogFunctionInvocationGlobals, HogFunctionType } from './types'
@@ -2220,6 +2223,26 @@ describe('CDP API', () => {
             tracking: { captureEngagement: true },
         }
 
+        const FILE_CONTENT = 'c2VjcmV0LWZpbGU='
+        const attachment = (overrides: Record<string, unknown> = {}) => ({
+            filename: 'report.pdf',
+            content: FILE_CONTENT,
+            contentType: 'application/pdf',
+            ...overrides,
+        })
+        const storedAttachments = async (filename: string): Promise<{ job_id: string; content: Buffer }[]> => {
+            const pool = new Pool({ connectionString: hub.CYCLOTRON_NODE_DATABASE_URL })
+            try {
+                return (
+                    await pool.query('SELECT job_id, content FROM cyclotron_email_attachments WHERE filename = $1', [
+                        filename,
+                    ])
+                ).rows
+            } finally {
+                await pool.end()
+            }
+        }
+
         beforeEach(async () => {
             produceOutput = jest.spyOn(api['emailStatusService']['outputs'], 'produce')
             mockQueueInvocations = jest.fn().mockResolvedValue(undefined)
@@ -2393,6 +2416,14 @@ describe('CDP API', () => {
                     ['transactional_email/send', { ...body, send: { ...codeSend, cc: ['manager@example.com'] } }],
                     ['transactional_email/send', { ...body, send: { ...codeSend, cc: ['someone-else@example.com'] } }],
                 ],
+                [
+                    'a changed file under the same name',
+                    ['transactional_email/send', { ...body, send: { ...codeSend, attachments: [attachment()] } }],
+                    [
+                        'transactional_email/send',
+                        { ...body, send: { ...codeSend, attachments: [attachment({ content: 'b3RoZXI=' })] } },
+                    ],
+                ],
             ] as const)(
                 'rejects the key for a request with %s',
                 async (_desc, [firstRoute, first], [route, payload]) => {
@@ -2512,12 +2543,58 @@ describe('CDP API', () => {
                 'more than 10 cc addresses',
                 { ...codeSend, cc: Array.from({ length: 11 }, (_, i) => `copy-${i}@example.com`) },
             ],
+            ['a file that is not base64', { ...codeSend, attachments: [attachment({ content: `${FILE_CONTENT}!` })] }],
+            ['an empty file', { ...codeSend, attachments: [attachment({ content: '' })] }],
+            [
+                'a file too long to check as base64',
+                { ...codeSend, attachments: [attachment({ content: 'A'.repeat(4_500_000) })] },
+            ],
+            [
+                'a file type mailbox providers block',
+                { ...codeSend, attachments: [attachment({ filename: 'setup.EXE' })] },
+            ],
+            ['a file type only SES refuses', { ...codeSend, attachments: [attachment({ filename: 'settings.reg' })] }],
+            ['a file type only Gmail blocks', { ...codeSend, attachments: [attachment({ filename: 'disk.iso' })] }],
+            [
+                'a filename with a path separator',
+                { ...codeSend, attachments: [attachment({ filename: '../report.pdf' })] },
+            ],
+            [
+                'a filename with a line break',
+                { ...codeSend, attachments: [attachment({ filename: 'report\r\n.pdf' })] },
+            ],
+            [
+                'a content type that is not a MIME type',
+                { ...codeSend, attachments: [attachment({ contentType: 'pdf' })] },
+            ],
+            ['a content id in angle brackets', { ...codeSend, attachments: [attachment({ contentId: '<logo>' })] }],
+            [
+                'a content id longer than SES accepts',
+                { ...codeSend, attachments: [attachment({ contentId: 'x'.repeat(79) })] },
+            ],
+            [
+                'a content type longer than SES accepts',
+                { ...codeSend, attachments: [attachment({ contentType: `application/${'x'.repeat(67)}` })] },
+            ],
+            ['a file field this version does not know', { ...codeSend, attachments: [attachment({ size: 11 })] }],
+            ['more than 10 files', { ...codeSend, attachments: Array.from({ length: 11 }, () => attachment()) }],
+            [
+                'files over 512 KB in total',
+                {
+                    ...codeSend,
+                    attachments: [
+                        attachment({ content: Buffer.alloc(300 * 1024).toString('base64') }),
+                        attachment({ content: `${Buffer.alloc(213 * 1024).toString('base64')}${FILE_CONTENT}` }),
+                    ],
+                },
+            ],
         ])('refuses a code send with %s', async (_desc, acceptedSend) => {
             const invalidBefore = await outcomeCount('invalid_request')
 
             const res = await sendAccepted(hogFlow.id, { ...body, send: acceptedSend })
 
             expect(res.status).toEqual(400)
+            expect(JSON.stringify(res.body)).not.toContain(FILE_CONTENT.slice(0, 12))
             expect(await outcomeCount('invalid_request')).toEqual(invalidBefore + 1)
             expect(mockQueueInvocations).not.toHaveBeenCalled()
         })
@@ -2554,6 +2631,65 @@ describe('CDP API', () => {
             expect(mockQueueInvocations).not.toHaveBeenCalled()
         })
 
+        it('stores the files of a code send outside the queued run and its snapshot, and a replay stores none', async () => {
+            const resultsOutputs = api['invocationResultsService'].invocationResultsRowsService['outputs']
+            const produceResultRow = jest.spyOn(resultsOutputs, 'produce')
+            const resultsEnabled = hub.HOG_INVOCATION_RESULTS_ENABLED
+            hub.HOG_INVOCATION_RESULTS_ENABLED = true
+            const filename = `report-${new UUIDT().toString()}.pdf`
+            const keyed = {
+                ...body,
+                idempotency_key: `key-${new UUIDT().toString()}`,
+                send: { ...codeSend, attachments: [attachment({ filename })] },
+            }
+
+            try {
+                const first = await sendAccepted(hogFlow.id, keyed)
+                const replay = await sendAccepted(hogFlow.id, keyed)
+
+                expect(first.status).toEqual(202)
+                expect(replay.body).toMatchObject({ invocation_id: first.body.invocation_id, replayed: true })
+                expect(await storedAttachments(filename)).toEqual([
+                    { job_id: first.body.invocation_id, content: Buffer.from('secret-file') },
+                ])
+                const [invocation] = mockQueueInvocations.mock.calls[0][0]
+                expect(invocation.state.send.attachments).toEqual([
+                    {
+                        filename,
+                        contentType: 'application/pdf',
+                        sizeBytes: 11,
+                        sha256: createHash('sha256').update('secret-file').digest('hex'),
+                    },
+                ])
+                expect(JSON.stringify(invocation)).not.toContain(FILE_CONTENT)
+                const snapshots = produceResultRow.mock.calls.filter(([output]) => output === 'hog_invocation_results')
+                expect(snapshots).not.toHaveLength(0)
+                for (const [, message] of snapshots) {
+                    const row = parseJSON(String(message.value))
+                    const globals = JSON.stringify(await decodeInvocationGlobals(row.invocation_globals))
+                    expect(globals).toContain(filename)
+                    expect(globals).not.toContain(FILE_CONTENT)
+                }
+            } finally {
+                hub.HOG_INVOCATION_RESULTS_ENABLED = resultsEnabled
+            }
+        })
+
+        it('keeps the stored files when the queue insert fails, as the job may be queued already', async () => {
+            mockQueueInvocations.mockRejectedValueOnce(new Error('queue down'))
+            const filename = `report-${new UUIDT().toString()}.pdf`
+
+            const res = await sendAccepted(hogFlow.id, {
+                ...body,
+                send: { ...codeSend, attachments: [attachment({ filename })] },
+            })
+
+            expect(res.status).toEqual(500)
+            expect(await storedAttachments(filename)).toEqual([
+                { job_id: expect.any(String), content: Buffer.from('secret-file') },
+            ])
+        })
+
         it.each<[string, () => { payload: Record<string, unknown>; verify?: () => void }]>([
             [
                 'the team is over its email quota',

nodejs/src/cdp/cdp-api.ts

@@ -57,12 +57,15 @@ import {
     AcceptedSend,
     AcceptedSendRequestSchema,
     CODE_SEND_TOKEN_CLAIMS,
+    QueuedSend,
     TEMPLATE_SEND_TOKEN_CLAIMS,
     TransactionalEmailOutcome,
     TransactionalEmailRequest,
     TransactionalEmailRequestSchema,
     TransactionalEmailStageTimer,
+    decodeAttachments,
     isRecipientFormatIssue,
+    queuedSend,
     transactionalEmailRecipients,
     transactionalEmailRequestsTotal,
 } from './services/hogflows/transactional-email'
@@ -71,6 +74,7 @@ import { JobQueue } from './services/job-queue/job-queue.interface'
 import { GroupsManagerService } from './services/managers/groups-manager.service'
 import { HogFunctionManagerService } from './services/managers/hog-function-manager.service'
 import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service'
+import { EmailAttachmentsService } from './services/messaging/email-attachments.service'
 import { EMAIL_QUEUE_PRIORITY, EmailQueuePriorityClass, cdpEmailQueuedTotal } from './services/messaging/email-priority'
 import { EmailStatusService } from './services/messaging/email-status.service'
 import { EmailSuppressionService } from './services/messaging/email-suppression.service'
@@ -178,7 +182,8 @@ const ACCEPTED_SEND_FINGERPRINT_MARKER = 'accepted_send'
 
 const transactionalEmailFingerprint = (
     hogFlowId: string,
-    { to, distinct_id, variables, send }: TransactionalEmailRequest
+    { to, distinct_id, variables }: TransactionalEmailRequest,
+    send: QueuedSend | undefined
 ): string => {
     const templateRequest = [hogFlowId, to, distinct_id, variables]
     return IdempotencyKeys.fingerprint(
@@ -191,7 +196,7 @@ type CallerChosenSender = Partial<NonNullable<AcceptedSend['from']>>
 const withSenderTheCallerChose = ({
     from,
     ...send
-}: AcceptedSend): Omit<AcceptedSend, 'from'> & { from?: CallerChosenSender } => ({
+}: QueuedSend): Omit<QueuedSend, 'from'> & { from?: CallerChosenSender } => ({
     ...send,
     from: senderTheCallerChose(from),
 })
@@ -236,6 +241,7 @@ export class CdpApi {
     private transactionalEmailJwt: ScopedServiceJwt
     private transactionalEmailIdempotencyKeys: IdempotencyKeys
     private emailSuppressionService: EmailSuppressionService
+    private emailAttachmentsService: EmailAttachmentsService
     private emailValidationService: EmailValidationService
     private teamWorkflowsConfigService: TeamWorkflowsConfigService
     private emailStatusService: EmailStatusService
@@ -253,6 +259,7 @@ export class CdpApi {
         this.hogFlowManager = services.hogFlowManager
         this.recipientTokensService = services.recipientTokensService
         this.emailSuppressionService = services.emailSuppressionService
+        this.emailAttachmentsService = services.emailAttachmentsService
         this.emailValidationService = services.emailValidationService
         this.teamWorkflowsConfigService = services.teamWorkflowsConfigService
         this.hogExecutorAsync = services.hogExecutorAsync
@@ -366,6 +373,7 @@ export class CdpApi {
             this.batchExportHogFunctionService.stop(),
             this.rerunJobManager?.disconnect() ?? Promise.resolve(),
             this.invocationResultsService.stop(),
+            this.emailAttachmentsService.stop(),
         ])
     }
 
@@ -1288,8 +1296,10 @@ export class CdpApi {
                 })
 
                 const invocation = createHogFlowInvocation(triggerGlobals, hogFlow, filterGlobals)
-                if (send) {
-                    invocation.state.send = send
+                const files = decodeAttachments(send)
+                const sendToQueue = send ? queuedSend(send, files) : undefined
+                if (sendToQueue) {
+                    invocation.state.send = sendToQueue
                 }
                 // The run skips the trigger pipeline that counts runs for event-triggered workflows, so the email
                 // worker counts it instead.
@@ -1301,7 +1311,7 @@ export class CdpApi {
                 invocation.queuePriority = EMAIL_QUEUE_PRIORITY[priorityClass]
 
                 const idempotencyKeys = this.transactionalEmailIdempotencyKeys
-                const fingerprint = transactionalEmailFingerprint(hogFlow.id, parsed.data)
+                const fingerprint = transactionalEmailFingerprint(hogFlow.id, parsed.data, sendToQueue)
                 if (idempotencyKey) {
                     // Checked before the quota, so a retry of a send that went out gets its answer even
                     // when the team has hit its quota since.
@@ -1341,6 +1351,10 @@ export class CdpApi {
                         }
                         return reply(refusal.outcome, refusal.status, { error: refusal.outcome })
                     }
+                    if (files.length) {
+                        stages.start('files')
+                        await this.emailAttachmentsService.store(invocation.id, team.id, files)
+                    }
                     stages.start('queue_insert')
                     this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running')
                     try {

nodejs/src/cdp/cdp-services.ts

@@ -35,6 +35,10 @@ import { IntegrationManagerService } from './services/managers/integration-manag
 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'
+import {
+    EMAIL_ATTACHMENTS_MAX_CONNECTIONS,
+    EmailAttachmentsService,
+} from './services/messaging/email-attachments.service'
 import { EmailStatusService } from './services/messaging/email-status.service'
 import { EmailSuppressionService } from './services/messaging/email-suppression.service'
 import { EmailValidationService } from './services/messaging/email-validation.service'
@@ -124,6 +128,7 @@ export interface CdpCoreServices {
     messageAssetsService: MessageAssetsService
     /** Buffers transactional email status rows. `invocationResultsService.flush()` produces them. */
     emailStatusService: EmailStatusService
+    emailAttachmentsService: EmailAttachmentsService
 }
 
 export type CdpCoreServicesConfig = Pick<
@@ -471,6 +476,10 @@ export function createCdpCoreServices(
     const teamEmailRateLimiter = deps.emailValidationValkey
         ? new RateLimiterService(deps.emailValidationValkey, { name: 'team-email' })
         : null
+    const emailAttachmentsService = new EmailAttachmentsService(
+        config.CYCLOTRON_NODE_DATABASE_URL,
+        EMAIL_ATTACHMENTS_MAX_CONNECTIONS
+    )
     const emailService = new EmailService(
         {
             sesAccessKeyId: config.SES_ACCESS_KEY_ID,
@@ -495,7 +504,8 @@ export function createCdpCoreServices(
         workflowsActivationReporter,
         messageAssetsService,
         workflowEmailRateLimiter,
-        teamEmailRateLimiter
+        teamEmailRateLimiter,
+        emailAttachmentsService
     )
     const recipientTokensService = new RecipientTokensService(config.ENCRYPTION_SALT_KEYS, config.SITE_URL)
     const hogInputsService = new HogInputsService(deps.integrationManager, recipientTokensService, deps.encryptedFields)
@@ -644,6 +654,7 @@ export function createCdpCoreServices(
         recipientTokensService,
         outputs,
         emailService,
+        emailAttachmentsService,
         messageAssetsService,
         emailStatusService,
     }

nodejs/src/cdp/consumers/cdp-base.consumer.ts

@@ -26,6 +26,7 @@ import { HogFunctionManagerService } from '../services/managers/hog-function-man
 import { HogFunctionTemplateManagerService } from '../services/managers/hog-function-template-manager.service'
 import { PersonsManagerService } from '../services/managers/persons-manager.service'
 import { RecipientsManagerService } from '../services/managers/recipients-manager.service'
+import { EmailAttachmentsService } from '../services/messaging/email-attachments.service'
 import { EmailService } from '../services/messaging/email.service'
 import { RecipientPreferencesService } from '../services/messaging/recipient-preferences.service'
 import { HogFunctionMonitoringService } from '../services/monitoring/hog-function-monitoring.service'
@@ -73,6 +74,7 @@ export abstract class CdpConsumerBase<TConfig extends CdpConsumerBaseConfig = Cd
     recipientsManager: RecipientsManagerService
 
     emailService: EmailService
+    emailAttachmentsService: EmailAttachmentsService
     hogFunctionMonitoringService: HogFunctionMonitoringService
     workflowsActivationReporter: WorkflowsActivationReporter
     cdpUsageReporter: CdpUsageReporterService
@@ -106,6 +108,7 @@ export abstract class CdpConsumerBase<TConfig extends CdpConsumerBaseConfig = Cd
         this.recipientPreferencesService = services.recipientPreferencesService
         this.hogFlowExecutor = services.hogFlowExecutor
         this.emailService = services.emailService
+        this.emailAttachmentsService = services.emailAttachmentsService
         this.hogFunctionMonitoringService = services.hogFunctionMonitoringService
         this.workflowsActivationReporter = services.workflowsActivationReporter
         this.cdpUsageReporter = services.cdpUsageReporter
@@ -155,6 +158,7 @@ export abstract class CdpConsumerBase<TConfig extends CdpConsumerBaseConfig = Cd
         // Release the per-consumer conversion-watcher pool created in createCdpCoreServices, so
         // graceful shutdown frees its cyclotron connections like the other cyclotron pools do.
         await this.invocationResultsService.stop()
+        await this.emailAttachmentsService.stop()
         logger.info('👍', `${this.name} - stopped!`)
     }