refactor(cdp): rename emailAssets to messageAssets

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

@@ -1617,7 +1617,7 @@ describe('CDP API', () => {
                         ],
                         capturedPostHogEvents: [],
                         warehouseWebhookPayloads: [],
-                        emailAssets: [],
+                        messageAssets: [],
                     })
                 )
         })

nodejs/src/cdp/services/hog-executor.service.test.ts

@@ -141,7 +141,7 @@ describe('Hog Executor', () => {
             expect(result).toEqual({
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
                 invocation: {
                     state: {
                         globals: invocation.state.globals,

nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts

@@ -505,7 +505,7 @@ describe('HogFunctionHandler', () => {
             metrics: [],
             capturedPostHogEvents: [],
             warehouseWebhookPayloads: [],
-            emailAssets: [],
+            messageAssets: [],
         })
 
         const invocationResult = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {

nodejs/src/cdp/services/hogflows/actions/hog_function.ts

@@ -61,7 +61,7 @@ export class HogFunctionHandler implements ActionHandler {
             ...functionResult.warehouseWebhookPayloads,
         ]
         result.metrics = [...result.metrics, ...functionResult.metrics]
-        result.emailAssets = [...result.emailAssets, ...functionResult.emailAssets]
+        result.messageAssets = [...result.messageAssets, ...functionResult.messageAssets]
 
         if (!functionResult.finished) {
             // Set the state of the function result on the substate of the flow for the next execution
@@ -158,7 +158,7 @@ export class HogFunctionHandler implements ActionHandler {
                 metrics,
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
             }
         }
 
@@ -187,7 +187,7 @@ export class HogFunctionHandler implements ActionHandler {
                 ],
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
             }
         }
 

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

@@ -280,7 +280,7 @@ describe('Hogflow Executor', () => {
             expect(result).toEqual({
                 capturedPostHogEvents: [],
                 warehouseWebhookPayloads: [],
-                emailAssets: [],
+                messageAssets: [],
                 invocation: {
                     state: {
                         actionStepCount: 1,

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

@@ -214,7 +214,7 @@ export class HogFlowExecutorService {
         const logs: MinimalLogEntry[] = []
         const capturedPostHogEvents: HogFunctionCapturedEvent[] = []
         const warehouseWebhookPayloads: WarehouseWebhookPayload[] = []
-        const emailAssets: MessageAssetRow[] = []
+        const messageAssets: MessageAssetRow[] = []
 
         const earlyExitResult = await this.shouldExitEarly(invocation, metrics, capturedPostHogEvents)
         if (earlyExitResult) {
@@ -252,7 +252,7 @@ export class HogFlowExecutorService {
             metrics.push(...result.metrics)
             capturedPostHogEvents.push(...result.capturedPostHogEvents)
             warehouseWebhookPayloads.push(...result.warehouseWebhookPayloads)
-            emailAssets.push(...result.emailAssets)
+            messageAssets.push(...result.messageAssets)
 
             if (this.shouldEndHogFlowExecution(result, logs)) {
                 break
@@ -263,7 +263,7 @@ export class HogFlowExecutorService {
         result.metrics = metrics
         result.capturedPostHogEvents = capturedPostHogEvents
         result.warehouseWebhookPayloads = warehouseWebhookPayloads
-        result.emailAssets = emailAssets
+        result.messageAssets = messageAssets
 
         return result
     }

nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.test.ts

@@ -52,7 +52,7 @@ describe('CyclotronJobQueuePostgresV2', () => {
             metrics: [],
             capturedPostHogEvents: [],
             warehouseWebhookPayloads: [],
-            emailAssets: [],
+            messageAssets: [],
             ...overrides,
         }
     }

nodejs/src/cdp/services/job-queue/shared.test.ts

@@ -266,7 +266,7 @@ describe('createInvocationSanitizer', () => {
             metrics: [],
             capturedPostHogEvents: [],
             warehouseWebhookPayloads: [],
-            emailAssets: [],
+            messageAssets: [],
         }
 
         const [sanitized] = sanitizer.sanitizeResults([result])

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

@@ -300,7 +300,7 @@ export class EmailService {
             }
 
             if (success && assetRow) {
-                result.emailAssets.push(assetRow)
+                result.messageAssets.push(assetRow)
             }
         }
 

nodejs/src/cdp/services/messaging/message-assets.service.test.ts

@@ -47,7 +47,7 @@ const invocationWithAction = (id: string, teamId = 7): CyclotronJobInvocationHog
 
 const resultWith = (assets: MessageAssetRow[]): CyclotronJobInvocationResult =>
     ({
-        emailAssets: assets,
+        messageAssets: assets,
     }) as unknown as CyclotronJobInvocationResult
 
 const producedRowAt = (outputs: jest.Mocked<IngestionOutputs<'message_assets'>>, index: number): MessageAssetRow => {
@@ -245,7 +245,7 @@ describe('MessageAssetsService', () => {
             expect(outputs.produce).toHaveBeenCalledTimes(1)
         })
 
-        it('skips results that carry an empty emailAssets array', async () => {
+        it('skips results that carry an empty messageAssets array', async () => {
             service.queueInvocationResults([resultWith([]), resultWith([])])
             await service.flush()
             expect(outputs.produce).not.toHaveBeenCalled()

nodejs/src/cdp/services/messaging/message-assets.service.ts

@@ -210,10 +210,10 @@ export class MessageAssetsService {
 
     queueInvocationResults(results: CyclotronJobInvocationResult[]): void {
         for (const result of results) {
-            if (!result.emailAssets || result.emailAssets.length === 0) {
+            if (!result.messageAssets || result.messageAssets.length === 0) {
                 continue
             }
-            for (const row of result.emailAssets) {
+            for (const row of result.messageAssets) {
                 this.queuedRows.push(row)
             }
         }

nodejs/src/cdp/services/messaging/push-notification.service.test.ts

@@ -243,11 +243,11 @@ describe('PushNotificationService', () => {
                 if (captured === 'nothing') {
                     // An asset is a snapshot of what a recipient received; a send that reached nobody
                     // has none, and email behaves the same way (it captures only on success).
-                    expect(result.emailAssets).toEqual([])
+                    expect(result.messageAssets).toEqual([])
                     return
                 }
-                expect(result.emailAssets).toHaveLength(1)
-                expect(result.emailAssets[0]).toMatchObject({
+                expect(result.messageAssets).toHaveLength(1)
+                expect(result.messageAssets[0]).toMatchObject({
                     kind: 'push',
                     status: captured,
                     action_id: 'action_push_1',
@@ -279,7 +279,7 @@ describe('PushNotificationService', () => {
 
                 expect(result.error).toBeUndefined()
                 expect(result.metrics).toContainEqual(expect.objectContaining({ metric_name: 'push_sent' }))
-                expect(result.emailAssets).toEqual([])
+                expect(result.messageAssets).toEqual([])
                 expect(result.logs.map((log) => log.message)).toContainEqual(
                     expect.stringContaining('could not be captured')
                 )
@@ -298,7 +298,7 @@ describe('PushNotificationService', () => {
 
                 expect(result.error).toBeUndefined()
                 expect(result.metrics).toEqual([])
-                expect(result.emailAssets).toEqual([])
+                expect(result.messageAssets).toEqual([])
                 expect(result.logs.map((log) => log.message)).toContainEqual(expect.stringContaining('accepted by FCM'))
             })
 
@@ -315,7 +315,7 @@ describe('PushNotificationService', () => {
 
                 const result = await serviceWithAssets.executeSendPushNotification(invocation)
 
-                expect(result.emailAssets).toHaveLength(1)
+                expect(result.messageAssets).toHaveLength(1)
             })
         })
 

nodejs/src/cdp/services/messaging/push-notification.service.ts

@@ -291,7 +291,7 @@ export class PushNotificationService {
             try {
                 const assetRow = this.messageAssetsService.buildRowForPush(invocation, params, [...deliveredPlatforms])
                 if (assetRow) {
-                    result.emailAssets.push(assetRow)
+                    result.messageAssets.push(assetRow)
                 }
             } catch (err) {
                 addLog('warn', 'The notification was delivered but could not be captured for the Assets tab.')

nodejs/src/cdp/services/monitoring/hog-invocation-results.service.test.ts

@@ -48,7 +48,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
 
@@ -230,7 +230,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
             await service.flush()
@@ -252,7 +252,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
             await service.flush()
@@ -276,7 +276,7 @@ describe('HogInvocationResultsService', () => {
                     metrics: [],
                     capturedPostHogEvents: [],
                     warehouseWebhookPayloads: [],
-                    emailAssets: [],
+                    messageAssets: [],
                 } as any,
             ])
             await service.flush()

nodejs/src/cdp/types.ts

@@ -304,7 +304,7 @@ export type CyclotronJobInvocationResult<T extends CyclotronJobInvocation = Cycl
     metrics: MinimalAppMetric[]
     capturedPostHogEvents: HogFunctionCapturedEvent[]
     warehouseWebhookPayloads: WarehouseWebhookPayload[]
-    emailAssets: MessageAssetRow[]
+    messageAssets: MessageAssetRow[]
     execResult?: unknown
 }
 

nodejs/src/cdp/utils/invocation-utils.ts

@@ -70,7 +70,7 @@ export function createInvocationResult<T extends CyclotronJobInvocation>(
         | 'finished'
         | 'capturedPostHogEvents'
         | 'warehouseWebhookPayloads'
-        | 'emailAssets'
+        | 'messageAssets'
         | 'logs'
         | 'metrics'
         | 'error'
@@ -82,7 +82,7 @@ export function createInvocationResult<T extends CyclotronJobInvocation>(
         finished: true,
         capturedPostHogEvents: [],
         warehouseWebhookPayloads: [],
-        emailAssets: [],
+        messageAssets: [],
         logs: [],
         metrics: [],
         ...resultParams,

nodejs/src/cdp/workflows-e2e.test.ts

@@ -3385,7 +3385,7 @@ describe('Workflows E2E (email queue)', () => {
     //
     // Email assets used to be produced one-at-a-time via a fire-and-forget Kafka call
     // from `email.service.ts → MessageAssetsService.captureSentEmail`. We've moved that
-    // to a buffer-then-flush pattern that drains `result.emailAssets` at the batch
+    // to a buffer-then-flush pattern that drains `result.messageAssets` at the batch
     // boundary and bulk-produces, gated on broker ack before the consumer commits
     // offsets. These tests pin the end-to-end behavior: one workflow → one asset row in
     // the `message_assets` Kafka topic with the right metadata, and a single batch with