diff --git a/nodejs/src/cdp/_tests/email-delivery-fakes.ts b/nodejs/src/cdp/_tests/email-delivery-fakes.ts index d2b034d875fc..871e88ebe289 100644 --- a/nodejs/src/cdp/_tests/email-delivery-fakes.ts +++ b/nodejs/src/cdp/_tests/email-delivery-fakes.ts @@ -1,12 +1,14 @@ import { SESv2Client } from '@aws-sdk/client-sesv2' import { createHash, randomBytes, randomUUID } from 'node:crypto' import { IncomingMessage, Server, ServerResponse, createServer, request as httpRequest } from 'node:http' -import { AddressInfo } from 'node:net' +import { AddressInfo, Socket, connect, createServer as createTcpServer } from 'node:net' import { FakeSes, FakeSesAttachment } from '~/cdp/_tests/fake-ses' import { getDefaultCdpConfig } from '~/cdp/config' import { EmailAttachmentMetadata } from '~/cdp/schema/cyclotron' +import { CyclotronJobQueuePostgresV2 } from '~/cdp/services/job-queue/job-queue-postgres-v2' import { AttachmentStore, buildAttachmentStore } from '~/cdp/services/messaging/attachment-store' +import { CyclotronJobInvocation, CyclotronJobInvocationResult, CyclotronJobQueueKind } from '~/cdp/types' export const attachmentStoreConfig = { ...getDefaultCdpConfig(), @@ -17,9 +19,9 @@ export const attachmentStore = buildAttachmentStore(attachmentStoreConfig)! export type StoredFile = { bytes: Buffer; metadata: EmailAttachmentMetadata } export const attachmentFile = ( - ref: Omit, 'sha256' | 'sizeBytes'> = {} + ref: Omit, 'sha256' | 'sizeBytes'> = {}, + bytes: Buffer = Buffer.from(`%PDF-1.4 ${randomBytes(16).toString('hex')}`) ): StoredFile => { - const bytes = Buffer.from(`%PDF-1.4 ${randomBytes(16).toString('hex')}`) return { bytes, metadata: { @@ -33,9 +35,10 @@ export const attachmentFile = ( export const uploadAttachment = async ( teamId: number, - ref: Omit, 'sha256' | 'sizeBytes'> + ref: Omit, 'sha256' | 'sizeBytes'>, + bytes?: Buffer ): Promise => { - const file = attachmentFile(ref) + const file = attachmentFile(ref, bytes) const [slot] = await attachmentStore.prepareUploads(teamId, [ { sha256: file.metadata.sha256, size: file.bytes.length }, ]) @@ -65,10 +68,12 @@ export const deliveredTo = (ses: FakeSes, address: string): DeliveredEmail[] => .filter((message) => message.to.some((to) => to.includes(address))) .map(({ to, attachments }) => ({ to, attachments })) -export type BucketAnswer = 'pass' | 'refuse' | 'hold' | 'holdFilesItHas' | 'stallFileBodies' +export type BucketAnswer = 'pass' | 'refuse' | 'hold' | 'holdCopies' | 'holdFilesItHas' | 'stallFileBodies' export class BucketProxy { answer: BucketAnswer = 'pass' + lastModifiedAgeMs = 0 + requests = 0 abandonedRequests = 0 downloads = 0 endpoint = '' @@ -112,6 +117,7 @@ export class BucketProxy { this.abandonedRequests++ } }) + this.requests++ if (request.method === 'GET') { this.downloads++ } @@ -126,7 +132,7 @@ export class BucketProxy { request.socket.destroy() return } - if (this.answer === 'hold') { + if (this.answer === 'hold' || (this.answer === 'holdCopies' && request.headers['x-amz-copy-source'])) { this.held.push(() => this.answerRequest(request, response)) return } @@ -146,6 +152,11 @@ export class BucketProxy { answer.resume() return } + if (request.method === 'HEAD' && this.lastModifiedAgeMs && answer.headers['last-modified']) { + answer.headers['last-modified'] = new Date( + Date.parse(answer.headers['last-modified']) - this.lastModifiedAgeMs + ).toUTCString() + } response.writeHead(answer.statusCode!, answer.headers) if (fileItHas && this.answer === 'stallFileBodies') { answer.once('data', (chunk: Buffer) => { @@ -173,3 +184,148 @@ export const sesClientFor = ( credentials: { accessKeyId: 'fake-ses', secretAccessKey: 'fake-ses' }, ...overrides, }) + +export class EmailQueueOfOneTest extends CyclotronJobQueuePostgresV2 { + constructor( + private readonly queueName: string, + config: ConstructorParameters[1] + ) { + super(10, config) + } + + public override startAsConsumer( + _queue: CyclotronJobQueueKind, + consumeBatch: (invocations: CyclotronJobInvocation[]) => Promise<{ backgroundTask: Promise }> + ): Promise { + return super.startAsConsumer(this.queueName as CyclotronJobQueueKind, (invocations) => + consumeBatch(invocations.map((invocation) => ({ ...invocation, queue: 'email' }))) + ) + } + + public override queueInvocations(invocations: CyclotronJobInvocation[]): Promise { + return super.queueInvocations(invocations.map((invocation) => this.onOwnQueue(invocation))) + } + + public override queueInvocationResults(results: CyclotronJobInvocationResult[]): Promise { + return super.queueInvocationResults( + results.map((result) => ({ ...result, invocation: this.onOwnQueue(result.invocation) })) + ) + } + + private onOwnQueue(invocation: CyclotronJobInvocation): CyclotronJobInvocation { + const queue = invocation.queue === 'email' ? this.queueName : `${this.queueName}/${invocation.queue}` + return { ...invocation, queue: queue as CyclotronJobQueueKind } + } +} + +export type HeldQuery = { release: () => void; drop: () => void } + +const SSL_REQUEST = 80877103 +const GSS_ENCRYPTION_REQUEST = 80877104 +const ENDS_A_REQUEST = new Set(['S', 'Q', 'X', 'p', 'H', 'c', 'f']) + +export class PostgresProxy { + url = '' + private readonly waiting: { matches: (request: string) => boolean; caught: (held: HeldQuery) => void }[] = [] + private readonly holding = new Set() + private readonly sockets = new Set() + private readonly server = createTcpServer((client) => this.forward(client)) + + static async start(upstream: URL): Promise { + const proxy = new PostgresProxy(upstream) + await new Promise((resolve) => proxy.server.listen(0, '127.0.0.1', resolve)) + const url = new URL(upstream) + url.host = `127.0.0.1:${(proxy.server.address() as AddressInfo).port}` + proxy.url = url.toString() + return proxy + } + + private constructor(private readonly upstream: URL) {} + + hold(matches: (request: string) => boolean): Promise { + return new Promise((caught) => this.waiting.push({ matches, caught })) + } + + releaseAll(): void { + this.waiting.splice(0) + for (const held of this.holding) { + held.release() + } + } + + async close(): Promise { + for (const socket of this.sockets) { + socket.destroy() + } + await new Promise((resolve) => this.server.close(resolve)) + } + + private forward(client: Socket): void { + const upstream = connect(Number(this.upstream.port || 5432), this.upstream.hostname) + for (const socket of [client, upstream]) { + this.sockets.add(socket) + socket.on('error', () => undefined) + socket.on('close', () => { + client.destroy() + upstream.destroy() + }) + } + upstream.pipe(client) + + let unread = Buffer.alloc(0) + let request: Buffer[] = [] + let startedUp = false + let paused = false + const forwardRequests = (): void => { + while (!paused) { + const typeBytes = startedUp ? 1 : 0 + if (unread.length < typeBytes + 4) { + return + } + const length = typeBytes + unread.readInt32BE(typeBytes) + if (unread.length < length) { + return + } + const message = unread.subarray(0, length) + unread = unread.subarray(length) + if (!startedUp) { + const code = message.readInt32BE(4) + startedUp = code !== SSL_REQUEST && code !== GSS_ENCRYPTION_REQUEST + upstream.write(message) + continue + } + request.push(message) + if (!ENDS_A_REQUEST.has(String.fromCharCode(message[0]))) { + continue + } + const whole = Buffer.concat(request) + request = [] + const index = this.waiting.findIndex(({ matches }) => matches(whole.toString('latin1'))) + if (index === -1) { + upstream.write(whole) + continue + } + const [{ caught }] = this.waiting.splice(index, 1) + paused = true + const held: HeldQuery = { + release: () => { + this.holding.delete(held) + upstream.write(whole) + paused = false + forwardRequests() + }, + drop: () => { + this.holding.delete(held) + client.destroy() + }, + } + this.holding.add(held) + caught(held) + } + } + client.on('data', (chunk: Buffer) => { + unread = Buffer.concat([unread, chunk]) + forwardRequests() + }) + } +} diff --git a/nodejs/src/cdp/cdp-api.serial.test.ts b/nodejs/src/cdp/cdp-api.serial.test.ts index 745a324874f1..36981c28dc13 100644 --- a/nodejs/src/cdp/cdp-api.serial.test.ts +++ b/nodejs/src/cdp/cdp-api.serial.test.ts @@ -4,19 +4,25 @@ import { mockFetch } from '../../tests/helpers/mocks/request.mock' import { Server } from 'http' import jwt from 'jsonwebtoken' -import { createHash, randomBytes } from 'node:crypto' +import { DateTime } from 'luxon' +import { createHash, randomBytes, randomUUID } from 'node:crypto' +import { Pool } from 'pg' +import pino from 'pino' import { Counter, register } from 'prom-client' import supertest from 'supertest' import express from 'ultimate-express' +import { CyclotronInputType } from '~/cdp/schema/cyclotron' import { HogFlow } from '~/cdp/schema/hogflow' import { PosthogJwtAudience } from '~/cdp/utils/jwt-utils' import { ScopedServiceJwt } from '~/cdp/utils/scoped-service-jwt' import { setupExpressApp } from '~/common/api/router' +import { PersonReadRepository } from '~/common/persons/repositories/person-repository' +import { PostgresPersonRepository } from '~/common/persons/repositories/postgres-person-repository' import { deleteKeysWithPrefix } from '~/common/redis/_tests/redis' import { RedisV2, createRedisV2PoolFromConfig } from '~/common/redis/redis-v2' import { closeHub, createHub } from '~/common/utils/db/hub' -import { PostgresUse } from '~/common/utils/db/postgres' +import { PostgresRouter, PostgresUse } from '~/common/utils/db/postgres' import { parseJSON } from '~/common/utils/json-parse' import { logger } from '~/common/utils/logger' import { UUIDT } from '~/common/utils/utils' @@ -26,8 +32,21 @@ import { createCdpConsumerDeps } from '../../tests/helpers/cdp' import { forSnapshot } from '../../tests/helpers/snapshots' import { createTeam, getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql' import { Hub, Team } from '../types' -import { FixtureHogFlowBuilder } from './_tests/builders/hogflow.builder' +import { FixtureHogFlowBuilder, SimpleHogFlowRepresentation } from './_tests/builders/hogflow.builder' +import { + BucketProxy, + EmailQueueOfOneTest, + HeldQuery, + PostgresProxy, + StoredFile, + attachmentFile, + attachmentStoreConfig, + deliveredTo, + sesClientFor, + uploadAttachment, +} from './_tests/email-delivery-fakes' import { HOG_EXAMPLES, HOG_FILTERS_EXAMPLES, HOG_INPUTS_EXAMPLES } from './_tests/examples' +import { startFakeSes } from './_tests/fake-ses' import { insertHogFunction as _insertHogFunction, createHogFunction, @@ -38,18 +57,33 @@ import { insertHogFlow as _insertHogFlow, hideHogFlowsFromReplica } from './_tes import { CdpApi } from './cdp-api' import { getDefaultCdpConfig } from './config' import { CdpConsumerBaseDeps } from './consumers/cdp-base.consumer' +import { CdpCyclotronWorkerEmail } from './consumers/cdp-cyclotron-worker-email.consumer' import { posthogFilterOutPlugin } from './legacy-plugins/_transformations/posthog-filter-out-plugin/template' import { IdempotencyKeys, createIdempotencyValkeyPool } from './services/hogflows/idempotency-keys' import { + ATTACHMENT_CONTENT_ID_MAX_LENGTH, + ATTACHMENT_CONTENT_TYPE_MAX_LENGTH, + ATTACHMENT_FILENAME_MAX_LENGTH, transactionalEmailRequestsTotal, transactionalEmailStageDurationMs, } from './services/hogflows/transactional-email' -import { S3AttachmentStore, buildAttachmentStore } from './services/messaging/attachment-store' -import { AttachmentUploads } from './services/messaging/attachment-uploads' +import { SEND_NEEDS_FILES_FOR_MS } from './services/messaging/attachment-horizon' +import { + ATTACHMENT_LIFETIME_MS, + DAY_MS, + S3AttachmentStore, + buildAttachmentStore, +} from './services/messaging/attachment-store' +import { ATTACHMENT_VERIFICATION_DEADLINE_MS, AttachmentUploads } from './services/messaging/attachment-uploads' import { SuppressionLookupError } from './services/messaging/email-suppression.service' +import { UTM_VALUE_LIMIT_CHARS } from './services/messaging/helpers/utm' +import { InFlightRegistry } from './services/messaging/in-flight-registry' +import { MESSAGE_MAX_BYTES, estimateMessageBytes } from './services/messaging/message-estimate' import { BASE_REDIS_KEY, HogWatcherState } from './services/monitoring/hog-watcher.service' +import { template as emailTemplate } from './templates/_destinations/email/email.template' import { compileHog } from './templates/compiler' import { HogFunctionInvocationGlobals, HogFunctionType } from './types' +import { EMAIL_BODY_TAG_LIMIT, LIQUID_RENDER_LIMIT_MS } from './utils/liquid' // Email MX validation runs on every email send, so without a mock the test-panel // email tests would do live DNS lookups for their fixture recipients (and @@ -64,6 +98,18 @@ jest.mock('node:dns/promises', () => ({ })), })) +const emittedLogLines = (): { lines: string[]; stop: () => void } => { + const target = logger as unknown as { pino: pino.Logger } + const original = target.pino + const lines: string[] = [] + target.pino = pino({ level: 'debug' }, { write: (line: string) => void lines.push(line) }) + return { + lines, + stop: () => { + target.pino = original + }, + } +} describe('CDP API', () => { let hub: Hub let cdpDeps: CdpConsumerBaseDeps @@ -2245,7 +2291,8 @@ describe('CDP API', () => { const transactionalFlow = ( trigger: 'api' | 'schedule', status: HogFlow['status'], - category: 'transactional' | 'marketing' = 'transactional' + category: 'transactional' | 'marketing' = 'transactional', + inputs: Record = {} ): HogFlow => new FixtureHogFlowBuilder() .withTeamId(team.id) @@ -2256,7 +2303,7 @@ describe('CDP API', () => { trigger: { type: 'trigger', config: { type: trigger } }, email: { type: 'function_email', - config: { template_id: 'template-email', inputs: {}, message_category_type: category }, + config: { template_id: 'template-email', inputs, message_category_type: category }, }, }, edges: [{ from: 'trigger', to: 'email', type: 'continue' }], @@ -2332,6 +2379,13 @@ describe('CDP API', () => { const res = await send(hogFlow.id, { Authorization: `Bearer ${mintToken(team.id, hogFlow.id)}` }) expect(res.status).toEqual(202) + expect(res.body).toEqual({ + status: 'queued', + invocation_id: expect.any(String), + replayed: false, + attachment_count: 0, + attachment_bytes: 0, + }) expect(await outcomeCount('queued')).toEqual(queuedBefore + 1) // Without an Idempotency-Key the request makes no idempotency call. expect(stageCountChanges(stagesBefore, await stageCounts())).toEqual({ @@ -2441,11 +2495,7 @@ describe('CDP API', () => { expect(sent.status).toEqual(202) expect(sent.body).toMatchObject({ replayed: false }) expect(replayed.status).toEqual(202) - expect(replayed.body).toEqual({ - status: 'queued', - invocation_id: sent.body.invocation_id, - replayed: true, - }) + expect(replayed.body).toEqual({ ...sent.body, replayed: true }) expect(mockQueueInvocations).toHaveBeenCalledTimes(1) expect(emailStatusRows()).toEqual([ expect.objectContaining({ message_id: sent.body.invocation_id, status: 'queued' }), @@ -2587,38 +2637,6 @@ describe('CDP API', () => { expect(mockQueueInvocations).not.toHaveBeenCalled() }) - it.each([ - ['a field this version does not know', { ...codeSend, priority: 'high' }], - [ - 'a content field this version does not know', - { ...codeSend, content: { ...codeSend.content, amp: '

' } }, - ], - [ - 'a sender field this version does not know', - { ...codeSend, from: { ...codeSend.from, replyName: 'Ops' } }, - ], - [ - 'a tracking field this version does not know', - { ...codeSend, tracking: { captureEngagement: true, pixel: false } }, - ], - ['content without a body', { ...codeSend, content: { subject: 'Hi', html: '', text: '' } }], - ['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 }], - [ - 'more than 10 cc addresses', - { ...codeSend, cc: Array.from({ length: 11 }, (_, i) => `copy-${i}@example.com`) }, - ], - ])('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(await outcomeCount('invalid_request')).toEqual(invalidBefore + 1) - expect(mockQueueInvocations).not.toHaveBeenCalled() - }) - it.each([ [ 'a bcc address is suppressed', @@ -2651,27 +2669,1122 @@ describe('CDP API', () => { expect(mockQueueInvocations).not.toHaveBeenCalled() }) - it('refuses a code send with files as a field it does not take, and leaves its key free for a send without them', async () => { - const idempotencyKey = `key-${new UUIDT().toString()}` - const file = { filename: 'report.pdf', content: 'c2VjcmV0LWZpbGU=', contentType: 'application/pdf' } + describe('queued on Cyclotron Postgres', () => { + let cyclotron: Pool + let posthogDb: PostgresProxy + let personsDb: PostgresProxy + let bucket: BucketProxy + let router: PostgresRouter + let queueName: string + let filesApi: CdpApi + let filesApp: express.Application + let filesServer: Server + let invoice: StoredFile + let receipt: StoredFile + let large: StoredFile + + const personsReadFrom = (persons: PostgresPersonRepository): PersonReadRepository => ({ + fetchPerson: (teamId, distinctId, callerTag) => persons.fetchPerson(teamId, distinctId, { callerTag }), + fetchPersonsByDistinctIds: (teamPersons, callerTag) => + persons.fetchPersonsByDistinctIds(teamPersons, true, callerTag), + fetchPersonsByPersonIds: (teamPersons, callerTag) => + persons.fetchPersonsByPersonIds(teamPersons, true, callerTag), + fetchDistinctIdsForPersons: (teamId, personIntIds, options) => + persons.fetchDistinctIdsForPersons(teamId, personIntIds, options), + }) + const apisOfOneTest: { cdpApi: CdpApi; server: Server }[] = [] + const buildFilesApi = async (settings: Partial = {}): Promise => { + const producer = new EmailQueueOfOneTest(queueName, hub) + await producer.startAsProducer() + return new CdpApi( + { + ...hub, + CDP_EMAIL_ATTACHMENTS_TEAMS: `${team.id}`, + CDP_EMAIL_ATTACHMENTS_S3_ENDPOINT: bucket.endpoint, + CDP_EMAIL_ATTACHMENTS_S3_PREFIX: attachmentStoreConfig.CDP_EMAIL_ATTACHMENTS_S3_PREFIX, + ...settings, + }, + { + ...cdpDeps, + postgres: router, + personRepository: personsReadFrom(new PostgresPersonRepository(router)), + idempotencyValkey: createIdempotencyValkeyPool(hub), + }, + { hogQueue: createMockJobQueue(), hogflowQueue: producer } + ) + } + const servedForOneTest = async (settings: Partial): Promise => { + const cdpApi = await buildFilesApi(settings) + const served = serve(cdpApi) + apisOfOneTest.push({ cdpApi, server: served.listen(0, () => {}) }) + return served + } + const serve = (cdpApi: CdpApi): express.Application => { + const served = setupExpressApp() + served.use('/', cdpApi.router()) + return served + } + + type QueuedJob = { id: string; status: string; send: Record; stored: string } + const queuedJobs = async (): Promise => { + const { rows } = await cyclotron.query( + 'SELECT id, status, state FROM cyclotron_jobs WHERE queue_name = $1 ORDER BY created', + [queueName] + ) + return rows.map((row) => ({ + id: row.id, + status: row.status, + send: parseJSON(row.state.toString()).state.send, + stored: row.state.toString('latin1'), + })) + } + const queuedIds = async (): Promise => (await queuedJobs()).map((job) => job.id) - const refused = await sendAccepted(hogFlow.id, { - ...body, - idempotency_key: idempotencyKey, - send: { ...codeSend, attachments: [file] }, + const reference = (file: StoredFile) => ({ + sha256: file.metadata.sha256, + filename: 'invoice.pdf', + content_type: 'application/pdf', }) - expect(refused.status).toEqual(400) - expect(parseJSON(refused.body.error)).toEqual([ - expect.objectContaining({ code: 'unrecognized_keys', keys: ['attachments'], path: ['send'] }), - ]) - expect(mockQueueInvocations).not.toHaveBeenCalled() + const newKey = () => `key-${new UUIDT().toString()}` + type SendOptions = { + acceptedSend?: object + flowId?: string + variables?: Record + distinctId?: string + to?: string + app?: express.Application + } + const sendOnFilesApi = ( + payload: Record, + { flowId = hogFlow.id, app: served = filesApp }: SendOptions = {} + ) => + supertest(served) + .post(`/api/projects/${team.id}/hog_flows/${flowId}/transactional_email/send`) + .set({ Authorization: `Bearer ${tokenFor('transactional_email/send', team.id, flowId)}` }) + .send(payload) + const sendWithFiles = ( + key: string, + attachments: object[], + { + acceptedSend = codeSend as object, + variables = body.variables as Record, + distinctId = body.distinct_id, + to = body.to, + ...options + }: SendOptions = {} + ) => + sendOnFilesApi( + { + ...body, + to, + distinct_id: distinctId, + variables, + idempotency_key: key, + send: { ...acceptedSend, attachments }, + }, + options + ) + + beforeAll(async () => { + cyclotron = new Pool({ connectionString: hub.CYCLOTRON_NODE_DATABASE_URL }) + posthogDb = await PostgresProxy.start(new URL(hub.DATABASE_URL)) + personsDb = await PostgresProxy.start(new URL(hub.PERSONS_DATABASE_URL || hub.DATABASE_URL)) + bucket = await BucketProxy.start() + router = new PostgresRouter({ + ...hub, + DATABASE_URL: posthogDb.url, + DATABASE_READONLY_URL: posthogDb.url, + PERSONS_DATABASE_URL: personsDb.url, + PERSONS_READONLY_DATABASE_URL: personsDb.url, + }) + queueName = `transactional-email-test-${randomUUID()}` + filesApi = await buildFilesApi() + filesApp = serve(filesApi) + filesServer = filesApp.listen(0, () => {}) + invoice = await uploadAttachment(team.id, { filename: 'invoice.pdf' }) + receipt = await uploadAttachment(team.id, { filename: 'receipt.pdf' }) + large = await uploadAttachment(team.id, { filename: 'large.pdf' }, Buffer.alloc(28_800_000, 'a')) + }) + + beforeEach(async () => { + await insertHogFunctionTemplate(hub.postgres, emailTemplate) + }) + + afterEach(async () => { + posthogDb.releaseAll() + personsDb.releaseAll() + bucket.lastModifiedAgeMs = 0 + bucket.release('pass') + await Promise.all( + apisOfOneTest.splice(0).map(({ cdpApi, server: served }) => { + served.close() + return cdpApi.stop() + }) + ) + await cyclotron.query('DELETE FROM cyclotron_jobs WHERE queue_name = $1', [queueName]) + }) + + afterAll(async () => { + filesServer.close() + await filesApi.stop() + await Promise.all([bucket.close(), posthogDb.close(), personsDb.close()]) + await router.end() + await cyclotron.end() + }) + + const withFile = (fields: Record) => ({ + ...codeSend, + attachments: [{ ...reference(invoice), ...fields }], + }) + const filenameOf = (length: number) => `${'x'.repeat(length - '.pdf'.length)}.pdf` + const contentTypeOf = (length: number) => `application/${'x'.repeat(length - 'application/'.length)}` + const contentIdOf = (length: number) => 'x'.repeat(length) + const filenameReason = `filename can be at most ${ATTACHMENT_FILENAME_MAX_LENGTH} characters` + const contentTypeReason = `content_type needs 1 to ${ATTACHMENT_CONTENT_TYPE_MAX_LENGTH} characters` + const contentIdReason = `content_id needs 1 to ${ATTACHMENT_CONTENT_ID_MAX_LENGTH} characters` + + it.each<[string, () => object, () => object, string]>([ + [ + 'a field this version does not know', + () => ({ ...codeSend, priority: 'high' }), + () => codeSend, + '"priority"', + ], + [ + 'a content field this version does not know', + () => ({ ...codeSend, content: { ...codeSend.content, amp: '

' } }), + () => codeSend, + '"amp"', + ], + [ + 'a sender field this version does not know', + () => ({ ...codeSend, from: { ...codeSend.from, replyName: 'Ops' } }), + () => codeSend, + '"replyName"', + ], + [ + 'a tracking field this version does not know', + () => ({ ...codeSend, tracking: { captureEngagement: true, pixel: false } }), + () => codeSend, + '"pixel"', + ], + [ + 'content without a body', + () => ({ ...codeSend, content: { subject: 'Hi', html: '', text: '' } }), + () => codeSend, + 'content needs html or text', + ], + [ + 'content with an empty subject', + () => ({ ...codeSend, content: { subject: '', text: 'Hi' } }), + () => codeSend, + '"subject"', + ], + [ + 'a sender without an integration', + () => ({ ...codeSend, from: { email: 'security@example.com' } }), + () => codeSend, + '"integrationId"', + ], + ['no tracking choice', () => ({ content: codeSend.content }), () => codeSend, '"tracking"'], + [ + 'more than 10 cc addresses', + () => ({ ...codeSend, cc: Array.from({ length: 11 }, (_, i) => `copy-${i}@example.com`) }), + () => codeSend, + '"cc"', + ], + [ + 'more than 10 files', + () => ({ ...codeSend, attachments: Array(11).fill(reference(invoice)) }), + () => codeSend, + '"attachments"', + ], + [ + 'a file hash that is not 64 lowercase hex characters', + () => withFile({ sha256: invoice.metadata.sha256.toUpperCase() }), + () => codeSend, + '"sha256"', + ], + ['a file without a name', () => withFile({ filename: '' }), () => codeSend, '"filename"'], + [ + 'a file field this version does not know', + () => withFile({ disposition: 'inline' }), + () => codeSend, + '"disposition"', + ], + [ + 'an inline file instead of a reference', + () => ({ + ...codeSend, + attachments: [ + { filename: 'report.pdf', content: 'c2VjcmV0LWZpbGU=', contentType: 'application/pdf' }, + ], + }), + () => codeSend, + '"sha256"', + ], + [ + 'a file name over the SES limit', + () => withFile({ filename: filenameOf(ATTACHMENT_FILENAME_MAX_LENGTH + 1) }), + () => withFile({ filename: filenameOf(ATTACHMENT_FILENAME_MAX_LENGTH) }), + filenameReason, + ], + [ + 'a content type over the SES limit', + () => withFile({ content_type: contentTypeOf(ATTACHMENT_CONTENT_TYPE_MAX_LENGTH + 1) }), + () => withFile({ content_type: contentTypeOf(ATTACHMENT_CONTENT_TYPE_MAX_LENGTH) }), + contentTypeReason, + ], + ['an empty content type', () => withFile({ content_type: '' }), () => withFile({}), contentTypeReason], + [ + 'a content id over the SES limit', + () => withFile({ content_id: contentIdOf(ATTACHMENT_CONTENT_ID_MAX_LENGTH + 1) }), + () => withFile({ content_id: contentIdOf(ATTACHMENT_CONTENT_ID_MAX_LENGTH) }), + contentIdReason, + ], + ['an empty content id', () => withFile({ content_id: '' }), () => withFile({}), contentIdReason], + [ + 'a file name that starts a new header line', + () => withFile({ filename: 'invoice.pdf\r\nBcc: someone@example.com' }), + () => withFile({ filename: 'invoice.pdf' }), + 'filename cannot contain control characters', + ], + [ + 'a content type that starts a new header line', + () => withFile({ content_type: 'application/pdf\r\nBcc: someone@example.com' }), + () => withFile({}), + 'content_type cannot contain control characters', + ], + [ + 'a content id with a control character', + () => withFile({ content_id: 'logo\n' }), + () => withFile({}), + 'content_id cannot contain control characters', + ], + [ + 'a file type SES does not send', + () => withFile({ filename: 'setup.exe' }), + () => withFile({ filename: 'setup.exe.pdf' }), + 'SES does not send .exe files, so setup.exe cannot be attached', + ], + [ + 'a file type SES does not send, in upper case', + () => withFile({ filename: 'Invoice.PDF.JS' }), + () => withFile({ filename: 'Invoice.PDF.JSON' }), + 'SES does not send .js files, so Invoice.PDF.JS cannot be attached', + ], + ])( + 'refuses a code send with %s, naming it, queues nothing, and its key sends once it is fixed', + async (_desc, refused, fixed, reason) => { + const invalidBefore = await outcomeCount('invalid_request') + const idempotency_key = newKey() + + const res = await sendOnFilesApi({ ...body, idempotency_key, send: refused() }) + const queuedOnRefusal = await queuedIds() + const retried = await sendOnFilesApi({ ...body, idempotency_key, send: fixed() }) + + expect(res.status).toEqual(400) + expect(res.body.error).toContain(reason) + expect(await outcomeCount('invalid_request')).toEqual(invalidBefore + 1) + expect(queuedOnRefusal).toEqual([]) + expect(retried.status).toEqual(202) + expect(await queuedIds()).toEqual([retried.body.invocation_id]) + } + ) + + it('queues the references of the files it verified, without their bytes, answers their totals, and the worker delivers those files', async () => { + const recipient = `${randomUUID()}@example.com` + await insertIntegration(hub.postgres, team.id, { + id: codeSend.from.integrationId, + kind: 'email', + config: { + email: codeSend.from.email, + name: codeSend.from.name, + domain: 'example.com', + verified: true, + provider: 'ses', + }, + }) + const logo = { + sha256: receipt.metadata.sha256, + filename: 'logo.png', + content_type: 'image/png', + content_id: 'logo', + } + + const generated = await flowThrough( + { exit }, + [step('trigger', 'email'), step('email', 'exit')], + 'transactional', + { + email: { + templating: 'liquid', + value: { + to: { email: '{{ event.properties.to | raw }}', name: '' }, + cc: '', + bcc: '', + from: { integrationId: null }, + replyTo: '', + subject: '', + preheader: '', + text: '', + html: '', + }, + }, + } + ) + + const res = await sendWithFiles(newKey(), [reference(invoice), logo], { + to: recipient, + flowId: generated.id, + }) + + expect(res.status).toBe(202) + expect(res.body).toEqual({ + status: 'queued', + invocation_id: expect.any(String), + replayed: false, + attachment_count: 2, + attachment_bytes: invoice.bytes.length + receipt.bytes.length, + }) + const [job] = await queuedJobs() + expect(job.id).toBe(res.body.invocation_id) + expect(job.send).toEqual({ + ...codeSend, + attachments: [ + { + sha256: invoice.metadata.sha256, + filename: 'invoice.pdf', + contentType: 'application/pdf', + sizeBytes: invoice.bytes.length, + }, + { + sha256: receipt.metadata.sha256, + filename: 'logo.png', + contentType: 'image/png', + contentId: 'logo', + sizeBytes: receipt.bytes.length, + }, + ], + }) + for (const file of [invoice, receipt]) { + expect(job.stored).not.toContain(file.bytes.toString('latin1')) + expect(job.stored).not.toContain(file.bytes.toString('base64')) + } + + const ses = await startFakeSes({ + [hub.SES_TRACKED_CONFIGURATION_SET]: { tracking: true }, + [hub.SES_UNTRACKED_CONFIGURATION_SET]: { tracking: false }, + }) + const workerConfig = { + ...hub, + CDP_CYCLOTRON_BATCH_DELAY_MS: 10, + CDP_EMAIL_ATTACHMENTS_S3_PREFIX: attachmentStoreConfig.CDP_EMAIL_ATTACHMENTS_S3_PREFIX, + } + const worker = new CdpCyclotronWorkerEmail( + workerConfig, + createCdpConsumerDeps(hub), + new EmailQueueOfOneTest(queueName, workerConfig), + new InFlightRegistry({ budgetBytes: 2 ** 40, heapLimitBytes: 2 ** 40, maxInFlight: 1 }, () => 0) + ) + worker.emailService.sesV2Client = sesClientFor(ses) + await worker.start() + try { + const delivered = await waitForExpect(() => { + expect(deliveredTo(ses, recipient)).toHaveLength(1) + return deliveredTo(ses, recipient)[0] + }) + expect(delivered.attachments).toEqual([ + { + filename: 'invoice.pdf', + contentType: 'application/pdf', + contentId: undefined, + inline: false, + bytes: invoice.bytes, + }, + { + filename: 'logo.png', + contentType: 'image/png', + contentId: 'logo', + inline: true, + bytes: receipt.bytes, + }, + ]) + } finally { + await worker.stop() + await ses.close() + } + }) + + it.each([ + ['the same files', () => [reference(invoice)], 202], + ['other files', () => [reference(receipt)], 422], + ])( + 'answers a retry of a keyed send with files that names %s from its first send, and queues one job', + async (_name, retryFiles, status) => { + const key = newKey() + const sent = await sendWithFiles(key, [reference(invoice)]) + + const retried = await sendWithFiles(key, retryFiles()) + + expect(sent.status).toBe(202) + expect(retried.status).toBe(status) + expect(retried.body).toEqual( + status === 202 ? { ...sent.body, replayed: true } : { error: 'idempotency_key_reused' } + ) + expect(await queuedIds()).toEqual([sent.body.invocation_id]) + } + ) + + const flowThrough = async ( + steps: SimpleHogFlowRepresentation['actions'], + edges: SimpleHogFlowRepresentation['edges'], + category: 'transactional' | 'marketing' = 'transactional', + emailInputs: Record = {}, + emailConfig: { skip_frequency_cap?: boolean } = {} + ): Promise => + await insertHogFlow( + new FixtureHogFlowBuilder() + .withTeamId(team.id) + .withStatus('active') + .withExitCondition('exit_only_at_end') + .withWorkflow({ + actions: { + trigger: { type: 'trigger', config: { type: 'api' } }, + ...steps, + email: { + type: 'function_email', + config: { + template_id: 'template-email', + inputs: emailInputs, + message_category_type: category, + ...emailConfig, + }, + }, + }, + edges, + }) + .build() + ) + const delay = { type: 'delay', config: { delay_duration: '1h' } } as const + const exit = { type: 'exit', config: {} } as const + const step = (from: string, to: string) => ({ from, to, type: 'continue' as const }) + + const shapesThatCannotCarryFiles: [string, () => Promise, string][] = [ + [ + 'a delay before its email step', + () => flowThrough({ delay }, [step('trigger', 'delay'), step('delay', 'email')]), + 'workflow_shape_not_supported', + ], + [ + 'an email step the frequency cap can hold', + () => flowThrough({}, [step('trigger', 'email')], 'marketing'), + 'workflow_can_wait', + ], + [ + 'a marketing email step that skips the frequency cap, which the team sending cap can still hold', + () => flowThrough({}, [step('trigger', 'email')], 'marketing', {}, { skip_frequency_cap: true }), + 'workflow_can_wait', + ], + ] + + it.each(shapesThatCannotCarryFiles)( + 'refuses a send with files to a workflow with %s, queues nothing, and releases the key', + async (_name, arrange, code) => { + const flow = await arrange() + const key = newKey() + const stagesBefore = await stageCounts() + + const res = await sendWithFiles(key, [reference(invoice)], { flowId: flow.id }) + const stagesOfTheRefusal = stageCountChanges(stagesBefore, await stageCounts()) + const queuedOnRefusal = await queuedIds() + const retry = await sendWithFiles(key, [reference(invoice)]) + + expect(res.status).toBe(400) + expect(res.body).toEqual({ + error: code, + reason: expect.any(String), + fix: expect.any(String), + }) + expect(stagesOfTheRefusal).toEqual({ + auth: 1, + validate: 1, + load: 1, + build: 1, + idempotency: 1, + total: 1, + }) + expect(queuedOnRefusal).toEqual([]) + expect(retry.status).toBe(202) + expect(retry.body).toMatchObject({ replayed: false }) + expect(await queuedIds()).toEqual([retry.body.invocation_id]) + } + ) + + it.each(shapesThatCannotCarryFiles)( + 'replays a keyed send with files after its workflow changed to %s', + async (_name, arrange) => { + const flow = await flowThrough({ exit }, [step('trigger', 'email'), step('email', 'exit')]) + const key = newKey() + const first = await sendWithFiles(key, [reference(invoice)], { flowId: flow.id }) + expect(first.status).toBe(202) + const changed = await arrange() + await hub.postgres.query( + PostgresUse.COMMON_WRITE, + `UPDATE posthog_hogflow SET actions = changed.actions, edges = changed.edges + FROM posthog_hogflow changed WHERE posthog_hogflow.id = $1 AND changed.id = $2`, + [flow.id, changed.id], + 'change-workflow-shape' + ) + filesApi['hogFlowManager']['onHogFlowsReloaded'](team.id, [flow.id]) + + const retry = await sendWithFiles(key, [reference(invoice)], { flowId: flow.id }) + + expect(retry.status).toBe(202) + expect(retry.body).toEqual({ ...first.body, replayed: true }) + expect(await queuedIds()).toEqual([first.body.invocation_id]) + } + ) + + it('refuses files from a team off the allowlist before it reserves the key, and queues nothing', async () => { + const offTheList = await servedForOneTest({ CDP_EMAIL_ATTACHMENTS_TEAMS: `${team.id + 1}` }) + const stagesBefore = await stageCounts() + + const res = await sendWithFiles(newKey(), [reference(invoice)], { app: offTheList }) + + expect(res.status).toBe(400) + expect(res.body).toEqual({ error: 'field_not_accepted' }) + expect(stageCountChanges(stagesBefore, await stageCounts())).toEqual({ auth: 1, validate: 1, total: 1 }) + expect(await queuedIds()).toEqual([]) + }) + + it.each<[string, () => Promise<{ file: StoredFile; fix: () => Promise }>]>([ + [ + 'a file that was never uploaded', + () => { + const file = attachmentFile() + return Promise.resolve({ file, fix: () => uploadAttachment(team.id, {}, file.bytes) }) + }, + ], + [ + 'a file only another team uploaded', + async () => { + const file = await uploadAttachment(await createTeam(hub.postgres, team.organization_id), {}) + return { file, fix: () => uploadAttachment(team.id, {}, file.bytes) } + }, + ], + ])( + 'refuses a send with %s, queues nothing, and its key sends once that is fixed', + async (_name, arrange) => { + const { file, fix } = await arrange() + const key = newKey() + + const refused = await sendWithFiles(key, [reference(file)]) + const queuedOnRefusal = await queuedIds() + await fix() + const retried = await sendWithFiles(key, [reference(file)]) + + expect(refused.status).toBe(400) + expect(refused.body).toEqual({ error: 'attachment_not_uploaded', sha256: file.metadata.sha256 }) + expect(queuedOnRefusal).toEqual([]) + expect(retried.status).toBe(202) + expect(await queuedIds()).toEqual([retried.body.invocation_id]) + } + ) + + const storedEmail = (email: Record): Record => ({ + email: { + templating: 'liquid', + value: { + to: { email: '{{ event.properties.to }}', name: '' }, + from: {}, + subject: 'Your report', + text: 'Your report is attached', + ...email, + }, + }, + }) + const storedTemplateRendering = async ( + email: Record, + chunk: string + ): Promise => { + const flow = await insertHogFlow( + transactionalFlow('api', 'active', 'transactional', storedEmail(email)) + ) + return { + flowId: flow.id, + variables: { chunk }, + acceptedSend: { tracking: { captureEngagement: false } }, + } + } + const storedTemplateOnAFreshApi = async (): Promise => ({ + options: await storedTemplateRendering({ html: '

{{ variables.chunk }}

' }, 'x'), + app: await servedForOneTest({}), + }) + const aPersonWithTheDistinctId = async (): Promise => { + const distinctId = `person-${randomUUID()}` + await hub.personRepository.createPerson( + DateTime.now(), + {}, + {}, + {}, + team.id, + null, + true, + new UUIDT().toString(), + { distinctId } + ) + return distinctId + } + + type HeldSend = { options?: SendOptions; app?: express.Application; fix?: () => void } + const caught = async (query: Promise, expire: () => void): Promise => { + const held = await query + expire() + return held + } + const realAbortSignalTimeout = AbortSignal.timeout.bind(AbortSignal) + let verificationDeadline = new AbortController() + const expireTheVerificationDeadline = (): void => + verificationDeadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + const queriesFor = (value: string) => (request: string) => request.includes(value) + + it.each<[string, () => Promise]>([ + [ + 'no attachment store', + async () => ({ app: await servedForOneTest({ CDP_EMAIL_ATTACHMENTS_S3_BUCKET: '' }) }), + ], + [ + 'an attachment store that does not answer', + () => { + bucket.answer = 'refuse' + return Promise.resolve({ fix: () => bucket.release('pass') }) + }, + ], + [ + 'an attachment store whose HEAD outlasts the deadline, on a file that needs keeping', + () => { + bucket.answer = 'hold' + bucket.lastModifiedAgeMs = ATTACHMENT_LIFETIME_MS - SEND_NEEDS_FILES_FOR_MS + DAY_MS + void waitForExpect(() => expect(bucket.heldRequests).toBe(1)).then( + expireTheVerificationDeadline + ) + return Promise.resolve({ fix: () => bucket.release('pass') }) + }, + ], + [ + 'an attachment store whose COPY outlasts the deadline, on a file that needs keeping', + () => { + bucket.answer = 'holdCopies' + bucket.lastModifiedAgeMs = ATTACHMENT_LIFETIME_MS - SEND_NEEDS_FILES_FOR_MS + DAY_MS + void waitForExpect(() => expect(bucket.heldRequests).toBe(1)).then( + expireTheVerificationDeadline + ) + return Promise.resolve({ fix: () => bucket.release('pass') }) + }, + ], + [ + 'the recipient checks before the files use up the deadline', + async () => { + const to = `${randomUUID()}@example.com` + const lookup = posthogDb.hold( + (request) => request.includes('posthog_messagesuppression') && request.includes(to) + ) + void lookup.then((held) => { + expireTheVerificationDeadline() + held.release() + }) + return Promise.resolve({ options: { to } }) + }, + ], + [ + 'the template lookup of the email step fails', + async () => { + void posthogDb.hold(queriesFor('posthog_hogfunctiontemplate')).then((held) => held.drop()) + return await storedTemplateOnAFreshApi() + }, + ], + [ + 'the template lookup of the email step outlasts the deadline', + async () => { + void caught( + posthogDb.hold(queriesFor('posthog_hogfunctiontemplate')), + expireTheVerificationDeadline + ) + return await storedTemplateOnAFreshApi() + }, + ], + [ + 'the person lookup fails', + async () => { + const distinctId = await aPersonWithTheDistinctId() + void personsDb.hold(queriesFor(distinctId)).then((held) => held.drop()) + return { options: { distinctId } } + }, + ], + [ + 'the person lookup outlasts the deadline', + async () => { + const distinctId = await aPersonWithTheDistinctId() + void caught(personsDb.hold(queriesFor(distinctId)), expireTheVerificationDeadline) + return { options: { distinctId } } + }, + ], + [ + 'the custom UTM tags of the email step render past the deadline', + async () => { + const distinctId = await aPersonWithTheDistinctId() + const flow = transactionalFlow('api', 'active', 'transactional', storedEmail({ html: 'x' })) + const emailStep = flow.actions.find((action) => action.type === 'function_email')! + const slowUtmTag = renderThatOutlastsTheLiquidLimit + Object.assign(emailStep.config, { + utm_tags_enabled: true, + utm_params: { + utm_source: slowUtmTag, + utm_medium: slowUtmTag, + utm_campaign: slowUtmTag, + utm_content: slowUtmTag, + }, + }) + const { id: flowId } = await insertHogFlow(flow) + const realNow = performance.now.bind(performance) + let skippedMs = 0 + const clock = jest.spyOn(performance, 'now').mockImplementation(() => realNow() + skippedMs) + void personsDb.hold(queriesFor(distinctId)).then((held) => { + skippedMs = ATTACHMENT_VERIFICATION_DEADLINE_MS - 2 * LIQUID_RENDER_LIMIT_MS + held.release() + }) + return { + options: { + flowId, + distinctId, + acceptedSend: { tracking: { captureEngagement: false } }, + }, + fix: () => clock.mockRestore(), + } + }, + ], + ])( + 'answers 503 when %s without waiting on held work, queues nothing, asks the store for nothing after the answer, and its key sends once that is fixed', + async (_name, arrange) => { + const deadlines = jest.spyOn(AbortSignal, 'timeout').mockImplementation((ms: number) => { + if (ms !== ATTACHMENT_VERIFICATION_DEADLINE_MS) { + return realAbortSignalTimeout(ms) + } + verificationDeadline = new AbortController() + return verificationDeadline.signal + }) + const { options, app: refusingApp, fix } = await arrange() + const key = newKey() + + const refused = await sendWithFiles(key, [reference(invoice)], { ...options, app: refusingApp }) + const bucketRequestsAtTheAnswer = bucket.requests + posthogDb.releaseAll() + personsDb.releaseAll() + fix?.() + await waitForExpect(() => expect(bucket.heldRequests).toBe(0)) + const queuedBeforeTheRetry = await queuedIds() + const bucketRequestsBeforeTheRetry = bucket.requests + deadlines.mockRestore() + const retried = await sendWithFiles(key, [reference(invoice)], options) + + expect(refused.status).toBe(503) + expect(refused.body).toEqual({ error: 'delivery_unavailable' }) + expect(queuedBeforeTheRetry).toEqual([]) + expect(bucketRequestsBeforeTheRetry).toBe(bucketRequestsAtTheAnswer) + expect(retried.status).toBe(202) + expect(await queuedIds()).toEqual([retried.body.invocation_id]) + } + ) + + const roomForBodies = MESSAGE_MAX_BYTES - estimateMessageBytes([28_800_000]) + const bodyThatOverflows = 'x'.repeat(roomForBodies) + const chunkThatFitsOnceNotTwice = 'x'.repeat(Math.ceil(roomForBodies / 2)) + const link = 'x' + const linksThatFitUntilTracked = link.repeat(Math.floor((roomForBodies * 0.7) / link.length)) + const contentOf = (html: string): SendOptions => ({ + acceptedSend: { ...codeSend, content: { subject: 'Your report', html } }, + }) + const chunkRenderedTwice = { html: '

{{ variables.chunk }}

{{ variables.chunk }}

' } + const chunkAsPreheader = { html: '

Your report

', preheader: '{{ variables.chunk }}' } + const renderThatOutlastsTheLiquidLimit = + '{% for a in (1..1000) %}{% for b in (1..1000) %}{% for c in (1..1000) %}x{% endfor %}{% endfor %}{% endfor %}' + const chunkRenderedTwiceBehindAnUnclosedTag = { html: `{% if variables.chunk %}${chunkRenderedTwice.html}` } + + it.each<[string, () => Promise]>([ + ['content', () => Promise.resolve(contentOf('x'))], + ['a stored template whose rendered body', () => storedTemplateRendering(chunkRenderedTwice, 'x')], + [ + 'a stored template that fails to parse at accept, counted as no body, whose body', + () => storedTemplateRendering(chunkRenderedTwiceBehindAnUnclosedTag, chunkThatFitsOnceNotTwice), + ], + [ + 'content on the generated workflow, a trigger, an email step and an exit, whose body', + async () => ({ + ...contentOf('x'), + flowId: (await flowThrough({ exit }, [step('trigger', 'email'), step('email', 'exit')])).id, + }), + ], + ])('accepts a send with files and %s fits beside them, and queues it', async (_name, arrange) => { + const res = await sendWithFiles(newKey(), [reference(large)], await arrange()) + + expect(res.status).toBe(202) + expect(res.body).toMatchObject({ attachment_count: 1, attachment_bytes: large.bytes.length }) + expect(await queuedIds()).toEqual([res.body.invocation_id]) + }) + + it.each<[string, () => Promise]>([ + ['content', () => Promise.resolve(contentOf(bodyThatOverflows))], + [ + 'content whose links grow once the worker tracks them', + () => Promise.resolve(contentOf(linksThatFitUntilTracked)), + ], + [ + 'a stored template that renders a chunk twice', + () => storedTemplateRendering(chunkRenderedTwice, chunkThatFitsOnceNotTwice), + ], + [ + 'a stored template whose preheader goes into the html', + () => storedTemplateRendering(chunkAsPreheader, bodyThatOverflows), + ], + [ + 'a stored template whose body grows with the person it sends to', + async () => ({ + ...(await storedTemplateRendering( + { html: '

Your report

{% if person.id %}{{ variables.chunk }}{% endif %}' }, + bodyThatOverflows + )), + distinctId: await aPersonWithTheDistinctId(), + }), + ], + [ + 'a stored template whose body renders only once the worker supplies its groups', + () => + storedTemplateRendering( + { html: '

Your report

{% if groups %}{{ variables.chunk }}{% endif %}' }, + bodyThatOverflows + ), + ], + [ + 'content whose subject', + () => + Promise.resolve({ + acceptedSend: { + ...codeSend, + content: { subject: bodyThatOverflows, html: '

Your report

' }, + }, + }), + ], + [ + 'a stored template whose rendered subject', + () => storedTemplateRendering({ subject: '{{ variables.chunk }}' }, bodyThatOverflows), + ], + ])( + 'refuses a send with files and %s too large to fit beside them, queues nothing, and its key sends smaller files', + async (_name, arrange) => { + const options = await arrange() + const key = newKey() + + const res = await sendWithFiles(key, [reference(large)], options) + const queuedOnRefusal = await queuedIds() + const retried = await sendWithFiles(key, [reference(invoice)], options) + + expect(res.status).toBe(400) + expect(res.body).toEqual({ + error: 'message_too_large', + message_bytes: expect.any(Number), + max_message_bytes: MESSAGE_MAX_BYTES, + attachment_bytes: large.bytes.length, + }) + expect(res.body.message_bytes).toBeGreaterThan(MESSAGE_MAX_BYTES) + expect(queuedOnRefusal).toEqual([]) + expect(retried.status).toBe(202) + expect(await queuedIds()).toEqual([retried.body.invocation_id]) + } + ) + + it('answers 503, not a template refusal, when the text of a text-only email step renders into the deadline, and queues nothing', async () => { + const distinctId = await aPersonWithTheDistinctId() + const { id: flowId } = await insertHogFlow( + transactionalFlow( + 'api', + 'active', + 'transactional', + storedEmail({ text: renderThatOutlastsTheLiquidLimit }) + ) + ) + const realNow = performance.now.bind(performance) + let skippedMs = 0 + const clock = jest.spyOn(performance, 'now').mockImplementation(() => realNow() + skippedMs) + void personsDb.hold(queriesFor(distinctId)).then((held) => { + skippedMs = ATTACHMENT_VERIFICATION_DEADLINE_MS - LIQUID_RENDER_LIMIT_MS / 2 + held.release() + }) + try { + const res = await sendWithFiles(newKey(), [reference(invoice)], { + flowId, + distinctId, + acceptedSend: { tracking: { captureEngagement: false } }, + }) + + expect(res.status).toBe(503) + expect(res.body).toEqual({ error: 'delivery_unavailable' }) + expect(await queuedIds()).toEqual([]) + } finally { + clock.mockRestore() + } + }) + + it('queues a send with files whose email step fails to render at accept, and logs none of its variables', async () => { + const secret = `reset-token-${randomUUID()}` + const options = await storedTemplateRendering({ html: '{% include variables.chunk %}' }, secret) + const emitted = emittedLogLines() + + let res: supertest.Response + try { + res = await sendWithFiles(newKey(), [reference(invoice)], options) + } finally { + emitted.stop() + } + + expect(res.status).toBe(202) + expect(await queuedIds()).toEqual([res.body.invocation_id]) + expect( + emitted.lines.some((line) => + String(parseJSON(line).msg).endsWith('Could not render the email step of a send with files') + ) + ).toBe(true) + expect(emitted.lines.join('')).not.toContain(secret) + }) + + const storedUtmValue = async (utmCampaign: string): Promise => { + const flow = transactionalFlow('api', 'active', 'transactional', storedEmail({ html: '

x

' })) + const emailStep = flow.actions.find((action) => action.type === 'function_email')! + Object.assign(emailStep.config, { utm_tags_enabled: true, utm_params: { utm_campaign: utmCampaign } }) + const { id: flowId } = await insertHogFlow(flow) + return { flowId, acceptedSend: { tracking: { captureEngagement: false } } } + } + + it.each<[string, string, () => Promise]>([ + [ + 'goes over a template limit', + 'email_render_limit_exceeded', + () => storedTemplateRendering({ html: renderThatOutlastsTheLiquidLimit }, 'x'), + ], + [ + 'has a body whose HTML and text together hold one Liquid tag over the limit', + 'email_body_too_many_tags', + () => + storedTemplateRendering( + { + html: '{{ variables.chunk }}'.repeat(EMAIL_BODY_TAG_LIMIT / 2 + 1), + text: '{% if variables.chunk %}x{% endif %}'.repeat(EMAIL_BODY_TAG_LIMIT / 4), + }, + 'x' + ), + ], + [ + 'has a body whose one Liquid block brings it one tag over the limit', + 'email_body_too_many_tags', + () => + storedTemplateRendering( + { html: `{% liquid\n${'echo variables.chunk\n'.repeat(EMAIL_BODY_TAG_LIMIT)}%}` }, + 'x' + ), + ], + [ + 'has a body whose quoted Liquid block opener comes before the Liquid tags over the limit', + 'email_body_too_many_tags', + () => + storedTemplateRendering( + { html: `{{ '{% liquid ' }}${'{{ variables.chunk }}'.repeat(EMAIL_BODY_TAG_LIMIT)}` }, + 'x' + ), + ], + [ + 'has a body whose one Liquid block, opened with an encoded space, brings it one tag over the limit', + 'email_body_too_many_tags', + () => + storedTemplateRendering( + { html: `{% liquid%20\n${'echo variables.chunk\n'.repeat(EMAIL_BODY_TAG_LIMIT)}%}` }, + 'x' + ), + ], + [ + 'has a body whose one Liquid block, opened after a blank only Liquid reads as blank, brings it one tag over the limit', + 'email_body_too_many_tags', + () => + storedTemplateRendering( + { html: `{%\u180Eliquid\n${'echo variables.chunk\n'.repeat(EMAIL_BODY_TAG_LIMIT)}%}` }, + 'x' + ), + ], + [ + 'has a body full of Liquid tags, which would take seconds to parse', + 'email_body_too_many_tags', + () => storedTemplateRendering({ html: '{{ x }}'.repeat(100_000) }, 'x'), + ], + [ + 'has a subject full of Liquid tags, which would take seconds to parse', + 'email_body_too_many_tags', + () => storedTemplateRendering({ subject: '{{ x }}'.repeat(100_000) }, 'x'), + ], + [ + 'has a UTM value one character over the length limit', + 'utm_value_too_long', + () => storedUtmValue('x'.repeat(UTM_VALUE_LIMIT_CHARS + 1)), + ], + [ + 'has a UTM value full of Liquid tags, which would take seconds to parse', + 'utm_value_too_long', + () => storedUtmValue('{{ x }}'.repeat(100_000)), + ], + ])( + 'refuses a send with files whose email step %s, queues nothing, and frees its key', + async (_name, error, arrange) => { + const options = await arrange() + const key = newKey() + + const refused = await sendWithFiles(key, [reference(invoice)], options) + const retried = await sendWithFiles(key, [reference(invoice)], options) + + expect(refused.status).toBe(400) + expect(refused.body).toEqual({ + error, + reason: expect.stringContaining('"function_email"'), + fix: expect.any(String), + }) + expect(retried.body.error).toBe(error) + expect(await queuedIds()).toEqual([]) + } + ) + + it('accepts a send with files whose email editor design state holds more Liquid tags than the step may render', async () => { + const inputs = storedEmail({ html: '

{{ variables.chunk }}

' }) + inputs.email.value.design = { body: '{{ variables.chunk }}'.repeat(EMAIL_BODY_TAG_LIMIT + 1) } + const { id: flowId } = await insertHogFlow(transactionalFlow('api', 'active', 'transactional', inputs)) + + const res = await sendWithFiles(newKey(), [reference(invoice)], { + flowId, + variables: { chunk: 'x' }, + acceptedSend: { tracking: { captureEngagement: false } }, + }) + + expect(res.status).toBe(202) + expect(await queuedIds()).toEqual([res.body.invocation_id]) + }) + + it('accepts a send without files whose email step body holds more Liquid tags than a send with files may', async () => { + const options = await storedTemplateRendering( + { html: '{{ variables.chunk }}'.repeat(EMAIL_BODY_TAG_LIMIT + 1) }, + 'x' + ) + + const res = await sendWithFiles(newKey(), [], options) - const withoutFiles = await sendAccepted(hogFlow.id, { - ...body, - idempotency_key: idempotencyKey, - send: codeSend, + expect(res.status).toBe(202) + expect(await queuedIds()).toEqual([res.body.invocation_id]) }) - expect(withoutFiles.status).toEqual(202) }) it.each<[string, () => { payload: Record; verify?: () => void }]>([ diff --git a/nodejs/src/cdp/cdp-api.ts b/nodejs/src/cdp/cdp-api.ts index ff6cb30fce73..6691381e21b1 100644 --- a/nodejs/src/cdp/cdp-api.ts +++ b/nodejs/src/cdp/cdp-api.ts @@ -46,8 +46,11 @@ import { HOGFLOW_BATCH_RESOLVE_QUEUE, serializeResolverState, } from './services/hogflows/batch-resolver.types' +import { EmailStep, filesWorkflowShape } from './services/hogflows/files-workflow-shape' import { HogFlowExecutorService, createHogFlowInvocation } from './services/hogflows/hogflow-executor.service' +import { HogFlowFunctionsService } from './services/hogflows/hogflow-functions.service' import { HogFlowManagerService } from './services/hogflows/hogflow-manager.service' +import { hogFlowPersonLookup, loadHogFlowPerson } from './services/hogflows/hogflow-person' import { checkHogFlowQuotaLimits } from './services/hogflows/hogflow-quota-limiting' import { TRANSACTIONAL_EMAIL_EVENT } from './services/hogflows/hogflow-sensitivity' import { matchesWaitUntilCondition } from './services/hogflows/hogflow-utils' @@ -63,21 +66,34 @@ import { TransactionalEmailRequest, TransactionalEmailRequestSchema, TransactionalEmailStageTimer, + acceptedSendEmailFields, isRecipientFormatIssue, transactionalEmailRecipients, transactionalEmailRequestsTotal, } from './services/hogflows/transactional-email' import { InvocationResultsService } from './services/invocation-results.service' import { JobQueue } from './services/job-queue/job-queue.interface' -import { GroupsManagerService } from './services/managers/groups-manager.service' +import { GroupsManagerService, GroupsMap } from './services/managers/groups-manager.service' import { HogFunctionManagerService } from './services/managers/hog-function-manager.service' +import { PersonsManagerService } from './services/managers/persons-manager.service' import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service' -import { AttachmentUploads, countUploadRequest } from './services/messaging/attachment-uploads' +import { + ATTACHMENT_VERIFICATION_DEADLINE_MS, + AttachmentUploads, + MessageBodies, + MessageBodySizing, + SendRefusal, + countUploadRequest, +} from './services/messaging/attachment-uploads' 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' import { EmailTrackingService } from './services/messaging/email-tracking.service' import { EmailValidationService } from './services/messaging/email-validation.service' +import { utmTagsFor } from './services/messaging/helpers/email-html' +import { sesHtmlMaxBytes } from './services/messaging/helpers/ses-html-max-bytes' +import { EmailTrackingCodeSigner } from './services/messaging/helpers/tracking-code' +import { UTM_VALUE_LIMIT_CHARS, utmValueOverLimit } from './services/messaging/helpers/utm' import { RecipientTokensService } from './services/messaging/recipient-tokens.service' import { HogWatcherService, @@ -88,7 +104,14 @@ import { import { NativeDestinationExecutorService } from './services/native-destination-executor.service' import { SegmentDestinationExecutorService } from './services/segment-destination-executor.service' import { HOG_FUNCTION_TEMPLATES } from './templates' -import { HogFunctionInvocationGlobals, HogFunctionType, MinimalLogEntry } from './types' +import { + CyclotronJobInvocationHogFlow, + CyclotronJobInvocationHogFunction, + CyclotronPerson, + HogFunctionInvocationGlobals, + HogFunctionType, + MinimalLogEntry, +} from './types' import { convertToHogFunctionInvocationGlobals, isConvertibleClickHouseEvent, @@ -97,10 +120,13 @@ import { sanitizeLogMessage, } from './utils' import { recordRequestOrigins } from './utils/cdp-fetch' +import { deadlineAfter, throwIfDeadlinePassed } from './utils/deadline' import { dualRead, dualWrite } from './utils/dual-store' +import { isHogLimitError, isHogVMError } from './utils/hog-error-classification' import { convertToHogFunctionFilterGlobal } from './utils/hog-function-filtering' import { buildHogFunctionInvocations } from './utils/invocation-utils' import { PosthogJwtAudience } from './utils/jwt-utils' +import { EMAIL_BODY_TAG_LIMIT, countLiquidTags } from './utils/liquid' import { ScopedServiceJwt } from './utils/scoped-service-jwt' import { parseWorkflowStepDispatchKey } from './utils/workflow-step-dispatch-key' @@ -176,6 +202,57 @@ function parseRealtimeCohort(value: unknown): BatchResolverState['realtimeCohort return { id: id as number, teamId: team_id as number, waitUntil: waitUntil.toISOString() } } +const NO_ATTACHMENTS = { attachment_count: 0, attachment_bytes: 0 } + +const emailRenderLimitExceeded = (emailStep: EmailStep): SendRefusal => ({ + status: 400, + body: { + error: 'email_render_limit_exceeded', + reason: `Rendering the "${emailStep.name}" email step went over a template limit on time or length, so the size of this email with its files could not be checked.`, + fix: 'Simplify the template of this email step, or send this email without files.', + }, +}) + +const utmValueTooLong = (emailStep: EmailStep, key: string): SendRefusal => ({ + status: 400, + body: { + error: 'utm_value_too_long', + reason: `The ${key} value of the "${emailStep.name}" email step is longer than ${UTM_VALUE_LIMIT_CHARS} characters, so the size of this email with its files could not be checked.`, + fix: `Shorten the ${key} value of this email step to ${UTM_VALUE_LIMIT_CHARS} characters or fewer, or send this email without files.`, + }, +}) + +const emailBodyTooManyTags = (emailStep: EmailStep): SendRefusal => ({ + status: 400, + body: { + error: 'email_body_too_many_tags', + reason: `The "${emailStep.name}" email step has more than ${EMAIL_BODY_TAG_LIMIT} Liquid tags, so the size of this email with its files could not be checked.`, + fix: `Use ${EMAIL_BODY_TAG_LIMIT} Liquid tags or fewer in this email step, or send this email without files.`, + }, +}) + +const liquidTagsIn = (value: unknown): number => { + if (typeof value === 'string') { + return countLiquidTags(value) + } + if (value && typeof value === 'object') { + return Object.values(value).reduce((total: number, item) => total + liquidTagsIn(item), 0) + } + return 0 +} + +const liquidTagsInStep = (emailStep: EmailStep): number => + Object.entries(emailStep.config.inputs).reduce((total: number, [key, input]) => { + if (input?.templating !== 'liquid') { + return total + } + const rendered = + key === 'email' && input.value && typeof input.value === 'object' + ? { ...input.value, design: undefined } + : input.value + return total + liquidTagsIn(rendered) + }, 0) + const ACCEPTED_SEND_FINGERPRINT_MARKER = 'accepted_send' const transactionalEmailFingerprint = ( @@ -191,6 +268,8 @@ const transactionalEmailFingerprint = ( type CallerChosenSender = Partial> +type WorkerRenderContext = { person: CyclotronPerson | undefined; groups: GroupsMap } + const withSenderTheCallerChose = ({ from, ...send @@ -206,6 +285,14 @@ const senderTheCallerChose = (from: AcceptedSend['from']): CallerChosenSender | return from.email ? { email: from.email, name: from.name } : { integrationId: from.integrationId, name: from.name } } +const sendBodies = (invocation: CyclotronJobInvocationHogFlow): MessageBodies => { + const sendFields = acceptedSendEmailFields(invocation.state.send) + return { + bodyBytes: [sendFields.html ?? '', sendFields.text ?? ''].map((body) => Buffer.byteLength(body)), + subject: sendFields.subject ?? '', + } +} + export class CdpApi { private hogExecutorAsync: HogExecutorAsyncService private hogInputsService: HogInputsService @@ -216,6 +303,7 @@ export class CdpApi { private hogFlowManager: HogFlowManagerService private hogFlowExecutor: HogFlowExecutorService + private hogFlowFunctionsService: HogFlowFunctionsService private hogWatcher: HogWatcherService private hogWatcherMirror: HogWatcherService private hogTransformer: HogTransformerService @@ -225,9 +313,11 @@ export class CdpApi { private hogQueue: JobQueue private hogflowQueue: JobQueue private emailTrackingService: EmailTrackingService + private emailTrackingCodeSigner: EmailTrackingCodeSigner private recipientTokensService: RecipientTokensService private batchExportHogFunctionService: BatchExportHogFunctionService private groupsManager: GroupsManagerService + private personsManager: PersonsManagerService private batchResolverProducer: CyclotronV2JobProducer | null // Scoped auth for the reschedule_parked and cancel routes (exempted from the shared // internal-secret middleware): Django mints per-call JWTs pinned to a team + workflow, @@ -263,12 +353,14 @@ export class CdpApi { this.hogExecutorAsync = services.hogExecutorAsync this.hogInputsService = services.hogInputsService this.hogFlowExecutor = services.hogFlowExecutor + this.hogFlowFunctionsService = services.hogFlowFunctionsService this.nativeDestinationExecutorService = services.nativeDestinationExecutorService this.segmentDestinationExecutorService = services.segmentDestinationExecutorService this.hogWatcher = services.hogWatcher this.hogWatcherMirror = services.hogWatcherMirror this.invocationResultsService = services.invocationResultsService this.emailStatusService = services.emailStatusService + this.personsManager = services.personsManager // API-only services. The hog-transformer's monitoring service reuses the same // resolved outputs registry as the core CDP services — no separate construction. @@ -284,6 +376,10 @@ export class CdpApi { jobQueues, new IdempotencyKeys(deps.idempotencyValkey ?? null, 'workflow-webhook') ) + this.emailTrackingCodeSigner = new EmailTrackingCodeSigner( + config.ENCRYPTION_SALT_KEYS, + config.CDP_EMAIL_TRACKING_URL + ) this.emailTrackingService = services.emailTrackingService this.groupsManager = new GroupsManagerService(deps.teamManager, deps.groupRepository) this.batchExportHogFunctionService = new BatchExportHogFunctionService( @@ -1269,6 +1365,7 @@ export class CdpApi { private postTransactionalEmail = (schema: z.ZodType, tokenClaims: Record) => async (req: ModifiedRequest, res: express.Response): Promise => { + const verificationDeadline = deadlineAfter(ATTACHMENT_VERIFICATION_DEADLINE_MS) const stages = new TransactionalEmailStageTimer() const reply = (outcome: TransactionalEmailOutcome, status: number, body: Record) => { transactionalEmailRequestsTotal.inc({ outcome }) @@ -1310,6 +1407,10 @@ export class CdpApi { return reply('invalid_request', 400, { error: parsed.error.message }) } const { to, distinct_id: distinctId, variables, idempotency_key: idempotencyKey, send } = parsed.data + const references = send?.attachments ?? [] + if (references.length > 0 && !this.attachmentUploads.isAllowed(parseInt(req.params.team_id))) { + return reply('field_not_accepted', 400, { error: 'field_not_accepted' }) + } stages.start('load') const teamLookup = this.deps.teamManager.getTeam(parseInt(req.params.team_id)).catch(() => null) @@ -1327,6 +1428,7 @@ export class CdpApi { if (hogFlow.trigger?.type !== 'api') { return reply('not_sendable', 400, { error: 'not_sendable' }) } + const filesShape = references.length > 0 ? filesWorkflowShape(hogFlow) : null stages.start('build') const event: HogFunctionInvocationGlobals['event'] = { @@ -1357,7 +1459,8 @@ export class CdpApi { const invocation = createHogFlowInvocation(triggerGlobals, hogFlow, filterGlobals) if (send) { - invocation.state.send = send + const { attachments: _references, ...queuedSend } = send + invocation.state.send = queuedSend } // The run skips the trigger pipeline that counts runs for event-triggered workflows, so the email // worker counts it instead. @@ -1370,9 +1473,13 @@ export class CdpApi { const idempotencyKeys = this.transactionalEmailIdempotencyKeys const fingerprint = transactionalEmailFingerprint(hogFlow.id, parsed.data, send) + const releaseKey = async () => { + if (idempotencyKey) { + stages.start('idempotency') + await idempotencyKeys.release(team.id, idempotencyKey, invocation.id) + } + } 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. stages.start('idempotency') const reservation = await idempotencyKeys.reserve( team.id, @@ -1386,6 +1493,8 @@ export class CdpApi { status: 'queued', invocation_id: reservation.invocationId, replayed: true, + ...NO_ATTACHMENTS, + ...reservation.replayFields, }) case 'in_flight': return reply('idempotency_key_in_flight', 409, { error: 'idempotency_key_in_flight' }) @@ -1395,6 +1504,11 @@ export class CdpApi { return reply('idempotency_unavailable', 503, { error: 'idempotency_unavailable' }) } } + if (filesShape && 'refused' in filesShape) { + await releaseKey() + return reply(filesShape.refused.error, 400, filesShape.refused) + } + let attachmentTotals = NO_ATTACHMENTS try { stages.start('checks') const refusal = await this.refuseTransactionalEmail( @@ -1403,12 +1517,32 @@ export class CdpApi { transactionalEmailRecipients(parsed.data) ) if (refusal) { - if (idempotencyKey) { - stages.start('idempotency') - await idempotencyKeys.release(team.id, idempotencyKey, invocation.id) - } + await releaseKey() return reply(refusal.outcome, refusal.status, { error: refusal.outcome }) } + if (send && filesShape) { + stages.start('attachments') + const files = await this.attachmentUploads.acceptSend( + team.id, + references.map((ref) => ({ + sha256: ref.sha256, + filename: ref.filename, + contentType: ref.content_type, + contentId: ref.content_id, + })), + (deadline) => this.messageBodyBytes(hogFlow, invocation, filesShape.emailStep, deadline), + verificationDeadline + ) + if ('refused' in files) { + await releaseKey() + return reply(files.refused.body.error, files.refused.status, files.refused.body) + } + invocation.state.send = { ...invocation.state.send!, attachments: files.accepted.files } + attachmentTotals = { + attachment_count: files.accepted.files.length, + attachment_bytes: files.accepted.attachmentBytes, + } + } stages.start('queue_insert') this.invocationResultsService.invocationResultsRowsService.queueLifecycleRow(invocation, 'running') try { @@ -1419,16 +1553,18 @@ export class CdpApi { } cdpEmailQueuedTotal.labels(priorityClass).inc() } catch (error) { - // Nothing was queued, so the caller's retry with the same key must be able to send. - if (idempotencyKey) { - stages.start('idempotency') - await idempotencyKeys.release(team.id, idempotencyKey, invocation.id) - } + await releaseKey() throw error } if (idempotencyKey) { stages.start('idempotency') - await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id) + await idempotencyKeys.markQueued( + team.id, + idempotencyKey, + fingerprint, + invocation.id, + attachmentTotals + ) } stages.start('metrics_flush') this.emailStatusService.queue({ @@ -1449,7 +1585,12 @@ export class CdpApi { await this.invocationResultsService.invocationResultsRowsService.handOffQueuedRows() void produced.catch(warnTelemetryFlushFailed) - return reply('queued', 202, { status: 'queued', invocation_id: invocation.id, replayed: false }) + return reply('queued', 202, { + status: 'queued', + invocation_id: invocation.id, + replayed: false, + ...attachmentTotals, + }) } catch (e) { logger.error('Error queueing transactional email', { error: e instanceof Error ? e.message : String(e), @@ -1460,6 +1601,112 @@ export class CdpApi { } } + private async messageBodyBytes( + hogFlow: HogFlow, + invocation: CyclotronJobInvocationHogFlow, + emailStep: EmailStep, + deadline: AbortSignal + ): Promise { + const longUtmValue = emailStep.config.utm_tags_enabled + ? utmValueOverLimit(emailStep.config.utm_params) + : undefined + if (longUtmValue) { + return { refused: utmValueTooLong(emailStep, longUtmValue) } + } + if (liquidTagsInStep(emailStep) > EMAIL_BODY_TAG_LIMIT) { + return { refused: emailBodyTooManyTags(emailStep) } + } + const renderContext = await this.workerRenderContext(hogFlow, invocation) + deadline.throwIfAborted() + try { + return await this.emailStepBodies(hogFlow, invocation, emailStep, renderContext, deadline) + } catch (error) { + if (isHogLimitError(error)) { + return { refused: emailRenderLimitExceeded(emailStep) } + } + if (!isHogVMError(error)) { + throw error + } + logger.warn('⚠️', 'Could not render the email step of a send with files', { + team_id: hogFlow.team_id, + hog_flow_id: hogFlow.id, + error: error instanceof Error ? error.name : 'UnknownError', + }) + return sendBodies(invocation) + } + } + + private async workerRenderContext( + hogFlow: HogFlow, + invocation: CyclotronJobInvocationHogFlow + ): Promise { + const [person, groups] = await Promise.all([ + this.recipientPerson(hogFlow, invocation), + this.groupsManager.getGroupsForEvent( + hogFlow.team_id, + invocation.state.event.properties ?? {}, + `${this.config.SITE_URL}/project/${hogFlow.team_id}` + ), + ]) + return { person, groups } + } + + private async recipientPerson( + hogFlow: HogFlow, + invocation: CyclotronJobInvocationHogFlow + ): Promise { + const lookup = hogFlowPersonLookup(hogFlow, invocation.state) + return lookup ? await loadHogFlowPerson(this.personsManager, hogFlow, lookup) : undefined + } + + private async emailStepBodies( + hogFlow: HogFlow, + invocation: CyclotronJobInvocationHogFlow, + step: EmailStep, + { person, groups }: WorkerRenderContext, + deadline: AbortSignal + ): Promise { + const hogFunction = await this.hogFlowFunctionsService.buildHogFunction(hogFlow, step.config, step.name) + deadline.throwIfAborted() + const atEmailStep = { + ...invocation, + state: { ...invocation.state, currentAction: { id: step.id, startedAtTimestamp: Date.now() } }, + } + const rendered = await this.hogFlowFunctionsService.buildHogFunctionInvocation( + atEmailStep, + hogFunction, + { event: invocation.state.event, person, groups, variables: invocation.state.variables }, + deadline + ) + deadline.throwIfAborted() + const email = { ...rendered.state.globals.inputs?.email, ...acceptedSendEmailFields(invocation.state.send) } + return { + bodyBytes: [ + typeof email.html === 'string' && email.html + ? this.sesHtmlMaxBytes(rendered, email.html, email.preheader, deadline) + : 0, + typeof email.text === 'string' ? Buffer.byteLength(email.text) : 0, + ], + subject: typeof email.subject === 'string' ? email.subject : '', + } + } + + private sesHtmlMaxBytes( + rendered: CyclotronJobInvocationHogFunction, + html: string, + preheader: unknown, + deadline: AbortSignal + ): number { + const utmTags = utmTagsFor(rendered, () => undefined, deadline) + throwIfDeadlinePassed(deadline) + return sesHtmlMaxBytes(html, { + utmTags, + invocation: rendered, + signer: this.emailTrackingCodeSigner, + preheader: typeof preheader === 'string' ? preheader : undefined, + }) + } + private postAttachmentUploads = async (req: ModifiedRequest, res: express.Response): Promise => { if (!this.transactionalEmailJwt.enabled) { countUploadRequest('not_configured') diff --git a/nodejs/src/cdp/consumers/cdp-cyclotron-worker-email.consumer.serial.test.ts b/nodejs/src/cdp/consumers/cdp-cyclotron-worker-email.consumer.serial.test.ts index d52eabbcf5e6..4dcfd18a51e5 100644 --- a/nodejs/src/cdp/consumers/cdp-cyclotron-worker-email.consumer.serial.test.ts +++ b/nodejs/src/cdp/consumers/cdp-cyclotron-worker-email.consumer.serial.test.ts @@ -3,13 +3,14 @@ import { mockProducerObserver } from '~/tests/helpers/mocks/producer.mock' import { SESv2Client } from '@aws-sdk/client-sesv2' import { DateTime } from 'luxon' import { randomUUID } from 'node:crypto' -import { AddressInfo, Socket, connect, createServer } from 'node:net' import { Client, Pool } from 'pg' import { Histogram, register } from 'prom-client' import { FixtureHogFlowBuilder } from '~/cdp/_tests/builders/hogflow.builder' import { BucketProxy, + EmailQueueOfOneTest, + PostgresProxy, StoredFile, attachmentStoreConfig, deliveredTo, @@ -23,7 +24,7 @@ import { HogFlow } from '~/cdp/schema/hogflow' import { CyclotronV2Manager } from '~/cdp/services/cyclotron-v2' import { createHogFlowInvocation } from '~/cdp/services/hogflows/hogflow-executor.service' import { TRANSACTIONAL_EMAIL_EVENT } from '~/cdp/services/hogflows/hogflow-sensitivity' -import { CONSUMER_STOP_DEADLINE_MS, CyclotronJobQueuePostgresV2 } from '~/cdp/services/job-queue/job-queue-postgres-v2' +import { CONSUMER_STOP_DEADLINE_MS } from '~/cdp/services/job-queue/job-queue-postgres-v2' import { LONGEST_WAIT_FOR_ROOM_MS } from '~/cdp/services/messaging/attachment-horizon' import { InFlightLimits, @@ -31,7 +32,7 @@ import { LARGEST_SEND_FILE_BYTES, estimateSend, } from '~/cdp/services/messaging/in-flight-registry' -import { CyclotronJobInvocation, CyclotronJobInvocationResult, CyclotronJobQueueKind } from '~/cdp/types' +import { CyclotronJobInvocation } from '~/cdp/types' import { convertToHogFunctionFilterGlobal } from '~/cdp/utils/hog-function-filtering' import { KAFKA_EMAIL_STATUS } from '~/common/config/kafka-topics' import { closeHub, createHub } from '~/common/utils/db/hub' @@ -55,151 +56,6 @@ jest.mock('node:dns/promises', () => ({ })), })) -class EmailQueueOfOneTest extends CyclotronJobQueuePostgresV2 { - constructor( - private readonly queueName: string, - config: ConstructorParameters[1] - ) { - super(10, config) - } - - public override startAsConsumer( - _queue: CyclotronJobQueueKind, - consumeBatch: (invocations: CyclotronJobInvocation[]) => Promise<{ backgroundTask: Promise }> - ): Promise { - return super.startAsConsumer(this.queueName as CyclotronJobQueueKind, (invocations) => - consumeBatch(invocations.map((invocation) => ({ ...invocation, queue: 'email' }))) - ) - } - - public override queueInvocations(invocations: CyclotronJobInvocation[]): Promise { - return super.queueInvocations(invocations.map((invocation) => this.onOwnQueue(invocation))) - } - - public override queueInvocationResults(results: CyclotronJobInvocationResult[]): Promise { - return super.queueInvocationResults( - results.map((result) => ({ ...result, invocation: this.onOwnQueue(result.invocation) })) - ) - } - - private onOwnQueue(invocation: CyclotronJobInvocation): CyclotronJobInvocation { - const queue = invocation.queue === 'email' ? this.queueName : `${this.queueName}/${invocation.queue}` - return { ...invocation, queue: queue as CyclotronJobQueueKind } - } -} - -type HeldQuery = { release: () => void; drop: () => void } - -const SSL_REQUEST = 80877103 -const GSS_ENCRYPTION_REQUEST = 80877104 -const ENDS_A_REQUEST = new Set(['S', 'Q', 'X', 'p', 'H', 'c', 'f']) - -class CyclotronProxy { - url = '' - private readonly waiting: { matches: (request: string) => boolean; caught: (held: HeldQuery) => void }[] = [] - private readonly holding = new Set() - private readonly sockets = new Set() - private readonly server = createServer((client) => this.forward(client)) - - static async start(upstream: URL): Promise { - const proxy = new CyclotronProxy(upstream) - await new Promise((resolve) => proxy.server.listen(0, '127.0.0.1', resolve)) - const url = new URL(upstream) - url.host = `127.0.0.1:${(proxy.server.address() as AddressInfo).port}` - proxy.url = url.toString() - return proxy - } - - private constructor(private readonly upstream: URL) {} - - hold(matches: (request: string) => boolean): Promise { - return new Promise((caught) => this.waiting.push({ matches, caught })) - } - - releaseAll(): void { - this.waiting.splice(0) - for (const held of this.holding) { - held.release() - } - } - - async close(): Promise { - for (const socket of this.sockets) { - socket.destroy() - } - await new Promise((resolve) => this.server.close(resolve)) - } - - private forward(client: Socket): void { - const upstream = connect(Number(this.upstream.port || 5432), this.upstream.hostname) - for (const socket of [client, upstream]) { - this.sockets.add(socket) - socket.on('error', () => undefined) - socket.on('close', () => { - client.destroy() - upstream.destroy() - }) - } - upstream.pipe(client) - - let unread = Buffer.alloc(0) - let request: Buffer[] = [] - let startedUp = false - let paused = false - const forwardRequests = (): void => { - while (!paused) { - const typeBytes = startedUp ? 1 : 0 - if (unread.length < typeBytes + 4) { - return - } - const length = typeBytes + unread.readInt32BE(typeBytes) - if (unread.length < length) { - return - } - const message = unread.subarray(0, length) - unread = unread.subarray(length) - if (!startedUp) { - const code = message.readInt32BE(4) - startedUp = code !== SSL_REQUEST && code !== GSS_ENCRYPTION_REQUEST - upstream.write(message) - continue - } - request.push(message) - if (!ENDS_A_REQUEST.has(String.fromCharCode(message[0]))) { - continue - } - const whole = Buffer.concat(request) - request = [] - const index = this.waiting.findIndex(({ matches }) => matches(whole.toString('latin1'))) - if (index === -1) { - upstream.write(whole) - continue - } - const [{ caught }] = this.waiting.splice(index, 1) - paused = true - const held: HeldQuery = { - release: () => { - this.holding.delete(held) - upstream.write(whole) - paused = false - forwardRequests() - }, - drop: () => { - this.holding.delete(held) - client.destroy() - }, - } - this.holding.add(held) - caught(held) - } - } - client.on('data', (chunk: Buffer) => { - unread = Buffer.concat([unread, chunk]) - forwardRequests() - }) - } -} - const completes = (request: string): boolean => request.includes("SET status = 'completed'") const fails = (request: string): boolean => request.includes("SET status = 'failed'") const dequeues = (request: string): boolean => request.includes("SET status = 'running'") @@ -238,7 +94,7 @@ describe('CdpCyclotronWorkerEmail on Cyclotron Postgres', () => { let recipient: string let ses: FakeSes let bucket: BucketProxy - let cyclotron: CyclotronProxy + let cyclotron: PostgresProxy let workers: CdpCyclotronWorkerEmail[] let producer: EmailQueueOfOneTest let invoice: StoredFile @@ -278,7 +134,7 @@ describe('CdpCyclotronWorkerEmail on Cyclotron Postgres', () => { [hub.SES_UNTRACKED_CONFIGURATION_SET]: { tracking: false }, }) bucket = await BucketProxy.start() - cyclotron = await CyclotronProxy.start(new URL(hub.CYCLOTRON_NODE_DATABASE_URL!)) + cyclotron = await PostgresProxy.start(new URL(hub.CYCLOTRON_NODE_DATABASE_URL!)) mockProducerObserver.resetKafkaProducer() integration = await insertIntegration(hub.postgres, team.id, { diff --git a/nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts b/nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts index 033ec59fab31..b3d645fb6399 100644 --- a/nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts +++ b/nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts @@ -1,19 +1,17 @@ import { DateTime } from 'luxon' import { HogFlow } from '~/cdp/schema/hogflow' -import { hasPushActions } from '~/cdp/services/hogflows/hogflow-utils' +import { + hogFlowPersonLookup, + loadHogFlowPerson, + resolvesByRepointedPerson, +} from '~/cdp/services/hogflows/hogflow-person' import { instrumented } from '~/common/tracing/tracing-utils' import { logger } from '~/common/utils/logger' import { PluginsServerConfig } from '~/types' -import { isRowScopedTrigger } from '../schema/hogflow' import { JobQueue } from '../services/job-queue/job-queue.interface' -import { - CyclotronJobInvocation, - CyclotronJobInvocationHogFlow, - CyclotronJobInvocationResult, - CyclotronPerson, -} from '../types' +import { CyclotronJobInvocation, CyclotronJobInvocationHogFlow, CyclotronJobInvocationResult } from '../types' import { convertToHogFunctionFilterGlobal } from '../utils/hog-function-filtering' import { createInvocationResult } from '../utils/invocation-utils' import { CdpConsumerBaseDeps } from './cdp-base.consumer' @@ -24,16 +22,6 @@ type LoadHogFlowsResult = { canceledResults: CyclotronJobInvocationResult[] } -// A transactional email's caller holds only a send key, which grants no access to person data, so its -// template never sees person properties, or the display name derived from them. The person stays attached -// so the send is recorded against them. -function personForFlow( - hogFlow: HogFlow, - person: CyclotronPerson | null | undefined -): CyclotronPerson | null | undefined { - return person && hogFlow.trigger?.type === 'api' ? { ...person, properties: {}, name: '' } : person -} - export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker { protected override name = 'CdpCyclotronWorkerHogFlow' @@ -138,54 +126,15 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker { const hogFlowInvocationState = item.state as CyclotronJobInvocationHogFlow['state'] - // Row-scoped invocations (a warehouse row, a Slack message, a GitHub event) don't have - // a real person — the delivery is the unit of work and person-dependent steps no-op for - // these flows. Explicitly skip the person lookup rather than relying on - // event.distinct_id being empty so future changes to the synthetic event shape don't - // accidentally re-enable the lookup. - const isRowScoped = isRowScopedTrigger(hogFlow.trigger) - // Account-audience batch invocations carry the account's group key as - // event.distinct_id; resolving it as a person distinct_id would attach an - // unrelated person to the run. Accounts have no person — skip the lookup. - // The state stamp wins over the live trigger, which may have been edited to a - // person audience while these children were queued; the trigger check remains - // as a fallback for jobs enqueued before the stamp existed. - const isAccountAudience = - hogFlowInvocationState.accountAudience === true || - (hogFlow.trigger?.type === 'batch' && hogFlow.trigger.filters?.audience_type === 'accounts') - // The matcher wrote this job's personId anchor: a merge repointed the distinct_id onto a - // survivor, or the distinct_id acquired its first person. Resolve by that personId so the step - // reads the right person — resolving by the distinct_id would hit its stale ~1min cache entry - // (the pre-merge person, or none at all) and e.g. drop an email. - const resolveByRepointedPerson = - hogFlowInvocationState.personIdRepointed === true && !!hogFlowInvocationState.personId + const personLookup = hogFlowPersonLookup(hogFlow, hogFlowInvocationState) // One-shot: consume the flag on this wake-resolution only. Later steps fall back to normal // distinct_id-first resolution, which self-heals to the latest survivor if the distinct_id is // repointed again (a second merge onto a non-wait step is out of processMoveBatch's scope). - if (resolveByRepointedPerson) { + if (resolvesByRepointedPerson(hogFlowInvocationState)) { delete hogFlowInvocationState.personIdRepointed } - const personIdOrDistinctId = - isRowScoped || isAccountAudience - ? undefined - : resolveByRepointedPerson - ? hogFlowInvocationState.personId - : hogFlowInvocationState.event.distinct_id || hogFlowInvocationState.personId - const kind = - resolveByRepointedPerson || !hogFlowInvocationState.event.distinct_id ? 'person_id' : 'distinct_id' - - // A push cannot be recalled, and the cached person can predate a device that - // unregistered since the last read, so a flow that sends push bypasses the cache here. - // A wait woken by a match advances without re-evaluating, so the steps after it would - // otherwise read the person cached when the wait parked, without the write that woke it. - const forceFreshPerson = - hasPushActions(hogFlow.actions) || hogFlowInvocationState.currentAction?.eventMatched === true - const [resolvedPerson, groups] = await Promise.all([ - personIdOrDistinctId - ? this.personsManager.getCyclotronPerson(hogFlow.team_id, personIdOrDistinctId, kind, { - forceFresh: forceFreshPerson, - }) - : undefined, + const [person, groups] = await Promise.all([ + personLookup ? loadHogFlowPerson(this.personsManager, hogFlow, personLookup) : undefined, this.groupsManager.getGroupsForEvent( hogFlow.team_id, hogFlowInvocationState.event.properties, @@ -193,8 +142,6 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker { ), ]) - const person = personForFlow(hogFlow, resolvedPerson) - if (!person && hogFlow.trigger?.type === 'event') { logger.warn('⚠️', 'Person not found for hog flow invocation', { hogFlowId: hogFlow.id, @@ -221,7 +168,7 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker { const filterGlobals = convertToHogFunctionFilterGlobal({ event: hogFlowInvocationState.event, - person: person ?? undefined, + person, groups, variables: hogFlowInvocationState.variables || {}, }) @@ -230,24 +177,22 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker { ...item, state: hogFlowInvocationState, hogFlow, - person: person ?? undefined, + person, groups, filterGlobals, } - if (personIdOrDistinctId) { + if (personLookup) { loaded.refreshPerson = async () => { - const fresh = personForFlow( - hogFlow, - await this.personsManager.getCyclotronPerson(hogFlow.team_id, personIdOrDistinctId, kind, { - forceFresh: true, - }) - ) + const fresh = await loadHogFlowPerson(this.personsManager, hogFlow, { + ...personLookup, + forceFresh: true, + }) return { - person: fresh ?? undefined, + person: fresh, filterGlobals: convertToHogFunctionFilterGlobal({ event: hogFlowInvocationState.event, - person: fresh ?? undefined, + person: fresh, groups, variables: hogFlowInvocationState.variables || {}, }), diff --git a/nodejs/src/cdp/services/hog-executor.service.ts b/nodejs/src/cdp/services/hog-executor.service.ts index f0ee25eb01c0..379c6052ea5b 100644 --- a/nodejs/src/cdp/services/hog-executor.service.ts +++ b/nodejs/src/cdp/services/hog-executor.service.ts @@ -104,13 +104,15 @@ export class HogExecutorService { hogFunction: HogFunctionType, globals: HogFunctionInvocationGlobals, additionalInputs?: Record, - preferencesTokenSource?: PreferencesTokenSource + preferencesTokenSource?: PreferencesTokenSource, + signal?: AbortSignal ): Promise { return this.hogInputsService.buildInputsWithGlobals( hogFunction, globals, additionalInputs, - preferencesTokenSource + preferencesTokenSource, + signal ) } diff --git a/nodejs/src/cdp/services/hogflows/files-workflow-shape.test.ts b/nodejs/src/cdp/services/hogflows/files-workflow-shape.test.ts new file mode 100644 index 000000000000..a248bedb1d52 --- /dev/null +++ b/nodejs/src/cdp/services/hogflows/files-workflow-shape.test.ts @@ -0,0 +1,161 @@ +import { FixtureHogFlowBuilder, SimpleHogFlowRepresentation } from '~/cdp/_tests/builders/hogflow.builder' +import { HogFlow } from '~/cdp/schema/hogflow' + +import { filesWorkflowShape } from './files-workflow-shape' + +type Actions = SimpleHogFlowRepresentation['actions'] +type Edges = SimpleHogFlowRepresentation['edges'] +type Expected = { emailStep: string } | { error: string; step: string | null } + +const trigger = { type: 'trigger', config: { type: 'api' } } as const +const email = ( + category: 'transactional' | 'marketing' = 'transactional', + config: { skip_frequency_cap?: boolean } = {} +) => + ({ + type: 'function_email', + config: { template_id: 'template-email', inputs: {}, message_category_type: category, ...config }, + }) as const +const genericEmailFunction = { + type: 'function', + config: { template_id: 'template-email', inputs: {} }, +} as const +const delay = { type: 'delay', config: { delay_duration: '1h' } } as const +const branch = { type: 'conditional_branch' as const, config: { conditions: [{ filters: {} }] } } +const exit = { type: 'exit', config: {} } as const +const step = (from: string, to: string) => ({ from, to, type: 'continue' as const }) +const namedById = (actions: Actions): Actions => + Object.fromEntries(Object.entries(actions).map(([id, action]) => [id, { ...action, name: id }])) +const flowOf = (actions: Actions, edges: Edges, settings: Partial = {}): HogFlow => ({ + ...new FixtureHogFlowBuilder().withWorkflow({ actions: namedById(actions), edges }).build(), + ...settings, +}) + +const longChainAfterEmail = (length: number): { actions: Actions; edges: Edges } => { + const ids = Array.from({ length }, (_, index) => `after_${index}`) + return { + actions: { trigger, email: email(), ...Object.fromEntries(ids.map((id) => [id, delay])) }, + edges: [ + step('trigger', 'email'), + ...['email', ...ids].slice(0, -1).map((from, index) => step(from, ids[index])), + ], + } +} + +describe('filesWorkflowShape', () => { + const chain = longChainAfterEmail(20_000) + + it.each<[string, Actions, Edges, Expected, Partial?]>([ + [ + 'the generated workflow, a trigger, an email step and an exit', + { trigger, email: email(), exit }, + [step('trigger', 'email'), step('email', 'exit')], + { emailStep: 'email' }, + ], + [ + 'the generated workflow without its exit', + { trigger, email: email() }, + [step('trigger', 'email')], + { emailStep: 'email' }, + ], + [ + 'the generated workflow beside a step the trigger never reaches', + { trigger, email: email(), exit, later: delay }, + [step('trigger', 'email'), step('email', 'exit'), step('later', 'exit')], + { emailStep: 'email' }, + ], + [ + 'a generic function step that runs the email template', + { trigger, send: genericEmailFunction, exit }, + [step('trigger', 'send'), step('send', 'exit')], + { error: 'workflow_shape_not_supported', step: 'send' }, + ], + [ + 'a delay before the email', + { trigger, wait: delay, email: email() }, + [step('trigger', 'wait'), step('wait', 'email')], + { error: 'workflow_shape_not_supported', step: 'wait' }, + ], + [ + 'a delay after the email', + { trigger, email: email(), wait: delay, exit }, + [step('trigger', 'email'), step('email', 'wait'), step('wait', 'exit')], + { error: 'workflow_shape_not_supported', step: 'wait' }, + ], + [ + 'a second email', + { trigger, email: email(), second: email() }, + [step('trigger', 'email'), step('email', 'second')], + { error: 'workflow_shape_not_supported', step: 'second' }, + ], + [ + 'a branch before the email', + { trigger, choose: branch, email: email() }, + [step('trigger', 'choose'), { from: 'choose', to: 'email', type: 'branch', index: 0 }], + { error: 'workflow_shape_not_supported', step: 'choose' }, + ], + [ + 'an email step that loops back to itself', + { trigger, email: email() }, + [step('trigger', 'email'), step('email', 'email')], + { error: 'workflow_shape_not_supported', step: 'email' }, + ], + [ + 'a trigger that also leads straight to an exit', + { trigger, email: email(), done: exit }, + [step('trigger', 'email'), step('trigger', 'done')], + { error: 'workflow_shape_not_supported', step: 'done' }, + ], + [ + 'no email step', + { trigger, exit }, + [step('trigger', 'exit')], + { error: 'workflow_shape_not_supported', step: null }, + ], + [ + 'a marketing email step whose frequency cap can hold it', + { trigger, email: email('marketing'), exit }, + [step('trigger', 'email'), step('email', 'exit')], + { error: 'workflow_can_wait', step: 'email' }, + ], + [ + 'a marketing email step that skips the frequency cap, which the team sending cap can still hold', + { trigger, email: email('marketing', { skip_frequency_cap: true }), exit }, + [step('trigger', 'email'), step('email', 'exit')], + { error: 'workflow_can_wait', step: 'email' }, + ], + [ + 'the generated workflow under a sending rate limit, which can hold the email', + { trigger, email: email(), exit }, + [step('trigger', 'email'), step('email', 'exit')], + { error: 'workflow_can_wait', step: null }, + { email_sending_rate_limit: { count: 100, period: 'hour' } }, + ], + [ + 'the generated workflow with its sending rate limit cleared', + { trigger, email: email(), exit }, + [step('trigger', 'email'), step('email', 'exit')], + { emailStep: 'email' }, + { email_sending_rate_limit: null }, + ], + [ + 'a long chain of steps after the email', + chain.actions, + chain.edges, + { error: 'workflow_shape_not_supported', step: 'after_0' }, + ], + ])('judges a send with files to %s', (_name, actions, edges, expected, settings = {}) => { + const shape = filesWorkflowShape(flowOf(actions, edges, settings)) + + if ('emailStep' in expected) { + expect('emailStep' in shape && shape.emailStep.id).toBe(expected.emailStep) + return + } + expect(shape).toEqual({ + refused: { error: expected.error, reason: expect.any(String), fix: expect.any(String) }, + }) + if ('refused' in shape && expected.step) { + expect(shape.refused.reason).toContain(`"${expected.step}"`) + } + }) +}) diff --git a/nodejs/src/cdp/services/hogflows/files-workflow-shape.ts b/nodejs/src/cdp/services/hogflows/files-workflow-shape.ts new file mode 100644 index 000000000000..23ed8e1ab38f --- /dev/null +++ b/nodejs/src/cdp/services/hogflows/files-workflow-shape.ts @@ -0,0 +1,116 @@ +import { HogFlow, HogFlowAction } from '~/cdp/schema/hogflow' + +export type EmailStep = Extract + +export type FilesWorkflowRefusal = { + error: 'workflow_shape_not_supported' | 'workflow_can_wait' + reason: string + fix: string +} + +export type FilesWorkflowShape = { emailStep: EmailStep } | { refused: FilesWorkflowRefusal } + +const SUPPORTED_SHAPE_FIX = + 'Send files through a workflow that has a trigger, then one transactional email step, then an exit, or send this email without files.' + +const shapeNotSupported = (reason: string): FilesWorkflowShape => ({ + refused: { error: 'workflow_shape_not_supported', reason, fix: SUPPORTED_SHAPE_FIX }, +}) + +const stepDoesNotFit = (action: HogFlowAction): FilesWorkflowShape => + shapeNotSupported(`The "${action.name}" step does not fit the workflow shape a send with files supports.`) + +const marketingEmailCanWait = (emailStep: EmailStep): FilesWorkflowShape => ({ + refused: { + error: 'workflow_can_wait', + reason: `The "${emailStep.name}" step sends marketing email. Sending limits can hold it before it goes out, so its files could expire first.`, + fix: 'Send files through a workflow whose transactional email step follows the trigger directly, or send this email without files.', + }, +}) + +const sendingRateLimitCanWait = (): FilesWorkflowShape => ({ + refused: { + error: 'workflow_can_wait', + reason: "This workflow's sending rate limit can hold this send before its email goes out, so its files could expire first.", + fix: 'Send files through a workflow without a sending rate limit, or send this email without files.', + }, +}) + +const successorsById = (hogFlow: HogFlow): Map => { + const successors = new Map() + for (const edge of hogFlow.edges) { + const known = successors.get(edge.from) + if (known) { + known.push(edge.to) + } else { + successors.set(edge.from, [edge.to]) + } + } + return successors +} + +const reachableFromTriggerInOrder = (hogFlow: HogFlow, successors: Map): HogFlowAction[] => { + const actionsById = new Map(hogFlow.actions.map((action) => [action.id, action])) + const reached = new Set() + const ordered: HogFlowAction[] = [] + const toVisit = hogFlow.actions.filter((action) => action.type === 'trigger').map((action) => action.id) + for (let next = 0; next < toVisit.length; next++) { + const id = toVisit[next] + const action = actionsById.get(id) + if (reached.has(id) || !action) { + continue + } + reached.add(id) + ordered.push(action) + toVisit.push(...(successors.get(id) ?? [])) + } + return ordered +} + +const firstStepOutOfPlace = ( + successors: Map, + reachable: HogFlowAction[], + emailStep: EmailStep +): HogFlowAction | undefined => { + const reachableById = new Map(reachable.map((action) => [action.id, action])) + const outOfPlaceAfter = (action: HogFlowAction, fits: (next: HogFlowAction) => boolean) => + (successors.get(action.id) ?? []) + .map((id) => reachableById.get(id)) + .find((next): next is HogFlowAction => next !== undefined && !fits(next)) + for (const action of reachable) { + const outOfPlace = + action.type === 'trigger' + ? outOfPlaceAfter(action, (next) => next.id === emailStep.id) + : action.id === emailStep.id + ? outOfPlaceAfter(action, (next) => next.type === 'exit') + : undefined + if (outOfPlace) { + return outOfPlace + } + } + return undefined +} + +export const filesWorkflowShape = (hogFlow: HogFlow): FilesWorkflowShape => { + const successors = successorsById(hogFlow) + const reachable = reachableFromTriggerInOrder(hogFlow, successors) + let emailStep: EmailStep | undefined + for (const action of reachable) { + if (action.type === 'function_email' && !emailStep) { + emailStep = action + } else if (action.type !== 'trigger' && action.type !== 'exit') { + return stepDoesNotFit(action) + } + } + if (!emailStep) { + return shapeNotSupported('This workflow has no email step after its trigger.') + } + const outOfPlace = firstStepOutOfPlace(successors, reachable, emailStep) + if (outOfPlace) { + return stepDoesNotFit(outOfPlace) + } + if (emailStep.config.message_category_type !== 'transactional') { + return marketingEmailCanWait(emailStep) + } + return hogFlow.email_sending_rate_limit ? sendingRateLimitCanWait() : { emailStep } +} diff --git a/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts b/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts index 32b52699a9c8..e08430e822b7 100644 --- a/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts +++ b/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts @@ -100,7 +100,8 @@ export class HogFlowFunctionsService { async buildHogFunctionInvocation( invocation: CyclotronJobInvocationHogFlow, hogFunction: HogFunctionType, - globals: Omit + globals: Omit, + signal?: AbortSignal ): Promise { const teamId = invocation.hogFlow.team_id const projectUrl = `${this.siteUrl}/project/${teamId}` @@ -138,7 +139,8 @@ export class HogFlowFunctionsService { workflowId: invocation.hogFlow.id, invocationId: invocation.id, workflowVersion: invocation.hogFlow.version, - } + }, + signal ), timings: [], attempts: 0, diff --git a/nodejs/src/cdp/services/hogflows/hogflow-person.ts b/nodejs/src/cdp/services/hogflows/hogflow-person.ts new file mode 100644 index 000000000000..7fea9fa3f285 --- /dev/null +++ b/nodejs/src/cdp/services/hogflows/hogflow-person.ts @@ -0,0 +1,77 @@ +import { HogFlow, isRowScopedTrigger } from '~/cdp/schema/hogflow' +import { hasPushActions } from '~/cdp/services/hogflows/hogflow-utils' +import type { PersonsManagerService } from '~/cdp/services/managers/persons-manager.service' +import { CyclotronJobInvocationHogFlow, CyclotronPerson } from '~/cdp/types' + +export type HogFlowPersonLookup = { + id: string + kind: 'distinct_id' | 'person_id' + forceFresh: boolean +} + +type PersonReader = Pick + +type HogFlowState = CyclotronJobInvocationHogFlow['state'] + +// The matcher wrote this job's personId anchor: a merge repointed the distinct_id onto a +// survivor, or the distinct_id acquired its first person. Resolve by that personId so the step +// reads the right person — resolving by the distinct_id would hit its stale ~1min cache entry +// (the pre-merge person, or none at all) and e.g. drop an email. +export const resolvesByRepointedPerson = (state: HogFlowState): boolean => + state.personIdRepointed === true && !!state.personId + +export function hogFlowPersonLookup(hogFlow: HogFlow, state: HogFlowState): HogFlowPersonLookup | null { + // Row-scoped invocations (a warehouse row, a Slack message, a GitHub event) don't have + // a real person — the delivery is the unit of work and person-dependent steps no-op for + // these flows. Explicitly skip the person lookup rather than relying on + // event.distinct_id being empty so future changes to the synthetic event shape don't + // accidentally re-enable the lookup. + const isRowScoped = isRowScopedTrigger(hogFlow.trigger) + // Account-audience batch invocations carry the account's group key as + // event.distinct_id; resolving it as a person distinct_id would attach an + // unrelated person to the run. Accounts have no person — skip the lookup. + // The state stamp wins over the live trigger, which may have been edited to a + // person audience while these children were queued; the trigger check remains + // as a fallback for jobs enqueued before the stamp existed. + const isAccountAudience = + state.accountAudience === true || + (hogFlow.trigger?.type === 'batch' && hogFlow.trigger.filters?.audience_type === 'accounts') + if (isRowScoped || isAccountAudience) { + return null + } + const byRepointedPerson = resolvesByRepointedPerson(state) + const id = byRepointedPerson ? state.personId : state.event.distinct_id || state.personId + if (!id) { + return null + } + return { + id, + kind: byRepointedPerson || !state.event.distinct_id ? 'person_id' : 'distinct_id', + // A push cannot be recalled, and the cached person can predate a device that + // unregistered since the last read, so a flow that sends push bypasses the cache here. + // A wait woken by a match advances without re-evaluating, so the steps after it would + // otherwise read the person cached when the wait parked, without the write that woke it. + forceFresh: hasPushActions(hogFlow.actions) || state.currentAction?.eventMatched === true, + } +} + +// A transactional email's caller holds only a send key, which grants no access to person data, so its +// template never sees person properties, or the display name derived from them. The person stays attached +// so the send is recorded against them. +function personForFlow(hogFlow: HogFlow, person: CyclotronPerson | null): CyclotronPerson | undefined { + if (!person) { + return undefined + } + return hogFlow.trigger?.type === 'api' ? { ...person, properties: {}, name: '' } : person +} + +export async function loadHogFlowPerson( + persons: PersonReader, + hogFlow: HogFlow, + lookup: HogFlowPersonLookup +): Promise { + const person = await persons.getCyclotronPerson(hogFlow.team_id, lookup.id, lookup.kind, { + forceFresh: lookup.forceFresh, + }) + return personForFlow(hogFlow, person) +} diff --git a/nodejs/src/cdp/services/hogflows/idempotency-keys.test.ts b/nodejs/src/cdp/services/hogflows/idempotency-keys.test.ts index d98fbb69e639..3eea1ae897d9 100644 --- a/nodejs/src/cdp/services/hogflows/idempotency-keys.test.ts +++ b/nodejs/src/cdp/services/hogflows/idempotency-keys.test.ts @@ -48,6 +48,17 @@ describe('IdempotencyKeys', () => { expect(await operations('test', 'reserve', 'replayed')).toEqual(replayedBefore + 1) }) + it('replays the fields recorded with the run', async () => { + await keys.reserve(1, key, 'request-a', 'run-1') + await keys.markQueued(1, key, 'request-a', 'run-1', { attachment_count: 2, attachment_bytes: 4_096 }) + + expect(await keys.reserve(1, key, 'request-a', 'run-2')).toEqual({ + status: 'replayed', + invocationId: 'run-1', + replayFields: { attachment_count: 2, attachment_bytes: 4_096 }, + }) + }) + it('frees a key only for the request that holds it', async () => { await keys.reserve(1, key, 'request-a', 'run-1') diff --git a/nodejs/src/cdp/services/hogflows/idempotency-keys.ts b/nodejs/src/cdp/services/hogflows/idempotency-keys.ts index cd4a0e78480a..c0e106e82111 100644 --- a/nodejs/src/cdp/services/hogflows/idempotency-keys.ts +++ b/nodejs/src/cdp/services/hogflows/idempotency-keys.ts @@ -114,14 +114,16 @@ const sortKeys = (value: unknown): unknown => { export type IdempotencyReservation = | { status: 'reserved' } // The key already queued a run for the same request. - | { status: 'replayed'; invocationId: string } + | { status: 'replayed'; invocationId: string; replayFields?: ReplayFields } // The first request with the key has not finished queueing yet. | { status: 'in_flight' } // The key was used for a different request. | { status: 'conflict' } | { status: 'unavailable' } -type StoredKey = { fingerprint: string; invocationId: string; queued: boolean } +export type ReplayFields = Record + +type StoredKey = { fingerprint: string; invocationId: string; queued: boolean; replayFields?: ReplayFields } type PendingCompletion = { redisKey: string; value: StoredKey; expiresAt: number } @@ -200,13 +202,24 @@ export class IdempotencyKeys { if (stored.fingerprint !== fingerprint) { return { status: 'conflict' } } - return stored.queued ? { status: 'replayed', invocationId: stored.invocationId } : { status: 'in_flight' } + if (!stored.queued) { + return { status: 'in_flight' } + } + return stored.replayFields + ? { status: 'replayed', invocationId: stored.invocationId, replayFields: stored.replayFields } + : { status: 'replayed', invocationId: stored.invocationId } } - async markQueued(teamId: number, key: string, fingerprint: string, invocationId: string): Promise { + async markQueued( + teamId: number, + key: string, + fingerprint: string, + invocationId: string, + replayFields?: ReplayFields + ): Promise { const completion: PendingCompletion = { redisKey: this.redisKey(teamId, key), - value: { fingerprint, invocationId, queued: true }, + value: { fingerprint, invocationId, queued: true, ...(replayFields && { replayFields }) }, expiresAt: Date.now() + QUEUED_TTL_SECONDS * 1000, } for (let attempt = 1; attempt <= MARK_QUEUED_INLINE_ATTEMPTS; attempt++) { diff --git a/nodejs/src/cdp/services/hogflows/transactional-email.ts b/nodejs/src/cdp/services/hogflows/transactional-email.ts index e50244ac3b7a..b6e6b42fa527 100644 --- a/nodejs/src/cdp/services/hogflows/transactional-email.ts +++ b/nodejs/src/cdp/services/hogflows/transactional-email.ts @@ -1,9 +1,17 @@ import { Counter, Histogram } from 'prom-client' import { z } from 'zod' -import { CyclotronInvocationQueueParametersEmailType, EmailAttachmentMetadata } from '~/cdp/schema/cyclotron' +import { + CyclotronInvocationQueueParametersEmailType, + EmailAttachmentMetadata, + MAX_EMAIL_ATTACHMENTS, +} from '~/cdp/schema/cyclotron' +import { SHA256_HEX } from '~/cdp/services/messaging/attachment-store' export const MAX_COPY_RECIPIENTS = 10 +export const ATTACHMENT_FILENAME_MAX_LENGTH = 255 +export const ATTACHMENT_CONTENT_TYPE_MAX_LENGTH = 78 +export const ATTACHMENT_CONTENT_ID_MAX_LENGTH = 78 export const RECIPIENT_ADDRESS_PATTERN = /^[a-zA-Z0-9!#$%&'*+/=?^_`{|}~-]+(?:\.[a-zA-Z0-9!#$%&'*+/=?^_`{|}~-]+)*@[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+$/ @@ -14,6 +22,121 @@ const CopyRecipientsSchema = z .readonly() .optional() +export const SES_UNSUPPORTED_ATTACHMENT_EXTENSIONS: ReadonlySet = new Set([ + '.ade', + '.adp', + '.app', + '.asp', + '.bas', + '.bat', + '.cer', + '.chm', + '.cmd', + '.com', + '.cpl', + '.crt', + '.csh', + '.der', + '.exe', + '.fxp', + '.gadget', + '.hlp', + '.hta', + '.inf', + '.ins', + '.isp', + '.its', + '.js', + '.jse', + '.ksh', + '.lib', + '.lnk', + '.mad', + '.maf', + '.mag', + '.mam', + '.maq', + '.mar', + '.mas', + '.mat', + '.mau', + '.mav', + '.maw', + '.mda', + '.mdb', + '.mde', + '.mdt', + '.mdw', + '.mdz', + '.msc', + '.msh', + '.msh1', + '.msh2', + '.mshxml', + '.msh1xml', + '.msh2xml', + '.msi', + '.msp', + '.mst', + '.ops', + '.pcd', + '.pif', + '.plg', + '.prf', + '.prg', + '.reg', + '.scf', + '.scr', + '.sct', + '.shb', + '.shs', + '.sys', + '.ps1', + '.ps1xml', + '.ps2', + '.ps2xml', + '.psc1', + '.psc2', + '.tmp', + '.url', + '.vb', + '.vbe', + '.vbs', + '.vps', + '.vsmacros', + '.vss', + '.vst', + '.vsw', + '.vxd', + '.ws', + '.wsc', + '.wsf', + '.wsh', + '.xnk', +]) + +const lastExtension = (filename: string): string => { + const dot = filename.lastIndexOf('.') + return dot === -1 ? '' : filename.slice(dot).toLowerCase() +} + +const CONTROL_CHARACTERS = /[\u0000-\u001f\u007f]/ +const withoutControlCharacters = (field: string) => + [ + (value: string) => !CONTROL_CHARACTERS.test(value), + { message: `${field} cannot contain control characters` }, + ] as const + +const attachmentHeaderValue = (field: string, maxLength: number) => { + const message = `${field} needs 1 to ${maxLength} characters` + return z + .string() + .min(1, { message }) + .max(maxLength, { message }) + .refine(...withoutControlCharacters(field)) + .optional() +} + export const AcceptedSendSchema = z .object({ content: z @@ -39,13 +162,40 @@ export const AcceptedSendSchema = z cc: CopyRecipientsSchema, bcc: CopyRecipientsSchema, tracking: z.object({ captureEngagement: z.boolean() }).strict().readonly(), + attachments: z + .array( + z + .object({ + sha256: z.string().regex(SHA256_HEX), + filename: z + .string() + .min(1) + .max(ATTACHMENT_FILENAME_MAX_LENGTH, { + message: `filename can be at most ${ATTACHMENT_FILENAME_MAX_LENGTH} characters`, + }) + .refine(...withoutControlCharacters('filename')) + .refine((filename) => !SES_UNSUPPORTED_ATTACHMENT_EXTENSIONS.has(lastExtension(filename)), { + error: (issue) => { + const filename = String(issue.input) + return `SES does not send ${lastExtension(filename)} files, so ${filename} cannot be attached` + }, + }), + content_type: attachmentHeaderValue('content_type', ATTACHMENT_CONTENT_TYPE_MAX_LENGTH), + content_id: attachmentHeaderValue('content_id', ATTACHMENT_CONTENT_ID_MAX_LENGTH), + }) + .strict() + .readonly() + ) + .max(MAX_EMAIL_ATTACHMENTS) + .readonly() + .optional(), }) .strict() .readonly() export type AcceptedSend = z.infer -export type QueuedSend = AcceptedSend & { attachments?: readonly EmailAttachmentMetadata[] } +export type QueuedSend = Omit & { attachments?: readonly EmailAttachmentMetadata[] } export const acceptedSendEmailFields = ( send: QueuedSend | undefined @@ -103,11 +253,20 @@ export type TransactionalEmailOutcome = | 'invalid_request' | 'not_found' | 'not_sendable' + | 'workflow_can_wait' + | 'workflow_shape_not_supported' | 'quota_limited' | 'sending_suspended' | 'sending_paused' | 'recipient_suppressed' | 'recipient_unreachable' + | 'field_not_accepted' + | 'attachment_not_uploaded' + | 'message_too_large' + | 'email_render_limit_exceeded' + | 'utm_value_too_long' + | 'email_body_too_many_tags' + | 'delivery_unavailable' | 'error' export const transactionalEmailRequestsTotal = new Counter({ @@ -123,6 +282,7 @@ export const TRANSACTIONAL_EMAIL_STAGES = [ 'build', 'idempotency', 'checks', + 'attachments', 'queue_insert', 'metrics_flush', ] as const @@ -130,7 +290,7 @@ export type TransactionalEmailStage = (typeof TRANSACTIONAL_EMAIL_STAGES)[number export const transactionalEmailStageDurationMs = new Histogram({ name: 'cdp_transactional_email_stage_duration_ms', - help: 'Milliseconds one transactional email request spends in each stage: auth (the JWT), validate (the body), load (the team and the workflow), build (the run and its idempotency fingerprint), idempotency (every Idempotency-Key call to Valkey), checks (suspension, pause, suppression, quota and MX), queue_insert (the email queue insert), metrics_flush (the hand-off of the result and status rows to the Kafka producer, without the wait for the broker), and total. A request records only the stages it reached, so a refusal shows in the stage that refused it.', + help: 'Milliseconds one transactional email request spends in each stage: auth (the JWT), validate (the body), load (the team and the workflow), build (the run and its idempotency fingerprint), idempotency (every Idempotency-Key call to Valkey), checks (suspension, pause, suppression, quota and MX), attachments (verifying the files and estimating the message, only for a send with files), queue_insert (the email queue insert), metrics_flush (the hand-off of the result and status rows to the Kafka producer, without the wait for the broker), and total. A request records only the stages it reached, so a refusal shows in the stage that refused it.', labelNames: ['stage'], buckets: [1, 2.5, 5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000], }) diff --git a/nodejs/src/cdp/services/messaging/attachment-store.ts b/nodejs/src/cdp/services/messaging/attachment-store.ts index 7110b9f872e7..555e1f2c4788 100644 --- a/nodejs/src/cdp/services/messaging/attachment-store.ts +++ b/nodejs/src/cdp/services/messaging/attachment-store.ts @@ -69,7 +69,12 @@ export class AttachmentMissingError extends Error { export interface AttachmentStore { prepareUploads(teamId: number, files: UploadRequest[]): Promise - acceptForSend(teamId: number, refs: AttachmentRef[], needsUntil: Date): Promise + acceptForSend( + teamId: number, + refs: AttachmentRef[], + needsUntil: Date, + deadline?: AbortSignal + ): Promise load(teamId: number, files: EmailAttachmentMetadata[], signal: AbortSignal): Promise } @@ -98,10 +103,11 @@ export class S3AttachmentStore implements AttachmentStore { async acceptForSend( teamId: number, refs: AttachmentRef[], - needsUntil: Date + needsUntil: Date, + deadline?: AbortSignal ): Promise { const found = await Promise.all( - refs.map(async (ref) => ({ ref, stored: await this.storedFile(teamId, ref.sha256) })) + refs.map(async (ref) => ({ ref, stored: await this.storedFile(teamId, ref.sha256, deadline) })) ) const present = found.filter((file): file is { ref: AttachmentRef; stored: StoredFile } => !!file.stored) const missing = found.find((file) => !file.stored) @@ -111,7 +117,8 @@ export class S3AttachmentStore implements AttachmentStore { } await this.keepUntil( present.map(({ stored }) => stored), - needsUntil + needsUntil, + deadline ) return { files: present.map(({ ref, stored }) => ({ ...ref, sizeBytes: stored.sizeBytes })) } } @@ -165,23 +172,24 @@ export class S3AttachmentStore implements AttachmentStore { } } - private async storedFile(teamId: number, sha256: Sha256Hex): Promise { + private async storedFile(teamId: number, sha256: Sha256Hex, deadline?: AbortSignal): Promise { const key = this.key(teamId, sha256) - const head = await this.head(key) + const head = await this.head(key, deadline) if (!head || (head.ChecksumSHA256 && head.ChecksumSHA256 !== Buffer.from(sha256, 'hex').toString('base64'))) { return null } return { key, sizeBytes: head.ContentLength ?? 0, expiresAt: expiryOf(head) } } - private async keepUntil(files: StoredFile[], needsUntil: Date): Promise { + private async keepUntil(files: StoredFile[], needsUntil: Date, deadline?: AbortSignal): Promise { const toCopy = files.filter(({ expiresAt }) => expiresAt < needsUntil) if (toCopy.length && needsUntil > this.latestKeepable()) { throw new RangeError( `A file can be kept until ${this.latestKeepable().toISOString()} at the latest, not ${needsUntil.toISOString()}` ) } - await Promise.all(toCopy.map(({ key }) => this.restartLifetimeBySelfCopy(key))) + deadline?.throwIfAborted() + await Promise.all(toCopy.map(({ key }) => this.restartLifetimeBySelfCopy(key, deadline))) attachmentStoreFiles.labels({ operation: 'accept', outcome: 'copied' }).inc(toCopy.length) attachmentStoreFiles.labels({ operation: 'accept', outcome: 'kept' }).inc(files.length - toCopy.length) } @@ -197,8 +205,13 @@ export class S3AttachmentStore implements AttachmentStore { return [this.config.prefix, 'team', teamId, sha256].filter((segment) => segment !== '').join('/') } - private async head(key: string): Promise { - const signal = AbortSignal.timeout(this.config.timeoutMs) + private callSignal(deadline?: AbortSignal): AbortSignal { + const timeout = AbortSignal.timeout(this.config.timeoutMs) + return deadline ? AbortSignal.any([timeout, deadline]) : timeout + } + + private async head(key: string, deadline?: AbortSignal): Promise { + const signal = this.callSignal(deadline) try { return await settleBeforeAbortEvenWhileCredentialsStall( signal, @@ -214,9 +227,9 @@ export class S3AttachmentStore implements AttachmentStore { } } - private async restartLifetimeBySelfCopy(key: string): Promise { + private async restartLifetimeBySelfCopy(key: string, deadline?: AbortSignal): Promise { const { bucket } = this.config - const signal = AbortSignal.timeout(this.config.timeoutMs) + const signal = this.callSignal(deadline) try { await settleBeforeAbortEvenWhileCredentialsStall( signal, diff --git a/nodejs/src/cdp/services/messaging/attachment-uploads.ts b/nodejs/src/cdp/services/messaging/attachment-uploads.ts index 55cb6feff621..eb3cd19dc594 100644 --- a/nodejs/src/cdp/services/messaging/attachment-uploads.ts +++ b/nodejs/src/cdp/services/messaging/attachment-uploads.ts @@ -1,12 +1,21 @@ import { Counter } from 'prom-client' import { z } from 'zod' -import { MAX_EMAIL_ATTACHMENTS } from '~/cdp/schema/cyclotron' +import { EmailAttachmentMetadata, MAX_EMAIL_ATTACHMENTS } from '~/cdp/schema/cyclotron' import { buildIntegerMatcher } from '~/common/config/matchers' import { RedisV2 } from '~/common/redis/redis-v2' import { logger } from '~/common/utils/logger' -import { AttachmentStore, AttachmentStoreUnavailableError, DAY_MS, SHA256_HEX, UploadSlot } from './attachment-store' +import { SEND_NEEDS_FILES_FOR_MS } from './attachment-horizon' +import { + AttachmentRef, + AttachmentStore, + AttachmentStoreUnavailableError, + DAY_MS, + SHA256_HEX, + UploadSlot, + settleBeforeAbortEvenWhileCredentialsStall, +} from './attachment-store' import { MESSAGE_MAX_BYTES, estimateMessageBytes } from './message-estimate' export const ATTACHMENT_UPLOAD_QUOTA_KEY_PREFIX = 'cdp-email-attachments-upload-bytes' @@ -29,6 +38,25 @@ const UploadsRequestSchema = z.object({ export type AttachmentUploadsSettings = { teams: string; maxBytesPerTeamDay: number } export type UploadsAnswer = { status: number; body: Record } +export type SendRefusal = { + status: number + body: { + error: + | 'attachment_not_uploaded' + | 'message_too_large' + | 'email_render_limit_exceeded' + | 'utm_value_too_long' + | 'email_body_too_many_tags' + | 'delivery_unavailable' + [key: string]: unknown + } +} +export type MessageBodies = { bodyBytes: number[]; subject: string } +export type MessageBodySizing = MessageBodies | { refused: SendRefusal } +export type AcceptedSendFiles = { files: EmailAttachmentMetadata[]; attachmentBytes: number } + +const DJANGO_SEND_TIMEOUT_MS = 10_000 +export const ATTACHMENT_VERIFICATION_DEADLINE_MS = DJANGO_SEND_TIMEOUT_MS / 2 const uploadRequestsTotal = new Counter({ name: 'cdp_email_attachments_upload_requests_total', @@ -40,10 +68,10 @@ export function countUploadRequest(outcome: string): void { uploadRequestsTotal.inc({ outcome }) } -const DELIVERY_UNAVAILABLE: UploadsAnswer = { status: 503, body: { error: 'delivery_unavailable' } } +const DELIVERY_UNAVAILABLE = { status: 503, body: { error: 'delivery_unavailable' } } as const satisfies SendRefusal export class AttachmentUploads { - private readonly isAllowed: (teamId: number) => boolean + public readonly isAllowed: (teamId: number) => boolean constructor( private readonly store: AttachmentStore | null, @@ -120,6 +148,82 @@ export class AttachmentUploads { return { status: 200, body: { files: slots.map(toAnswer) } } } + public async acceptSend( + teamId: number, + refs: AttachmentRef[], + sizeMessageBodies: (deadline: AbortSignal) => Promise, + deadline: AbortSignal + ): Promise<{ accepted: AcceptedSendFiles } | { refused: SendRefusal }> { + const store = this.store + if (!store) { + return { refused: unavailable(teamId, 'No attachment store to verify the files of a send', null) } + } + try { + return await settleBeforeAbortEvenWhileCredentialsStall( + deadline, + this.verifySend(store, teamId, refs, sizeMessageBodies, deadline) + ) + } catch (error) { + if (deadline.aborted) { + return { refused: unavailable(teamId, 'Could not verify the files of a send in time', error) } + } + throw error + } + } + + private async verifySend( + store: AttachmentStore, + teamId: number, + refs: AttachmentRef[], + sizeMessageBodies: (deadline: AbortSignal) => Promise, + deadline: AbortSignal + ): Promise<{ accepted: AcceptedSendFiles } | { refused: SendRefusal }> { + deadline.throwIfAborted() + let verified: Awaited> + try { + const needsUntil = new Date(Date.now() + SEND_NEEDS_FILES_FOR_MS) + verified = await store.acceptForSend(teamId, refs, needsUntil, deadline) + } catch (error) { + deadline.throwIfAborted() + if (error instanceof AttachmentStoreUnavailableError) { + return { refused: unavailable(teamId, 'Could not verify the files of a send', error) } + } + throw error + } + if ('refusal' in verified) { + return { refused: { status: 400, body: { error: verified.refusal, sha256: verified.sha256 } } } + } + deadline.throwIfAborted() + let sizing: MessageBodySizing + try { + sizing = await sizeMessageBodies(deadline) + } catch (error) { + deadline.throwIfAborted() + return { refused: unavailable(teamId, 'Could not load the email step of a send', error) } + } + deadline.throwIfAborted() + if ('refused' in sizing) { + return sizing + } + const sizes = verified.files.map((file) => file.sizeBytes) + const attachmentBytes = sizes.reduce((total, size) => total + size, 0) + const messageBytes = estimateMessageBytes([...sizes, ...sizing.bodyBytes], [], sizing.subject) + if (messageBytes > MESSAGE_MAX_BYTES) { + return { + refused: { + status: 400, + body: { + error: 'message_too_large', + message_bytes: messageBytes, + max_message_bytes: MESSAGE_MAX_BYTES, + attachment_bytes: attachmentBytes, + }, + }, + } + } + return { accepted: { files: verified.files, attachmentBytes } } + } + private async chargeDailyUploadBytes(counter: RedisV2, teamId: number, bytes: number): Promise { const key = `${ATTACHMENT_UPLOAD_QUOTA_KEY_PREFIX}:${teamId}:${new Date().toISOString().slice(0, 10)}` const charged = await counter.useClient({ name: 'email-attachments-upload-quota' }, (client) => @@ -129,8 +233,11 @@ export class AttachmentUploads { } } -function unavailable(teamId: number, message: string, error: unknown): UploadsAnswer { - logger.warn('⚠️', message, { team_id: teamId, error: error instanceof Error ? error.message : String(error) }) +function unavailable(teamId: number, message: string, error: unknown): typeof DELIVERY_UNAVAILABLE { + logger.warn('⚠️', message, { + team_id: teamId, + ...(error !== null && { error: error instanceof Error ? error.message : String(error) }), + }) return DELIVERY_UNAVAILABLE } diff --git a/nodejs/src/cdp/services/messaging/email-tracking.service.ts b/nodejs/src/cdp/services/messaging/email-tracking.service.ts index 7ae601aacf9e..ce81942690a4 100644 --- a/nodejs/src/cdp/services/messaging/email-tracking.service.ts +++ b/nodejs/src/cdp/services/messaging/email-tracking.service.ts @@ -206,6 +206,8 @@ const isJavascriptHref = (href: string): boolean => { return url.slice(start, start + 'javascript:'.length).toLowerCase() === 'javascript:' } +export const openPixelTag = (pixelUrl: string): string => `` + export const addTrackingToEmail = ( html: string, invocation: CyclotronJobInvocationHogFunction, @@ -270,7 +272,7 @@ export const addTrackingToEmail = ( return m.replace(/\bhref\s*=\s*(?:"[^"]*"|'[^']*'|[^'">\s]+)/i, `href="${tracked}"`) }) - html = html.replace('', ``) + html = html.replace('', `${openPixelTag(trackingUrl)}`) return html } diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index 5a6cde966a3e..d3cedbc1c6f8 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -72,6 +72,7 @@ import { MailDevAPI, mailDevWebUrl } from './helpers/maildev' import { EmailTrackingCodeSigner } from './helpers/tracking-code' import { InFlightRegistry } from './in-flight-registry' import { MessageAssetsService } from './message-assets.service' +import { MESSAGE_MAX_BYTES } from './message-estimate' import { RecipientTokensService } from './recipient-tokens.service' async function metricValue(name: string, labels: Record, metricName = name): Promise { @@ -1793,6 +1794,24 @@ describe('EmailService', () => { }) }) + it.each(['preheader', 'subject'])( + 'sends nothing and fails as message_too_large when the %s carries the final message past the size limit', + async (field) => { + invocation.queueParameters = createEmailParams({ + from: { integrationId: 1 }, + to: { email: recipient, name: 'Test User' }, + attachments: [pdf.metadata], + [field]: 'x'.repeat((MESSAGE_MAX_BYTES * 3) / 4), + }) + + const result = await sender().executeSendEmail(invocation) + + expect(result.finished).toBe(true) + expect(result.error).toMatch(/^message_too_large: /) + expect(ses.sent).toEqual([]) + } + ) + it.each([ ['was never uploaded', () => Promise.resolve(attachmentFile())], ['belongs to another team', async () => await uploadAttachment(team.id + 1, { filename: 'a.pdf' })], diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index 0102070d2cb3..87041c8b371b 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -48,6 +48,7 @@ import { quoteMailbox, removeSuppressedRecipients, unquoteMailbox } from './help import { EmailTrackingCodeSigner, TRACKING_CODE_HEADER_NAME } from './helpers/tracking-code' import { callsSettledBefore, declaredFileBytes, exceedsLargestSend } from './in-flight-registry' import { MessageAssetsService } from './message-assets.service' +import { MESSAGE_MAX_BYTES, estimateMessageBytes } from './message-estimate' import { PreferencesTokenSource, RecipientTokensService, @@ -185,7 +186,7 @@ class EmailSendDeadlineError extends Error { class MessageTooLargeError extends Error { constructor() { - super('The attachment files of this email are larger than one email can carry') + super('This email and its attachment files are larger than one email can carry') this.name = 'MessageTooLargeError' } } @@ -238,7 +239,7 @@ function describeAttachmentFailure( if (error instanceof MessageTooLargeError) { return { reason: EMAIL_MESSAGE_TOO_LARGE, - message: `${EMAIL_MESSAGE_TOO_LARGE}: the attachment files of this email are larger than one email can carry, so it was not sent.`, + message: `${EMAIL_MESSAGE_TOO_LARGE}: this email and its attachment files are larger than one email can carry, so it was not sent.`, } } if (isAttachmentLoadFailure(error)) { @@ -1446,6 +1447,16 @@ export class EmailService { params.preheader ) : null + if ( + files.length > 0 && + estimateMessageBytes( + files.map(({ bytes }) => bytes.byteLength), + [html ?? '', params.text ?? ''], + sanitizeEmailSubject(params.subject) + ) > MESSAGE_MAX_BYTES + ) { + throw new MessageTooLargeError() + } const htmlBody = html ? { Html: { Data: html, Charset: 'UTF-8' } } : {} const sendEmailParams: SendEmailCommandInput = { diff --git a/nodejs/src/cdp/services/messaging/helpers/email-html.ts b/nodejs/src/cdp/services/messaging/helpers/email-html.ts index 00ed26378257..1c8316a0b3ec 100644 --- a/nodejs/src/cdp/services/messaging/helpers/email-html.ts +++ b/nodejs/src/cdp/services/messaging/helpers/email-html.ts @@ -26,7 +26,8 @@ export type SesTracking = { export const utmTagsFor = ( invocation: CyclotronJobInvocationHogFunction, - onRenderError: (key: string, message: string) => void + onRenderError: (key: string, message: string) => void, + deadline?: AbortSignal ): UtmTags | null => { const metadata = invocation.hogFunction.metadata if (metadata?.utm_tags_enabled !== true) { @@ -39,7 +40,7 @@ export const utmTagsFor = ( utm_campaign: metadata.hog_flow_name ?? invocation.hogFunction.name, utm_content: metadata.hog_flow_action_name ?? '', }, - renderUtmOverrides(metadata.utm_params, invocation.state.globals, onRenderError) + renderUtmOverrides(metadata.utm_params, invocation.state.globals, onRenderError, deadline) ) } diff --git a/nodejs/src/cdp/services/messaging/helpers/escape-html.ts b/nodejs/src/cdp/services/messaging/helpers/escape-html.ts index 869e51418ed5..ef6c8c88ef0b 100644 --- a/nodejs/src/cdp/services/messaging/helpers/escape-html.ts +++ b/nodejs/src/cdp/services/messaging/helpers/escape-html.ts @@ -6,5 +6,17 @@ const HTML_ESCAPE: Record = { "'": ''', } +const ESCAPE_GROWTH_BY_CHAR_CODE: ReadonlyMap = new Map( + Object.entries(HTML_ESCAPE).map(([character, escaped]) => [character.charCodeAt(0), escaped.length - 1]) +) + // Covers `& < > " '`, which is enough for a text node and for a quoted attribute value. export const escapeHtml = (text: string): string => text.replace(/[&<>"']/g, (c) => HTML_ESCAPE[c]) + +export const escapedHtmlByteLength = (text: string): number => { + let growth = 0 + for (let index = 0; index < text.length; index++) { + growth += ESCAPE_GROWTH_BY_CHAR_CODE.get(text.charCodeAt(index)) ?? 0 + } + return Buffer.byteLength(text, 'utf8') + growth +} diff --git a/nodejs/src/cdp/services/messaging/helpers/ses-html-max-bytes.test.ts b/nodejs/src/cdp/services/messaging/helpers/ses-html-max-bytes.test.ts new file mode 100644 index 000000000000..6b42055c1caa --- /dev/null +++ b/nodejs/src/cdp/services/messaging/helpers/ses-html-max-bytes.test.ts @@ -0,0 +1,87 @@ +import { createExampleInvocation } from '~/cdp/_tests/fixtures' +import { defaultConfig } from '~/common/config/config' + +import { htmlForSes, tagEmailLinks } from './email-html' +import { escapeHtml, escapedHtmlByteLength } from './escape-html' +import { sesHtmlMaxBytes } from './ses-html-max-bytes' +import { EmailTrackingCodeSigner } from './tracking-code' + +const SITE_URL = 'https://us.posthog.com' +const PREHEADER = 'Your files are attached' +const UTM = { + tags: { utm_source: 'posthog', utm_medium: 'email', utm_campaign: 'Spring sale', utm_content: 'cta' }, + siteUrl: SITE_URL, + fixedScanner: true, +} +const unclosedAnchors = (count: number): string => + `${`x`.repeat(count)}` + +describe('sesHtmlMaxBytes', () => { + const invocation = createExampleInvocation() + const signer = new EmailTrackingCodeSigner(defaultConfig.ENCRYPTION_SALT_KEYS, 'https://track.example.com') + const maxBytes = (html: string, utm: typeof UTM | null): number => + sesHtmlMaxBytes(html, { utmTags: utm?.tags ?? null, invocation, signer, preheader: PREHEADER }) + + it.each<{ case: string; html: string; utm?: typeof UTM | null }>([ + { case: 'a plain link', html: 'x' }, + { case: 'an unquoted href', html: 'x' }, + { case: 'raw ampersands', html: "x" }, + { case: 'quotes and brackets', html: `x` }, + { + case: 'raw ampersands without UTM tags', + html: `x`, + utm: null, + }, + { + case: 'raw quotes without UTM tags', + html: `x`, + utm: null, + }, + { + case: 'raw quotes in a closed body without UTM tags', + html: `x`, + utm: null, + }, + { case: 'entity-encoded characters', html: 'x' }, + { case: 'padding whitespace', html: 'x' }, + { case: 'an earlier run id', html: 'x' }, + { case: 'a body tag inside the href', html: 'x' }, + { + case: 'many links with an existing SES tag', + html: `${'y'.repeat(12)}`, + }, + { case: 'many anchors that never close', html: unclosedAnchors(1_000) }, + { + case: 'UTM values with apostrophes, reserved and multi-byte characters', + html: 'xy', + utm: { + ...UTM, + tags: { + utm_source: "it's & ", + utm_medium: 'é😀中', + utm_campaign: 'a b/c?d=e#f', + utm_content: "'()*!~-_.", + }, + }, + }, + ])('bounds the bytes SES gets for $case', ({ html, utm = UTM }) => { + const sent = htmlForSes( + tagEmailLinks(html, utm, { runId: invocation.id, senderEmail: 'hi@example.com', siteUrl: SITE_URL }), + { invocation, signer, isTest: false }, + PREHEADER + ) + + expect(Buffer.byteLength(sent, 'utf8')).toBeLessThanOrEqual(maxBytes(html, utm)) + }) + + it.each<{ case: string; html: string }>([ + { case: 'nothing', html: '' }, + { case: 'no escaped character', html: 'plain text' }, + { case: 'every escaped character', html: `&<>"'` }, + { case: 'escaped characters between multi-byte ones', html: `é&ü<€>"😀'中` }, + { case: 'a lone surrogate beside an ampersand', html: '\ud800&' }, + { case: 'a body of ampersands', html: '&'.repeat(10_000) }, + ])('counts the escaped bytes of $case as escaping it would', ({ html }) => { + expect(escapedHtmlByteLength(html)).toBe(Buffer.byteLength(escapeHtml(html), 'utf8')) + }) +}) diff --git a/nodejs/src/cdp/services/messaging/helpers/ses-html-max-bytes.ts b/nodejs/src/cdp/services/messaging/helpers/ses-html-max-bytes.ts new file mode 100644 index 000000000000..2c027e2fc156 --- /dev/null +++ b/nodejs/src/cdp/services/messaging/helpers/ses-html-max-bytes.ts @@ -0,0 +1,100 @@ +import { CyclotronJobInvocationHogFunction } from '~/cdp/types' + +import { openPixelTag, resolveEmailEngagementDistinctId } from '../email-tracking.service' +import { escapedHtmlByteLength } from './escape-html' +import { WORKFLOW_RUN_PARAM } from './link-tags' +import { maybeAddPreheaderToEmail } from './preheader' +import { SES_LINK_INDEX_TAG } from './ses' +import { EmailTrackingCodeSigner } from './tracking-code' +import { UtmTags } from './utm' + +export type SesHtmlInputs = { + utmTags: UtmTags | null + invocation: CyclotronJobInvocationHogFunction + signer: EmailTrackingCodeSigner + preheader?: string +} + +const QUOTES_A_RETAGGED_HREF_GAINS = 2 +const SES_LINK_ATTRIBUTE_GROWTH = Math.max(` ses:tags="${SES_LINK_INDEX_TAG}:"`.length, ' ses:no-track'.length) +const LONGEST_WORKFLOW_VERSION = Number.MAX_SAFE_INTEGER +const anchorCount = (html: string): number => { + const anchorStart = / + escapedHtmlByteLength(encodeURIComponent(String.fromCharCode(code))) +) +const isHighSurrogate = (code: number): boolean => code >= 0xd800 && code <= 0xdbff +const isLowSurrogate = (code: number): boolean => code >= 0xdc00 && code <= 0xdfff + +const escapedUriComponentBytes = (value: string): number => { + let bytes = 0 + for (let index = 0; index < value.length; index++) { + const code = value.charCodeAt(index) + if (code < 0x80) { + bytes += ASCII_ESCAPED_URI_COMPONENT_BYTES[code] + } else if (code < 0x800) { + bytes += 6 + } else if (isHighSurrogate(code) && isLowSurrogate(value.charCodeAt(index + 1))) { + bytes += 12 + index++ + } else if (isHighSurrogate(code) || isLowSurrogate(code)) { + throw new URIError('URI malformed') + } else { + bytes += 9 + } + } + return bytes +} + +const ESCAPED_AMPERSAND_BYTES = escapedHtmlByteLength('&') + +const escapedQueryBytes = (params: Record): number => + Object.entries(params) + .filter(([, value]) => value) + .reduce( + (total, [key, value]) => + total + ESCAPED_AMPERSAND_BYTES + escapedHtmlByteLength(`${key}=`) + escapedUriComponentBytes(value), + 0 + ) + +const growthPerAnchor = (utmTags: UtmTags | null, runId: string, anchors: number): number => + (utmTags ? QUOTES_A_RETAGGED_HREF_GAINS + escapedQueryBytes(utmTags) : 0) + + QUOTES_A_RETAGGED_HREF_GAINS + + escapedQueryBytes({ [WORKFLOW_RUN_PARAM]: runId }) + + SES_LINK_ATTRIBUTE_GROWTH + + String(anchors).length + +const longestOpenPixelBytes = ( + invocation: CyclotronJobInvocationHogFunction, + signer: EmailTrackingCodeSigner +): number => + Buffer.byteLength( + openPixelTag( + signer.pixelUrl( + { + ...invocation, + distinctId: resolveEmailEngagementDistinctId(invocation), + workflowVersion: LONGEST_WORKFLOW_VERSION, + }, + true + ) + ), + 'utf8' + ) + +export const sesHtmlMaxBytes = (html: string, { utmTags, invocation, signer, preheader }: SesHtmlInputs): number => { + const anchors = anchorCount(html) + return ( + escapedHtmlByteLength(html) + + (anchors === 0 ? 0 : anchors * growthPerAnchor(utmTags, invocation.id, anchors)) + + longestOpenPixelBytes(invocation, signer) + + Buffer.byteLength(maybeAddPreheaderToEmail('', preheader), 'utf8') + ) +} diff --git a/nodejs/src/cdp/services/messaging/helpers/utm.ts b/nodejs/src/cdp/services/messaging/helpers/utm.ts index a07451725d8b..5195b9bec04b 100644 --- a/nodejs/src/cdp/services/messaging/helpers/utm.ts +++ b/nodejs/src/cdp/services/messaging/helpers/utm.ts @@ -1,7 +1,8 @@ import { getDomain } from 'tldts' import { HogFunctionInvocationGlobalsWithInputs } from '~/cdp/types' -import { LiquidRenderer } from '~/cdp/utils/liquid' +import { throwIfDeadlinePassed } from '~/cdp/utils/deadline' +import { LiquidRenderBudget, LiquidRenderer } from '~/cdp/utils/liquid' import { decodeHtmlEntitiesInHref } from '../email-tracking.service' import { escapeHtml } from './escape-html' @@ -25,6 +26,16 @@ export type WorkflowRunTag = { const UTM_KEYS = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_content'] as const +export const UTM_VALUE_LIMIT_CHARS = 1_000 + +export const utmValueOverLimit = (raw: unknown): (typeof UTM_KEYS)[number] | undefined => + raw && typeof raw === 'object' + ? UTM_KEYS.find((key) => { + const value = (raw as Record)[key] + return typeof value === 'string' && value.length > UTM_VALUE_LIMIT_CHARS + }) + : undefined + // One attribute per match, so `href` and `data-ph-no-utm` count only as attribute names, never as text // inside another attribute's value or as part of a longer name like `data-href`. const ATTRIBUTE_REGEX = /\s([^\s"'<>/=]+)(?:\s*=\s*(?:"([^"]*)"|'([^']*)'|([^\t\n\f\r >]+)))?/g @@ -57,7 +68,8 @@ const isOptedOutOfUtmTags = ({ attributes }: Anchor): boolean => export const renderUtmOverrides = ( raw: unknown, globals: HogFunctionInvocationGlobalsWithInputs, - onError: (key: string, message: string) => void + onError: (key: string, message: string) => void, + deadline?: AbortSignal ): Record => { const rendered: Record = {} if (!raw || typeof raw !== 'object') { @@ -68,9 +80,12 @@ export const renderUtmOverrides = ( if (typeof value !== 'string' || !value.trim()) { continue } + throwIfDeadlinePassed(deadline) try { // Liquid escapes HTML in what it renders, and a URL needs the plain text. - rendered[key] = decodeHtmlEntitiesInHref(LiquidRenderer.renderWithHogFunctionGlobals(value, globals)) + rendered[key] = decodeHtmlEntitiesInHref( + LiquidRenderer.renderWithHogFunctionGlobals(value, globals, new LiquidRenderBudget(deadline)) + ) } catch (error) { onError(key, error instanceof Error ? error.message : String(error)) } diff --git a/nodejs/src/cdp/services/messaging/message-estimate.test.ts b/nodejs/src/cdp/services/messaging/message-estimate.test.ts new file mode 100644 index 000000000000..9fcdc1ba175f --- /dev/null +++ b/nodejs/src/cdp/services/messaging/message-estimate.test.ts @@ -0,0 +1,22 @@ +import { estimateMessageBytes } from './message-estimate' + +describe('estimateMessageBytes', () => { + const withoutBodies = estimateMessageBytes([]) + + it.each([ + ['an ASCII body', 'x'.repeat(57), 78], + ['a body of two-byte characters', 'é'.repeat(57), 156], + ['an empty body', '', 0], + ])('counts %s at its base64 transfer size', (_name, body, transferBytes) => { + expect(estimateMessageBytes([], [body]) - withoutBodies).toBe(transferBytes) + }) + + it.each([ + ['an ASCII subject', 'x'.repeat(42), 75], + ['a subject one byte past an encoded word', 'x'.repeat(43), 150], + ['a subject of two-byte characters', 'é'.repeat(21), 75], + ['an empty subject', '', 0], + ])('counts %s at its folded encoded-word size', (_name, subject, headerBytes) => { + expect(estimateMessageBytes([], [], subject) - withoutBodies).toBe(headerBytes) + }) +}) diff --git a/nodejs/src/cdp/services/messaging/message-estimate.ts b/nodejs/src/cdp/services/messaging/message-estimate.ts index ded7ab09c5dd..3074f489ed51 100644 --- a/nodejs/src/cdp/services/messaging/message-estimate.ts +++ b/nodejs/src/cdp/services/messaging/message-estimate.ts @@ -1,8 +1,21 @@ export const MESSAGE_MAX_BYTES = 40_000_000 const HEADERS_AND_MIME_BOUNDARIES_BYTES = 64 * 1024 +const SUBJECT_BYTES_PER_ENCODED_WORD = 42 +const FOLDED_ENCODED_WORD_BYTES = 75 -const encodedFileBytes = (size: number): number => Math.ceil((Math.ceil(size / 3) * 4 * 78) / 76) +const base64TransferBytes = (size: number): number => Math.ceil((Math.ceil(size / 3) * 4 * 78) / 76) -export function estimateMessageBytes(fileSizes: readonly number[]): number { - return fileSizes.reduce((total, size) => total + encodedFileBytes(size), HEADERS_AND_MIME_BOUNDARIES_BYTES) +const encodedSubjectBytes = (subject: string): number => + Math.ceil(Buffer.byteLength(subject, 'utf8') / SUBJECT_BYTES_PER_ENCODED_WORD) * FOLDED_ENCODED_WORD_BYTES + +export function estimateMessageBytes( + fileSizes: readonly number[], + bodies: readonly string[] = [], + subject = '' +): number { + const bodySizes = bodies.map((body) => Buffer.byteLength(body, 'utf8')) + return [...fileSizes, ...bodySizes].reduce( + (total, size) => total + base64TransferBytes(size), + HEADERS_AND_MIME_BOUNDARIES_BYTES + encodedSubjectBytes(subject) + ) } diff --git a/nodejs/src/cdp/utils/hog-error-classification.ts b/nodejs/src/cdp/utils/hog-error-classification.ts index 893098901ed3..cf41017a7ff3 100644 --- a/nodejs/src/cdp/utils/hog-error-classification.ts +++ b/nodejs/src/cdp/utils/hog-error-classification.ts @@ -55,6 +55,10 @@ function findHogVMException(error: unknown): KindedError | undefined { return undefined } +export const isHogVMError = (error: unknown): boolean => findHogVMException(error) !== undefined + +export const isHogLimitError = (error: unknown): boolean => findHogVMException(error)?.kind === 'limit' + const BYTECODE_CONTRACT = 'bytecodeContract' /** diff --git a/nodejs/src/cdp/utils/liquid.ts b/nodejs/src/cdp/utils/liquid.ts index 7331ef12ef3e..b1a5a60a70fa 100644 --- a/nodejs/src/cdp/utils/liquid.ts +++ b/nodejs/src/cdp/utils/liquid.ts @@ -25,6 +25,41 @@ const LIQUID_TAG_CLOSERS: Record = { '{': '}}', '%': '%}' } // with a throwing filter rather than unregistered, because an unknown filter silently acts as identity. const LIQUID_UNBOUNDED_FILTERS = ['where_exp', 'find_exp', 'find_index_exp', 'reject_exp', 'group_by_exp'] +export const EMAIL_BODY_TAG_LIMIT = 10_000 + +const LIQUID_BLOCK_OPENER = /-?[\s\u180E]*liquid/y + +const countStatementLines = (body: string): number => { + let count = 0 + for (const line of body.split('\n')) { + if (line.trim() !== '') { + count++ + } + } + return count +} + +export const countLiquidTags = (template: string): number => { + const parsed = decodeLiquidTags(template) + let count = 0 + let statementsCountedUntil = 0 + for (let at = parsed.indexOf('{'); at !== -1; at = parsed.indexOf('{', at + 1)) { + const next = parsed[at + 1] + if (next !== '{' && next !== '%') { + continue + } + count++ + at++ + LIQUID_BLOCK_OPENER.lastIndex = at + 1 + if (next === '%' && at >= statementsCountedUntil && LIQUID_BLOCK_OPENER.test(parsed)) { + const end = parsed.indexOf('%}', LIQUID_BLOCK_OPENER.lastIndex) + statementsCountedUntil = end === -1 ? parsed.length : end + count += countStatementLines(parsed.slice(LIQUID_BLOCK_OPENER.lastIndex, statementsCountedUntil)) + } + } + return count +} + export class LiquidRenderBudget { private usedMs = 0 private outputChars = 0