feat(cdp): record the queued and sent status of each transactional email

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

@@ -2148,7 +2148,14 @@ describe('CDP API', () => {
                 .set(headers)
                 .send(payload)
 
+        let produceOutput: jest.SpyInstance
+        const emailStatusRows = (): Record<string, unknown>[] =>
+            produceOutput.mock.calls
+                .filter(([output]) => output === 'email_status')
+                .map(([, message]) => parseJSON(message.value.toString()))
+
         beforeEach(async () => {
+            produceOutput = jest.spyOn(api['emailStatusService']['outputs'], 'produce')
             mockQueueInvocations = jest.fn().mockResolvedValue(undefined)
             api['hogflowQueue'] = { queueInvocations: mockQueueInvocations } as any
             hogFlow = await insertHogFlow(transactionalFlow('api', 'active'))
@@ -2186,6 +2193,9 @@ describe('CDP API', () => {
                 expect.objectContaining({ app_source_id: hogFlow.id, metric_name: 'triggered', count: 1 }),
                 'hog_flow'
             )
+            expect(emailStatusRows()).toEqual([
+                expect.objectContaining({ team_id: team.id, message_id: res.body.invocation_id, status: 'queued' }),
+            ])
         })
 
         describe('with an idempotency key', () => {
@@ -2211,6 +2221,9 @@ describe('CDP API', () => {
                     replayed: true,
                 })
                 expect(mockQueueInvocations).toHaveBeenCalledTimes(1)
+                expect(emailStatusRows()).toEqual([
+                    expect.objectContaining({ message_id: first.body.invocation_id, status: 'queued' }),
+                ])
             })
 
             it('rejects the key for a different request', async () => {
@@ -2333,6 +2346,7 @@ describe('CDP API', () => {
             expect(refused.body).toEqual({ error: code })
             expect(await outcomeCount(code)).toEqual(refusedBefore + 1)
             expect(mockQueueInvocations).not.toHaveBeenCalled()
+            expect(emailStatusRows()).toEqual([])
 
             const retry = await send(hogFlow.id, headers, keyed)
 

nodejs/src/cdp/cdp-api.ts

@@ -64,6 +64,7 @@ import { GroupsManagerService } from './services/managers/groups-manager.service
 import { HogFunctionManagerService } from './services/managers/hog-function-manager.service'
 import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service'
 import { EMAIL_QUEUE_PRIORITY } from './services/messaging/email-priority'
+import { EmailStatusService } from './services/messaging/email-status.service'
 import { EmailSuppressionService } from './services/messaging/email-suppression.service'
 import { EmailTrackingService } from './services/messaging/email-tracking.service'
 import { EmailValidationService } from './services/messaging/email-validation.service'
@@ -202,6 +203,7 @@ export class CdpApi {
     private emailValidationService: EmailValidationService
     private teamWorkflowsConfigService: TeamWorkflowsConfigService
     private hogFunctionMonitoringService: HogFunctionMonitoringService
+    private emailStatusService: EmailStatusService
     private resumeBatchJwt: ScopedServiceJwt
 
     constructor(
@@ -226,6 +228,7 @@ export class CdpApi {
         this.hogWatcher = services.hogWatcher
         this.hogWatcherMirror = services.hogWatcherMirror
         this.invocationResultsService = services.invocationResultsService
+        this.emailStatusService = services.emailStatusService
 
         // API-only services. The hog-transformer's monitoring service reuses the same
         // resolved outputs registry as the core CDP services — no separate construction.
@@ -1260,7 +1263,12 @@ export class CdpApi {
             if (idempotencyKey) {
                 await idempotencyKeys.markQueued(team.id, idempotencyKey, fingerprint, invocation.id)
             }
-            await this.invocationResultsService.invocationResultsRowsService.flush()
+            this.emailStatusService.queue({ teamId: team.id, messageId: invocation.id, status: 'queued' })
+            // Flushed together, and bounded, so the best-effort status row can't hold up the 202.
+            await Promise.all([
+                this.invocationResultsService.invocationResultsRowsService.flush(),
+                this.emailStatusService.flush(),
+            ])
             // The run skips the trigger pipeline that counts runs for event-triggered workflows.
             this.hogFunctionMonitoringService.queueAppMetric(
                 {

nodejs/src/cdp/cdp-services.ts

@@ -34,6 +34,7 @@ import { IntegrationManagerService } from './services/managers/integration-manag
 import { PersonsManagerService } from './services/managers/persons-manager.service'
 import { RecipientsManagerService } from './services/managers/recipients-manager.service'
 import { TeamWorkflowsConfigService } from './services/managers/team-workflows-config.service'
+import { EmailStatusService } from './services/messaging/email-status.service'
 import { EmailSuppressionService } from './services/messaging/email-suppression.service'
 import { EmailValidationService } from './services/messaging/email-validation.service'
 import { EmailService, parseTeamEmailCapMode, parseTierCaps } from './services/messaging/email.service'
@@ -118,6 +119,8 @@ export interface CdpCoreServices {
     emailService: EmailService
     /** Buffers rendered-email asset rows and bulk-flushes them at the batch boundary. */
     messageAssetsService: MessageAssetsService
+    /** Buffers transactional email status rows. `invocationResultsService.flush()` produces them. */
+    emailStatusService: EmailStatusService
 }
 
 export type CdpCoreServicesConfig = Pick<
@@ -188,6 +191,9 @@ export type CdpCoreServicesConfig = Pick<
         | 'HOG_INVOCATION_RESULTS_ENABLED'
         | 'MESSAGE_ASSETS_TOPIC'
         | 'MESSAGE_ASSETS_PRODUCER'
+        | 'EMAIL_STATUS_TOPIC'
+        | 'EMAIL_STATUS_PRODUCER'
+        | 'EMAIL_STATUS_ENABLED'
         | 'CDP_WAREHOUSE_SOURCE_WEBHOOKS_TOPIC'
         | 'CDP_WAREHOUSE_SOURCE_WEBHOOKS_PRODUCER'
         | 'CDP_DLQ_ENABLED'
@@ -435,6 +441,7 @@ export function createCdpCoreServices(
     const teamWorkflowsConfigService = new TeamWorkflowsConfigService(deps.postgres, deps.pubSub)
     const outputs = createCdpOutputsRegistry().build(deps.cdpProducerRegistry, config)
     const messageAssetsService = new MessageAssetsService(outputs)
+    const emailStatusService = new EmailStatusService(outputs, config)
     // Constructed here (rather than below with the other messaging services) so it can be threaded
     // into EmailService — the pre-send suppression check lives there so every send path shares one
     // choke point regardless of whether the invocation came from a workflow action or a hog function.
@@ -472,6 +479,7 @@ export function createCdpCoreServices(
         trackingCodeSigner,
         emailSuppressionService,
         recipientsManager,
+        emailStatusService,
         messageAssetsService,
         workflowEmailRateLimiter,
         teamEmailRateLimiter
@@ -576,7 +584,8 @@ export function createCdpCoreServices(
         warehouseWebhooksService,
         capturedEventsService,
         messageAssetsService,
-        conversionWatchersService
+        conversionWatchersService,
+        emailStatusService
     )
 
     const nativeDestinationExecutorService = new NativeDestinationExecutorService(config)
@@ -612,5 +621,6 @@ export function createCdpCoreServices(
         outputs,
         emailService,
         messageAssetsService,
+        emailStatusService,
     }
 }

nodejs/src/cdp/config.ts

@@ -1,6 +1,7 @@
 import {
     KAFKA_APP_METRICS_2,
     KAFKA_CDP_EVENTS_DLQ,
+    KAFKA_EMAIL_STATUS,
     KAFKA_EVENTS_JSON,
     KAFKA_HOG_INVOCATION_RESULTS,
     KAFKA_LOG_ENTRIES,
@@ -133,6 +134,10 @@ export type CdpConfig = ClickhouseConfig & {
     // "Assets" tab.
     MESSAGE_ASSETS_TOPIC: string
     MESSAGE_ASSETS_PRODUCER: CdpProducerName
+    // One row per status change of a transactional email, for the email_status ClickHouse table.
+    EMAIL_STATUS_TOPIC: string
+    EMAIL_STATUS_PRODUCER: CdpProducerName
+    EMAIL_STATUS_ENABLED: boolean
     HOG_INVOCATION_RERUN_MAX_COUNT: number
     // How many rerun wrapper jobs the worker dequeues per cyclotron-v2 poll.
     // Kept small by default — each job runs a full ClickHouse query per page.
@@ -374,6 +379,11 @@ export function getDefaultCdpConfig(): CdpConfig {
         // Same cyclotron Warpstream cluster as hog_invocation_results — ClickHouse
         // consumes message_assets from the warpstream_cyclotron named collection.
         MESSAGE_ASSETS_PRODUCER: WARPSTREAM_CYCLOTRON_PRODUCER,
+        EMAIL_STATUS_TOPIC: KAFKA_EMAIL_STATUS,
+        // ClickHouse consumes email_status from the warpstream_cyclotron named collection.
+        EMAIL_STATUS_PRODUCER: WARPSTREAM_CYCLOTRON_PRODUCER,
+        // Off by default in production, so that writes start only once the email_status table exists.
+        EMAIL_STATUS_ENABLED: isDevEnv() || isTestEnv(),
         // Hard cap on rows a single rerun wrapper job will drain. Mirrors the
         // Django serializer's HOG_INVOCATION_RERUN_MAX_COUNT (same env var).
         HOG_INVOCATION_RERUN_MAX_COUNT: 10000,

nodejs/src/cdp/outputs/outputs.ts

@@ -12,3 +12,6 @@ export type WarehouseSourceWebhooksOutput = typeof WAREHOUSE_SOURCE_WEBHOOKS_OUT
 
 export const CDP_EVENTS_DLQ_OUTPUT = 'cdp_events_dlq' as const
 export type CdpEventsDlqOutput = typeof CDP_EVENTS_DLQ_OUTPUT
+
+export const EMAIL_STATUS_OUTPUT = 'email_status' as const
+export type EmailStatusOutput = typeof EMAIL_STATUS_OUTPUT

nodejs/src/cdp/outputs/registry.ts

@@ -6,7 +6,7 @@ import {
 } from '~/common/outputs'
 import { IngestionOutputsBuilder } from '~/common/outputs/ingestion-outputs-builder'
 
-import { CDP_EVENTS_DLQ_OUTPUT, WAREHOUSE_SOURCE_WEBHOOKS_OUTPUT } from './outputs'
+import { CDP_EVENTS_DLQ_OUTPUT, EMAIL_STATUS_OUTPUT, WAREHOUSE_SOURCE_WEBHOOKS_OUTPUT } from './outputs'
 
 /**
  * Outputs the CDP deployments write to. Each output's topic + producer are
@@ -18,6 +18,7 @@ import { CDP_EVENTS_DLQ_OUTPUT, WAREHOUSE_SOURCE_WEBHOOKS_OUTPUT } from './outpu
  *   `clickhouse_app_metrics2` schema).
  * - `WAREHOUSE_SOURCE_WEBHOOKS_OUTPUT` — warehouse source webhook payloads.
  * - `CDP_EVENTS_DLQ_OUTPUT` — events that matched a function but produced no invocation.
+ * - `EMAIL_STATUS_OUTPUT` — status changes of transactional emails.
  */
 export function createCdpOutputsRegistry() {
     return new IngestionOutputsBuilder()
@@ -45,4 +46,8 @@ export function createCdpOutputsRegistry() {
             topicKey: 'CDP_EVENTS_DLQ_TOPIC',
             producerKey: 'CDP_EVENTS_DLQ_PRODUCER',
         })
+        .register(EMAIL_STATUS_OUTPUT, {
+            topicKey: 'EMAIL_STATUS_TOPIC',
+            producerKey: 'EMAIL_STATUS_PRODUCER',
+        })
 }

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

@@ -88,7 +88,8 @@ describe('Hog Executor', () => {
             hub.SITE_URL,
             new EmailTrackingCodeSigner(hub.ENCRYPTION_SALT_KEYS, hub.CDP_EMAIL_TRACKING_URL),
             new EmailSuppressionService(hub.postgres, emailSuppressionConfigFromEnv()),
-            new RecipientsManagerService(hub.postgres)
+            new RecipientsManagerService(hub.postgres),
+            null
         )
         const recipientTokensService = new RecipientTokensService(hub.ENCRYPTION_SALT_KEYS, hub.SITE_URL)
         return new HogExecutorAsyncService(

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

@@ -87,7 +87,8 @@ describe('HogFunctionHandler', () => {
             hub.SITE_URL,
             new EmailTrackingCodeSigner(hub.ENCRYPTION_SALT_KEYS, hub.CDP_EMAIL_TRACKING_URL),
             new EmailSuppressionService(hub.postgres, emailSuppressionConfigFromEnv()),
-            new RecipientsManagerService(hub.postgres)
+            new RecipientsManagerService(hub.postgres),
+            null
         )
         mockHogFunctionExecutor = new HogExecutorAsyncService(
             new HogExecutorService({ executionTimeoutMs: hub.CDP_WATCHER_HOG_COST_TIMING_UPPER_MS }, hogInputsService),

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

@@ -99,7 +99,8 @@ describe('Hogflow Executor', () => {
             hub.SITE_URL,
             new EmailTrackingCodeSigner(hub.ENCRYPTION_SALT_KEYS, hub.CDP_EMAIL_TRACKING_URL),
             emailSuppressionService,
-            new RecipientsManagerService(hub.postgres)
+            new RecipientsManagerService(hub.postgres),
+            null
         )
         const recipientTokensService = new RecipientTokensService(hub.ENCRYPTION_SALT_KEYS, hub.SITE_URL)
         const hogExecutor = new HogExecutorAsyncService(

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

@@ -2,6 +2,7 @@ import { CyclotronJobInvocationResult } from '../types'
 import { CapturedEventsService } from './captured-events/captured-events.service'
 import { ConversionWatchersService } from './conversion-watchers/conversion-watchers.service'
 import { InvocationResultsService } from './invocation-results.service'
+import { EmailStatusService } from './messaging/email-status.service'
 import { MessageAssetsService } from './messaging/message-assets.service'
 import { HogFunctionMonitoringService } from './monitoring/hog-function-monitoring.service'
 import { HogInvocationResultsService } from './monitoring/hog-invocation-results.service'
@@ -14,6 +15,7 @@ describe('InvocationResultsService', () => {
     let capturedEventsService: jest.Mocked<CapturedEventsService>
     let messageAssetsService: jest.Mocked<MessageAssetsService>
     let conversionWatchersService: jest.Mocked<ConversionWatchersService>
+    let emailStatusService: jest.Mocked<EmailStatusService>
     let service: InvocationResultsService
 
     const results = [
@@ -52,13 +54,18 @@ describe('InvocationResultsService', () => {
             flush: jest.fn().mockResolvedValue(undefined),
         } as unknown as jest.Mocked<ConversionWatchersService>
 
+        emailStatusService = {
+            flush: jest.fn().mockResolvedValue(undefined),
+        } as unknown as jest.Mocked<EmailStatusService>
+
         service = new InvocationResultsService(
             monitoringService,
             invocationResultsRowsService,
             warehouseWebhooksService,
             capturedEventsService,
             messageAssetsService,
-            conversionWatchersService
+            conversionWatchersService,
+            emailStatusService
         )
     })
 
@@ -114,6 +121,7 @@ describe('InvocationResultsService', () => {
             expect(warehouseWebhooksService.flush).toHaveBeenCalledTimes(1)
             expect(capturedEventsService.flush).toHaveBeenCalledTimes(1)
             expect(messageAssetsService.flush).toHaveBeenCalledTimes(1)
+            expect(emailStatusService.flush).toHaveBeenCalledTimes(1)
         })
 
         it('flushes all in parallel (all start before any finish)', async () => {

nodejs/src/cdp/services/invocation-results.service.ts

@@ -3,6 +3,7 @@ import { instrumentFn } from '~/common/tracing/tracing-utils'
 import { CyclotronJobInvocationResult } from '../types'
 import { CapturedEventsService } from './captured-events/captured-events.service'
 import { ConversionWatchersService } from './conversion-watchers/conversion-watchers.service'
+import { EmailStatusService } from './messaging/email-status.service'
 import { MessageAssetsService } from './messaging/message-assets.service'
 import { HogFunctionMonitoringService } from './monitoring/hog-function-monitoring.service'
 import { HogInvocationResultsService } from './monitoring/hog-invocation-results.service'
@@ -32,6 +33,8 @@ export interface ResultSink {
  *                                    Assets tab
  * - `ConversionWatchersService`   — per-run conversion watchers, which outlive the
  *                                    run so a late conversion stays observable
+ * - `EmailStatusService`          — status rows of transactional emails that the
+ *                                    email service queues at send time
  *
  * Callers interact with this one service instead of coordinating queue/flush
  * calls across the five individually. `queueInvocationResultsAndFlush` is the
@@ -48,7 +51,8 @@ export class InvocationResultsService {
         public readonly warehouseWebhooksService: WarehouseWebhooksService,
         public readonly capturedEventsService: CapturedEventsService,
         public readonly messageAssetsService: MessageAssetsService,
-        public readonly conversionWatchersService: ConversionWatchersService
+        public readonly conversionWatchersService: ConversionWatchersService,
+        public readonly emailStatusService: EmailStatusService
     ) {
         this.sinks = [
             monitoringService,
@@ -57,6 +61,7 @@ export class InvocationResultsService {
             capturedEventsService,
             messageAssetsService,
             conversionWatchersService,
+            emailStatusService,
         ]
     }