feat(messaging): send cc, bcc, headers and tags with a code send

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

@@ -2253,6 +2253,26 @@ describe('CDP API', () => {
                     { send: codeSend },
                     { send: { ...codeSend, from: { ...codeSend.from, integrationId: 13 } } },
                 ],
+                [
+                    'another cc',
+                    { send: { ...codeSend, cc: ['first@example.com'] } },
+                    { send: { ...codeSend, cc: ['second@example.com'] } },
+                ],
+                [
+                    'another bcc',
+                    { send: { ...codeSend, bcc: ['first@example.com'] } },
+                    { send: { ...codeSend, bcc: ['second@example.com'] } },
+                ],
+                [
+                    'other headers',
+                    { send: { ...codeSend, headers: [{ name: 'X-Entity-Ref-ID', value: 'order-1' }] } },
+                    { send: { ...codeSend, headers: [{ name: 'X-Entity-Ref-ID', value: 'order-2' }] } },
+                ],
+                [
+                    'other tags',
+                    { send: { ...codeSend, tags: [{ name: 'plan', value: 'free' }] } },
+                    { send: { ...codeSend, tags: [{ name: 'plan', value: 'pro' }] } },
+                ],
             ])('rejects the key for %s', async (_desc, first, retry) => {
                 await sendKeyed({ ...keyed, ...first })
 
@@ -2346,6 +2366,10 @@ describe('CDP API', () => {
                 { ...codeSend, content: { subject: 'Hi', html: '<img src="https://example.com/a.png" alt="Hi">' } },
             ],
             ['a code send for a local template', { tracking: { captureEngagement: false } }],
+            [
+                'a code send with copies to addresses with punctuation',
+                { ...codeSend, cc: ['archive=orders@example.com'], bcc: ["o'brien+audit@example.com"] },
+            ],
         ])('stamps %s on the queued run', async (_desc, acceptedSend) => {
             const res = await send(
                 hogFlow.id,
@@ -2363,6 +2387,7 @@ describe('CDP API', () => {
             ['content with an empty subject', { ...codeSend, content: { subject: '', text: 'Hi' } }],
             ['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'] }],
         ])('refuses a code send with %s', async (_desc, acceptedSend) => {
             const invalidBefore = await outcomeCount('invalid_request')
 
@@ -2394,7 +2419,7 @@ describe('CDP API', () => {
         })
 
         describe('send checks', () => {
-            type Target = { teamId: number; flowId: string; to: string }
+            type Target = { teamId: number; flowId: string; to: string; send?: Record<string, unknown> }
 
             const suppress = async (address: string): Promise<void> => {
                 await hub.postgres.query(
@@ -2463,23 +2488,35 @@ describe('CDP API', () => {
                 return { teamId: team.id, flowId: flow.id, to: body.to }
             }
 
+            const suppressedCopy = async (): Promise<Target> => {
+                const { teamId, flowId } = await suppressedRecipient()
+                return { teamId, flowId, to: body.to, send: { ...codeSend, cc: ['bounced@example.com'] } }
+            }
+
+            const suppressedRecipientAndCopy = async (): Promise<Target> => {
+                const target = await suppressedRecipient()
+                return { ...target, send: { ...codeSend, cc: ['bounced@example.com'] } }
+            }
+
             const domainWithoutMailServers = (): Promise<Target> =>
                 Promise.resolve({ teamId: team.id, flowId: hogFlow.id, to: 'someone@no-mail.invalid' })
 
             it.each([
-                ['invalid_recipient', domainWithoutMailServers, 'has no mail servers'],
-                ['recipient_suppressed', suppressedRecipient, 'bounced or was marked as spam'],
-                ['template_paused', pausedWorkflow, 'Email sending is paused for this email'],
-                ['sending_suspended', suspendedProject, 'Email sending is suspended for this project'],
-                ['sender_not_verified', unverifiedSender, "The sender's domain is not verified"],
-            ])('refuses a send that fails %s before it queues', async (code, setUp, why) => {
-                const { teamId, flowId, to } = await setUp()
+                ['invalid_recipient', domainWithoutMailServers, 'has no mail servers', 'to'],
+                ['recipient_suppressed', suppressedRecipient, 'bounced or was marked as spam', 'to'],
+                ['recipient_suppressed', suppressedCopy, 'bounced or was marked as spam', 'cc'],
+                ['recipient_suppressed', suppressedRecipientAndCopy, 'bounced or was marked as spam', null],
+                ['template_paused', pausedWorkflow, 'Email sending is paused for this email', undefined],
+                ['sending_suspended', suspendedProject, 'Email sending is suspended for this project', undefined],
+                ['sender_not_verified', unverifiedSender, "The sender's domain is not verified", undefined],
+            ])('refuses a send that fails %s before it queues, set up by %p', async (code, setUp, why, attr) => {
+                const { teamId, flowId, to, send } = await setUp()
                 const rejectedBefore = await outcomeCount(code)
 
                 const res = await supertest(app)
                     .post(`/api/projects/${teamId}/hog_flows/${flowId}/transactional_email`)
                     .set({ Authorization: `Bearer ${mintToken(teamId, flowId)}` })
-                    .send({ ...body, to })
+                    .send({ ...body, to, ...(send && { send }) })
 
                 expect(res.status).toEqual(400)
                 expect(res.body).toEqual({
@@ -2488,6 +2525,7 @@ describe('CDP API', () => {
                     detail: expect.not.stringMatching(/^Skipping send/),
                     why: expect.stringContaining(why),
                     fix: expect.any(String),
+                    attr,
                 })
                 expect(await outcomeCount(code)).toEqual(rejectedBefore + 1)
                 expect(mockQueueInvocations).not.toHaveBeenCalled()

nodejs/src/cdp/cdp-api.ts

@@ -57,6 +57,7 @@ import {
     NOT_SENDABLE,
     TransactionalEmailOutcome,
     TransactionalEmailRequestSchema,
+    refusedRecipientField,
     transactionalEmailRequestsTotal,
     transactionalSendCandidate,
     transactionalSendFingerprint,
@@ -1251,7 +1252,11 @@ export class CdpApi {
                         await idempotencyKeys.release(team.id, idempotencyKey, invocation.id)
                     }
                     const { code, detail, why, fix } = decision
-                    return reply(code, 400, { error: code, code, detail, why, fix })
+                    const attr =
+                        'refusedAddresses' in decision
+                            ? { attr: refusedRecipientField(decision.refusedAddresses, to, send) }
+                            : {}
+                    return reply(code, 400, { error: code, code, detail, why, fix, ...attr })
                 }
                 const { isLimited } = await checkHogFlowQuotaLimits(hogFlow, team.id, this.deps.quotaLimiting)
                 if (isLimited) {

nodejs/src/cdp/schema/cyclotron.test.ts

@@ -50,5 +50,23 @@ describe('cyclotron schema', () => {
 
             expect(CyclotronInvocationQueueParametersEmailSchema.shape.from.safeParse(from).success).toBe(false)
         })
+
+        it.each([
+            {
+                name: 'a header the caller may set',
+                headers: [{ name: 'X-Entity-Ref-ID', value: 'order-123' }],
+                ok: true,
+            },
+            { name: 'a header PostHog sets', headers: [{ name: 'bcc', value: 'a@example.com' }], ok: false },
+            { name: 'a header in a family PostHog sets', headers: [{ name: 'Content-Type', value: 'x' }], ok: false },
+            {
+                name: 'a line break in a value',
+                headers: [{ name: 'X-Ref', value: '1\r\nBcc: a@example.com' }],
+                ok: false,
+            },
+            { name: 'a colon in a name', headers: [{ name: 'X-Ref: 1\r\nBcc', value: 'a' }], ok: false },
+        ])('sends $name only when it is safe: $ok', ({ headers, ok }) => {
+            expect(CyclotronInvocationQueueParametersEmailSchema.shape.headers.safeParse(headers).success).toBe(ok)
+        })
     })
 })

nodejs/src/cdp/schema/cyclotron.ts

@@ -1,5 +1,6 @@
 import { z } from 'zod'
 
+import { EmailHeaderSchema, EmailTagSchema, MAX_EMAIL_HEADERS, MAX_EMAIL_TAGS } from './emailFields'
 import { PushNotificationPayloadSchema } from './pushNotification'
 
 export const CyclotronInputSchema = z.object({
@@ -165,6 +166,9 @@ export const CyclotronInvocationQueueParametersEmailSchema = z.object({
     }),
     cc: z.string().optional(),
     bcc: z.string().optional(),
+    // Set only by a code send of the transactional email API.
+    headers: z.array(EmailHeaderSchema).max(MAX_EMAIL_HEADERS).optional(),
+    tags: z.array(EmailTagSchema).max(MAX_EMAIL_TAGS).optional(),
     subject: z.string(),
     preheader: z.string().optional(),
     // Emails authored through the API or MCP often carry html only.

nodejs/src/cdp/schema/emailFields.ts

@@ -1,7 +1,60 @@
 import { z } from 'zod'
 
+// Django refuses these with a reason before the send is queued. The worker checks again, because a
+// header PostHog sets itself, or a line break in a value, must never reach a provider from any path.
+const HEADERS_POSTHOG_SETS = new Set([
+    'auto-submitted',
+    'bcc',
+    'cc',
+    'date',
+    'dkim-signature',
+    'from',
+    'message-id',
+    'mime-version',
+    'received',
+    'reply-to',
+    'return-path',
+    'sender',
+    'subject',
+    'to',
+])
+const HEADER_PREFIXES_POSTHOG_SETS = ['content-', 'x-posthog-', 'x-ses-']
+
+export const MAX_EMAIL_COPIES = 10
+export const MAX_EMAIL_HEADERS = 10
+export const MAX_EMAIL_TAGS = 10
+export const RESERVED_EMAIL_TAG_PREFIX = 'ph_'
+
+const isHeaderPostHogSets = (name: string): boolean => {
+    const lowered = name.toLowerCase()
+    return (
+        HEADERS_POSTHOG_SETS.has(lowered) || HEADER_PREFIXES_POSTHOG_SETS.some((prefix) => lowered.startsWith(prefix))
+    )
+}
+
+export const EmailHeaderSchema = z.object({
+    name: z
+        .string()
+        .regex(/^[!-9;-~]{1,126}$/)
+        .refine((name) => !isHeaderPostHogSets(name), 'PostHog sets this header itself'),
+    value: z.string().regex(/^[ -~]{0,870}$/),
+})
+
 // One bare address with a dotted domain. Spaces, quotes, brackets and list separators are refused, so a copy never
 // names a second recipient.
 export const BareEmailAddressSchema = z
     .string()
     .regex(/^[^\s@"<>,;:()[\]\\]+@[^\s@"<>,;:()[\]\\]+\.[^\s@"<>,;:()[\]\\]+$/)
+
+const TagPart = z.string().regex(/^[A-Za-z0-9_-]{1,256}$/)
+
+export const EmailTagSchema = z.object({
+    name: TagPart.refine(
+        (name) => !name.toLowerCase().startsWith(RESERVED_EMAIL_TAG_PREFIX),
+        'PostHog reserves tag names that start with ph_'
+    ),
+    value: TagPart,
+})
+
+export type EmailHeader = z.infer<typeof EmailHeaderSchema>
+export type EmailTag = z.infer<typeof EmailTagSchema>

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

@@ -3311,6 +3311,31 @@ describe('Hogflow Executor', () => {
                     html: '<p>Hi {{ variables.name }}</p>',
                 }),
             },
+            {
+                name: "a code send with the caller's copies, headers and tags over the step",
+                step: codeKeyStep,
+                send: (): AcceptedSend => ({
+                    ...codeSend(),
+                    cc: ['jane@example.com', 'joe@example.com'],
+                    bcc: ['audit@example.com'],
+                    headers: [{ name: 'X-Entity-Ref-ID', value: 'order-123' }],
+                    tags: [{ name: 'category', value: 'password_reset' }],
+                }),
+                expected: () => ({
+                    type: 'email',
+                    to: { email: 'recipient@example.com', name: 'Recipient' },
+                    from: { integrationId, email: 'security@posthog.com', name: 'Security' },
+                    cc: 'jane@example.com, joe@example.com',
+                    bcc: 'audit@example.com',
+                    headers: [{ name: 'X-Entity-Ref-ID', value: 'order-123' }],
+                    tags: [{ name: 'category', value: 'password_reset' }],
+                    replyTo: 'help@example.com, support@example.com',
+                    subject: 'Reset your password',
+                    preheader: '',
+                    text: 'Hi',
+                    html: '<p>Hi {{ variables.name }}</p>',
+                }),
+            },
         ])('hands the email queue $name', async ({ step, send, expected }) => {
             const invocation = flowInvocation(emailFlow(step()))
             invocation.state.variables = { name: 'Ada' }

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

@@ -32,6 +32,10 @@ const layAcceptedSendOver = (
             }),
             ...(send.from && { from: { ...send.from } }),
             ...(send.replyTo && { replyTo: send.replyTo.join(', ') }),
+            ...(send.cc && { cc: send.cc.join(', ') }),
+            ...(send.bcc && { bcc: send.bcc.join(', ') }),
+            ...(send.headers && { headers: send.headers.map((header) => ({ ...header })) }),
+            ...(send.tags && { tags: send.tags.map((tag) => ({ ...tag })) }),
         },
     },
 })

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

@@ -1,7 +1,14 @@
 import { Counter } from 'prom-client'
 import { z } from 'zod'
 
-import { BareEmailAddressSchema } from '~/cdp/schema/emailFields'
+import {
+    BareEmailAddressSchema,
+    EmailHeaderSchema,
+    EmailTagSchema,
+    MAX_EMAIL_COPIES,
+    MAX_EMAIL_HEADERS,
+    MAX_EMAIL_TAGS,
+} from '~/cdp/schema/emailFields'
 import { HogFlow } from '~/cdp/schema/hogflow'
 
 import { IntegrationManagerService } from '../managers/integration-manager.service'
@@ -34,6 +41,10 @@ export const AcceptedSendSchema = z
             .readonly()
             .optional(),
         replyTo: z.array(z.string()).readonly().optional(),
+        cc: z.array(BareEmailAddressSchema).max(MAX_EMAIL_COPIES).readonly().optional(),
+        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(),
         tracking: z.object({ captureEngagement: z.boolean() }).readonly(),
     })
     .readonly()
@@ -102,7 +113,7 @@ export const transactionalEmailRequestsTotal = new Counter({
 
 /**
  * What the delivery's send checks will see, known when the send is accepted: the sender the email step
- * resolves to for this run, and the recipient. The caller's `send.from` replaces the step's sender, as it
+ * resolves to for this run, and the recipients. The caller's `send.from` replaces the step's sender, as it
  * does at delivery.
  */
 export async function transactionalSendCandidate(
@@ -117,10 +128,26 @@ export async function transactionalSendCandidate(
         teamId: hogFlow.team_id,
         workflowMetadata: emailPauseMetadata(hogFlow),
         sender: from ? await integrationManager.get(selectEmailSenderIntegrationId(invocationId, from)) : null,
-        recipients: [to],
+        // Every copy is a recipient, so suppression covers the whole send.
+        recipients: [to, ...(send?.cc ?? []), ...(send?.bcc ?? [])],
     }
 }
 
+type RecipientField = 'to' | 'cc' | 'bcc'
+
+/** The request field that holds every refused address, or null when they sit in more than one field. */
+export function refusedRecipientField(
+    refusedAddresses: string[],
+    to: string,
+    send: AcceptedSend | undefined
+): RecipientField | null {
+    const held: Record<RecipientField, readonly string[]> = { to: [to], cc: send?.cc ?? [], bcc: send?.bcc ?? [] }
+    const holding = (Object.keys(held) as RecipientField[]).filter((field) =>
+        refusedAddresses.some((address) => held[field].includes(address))
+    )
+    return holding.length === 1 ? holding[0] : null
+}
+
 function emailStepSender(hogFlow: HogFlow): EmailSenderConfiguration | null {
     const step = hogFlow.actions.find((action) => action.type === 'function_email')
     const from = step?.type === 'function_email' ? step.config.inputs?.email?.value?.from : undefined

nodejs/src/cdp/services/messaging/email-send-policy.ts

@@ -20,7 +20,8 @@ type Rejection = {
 }
 
 export type PolicyRejection =
-    | (Rejection & { code: Exclude<PolicyCode, 'sending_suspended'> })
+    | (Rejection & { code: 'template_paused' | 'sender_not_verified' })
+    | (Rejection & { code: 'invalid_recipient' | 'recipient_suppressed'; refusedAddresses: string[] })
     | (Rejection & { code: 'sending_suspended'; cause: EmailSendingSuspensionCause })
 
 export type PolicyDecision = { ok: true; sender: IntegrationType } | PolicyRejection
@@ -93,6 +94,7 @@ export class EmailSendPolicy {
         return {
             ok: false,
             code: 'recipient_suppressed',
+            refusedAddresses: suppressed,
             detail: `Recipient(s) on the suppression list: ${suppressed.join(', ')}`,
             why: `An earlier email to ${suppressed.join(', ')} bounced or was marked as spam, so PostHog no longer sends to it.`,
             fix: 'Send to another address, or remove the address from the suppression list once it can receive email again.',
@@ -207,6 +209,7 @@ function invalidRecipient(undeliverable: Undeliverable): PolicyRejection {
     return {
         ok: false,
         code: 'invalid_recipient',
+        refusedAddresses: [undeliverable.address],
         detail: undeliverable.message,
         why,
         fix: 'Check the address for a typo. If it is right, the recipient cannot receive email, so skip them.',