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/hog-executor.service.ts b/nodejs/src/cdp/services/hog-executor.service.ts index 42a0a9b17e9d..cd9fc3ccfac7 100644 --- a/nodejs/src/cdp/services/hog-executor.service.ts +++ b/nodejs/src/cdp/services/hog-executor.service.ts @@ -488,7 +488,12 @@ export class HogExecutorService { result = await this.executeFetch(nextInvocation, options) } } else if (queueParamsType === 'sendPushNotification') { - result = await this.pushNotificationService.executeSendPushNotification(nextInvocation) + // Same signal the email branch uses below: `sendEmailsInline` is only set by the + // test panel, so it doubles as "this is a test send". + result = await this.pushNotificationService.executeSendPushNotification( + nextInvocation, + options?.sendEmailsInline ?? false + ) } else if (queueParamsType === 'email') { // Route to the email queue only if we're not already there and the // caller hasn't asked for inline-only execution (e.g. the test panel). 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 1406c7a90eec..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,12 +279,29 @@ 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') ) }) + it('records neither a metric nor an asset for a test send', async () => { + // "Run test" really delivers, but it must not show up as a workflow send: email + // already skips both, and a test row on the Assets tab reads as a real delivery. + respondWith(200) + const invocation = createSendPushNotificationInvocation({ + '$device_push_subscription_test-project': encryptedFields.encrypt('device-token-123'), + }) + invocation.state.actionId = 'action_push_1' + + const result = await serviceWithAssets.executeSendPushNotification(invocation, true) + + expect(result.error).toBeUndefined() + expect(result.metrics).toEqual([]) + expect(result.messageAssets).toEqual([]) + expect(result.logs.map((log) => log.message)).toContainEqual(expect.stringContaining('accepted by FCM')) + }) + it('captures one asset per notification, not one per delivered channel', async () => { respondWith(200) integrationManager.get = jest @@ -298,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 5fc2b4258cd4..cfb904b4786b 100644 --- a/nodejs/src/cdp/services/messaging/push-notification.service.ts +++ b/nodejs/src/cdp/services/messaging/push-notification.service.ts @@ -136,7 +136,8 @@ export class PushNotificationService { @instrumented('push-notification.executeSendPushNotification') async executeSendPushNotification( - invocation: CyclotronJobInvocationHogFunction + invocation: CyclotronJobInvocationHogFunction, + isTest = false ): Promise> { if (invocation.queueParameters?.type !== 'sendPushNotification') { throw new Error('Bad invocation') @@ -154,7 +155,9 @@ export class PushNotificationService { // Business metrics are emitted once, at the terminal outcome below, rather than per channel — a // rescheduled attempt must not re-count the same notification's skips or failures on every retry. const pushMetric = (metricName: 'push_sent' | 'push_skipped' | 'push_failed', count: number): void => { - if (count <= 0) { + // A test send from the editor's "Run test" must not land in the workflow's Metrics tab, + // matching what the email path already does. + if (count <= 0 || isTest) { return } result.metrics.push({ @@ -278,7 +281,9 @@ export class PushNotificationService { // Only a delivered notification is captured, matching email: an asset is a snapshot of what a // recipient received, and a skip has no recipient. Skips stay visible as `push_skipped` plus the // per-channel run log explaining why. - if (this.messageAssetsService && successCount > 0) { + // Skipped for a test send for the same reason the metrics are: the Assets tab should show + // what real recipients were sent, not what an editor preview produced. + if (this.messageAssetsService && successCount > 0 && !isTest) { // Best-effort: the notification is already delivered by this point, so a capture failure // must not fail the invocation. Throwing here would send the whole batch back for a retry // and deliver every notification in it a second time. Losing an Assets row is the cheaper @@ -286,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