diff --git a/nodejs/src/cdp/cdp-api.test.ts b/nodejs/src/cdp/cdp-api.test.ts index 71a72330099b..7790390c5594 100644 --- a/nodejs/src/cdp/cdp-api.test.ts +++ b/nodejs/src/cdp/cdp-api.test.ts @@ -1617,7 +1617,7 @@ describe('CDP API', () => { ], capturedPostHogEvents: [], warehouseWebhookPayloads: [], - emailAssets: [], + messageAssets: [], }) ) }) diff --git a/nodejs/src/cdp/services/hog-executor.service.test.ts b/nodejs/src/cdp/services/hog-executor.service.test.ts index 5169d7a45d52..f2d421d5abe8 100644 --- a/nodejs/src/cdp/services/hog-executor.service.test.ts +++ b/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, 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 fc26611b8662..e278ecde7d25 100644 --- a/nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts +++ b/nodejs/src/cdp/services/hogflows/actions/hog_function.test.ts @@ -505,7 +505,7 @@ describe('HogFunctionHandler', () => { metrics: [], capturedPostHogEvents: [], warehouseWebhookPayloads: [], - emailAssets: [], + messageAssets: [], }) const invocationResult = createInvocationResult(invocation, { diff --git a/nodejs/src/cdp/services/hogflows/actions/hog_function.ts b/nodejs/src/cdp/services/hogflows/actions/hog_function.ts index f88a37dec05e..e4dfd2bc4200 100644 --- a/nodejs/src/cdp/services/hogflows/actions/hog_function.ts +++ b/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: [], } } 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 97ce3d0998d6..a5d79a963611 100644 --- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.test.ts +++ b/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, diff --git a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts b/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts index 919714401b72..19a0a2140904 100644 --- a/nodejs/src/cdp/services/hogflows/hogflow-executor.service.ts +++ b/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 } diff --git a/nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.test.ts b/nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.test.ts index a7655ce2fcb4..14bcb321ebaf 100644 --- a/nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.test.ts +++ b/nodejs/src/cdp/services/job-queue/job-queue-postgres-v2.test.ts @@ -52,7 +52,7 @@ describe('CyclotronJobQueuePostgresV2', () => { metrics: [], capturedPostHogEvents: [], warehouseWebhookPayloads: [], - emailAssets: [], + messageAssets: [], ...overrides, } } diff --git a/nodejs/src/cdp/services/job-queue/shared.test.ts b/nodejs/src/cdp/services/job-queue/shared.test.ts index 1806437d5e6e..e36f10ce6604 100644 --- a/nodejs/src/cdp/services/job-queue/shared.test.ts +++ b/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]) diff --git a/nodejs/src/cdp/services/messaging/email.service.ts b/nodejs/src/cdp/services/messaging/email.service.ts index 4c63ec1a00ef..e5a85cd27d09 100644 --- a/nodejs/src/cdp/services/messaging/email.service.ts +++ b/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) } } diff --git a/nodejs/src/cdp/services/messaging/message-assets.service.test.ts b/nodejs/src/cdp/services/messaging/message-assets.service.test.ts index b93c2cf15ddb..595d19340105 100644 --- a/nodejs/src/cdp/services/messaging/message-assets.service.test.ts +++ b/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>, 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() diff --git a/nodejs/src/cdp/services/messaging/message-assets.service.ts b/nodejs/src/cdp/services/messaging/message-assets.service.ts index 28a177864fa7..d808561e22f3 100644 --- a/nodejs/src/cdp/services/messaging/message-assets.service.ts +++ b/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) } } diff --git a/nodejs/src/cdp/services/messaging/push-notification.service.test.ts b/nodejs/src/cdp/services/messaging/push-notification.service.test.ts index fb447e637b8f..34b16ca40351 100644 --- a/nodejs/src/cdp/services/messaging/push-notification.service.test.ts +++ b/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) }) }) diff --git a/nodejs/src/cdp/services/messaging/push-notification.service.ts b/nodejs/src/cdp/services/messaging/push-notification.service.ts index bfb2310ee315..cfb904b4786b 100644 --- a/nodejs/src/cdp/services/messaging/push-notification.service.ts +++ b/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.') diff --git a/nodejs/src/cdp/services/monitoring/hog-invocation-results.service.test.ts b/nodejs/src/cdp/services/monitoring/hog-invocation-results.service.test.ts index baa26582ec0b..3f02c32157fb 100644 --- a/nodejs/src/cdp/services/monitoring/hog-invocation-results.service.test.ts +++ b/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() diff --git a/nodejs/src/cdp/types.ts b/nodejs/src/cdp/types.ts index 0da69755902f..ec72426f936f 100644 --- a/nodejs/src/cdp/types.ts +++ b/nodejs/src/cdp/types.ts @@ -304,7 +304,7 @@ export type CyclotronJobInvocationResult( | 'finished' | 'capturedPostHogEvents' | 'warehouseWebhookPayloads' - | 'emailAssets' + | 'messageAssets' | 'logs' | 'metrics' | 'error' @@ -82,7 +82,7 @@ export function createInvocationResult( finished: true, capturedPostHogEvents: [], warehouseWebhookPayloads: [], - emailAssets: [], + messageAssets: [], logs: [], metrics: [], ...resultParams, diff --git a/nodejs/src/cdp/workflows-e2e.test.ts b/nodejs/src/cdp/workflows-e2e.test.ts index 2d7686554ed9..90efd493bdf6 100644 --- a/nodejs/src/cdp/workflows-e2e.test.ts +++ b/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