From c11a2ae805438efbe4bb8cb08b3c3a59f7c45b56 Mon Sep 17 00:00:00 2001 From: Harley Alexander <43975092+mayteio@users.noreply.github.com> Date: Tue, 27 Oct 2026 10:36:02 +0000 Subject: [PATCH] feat(workflows): tie email events to the send and version that caused them Add $workflow_invocation_id to $workflows_email_sent, $workflows_email_failed and every SES callback event, and $email_message_id to accepted SES sends and their callbacks, so each email event joins to the run and send that caused it. The sent event from a workflow step now carries $workflow_version. The step records the version it rendered under, and the send, tracking code, email metrics, billable count and conversion attribution all use that version, so a send that spans a republish no longer splits across two versions. Copied unchanged from #123128 so it can get a PostHog Review pass from a branch in this repository. Co-authored-by: Reichenbach, Michael <755327+Silthus@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 Generated-By: PostHog Desktop Task-Id: e72a2c10-f145-4f7c-8754-a4fcc06a18be --- .../hogflows/actions/hog_function.test.ts | 22 +++++++--- .../services/hogflows/actions/hog_function.ts | 4 +- .../cdp/services/hogflows/billing-utils.ts | 5 +++ .../hogflows/hogflow-executor.service.test.ts | 29 ++++++++++++++ .../hogflows/hogflow-functions.service.ts | 31 ++++++++------ .../messaging/email-tracking.service.test.ts | 40 +++++++++++++++++-- .../messaging/email-tracking.service.ts | 24 ++++------- .../services/messaging/email.service.test.ts | 23 +++++++++-- .../cdp/services/messaging/email.service.ts | 18 +++++++-- .../services/messaging/helpers/ses.test.ts | 9 +++-- .../src/cdp/services/messaging/helpers/ses.ts | 1 + .../hog-function-monitoring.service.ts | 3 +- nodejs/src/cdp/types.ts | 1 + 13 files changed, 160 insertions(+), 50 deletions(-) diff --git a/nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts b/nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts index 60e53e0ed5f7..ad238490d987 100644 --- a/nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts +++ b/nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts @@ -652,12 +652,13 @@ describe('HogFunctionHandler', () => { // received — the same version `email_sent` is counted under — so a send re-pins the attribution // version. A non-message step must leave it alone. it.each([ - { billingType: 'email' as const, expected: 3 }, - { billingType: 'push' as const, expected: 3 }, - { billingType: 'fetch' as const, expected: 1 }, + { billingType: 'email' as const, renderedVersion: 3, expected: 3, billedVersion: 3 }, + { billingType: 'email' as const, renderedVersion: 2, expected: 2, billedVersion: 2 }, + { billingType: 'push' as const, renderedVersion: 2, expected: 3, billedVersion: undefined }, + { billingType: 'fetch' as const, renderedVersion: 2, expected: 1, billedVersion: undefined }, ])( - 'a completed $billingType step leaves the attribution version at $expected', - async ({ billingType, expected }) => { + 'a completed $billingType step rendered under v$renderedVersion leaves the attribution version at $expected', + async ({ billingType, renderedVersion, expected, billedVersion }) => { const handler = new HogFunctionHandler( mockHogFlowFunctionsService, mockRecipientPreferencesService, @@ -670,6 +671,13 @@ describe('HogFunctionHandler', () => { hogFlow: { ...invocation.hogFlow, version: 3 }, state: { ...invocation.state, flowVersion: 1 }, } + const build = mockHogFlowFunctionsService.buildHogFunctionInvocation.bind(mockHogFlowFunctionsService) + jest.spyOn(mockHogFlowFunctionsService, 'buildHogFunctionInvocation').mockImplementation( + async (flowInvocation, hogFunction, globals) => { + const built = await build(flowInvocation, hogFunction, globals) + return { ...built, state: { ...built.state, renderedWorkflowVersion: renderedVersion } } + } + ) const invocationResult = createInvocationResult(republished, { queue: 'hog', @@ -679,6 +687,10 @@ describe('HogFunctionHandler', () => { await handler.execute({ invocation: republished, action, result: invocationResult }) expect(invocationResult.invocation.state.flowVersion).toBe(expected) + expect( + invocationResult.metrics.find((metric) => metric.metric_name === 'billable_invocation') + ?.app_source_version?.version + ).toBe(billedVersion) } ) diff --git a/nodejs/src/cdp/services/hogflows/actions/hog_function.ts b/nodejs/src/cdp/services/hogflows/actions/hog_function.ts index 35f1b3d1fffa..7f3d19260c3c 100644 --- a/nodejs/src/cdp/services/hogflows/actions/hog_function.ts +++ b/nodejs/src/cdp/services/hogflows/actions/hog_function.ts @@ -267,7 +267,9 @@ export class HogFunctionHandler implements ActionHandler { // published — and the conversion belongs to the version whose message the person // received, which is also the version `email_sent` was counted under. Leaving the // run-start stamp here would split a rate across two versions. - if (this.hogFlowActionBillingType === 'email' || this.hogFlowActionBillingType === 'push') { + if (this.hogFlowActionBillingType === 'email') { + result.invocation.state.flowVersion = functionResult.invocation.state.renderedWorkflowVersion + } else if (this.hogFlowActionBillingType === 'push') { result.invocation.state.flowVersion = invocation.hogFlow.version } } diff --git a/nodejs/src/cdp/services/hogflows/billing-utils.ts b/nodejs/src/cdp/services/hogflows/billing-utils.ts index c8557cfe812a..e133704149e2 100644 --- a/nodejs/src/cdp/services/hogflows/billing-utils.ts +++ b/nodejs/src/cdp/services/hogflows/billing-utils.ts @@ -1,5 +1,7 @@ import { CyclotronJobInvocationHogFunction, CyclotronJobInvocationResult } from '~/cdp/types' +import { resolveEmailSendingAppSourceVersion } from '../messaging/email-tracking.service' + type HogFlowBillingMetricData = { invocation: CyclotronJobInvocationHogFunction billingMetricType: 'fetch' | 'email' | 'sms' | 'push' @@ -15,6 +17,8 @@ export const trackHogFlowBillableInvocation = ( result: CyclotronJobInvocationResult, data: HogFlowBillingMetricData ) => { + const appSourceVersion = + data.billingMetricType === 'email' ? resolveEmailSendingAppSourceVersion(data.invocation) : undefined result.metrics.push({ team_id: data.invocation.teamId, app_source_id: data.invocation.functionId, @@ -22,5 +26,6 @@ export const trackHogFlowBillableInvocation = ( metric_kind: data.billingMetricType, metric_name: 'billable_invocation', count: 1, + ...(appSourceVersion ? { app_source_version: appSourceVersion } : {}), }) } diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts index 5f6df00a7697..a9ba709d2753 100644 --- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts +++ b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts @@ -428,6 +428,35 @@ describe('Hogflow Executor', () => { }) }) + it.each([ + ['keeps the version a step rendered under', true, 2], + ['uses the live version for a step saved before the rendered version was stored', false, 3], + ])('%s when the workflow is republished before it sends', async (_name, storesRenderedVersion, expected) => { + await insertHogFunctionTemplate(hub.postgres, customerTaskTemplate) + const config = { template_id: customerTaskTemplate.id, inputs: { name: { value: 'Follow up' } } } + const renderedFlow = { ...hogFlow, version: 2 } + const invocation = createExampleHogFlowInvocation(renderedFlow, { + currentAction: { id: 'function_id_1', startedAtTimestamp: Date.now() }, + }) + const hogFunction = await hogFlowFunctionsService.buildHogFunction(renderedFlow, config) + const rendered = await hogFlowFunctionsService.buildHogFunctionInvocation(invocation, hogFunction, { + event: invocation.state.event, + person: invocation.person, + }) + const { renderedWorkflowVersion, ...legacyState } = rendered.state + invocation.state.currentAction!.hogFunctionState = parseJSON( + JSON.stringify(storesRenderedVersion ? { ...legacyState, renderedWorkflowVersion } : legacyState) + ) + + invocation.hogFlow = { ...renderedFlow, version: 3 } + const resumed = await hogFlowFunctionsService.buildHogFunctionInvocation(invocation, hogFunction, { + event: invocation.state.event, + person: invocation.person, + }) + + expect(resumed.state.renderedWorkflowVersion).toBe(expected) + }) + it.each([false, true])('keeps legacy task keys with saved function state: %s', async (hasSavedState) => { await insertHogFunctionTemplate(hub.postgres, customerTaskTemplate) const config = { diff --git a/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts b/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts index 2cdd30cf8ed6..36185b1a3fc6 100644 --- a/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts +++ b/nodejs/src/cdp/services/hogflows/hogflow-functions.service.ts @@ -121,21 +121,28 @@ export class HogFlowFunctionsService { }, } + const savedState = invocation.state.currentAction?.hogFunctionState const hogFunctionInvocation: CyclotronJobInvocationHogFunction = { ...invocation, hogFunction, - state: invocation.state.currentAction?.hogFunctionState ?? { - globals: await this.hogFunctionExecutor.hogExecutor.buildInputsWithGlobals( - hogFunction, - globalsWithSource - ), - timings: [], - attempts: 0, - actionId: invocation.state.currentAction?.id, - actionStepCount: invocation.state.actionStepCount, - customerTaskIdempotencyVersion: invocation.state.customerTaskIdempotencyVersion, - rerunAttempts: invocation.state.rerunAttempts, - }, + state: savedState + ? { + ...savedState, + renderedWorkflowVersion: savedState.renderedWorkflowVersion ?? invocation.hogFlow.version, + } + : { + globals: await this.hogFunctionExecutor.hogExecutor.buildInputsWithGlobals( + hogFunction, + globalsWithSource + ), + timings: [], + attempts: 0, + actionId: invocation.state.currentAction?.id, + actionStepCount: invocation.state.actionStepCount, + customerTaskIdempotencyVersion: invocation.state.customerTaskIdempotencyVersion, + rerunAttempts: invocation.state.rerunAttempts, + renderedWorkflowVersion: invocation.hogFlow.version, + }, } return hogFunctionInvocation diff --git a/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts b/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts index c76983c6f596..4af76046dca0 100644 --- a/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email-tracking.service.test.ts @@ -1,6 +1,6 @@ import { createMockJobQueue } from '~/tests/helpers/mocks/job-queue.mock' import { mockProducerObserver } from '~/tests/helpers/mocks/producer.mock' -import { mockFetch } from '~/tests/helpers/mocks/request.mock' +import { mockFetch, mockInternalFetch } from '~/tests/helpers/mocks/request.mock' import { Server } from 'http' import supertest from 'supertest' @@ -17,11 +17,13 @@ import { KAFKA_APP_METRICS_2, KAFKA_LOG_ENTRIES } from '~/common/config/kafka-to import { closeHub, createHub } from '~/common/utils/db/hub' import { PostgresUse } from '~/common/utils/db/postgres' import * as envUtils from '~/common/utils/env-utils' +import { parseJSON } from '~/common/utils/json-parse' import { createCdpConsumerDeps } from '~/tests/helpers/cdp' import { waitForExpect } from '~/tests/helpers/expectations' import { createTestTeamFixture } from '~/tests/helpers/sql' import { Hub, Team } from '../../../types' +import { TeamWorkflowsConfigService } from '../managers/team-workflows-config.service' import { METRIC_NAME_TO_EVENT_NAME, PIXEL_GIF, @@ -56,6 +58,7 @@ describe('EmailTrackingService', () => { functionId: 'fn-1', id: 'inv-1', teamId: 1, + state: {}, } as any const extractTarget = (html: string): string => { @@ -95,7 +98,7 @@ describe('EmailTrackingService', () => { // clicks we record ourselves, so a version missing here means those engagement metrics never // split by version however the send path is configured. it('carries the sending version into the pixel and redirect codes', () => { - const flowInvocation = { ...invocation, hogFlow: { id: 'flow-1', version: 7 } } as any + const flowInvocation = { ...invocation, state: { ...invocation.state, renderedWorkflowVersion: 7 } } const html = 'x' const out = addTrackingToEmail(html, flowInvocation, signer) @@ -429,30 +432,35 @@ describe('EmailTrackingService', () => { // The route enforces a real SNS signature; verifying it needs AWS's private key, so we // stub the check and let a posted SNS envelope flow through the real handler + service. let verifySignatureSpy: jest.SpyInstance + let captureEngagementSpy: jest.SpyInstance beforeEach(() => { verifySignatureSpy = jest .spyOn(SesWebhookHandler.prototype as any, 'verifySnsSignature') .mockResolvedValue(true) + captureEngagementSpy = jest.spyOn(TeamWorkflowsConfigService.prototype, 'shouldCaptureEngagementEvents') }) afterEach(() => { verifySignatureSpy.mockRestore() + captureEngagementSpy.mockRestore() }) const postBounce = async ({ functionId, parentRunId, workflowVersion, + distinctId, }: { functionId: string parentRunId?: string workflowVersion?: number + distinctId?: string }): Promise => { // Third arg opts into the versioned payload, which `generate` won't emit by default // until phase two of the rollout — see EMIT_VERSIONED_PAYLOAD in tracking-code.ts. const trackingCode = signer.generate( - { functionId, id: invocationId, teamId: team.id, parentRunId, workflowVersion }, + { functionId, id: invocationId, teamId: team.id, parentRunId, workflowVersion, distinctId }, false, workflowVersion !== undefined ) @@ -562,6 +570,32 @@ describe('EmailTrackingService', () => { }) }) + it('captures the bounce event with the send it came from', async () => { + captureEngagementSpy.mockResolvedValue(true) + const hogFlow = await insertHogFlow(hub.postgres, { + ...new FixtureHogFlowBuilder().withTeamId(team.id).build(), + version: 5, + }) + + const res = await postBounce({ functionId: hogFlow.id, workflowVersion: 2, distinctId: 'user-1' }) + expect(res.status).toBe(200) + + await waitForExpect(() => { + const captured = mockInternalFetch.mock.calls.map(([, options]) => parseJSON(options.body)) + expect(captured).toHaveLength(1) + expect(captured[0]).toMatchObject({ + event: '$workflows_email_bounced', + distinct_id: 'user-1', + properties: { + $workflow_id: hogFlow.id, + $workflow_invocation_id: invocationId, + $workflow_version: 2, + $email_message_id: 'ses-message-id', + }, + }) + }) + }) + it('keys the log entry under parentRunId for batch-triggered runs', async () => { const hogFlow = await insertHogFlow( hub.postgres, diff --git a/nodejs/src/cdp/services/messaging/email-tracking.service.ts b/nodejs/src/cdp/services/messaging/email-tracking.service.ts index 645866c70885..de1baeb7fb83 100644 --- a/nodejs/src/cdp/services/messaging/email-tracking.service.ts +++ b/nodejs/src/cdp/services/messaging/email-tracking.service.ts @@ -3,13 +3,7 @@ import { Counter } from 'prom-client' import express from 'ultimate-express' import { HogFlow } from '~/cdp/schema/hogflow' -import { - CyclotronJobInvocationHogFlow, - CyclotronJobInvocationHogFunction, - LogEntry, - LogEntryLevel, - MinimalAppMetric, -} from '~/cdp/types' +import { CyclotronJobInvocationHogFunction, LogEntry, LogEntryLevel, MinimalAppMetric } from '~/cdp/types' import { ModifiedRequest } from '~/common/api/router' import { isDevEnv, isTestEnv } from '~/common/utils/env-utils' import { parseJSON } from '~/common/utils/json-parse' @@ -116,14 +110,11 @@ export const resolveEmailEngagementDistinctId = ( return invocation.state?.globals?.event?.distinct_id || undefined } -// The workflow version that is sending this message, for the tracking code minted below. A flow's -// email runs as a hog function invocation built by spreading the flow invocation, so `hogFlow` is -// present at runtime even though the type is the narrower hog function shape. A hog function send -// has none, and its engagement lands in the version-agnostic series alone. -export const resolveEmailSendingVersion = (invocation: CyclotronJobInvocationHogFunction): number | undefined => { - return 'hogFlow' in invocation - ? (invocation as unknown as CyclotronJobInvocationHogFlow).hogFlow.version - : undefined +export const resolveEmailSendingAppSourceVersion = ( + invocation: CyclotronJobInvocationHogFunction +): { id: string; version: number } | undefined => { + const version = invocation.state.renderedWorkflowVersion + return version === undefined ? undefined : { id: invocation.functionId, version } } // HTML attribute values arrive entity-encoded (e.g. `&`, `&`). Decode before @@ -159,7 +150,7 @@ export const addTrackingToEmail = ( // The pixel and redirect codes carry the sending version too. SES-attributed opens and clicks // read the version off the signed header instead, but a provider we track ourselves has these // URLs as its only carrier, and without the version its engagement cannot split by version. - const trackingInvocation = { ...invocation, distinctId, workflowVersion: resolveEmailSendingVersion(invocation) } + const trackingInvocation = { ...invocation, distinctId, workflowVersion: invocation.state.renderedWorkflowVersion } const trackingUrl = signer.pixelUrl(trackingInvocation, isTest) // Incremented for every anchor LINK_REGEX matches, including ones that turn out to be opted @@ -395,6 +386,7 @@ export class EmailTrackingService { properties: { $workflow_id: appSourceId, $workflow_action_id: actionId, + $workflow_invocation_id: invocationId, ...(workflowVersion !== undefined ? { $workflow_version: workflowVersion } : {}), ...properties, }, diff --git a/nodejs/src/cdp/services/messaging/email.service.test.ts b/nodejs/src/cdp/services/messaging/email.service.test.ts index 7c3cd05338a0..637176ff34e2 100644 --- a/nodejs/src/cdp/services/messaging/email.service.test.ts +++ b/nodejs/src/cdp/services/messaging/email.service.test.ts @@ -1655,10 +1655,15 @@ describe('EmailService', () => { }) it('records an email_untracked app metric for untracked sends but not for test sends', async () => { + invocation.state.renderedWorkflowVersion = 3 const normal = await service.executeSendEmail(invocation) expect(normal.metrics.map((m) => m.metric_name)).toEqual( expect.arrayContaining(['email_sent', 'email_untracked']) ) + expect(normal.metrics.find((m) => m.metric_name === 'email_untracked')?.app_source_version).toEqual({ + id: invocation.functionId, + version: 3, + }) const testSend = await service.executeSendEmail(invocation, true) expect(testSend.metrics).toEqual([]) @@ -1816,17 +1821,19 @@ describe('EmailService', () => { }) it.each([ - ['an ordinary template', {}, 'Test Subject'], - ['a sensitive template', { sensitive: true }, undefined], + ['an ordinary template', {}, 'Test Subject', undefined], + ['a sensitive template', { sensitive: true }, undefined, undefined], + ['a workflow step', {}, 'Test Subject', 3], ])( - 'should capture a $workflows_email_sent PostHog event on success, for %s', - async (_name, extraMetadata, subject) => { + 'should capture a $workflows_email_sent PostHog event and version the email_sent metric on success, for %s', + async (_name, extraMetadata, subject, workflowVersion) => { // Engagement capture is team-opt-in; enable it for this team so the captured event is emitted. jest.spyOn( (service as any).teamWorkflowsConfigService, 'shouldCaptureEngagementEvents' ).mockResolvedValue(true) invocation.hogFunction.metadata = { ...invocation.hogFunction.metadata, ...extraMetadata } + invocation.state.renderedWorkflowVersion = workflowVersion sendEmailSpy.mockResolvedValue({ MessageId: 'test-message-id' }) const result = await service.executeSendEmail(invocation) expect(result.error).toBeUndefined() @@ -1838,11 +1845,19 @@ describe('EmailService', () => { properties: { $workflow_id: invocation.functionId, $workflow_action_id: invocation.state.actionId, + $workflow_invocation_id: 'invocation-1', + $email_message_id: 'test-message-id', $email_to: 'test@example.com', $email_tracking_enabled: true, }, }) expect(result.capturedPostHogEvents[0].properties.$email_subject).toEqual(subject) + expect(result.capturedPostHogEvents[0].properties.$workflow_version).toEqual(workflowVersion) + expect( + result.metrics.find((metric) => metric.metric_name === 'email_sent')?.app_source_version + ).toEqual( + workflowVersion === undefined ? undefined : { id: invocation.functionId, version: workflowVersion } + ) } ) diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index ff83d8746de4..0ba4e3bdf50f 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/nodejs/src/cdp/services/messaging/email.service.ts @@ -25,7 +25,7 @@ import { EmailSuppressionService, SuppressionCheckOptions } from './email-suppre import { addTrackingToEmail, resolveEmailEngagementDistinctId, - resolveEmailSendingVersion, + resolveEmailSendingAppSourceVersion, } from './email-tracking.service' import { mailDevTransport, mailDevWebUrl } from './helpers/maildev' import { maybeAddPreheaderToEmail } from './helpers/preheader' @@ -403,6 +403,7 @@ export class EmailService { let throttled: boolean = false let assetRow: MessageAssetRow | null = null let trackingEnabled = true + let messageId: string | undefined try { // Team-level kill switches: staff suspend all workflow email for a team whose sender @@ -644,7 +645,7 @@ export class EmailService { await this.sendEmailWithMaildev(result, sendParams, from, trackingEnabled, isTest) break case 'ses': - await this.sendEmailWithSES(result, sendParams, from, trackingEnabled, isTest) + messageId = await this.sendEmailWithSES(result, sendParams, from, trackingEnabled, isTest) break case 'unsupported': @@ -696,6 +697,8 @@ export class EmailService { // Test sends (from the editor's "Run test") must not record metrics — keep them out of the // workflow's Metrics tab, mirroring the isTest skip the SES webhook applies to delivery/open/click. if (!isTest) { + const appSourceVersion = resolveEmailSendingAppSourceVersion(invocation) + const sendVersion = appSourceVersion ? { app_source_version: appSourceVersion } : {} result.metrics.push({ team_id: invocation.teamId, app_source_id: invocation.parentRunId ?? invocation.functionId, @@ -703,6 +706,7 @@ export class EmailService { metric_kind: 'email', metric_name: success ? 'email_sent' : 'email_failed', count: 1, + ...sendVersion, }) // Untracked sends can never produce opens/clicks, so record them separately: open/click @@ -715,6 +719,7 @@ export class EmailService { metric_kind: 'email', metric_name: 'email_untracked', count: 1, + ...sendVersion, }) } @@ -724,6 +729,7 @@ export class EmailService { } const distinctId = resolveEmailEngagementDistinctId(invocation) + const workflowVersion = invocation.state.renderedWorkflowVersion if ( distinctId && !isTest && @@ -737,6 +743,9 @@ export class EmailService { properties: { $workflow_id: invocation.functionId, $workflow_action_id: invocation.state.actionId, + $workflow_invocation_id: invocation.id, + ...(workflowVersion !== undefined ? { $workflow_version: workflowVersion } : {}), + ...(messageId ? { $email_message_id: messageId } : {}), $email_to: params.to.email, // A sensitive template's subject can carry a login code, so its events leave it out. ...(invocation.hogFunction?.metadata?.sensitive === true ? {} : { $email_subject: params.subject }), @@ -1091,7 +1100,7 @@ export class EmailService { from: { email: string; name: string }, trackingEnabled: boolean, isTest = false - ): Promise { + ): Promise { if (!this.sesV2Client) { throw new Error('SES is not configured - set SES_REGION and AWS credentials') } @@ -1099,7 +1108,7 @@ export class EmailService { // Full signed code (with distinct_id + isTest) rides in the header; the short unsigned // carrier (no distinct_id/isTest) goes in the SES EmailTag, guaranteed under the 256-char // tag-value limit. The webhook reads the header first and only falls back to the tag. - const workflowVersion = resolveEmailSendingVersion(result.invocation) + const workflowVersion = result.invocation.state.renderedWorkflowVersion const trackingCode = this.trackingCodeSigner.generate( { ...result.invocation, distinctId, workflowVersion }, isTest @@ -1203,6 +1212,7 @@ export class EmailService { if (!response.MessageId) { throw new Error('No messageId returned from SES') } + return response.MessageId } catch (error: unknown) { if (isSesThrottleError(error)) { sesThrottleResponsesTotal.inc({ error_code: error.name }) diff --git a/nodejs/src/cdp/services/messaging/helpers/ses.test.ts b/nodejs/src/cdp/services/messaging/helpers/ses.test.ts index ace2fd9dcbb6..2bb583ddc6f9 100644 --- a/nodejs/src/cdp/services/messaging/helpers/ses.test.ts +++ b/nodejs/src/cdp/services/messaging/helpers/ses.test.ts @@ -62,7 +62,7 @@ describe('SesWebhookHandler', () => { distinctId: 'user-123', teamId: '1', metricName: 'email_opened', - properties: { $email_to: 'to@example.com' }, + properties: { $email_to: 'to@example.com', $email_message_id: 'msg-123' }, timestamp: '2025-10-03T12:01:00Z', }, ]) @@ -104,6 +104,7 @@ describe('SesWebhookHandler', () => { expect(result.metrics?.[0].distinctId).toBe('user-123') expect(result.metrics?.[0].properties).toEqual({ $email_to: 'to@example.com', + $email_message_id: 'msg-123', $link_url: 'https://example.com', }) expect(result.metrics?.[0].timestamp).toBe('2025-10-03T12:02:00Z') @@ -127,6 +128,7 @@ describe('SesWebhookHandler', () => { expect(result.status).toBe(200) expect(result.metrics?.[0].properties).toEqual({ $email_to: 'to@example.com', + $email_message_id: 'msg-123', $link_url: 'https://example.com/pricing?a=1', $link_index: '2', }) @@ -151,6 +153,7 @@ describe('SesWebhookHandler', () => { const result = await handler.handleWebhook({ body, headers: {} }) expect(result.metrics?.[0].properties).toEqual({ $email_to: 'to@example.com', + $email_message_id: 'msg-123', $link_url: 'https://app.posthog.com', $link_index: '1', }) @@ -495,7 +498,7 @@ describe('SesWebhookHandler', () => { distinctId: 'user-123', teamId: '1', metricName: 'email_delivered', - properties: { $email_to: 'to@example.com' }, + properties: { $email_to: 'to@example.com', $email_message_id: 'msg-123' }, timestamp: '2025-10-03T12:03:00Z', }, ]) @@ -790,7 +793,7 @@ describe('SesWebhookHandler', () => { parentRunId: 'batch-run-id', teamId: '1', metricName: 'email_opened', - properties: { $email_to: 'to@example.com' }, + properties: { $email_to: 'to@example.com', $email_message_id: 'msg-123' }, timestamp: '2025-10-03T12:01:00Z', }, ]) diff --git a/nodejs/src/cdp/services/messaging/helpers/ses.ts b/nodejs/src/cdp/services/messaging/helpers/ses.ts index 04a6ced57eac..50fac38d279f 100644 --- a/nodejs/src/cdp/services/messaging/helpers/ses.ts +++ b/nodejs/src/cdp/services/messaging/helpers/ses.ts @@ -756,6 +756,7 @@ export class SesWebhookHandler { const isSensitive = rec.mail.tags?.[SES_SENSITIVE_TAG]?.[0] === 'true' const properties: Record = { $email_to: rec.mail.destination?.[0], + $email_message_id: rec.mail.messageId, ...(isSensitive ? {} : { $email_subject: rec.mail.commonHeaders?.subject }), } diff --git a/nodejs/src/cdp/services/monitoring/hog-function-monitoring.service.ts b/nodejs/src/cdp/services/monitoring/hog-function-monitoring.service.ts index 65c58f842fb5..395c9ec42d4b 100644 --- a/nodejs/src/cdp/services/monitoring/hog-function-monitoring.service.ts +++ b/nodejs/src/cdp/services/monitoring/hog-function-monitoring.service.ts @@ -165,8 +165,7 @@ export class HogFunctionMonitoringService { : result.invocation.functionId // Stamped here rather than at each of the ~10 sites that push onto `result.metrics` // (executor, action handlers, billing). The workflow is re-read from the manager on every - // dequeue, so this is the version that actually executed the step — a run started on v2 - // that reaches its email step after v3 is published attributes that step to v3. + // dequeue, so this is the version that actually executed the step. const appSourceVersion = isHogFlowResult(result) ? { id: result.invocation.hogFlow.id, version: result.invocation.hogFlow.version } : undefined diff --git a/nodejs/src/cdp/types.ts b/nodejs/src/cdp/types.ts index 79aa6c4a11f3..f55d7469b163 100644 --- a/nodejs/src/cdp/types.ts +++ b/nodejs/src/cdp/types.ts @@ -423,6 +423,7 @@ export type CyclotronJobInvocationHogFunctionContext = { actionId?: string // The hogflow action node ID, used for metrics instance_id when executing within a workflow actionStepCount?: number customerTaskIdempotencyVersion?: 1 + renderedWorkflowVersion?: number } export type WorkflowStepResumeStatus = 'completed' | 'failed' | 'cancelled'