feat(workflows): tie email events to the send and version that caused them

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<CyclotronJobInvocationHogFlow>(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)
         }
     )
 

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
             }
         }

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,12 +17,15 @@ 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,
         instance_id: data.invocation.state.actionId || data.invocation.id,
         metric_kind: data.billingMetricType,
         metric_name: 'billable_invocation',
         count: 1,
+        ...(appSourceVersion ? { app_source_version: appSourceVersion } : {}),
     })
 }

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 = {

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

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 = '<body><a href="https://example.com/">x</a></body>'
 
             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<supertest.Response> => {
                 // 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,

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,
                 },

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 }
+                )
             }
         )