feat(messaging): send attachments with a code send

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

@@ -2273,6 +2273,11 @@ describe('CDP API', () => {
                     { send: { ...codeSend, tags: [{ name: 'plan', value: 'free' }] } },
                     { send: { ...codeSend, tags: [{ name: 'plan', value: 'pro' }] } },
                 ],
+                [
+                    'another attachment',
+                    { send: { ...codeSend, attachments: [{ filename: 'invoice.pdf', content: 'Zmlyc3Q=' }] } },
+                    { send: { ...codeSend, attachments: [{ filename: 'invoice.pdf', content: 'c2Vjb25k' }] } },
+                ],
             ])('rejects the key for %s', async (_desc, first, retry) => {
                 await sendKeyed({ ...keyed, ...first })
 
@@ -2388,6 +2393,13 @@ describe('CDP API', () => {
             ['a sender without an integration', { ...codeSend, from: { email: 'security@example.com' } }],
             ['no tracking choice', { content: codeSend.content }],
             ['a copy that names two recipients', { ...codeSend, cc: ['jane@example.com, joe@example.com'] }],
+            [
+                'attachments over the size cap',
+                {
+                    ...codeSend,
+                    attachments: [{ filename: 'big.pdf', content: Buffer.alloc(512 * 1024 + 1).toString('base64') }],
+                },
+            ],
         ])('refuses a code send with %s', async (_desc, acceptedSend) => {
             const invalidBefore = await outcomeCount('invalid_request')
 

nodejs/src/cdp/rerun/rerun-paginator.replay-fidelity.test.ts

@@ -6,7 +6,7 @@ import { CyclotronJobQueuePostgresV2 } from '../services/job-queue/job-queue-pos
 import { JobQueue } from '../services/job-queue/job-queue.interface'
 import { HogFunctionManagerService } from '../services/managers/hog-function-manager.service'
 import { HogFunctionMonitoringService } from '../services/monitoring/hog-function-monitoring.service'
-import { HogInvocationResultsService } from '../services/monitoring/hog-invocation-results.service'
+import { ATTACHMENTS_OMITTED, HogInvocationResultsService } from '../services/monitoring/hog-invocation-results.service'
 import { CyclotronJobInvocationHogFlow } from '../types'
 import { RerunJobState } from './rerun-job.types'
 import { RerunPaginatorService } from './rerun-paginator.service'
@@ -145,6 +145,38 @@ describe('RerunPaginatorService replay fidelity (hog_flow)', () => {
         expect(invocation.state?.rerunAttempts).toBe(1)
     })
 
+    it('skips a code send whose attachments were not stored and reruns the rest of the page', async () => {
+        const row = (invocationId: string, persistedState: Record<string, unknown>): Record<string, unknown> => ({
+            invocation_id: invocationId,
+            parent_run_id: '',
+            attempts: 0,
+            last_scheduled_at: '2026-06-01 09:00:00.000000',
+            first_scheduled_at: '2026-06-01 09:00:00.000000',
+            invocation_globals: JSON.stringify(persistedState),
+        })
+        const event = { uuid: 'evt-3', distinct_id: 'd-3', properties: {}, timestamp: '2026-06-01T09:00:00Z' }
+        const rows = [
+            row('inv-attachments', {
+                event,
+                send: {
+                    tracking: { captureEngagement: true },
+                    attachments: [{ filename: 'invite.ics', contentType: 'text/calendar', size: 32 }],
+                },
+                [ATTACHMENTS_OMITTED]: true,
+            }),
+            row('inv-plain', { event, actionStepCount: 0, variables: {} }),
+        ]
+        const hogFlowQueue = {
+            queueInvocations: jest.fn().mockResolvedValue(undefined),
+        } as unknown as jest.Mocked<CyclotronJobQueuePostgresV2>
+
+        const paginator = buildPaginator(fakeClickhouse(rows), hogFlowQueue)
+        await paginator.processPage(teamId, state, { jobId: 'rerun-3', createdAt: DateTime.now() })
+
+        const enqueued = hogFlowQueue.queueInvocations.mock.calls.flatMap(([invocations]) => invocations)
+        expect(enqueued.map((invocation) => invocation.id)).toEqual(['inv-plain'])
+    })
+
     it('leaves currentAction undefined for a flow that never progressed (fresh start is correct there)', async () => {
         const persistedState = {
             event: { uuid: 'evt-2', distinct_id: 'd-2', properties: {}, timestamp: '2026-06-01T09:00:00Z' },

nodejs/src/cdp/rerun/rerun-paginator.service.ts

@@ -13,6 +13,7 @@ import { JobQueue } from '../services/job-queue/job-queue.interface'
 import { HogFunctionManagerService } from '../services/managers/hog-function-manager.service'
 import { HogFunctionMonitoringService } from '../services/monitoring/hog-function-monitoring.service'
 import {
+    ATTACHMENTS_OMITTED,
     HogInvocationResultsService,
     SENSITIVE_VARIABLES_OMITTED,
     decodeInvocationGlobals,
@@ -776,6 +777,14 @@ export class RerunPaginatorService {
                 })
                 return null
             }
+            if (persistedState[ATTACHMENTS_OMITTED]) {
+                logger.warn('⚠️', 'Skipping rerun of a code send whose attachments were not stored', {
+                    functionId,
+                    teamId,
+                    invocation_id: row.invocation_id,
+                })
+                return null
+            }
             const eventForFilter = persistedState.event ?? {}
             const filterGlobals: HogFunctionFilterGlobals = convertToHogFunctionFilterGlobal({
                 event: eventForFilter,

nodejs/src/cdp/schema/emailFields.ts

@@ -58,3 +58,42 @@ export const EmailTagSchema = z.object({
 
 export type EmailHeader = z.infer<typeof EmailHeaderSchema>
 export type EmailTag = z.infer<typeof EmailTagSchema>
+
+export const MAX_EMAIL_ATTACHMENTS = 10
+export const MAX_EMAIL_ATTACHMENTS_BYTES = 512 * 1024
+
+export const EmailAttachmentSchema = z
+    .object({
+        filename: z
+            .string()
+            .min(1)
+            .max(255)
+            .regex(/^[^\x00-\x1f\x7f/\\]+$/),
+        content: z
+            .string()
+            .min(1)
+            .regex(/^[A-Za-z0-9+/]+={0,2}$/),
+        contentType: z
+            .string()
+            .max(78)
+            .regex(/^[!-~]+\/[!-~]+(;[ -~]*)?$/)
+            .optional(),
+        contentId: z
+            .string()
+            .max(78)
+            .regex(/^[!-;=?-~]+$/)
+            .optional(),
+    })
+    .readonly()
+
+export const EmailAttachmentsSchema = z
+    .array(EmailAttachmentSchema)
+    .max(MAX_EMAIL_ATTACHMENTS)
+    .refine(
+        (attachments) =>
+            attachments.reduce((total, { content }) => total + Buffer.byteLength(content, 'base64'), 0) <=
+            MAX_EMAIL_ATTACHMENTS_BYTES,
+        `Attachments can hold ${MAX_EMAIL_ATTACHMENTS_BYTES / 1024}KB in total`
+    )
+
+export type EmailAttachment = z.infer<typeof EmailAttachmentSchema>

nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts

@@ -16,6 +16,8 @@ import { PostgresUse } from '~/common/utils/db/postgres'
 import { parseJSON } from '~/common/utils/json-parse'
 import { logger } from '~/common/utils/logger'
 import { FetchResponse, fetch, internalFetch } from '~/common/utils/request'
+import { UUIDT } from '~/common/utils/utils'
+import { waitForExpect } from '~/tests/helpers/expectations'
 import { createTestTeamFixture, uniqueTestId } from '~/tests/helpers/sql'
 
 import { Hub, Team } from '../../../types'
@@ -32,6 +34,7 @@ import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.se
 import { EmailSuppressionService, emailSuppressionConfigFromEnv } from '../messaging/email-suppression.service'
 import { EmailValidationService } from '../messaging/email-validation.service'
 import { EmailService } from '../messaging/email.service'
+import { MailDevAPI } from '../messaging/helpers/maildev'
 import { EmailTrackingCodeSigner } from '../messaging/helpers/tracking-code'
 import { RecipientPreferencesService } from '../messaging/recipient-preferences.service'
 import { RecipientTokensService } from '../messaging/recipient-tokens.service'
@@ -3369,6 +3372,38 @@ describe('Hogflow Executor', () => {
             expect(metrics.filter((m) => m.metric_name === 'email_sent')).toHaveLength(sent)
         })
 
+        it("delivers a code send's attachments", async () => {
+            const subject = `Your invite ${new UUIDT().toString()}`
+            const invocation = flowInvocation(emailFlow(codeKeyStep()))
+            invocation.state.send = {
+                ...codeSend(),
+                content: { subject, html: '<p>See you there</p>', text: 'See you there' },
+                attachments: [
+                    {
+                        filename: 'invite.ics',
+                        content: Buffer.from('BEGIN:VCALENDAR\r\nEND:VCALENDAR\r\n').toString('base64'),
+                        contentType: 'text/calendar; method=REQUEST',
+                    },
+                ],
+            }
+
+            let result = await executor.execute(invocation)
+            while (!result.finished) {
+                result = await executor.execute(result.invocation)
+            }
+
+            expect(result.error).toBeUndefined()
+            mockFetch.mockImplementation(jest.requireActual('~/common/utils/request').fetch)
+            const mailDev = new MailDevAPI()
+            await waitForExpect(async () =>
+                expect((await mailDev.getEmails()).filter((email) => email.subject === subject)).toHaveLength(1)
+            )
+            const [delivered] = (await mailDev.getEmails()).filter((email) => email.subject === subject)
+            expect(
+                delivered.attachments.map((a: { fileName: string; contentType: string }) => [a.fileName, a.contentType])
+            ).toEqual([['invite.ics', 'text/calendar']])
+        })
+
         it('should complete the full round-trip: hogflow → email queue → email sent → workflow continues', async () => {
             const invocation = flowInvocation(emailFlow(storedStep()))
 

nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts

@@ -161,7 +161,11 @@ export class HogFlowFunctionsService {
             ...invocation,
             hogFunction: {
                 ...hogFunction,
-                metadata: { ...hogFunction.metadata, capture_engagement: send?.tracking.captureEngagement ?? null },
+                metadata: {
+                    ...hogFunction.metadata,
+                    capture_engagement: send?.tracking.captureEngagement ?? null,
+                    attachments: send?.attachments ?? null,
+                },
             },
             state: invocation.state.currentAction?.hogFunctionState ?? {
                 globals: await this.buildInputsWithGlobals(hogFunction, globalsWithSource, send),

nodejs/src/cdp/services/hogflows/transactional-email.ts

@@ -3,6 +3,7 @@ import { z } from 'zod'
 
 import {
     BareEmailAddressSchema,
+    EmailAttachmentsSchema,
     EmailHeaderSchema,
     EmailTagSchema,
     MAX_EMAIL_COPIES,
@@ -45,6 +46,7 @@ export const AcceptedSendSchema = z
         bcc: z.array(BareEmailAddressSchema).max(MAX_EMAIL_COPIES).readonly().optional(),
         headers: z.array(EmailHeaderSchema.readonly()).max(MAX_EMAIL_HEADERS).readonly().optional(),
         tags: z.array(EmailTagSchema.readonly()).max(MAX_EMAIL_TAGS).readonly().optional(),
+        attachments: EmailAttachmentsSchema.readonly().optional(),
         tracking: z.object({ captureEngagement: z.boolean() }).readonly(),
     })
     .readonly()

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

@@ -1589,6 +1589,69 @@ describe('EmailService', () => {
             )
         })
 
+        it("sends a code send's attachments beside its recipients and PostHog's headers", async () => {
+            sendEmailSpy.mockResolvedValue({ MessageId: 'test-message-id' })
+            const invite = 'BEGIN:VCALENDAR\r\nMETHOD:REQUEST\r\nEND:VCALENDAR\r\n'
+            invocation.hogFunction.metadata = {
+                message_category_type: 'transactional',
+                attachments: [
+                    {
+                        filename: 'invite.ics',
+                        content: Buffer.from(invite).toString('base64'),
+                        contentType: 'text/calendar; method=REQUEST',
+                    },
+                    { filename: 'logo.png', content: Buffer.from('png').toString('base64'), contentId: 'logo' },
+                ],
+            }
+            invocation.queueParameters = createEmailParams({
+                html: '<p>Hi</p><img src="cid:logo">',
+                cc: 'jane@example.com',
+                bcc: 'audit@example.com',
+                replyTo: 'help@example.com',
+                headers: [{ name: 'X-Entity-Ref-ID', value: 'order-123' }],
+                tags: [{ name: 'category', value: 'invite' }],
+            })
+
+            const result = await service.executeSendEmail(invocation)
+
+            expect(result.error).toBeUndefined()
+            const { input } = sendEmailSpy.mock.calls[0][0] as { input: any }
+            expect(input).toMatchObject({
+                FromEmailAddress: '"Test User" <test@posthog-test.com>',
+                Destination: {
+                    ToAddresses: ['"Test User" <test@example.com>'],
+                    CcAddresses: ['jane@example.com'],
+                    BccAddresses: ['audit@example.com'],
+                },
+                ReplyToAddresses: ['help@example.com'],
+                TenantName: `team-${team.id}`,
+            })
+            expect(input.EmailTags.map((t: { Name: string }) => t.Name)).toEqual(['ph_id', 'category'])
+            expect(input.Content.Simple.Headers.map((h: { Name: string }) => h.Name)).toEqual([
+                'X-Entity-Ref-ID',
+                'X-PostHog-Tracking-Code',
+                'Auto-Submitted',
+            ])
+            expect(input.Content.Simple.Attachments).toEqual([
+                {
+                    FileName: 'invite.ics',
+                    RawContent: Buffer.from(invite),
+                    ContentType: 'text/calendar; method=REQUEST',
+                    ContentId: undefined,
+                    ContentDisposition: 'ATTACHMENT',
+                    ContentTransferEncoding: 'BASE64',
+                },
+                {
+                    FileName: 'logo.png',
+                    RawContent: Buffer.from('png'),
+                    ContentType: undefined,
+                    ContentId: 'logo',
+                    ContentDisposition: 'INLINE',
+                    ContentTransferEncoding: 'BASE64',
+                },
+            ])
+        })
+
         it('attaches the X-PostHog-Tracking-Code header carrying the full signed code', async () => {
             // The header is the authoritative tracking-code carrier (the EmailTag is the
             // bounded backwards-compat fallback). It rides on every outbound message,

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

@@ -1,10 +1,11 @@
-import { MessageHeader, SESv2Client, SendEmailCommand, SendEmailCommandInput } from '@aws-sdk/client-sesv2'
+import { Attachment, MessageHeader, SESv2Client, SendEmailCommand, SendEmailCommandInput } from '@aws-sdk/client-sesv2'
 import { DateTime } from 'luxon'
 import { SendMailOptions } from 'nodemailer'
+import Mail from 'nodemailer/lib/mailer'
 import { Counter, Histogram } from 'prom-client'
 
 import { CyclotronInvocationQueueParametersEmailType } from '~/cdp/schema/cyclotron'
-import { EmailTag } from '~/cdp/schema/emailFields'
+import { EmailAttachment, EmailAttachmentsSchema, EmailTag } from '~/cdp/schema/emailFields'
 import { HogFlowEmailSendingRateLimit, HogFlowEmailSendingRateLimitSchema } from '~/cdp/schema/hogflow'
 import {
     CyclotronJobInvocationHogFunction,
@@ -321,6 +322,27 @@ export function emailTagProperties(tags: readonly EmailTag[]): Record<string, st
     return Object.fromEntries(tags.map((tag) => [tag.name, tag.value]))
 }
 
+// The workflow puts a code send's attachments on the email step's hog function. Checked again here so
+// an attachment from any path reaches the provider only in a shape the send route accepts.
+function codeSendAttachments(invocation: CyclotronJobInvocationHogFunction): EmailAttachment[] {
+    return EmailAttachmentsSchema.parse(invocation.hogFunction.metadata?.attachments ?? [])
+}
+
+function toSesAttachment({ filename, content, contentType, contentId }: EmailAttachment): Attachment {
+    return {
+        FileName: filename,
+        RawContent: Buffer.from(content, 'base64'),
+        ContentType: contentType,
+        ContentId: contentId,
+        ContentDisposition: contentId ? 'INLINE' : 'ATTACHMENT',
+        ContentTransferEncoding: 'BASE64',
+    }
+}
+
+function toMailAttachment({ filename, content, contentType, contentId }: EmailAttachment): Mail.Attachment {
+    return { filename, content: Buffer.from(content, 'base64'), contentType, cid: contentId }
+}
+
 export function parseAddressList(value?: string): string[] | undefined {
     if (!value || !value.trim()) {
         return undefined
@@ -994,7 +1016,13 @@ export class EmailService {
         const ccAddresses = parseAddressList(params.cc)
         const bccAddresses = parseAddressList(params.bcc)
         const replyToAddresses = parseAddressList(params.replyTo)
+        const attachments = codeSendAttachments(result.invocation)
 
+        if (attachments.length) {
+            mailOptions.attachments = attachments.map(toMailAttachment)
+            mailOptions.disableFileAccess = true
+            mailOptions.disableUrlAccess = true
+        }
         if (ccAddresses) {
             mailOptions.cc = ccAddresses
         }
@@ -1060,6 +1088,7 @@ export class EmailService {
               }
             : {}
 
+        const attachments = codeSendAttachments(result.invocation)
         const sendEmailParams: SendEmailCommandInput = {
             FromEmailAddress: formatMailbox(from.name, from.email),
             Destination: {
@@ -1075,6 +1104,7 @@ export class EmailService {
                         ...(params.text ? { Text: { Data: params.text, Charset: 'UTF-8' } } : {}),
                         ...htmlBody,
                     },
+                    ...(attachments.length ? { Attachments: attachments.map(toSesAttachment) } : {}),
                 },
             },
             ConfigurationSetName: this.resolveConfigurationSetName(trackingEnabled, result.invocation),