feat(cdp): record skipped, failed and delivered transactional emails

nodejs/src/cdp/cdp-api.ts

@@ -252,7 +252,8 @@ export class CdpApi {
             services.capturedEventsService,
             services.teamWorkflowsConfigService,
             new EmailTrackingCodeSigner(config.ENCRYPTION_SALT_KEYS, config.CDP_EMAIL_TRACKING_URL),
-            services.emailSuppressionService
+            services.emailSuppressionService,
+            services.emailStatusService
         )
         this.groupsManager = new GroupsManagerService(deps.teamManager, deps.groupRepository)
         this.batchExportHogFunctionService = new BatchExportHogFunctionService(

nodejs/src/cdp/cdp-services.ts

@@ -565,6 +565,7 @@ export function createCdpCoreServices(
         emailValidationService,
         cohortMembershipRepository,
         deps.integrationManager,
+        emailStatusService,
         hogFlowDuplicateObserver,
         cdpUsageReporter,
         { awaitedStepsEnabled: config.CDP_HOGFLOW_AWAITED_STEPS_ENABLED }

nodejs/src/cdp/consumers/cdp-cyclotron-worker-hogflow.consumer.ts

@@ -110,6 +110,7 @@ export class CdpCyclotronWorkerHogFlow extends CdpCyclotronWorker {
                     })
 
                     failedInvocations.push(item)
+                    this.invocationResultsService.emailStatusService.queueUnavailableRun(item)
 
                     return
                 }

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

@@ -6,13 +6,16 @@ import { register } from 'prom-client'
 import { FixtureHogFlowBuilder } from '~/cdp/_tests/builders/hogflow.builder'
 import { insertHogFunctionTemplate, insertIntegration } from '~/cdp/_tests/fixtures'
 import { createExampleHogFlowInvocation } from '~/cdp/_tests/fixtures-hogflows'
+import { EmailStatusOutput } from '~/cdp/outputs/outputs'
 import { HogFlowAction } from '~/cdp/schema/hogflow'
 import { createInvocationResult } from '~/cdp/utils/invocation-utils'
 import { PosthogJwtAudience } from '~/cdp/utils/jwt-utils'
 import { ScopedServiceJwt } from '~/cdp/utils/scoped-service-jwt'
 import { workflowStepDispatchKeyFromInvocation } from '~/cdp/utils/workflow-step-dispatch-key'
+import { IngestionOutputs } from '~/common/outputs/ingestion-outputs'
 import { closeHub, createHub } from '~/common/utils/db/hub'
 import { parseJSON } from '~/common/utils/json-parse'
+import { UUIDT } from '~/common/utils/utils'
 import { createTestTeamFixture } from '~/tests/helpers/sql'
 import { Hub, Team } from '~/types'
 
@@ -23,6 +26,7 @@ import { HogInputsService } from '../../hog-inputs.service'
 import { HogFunctionTemplateManagerService } from '../../managers/hog-function-template-manager.service'
 import { RecipientsManagerService } from '../../managers/recipients-manager.service'
 import { TeamWorkflowsConfigService } from '../../managers/team-workflows-config.service'
+import { EmailStatusService } from '../../messaging/email-status.service'
 import { EmailSuppressionService, emailSuppressionConfigFromEnv } from '../../messaging/email-suppression.service'
 import { EmailValidationService } from '../../messaging/email-validation.service'
 import { EmailService } from '../../messaging/email.service'
@@ -31,6 +35,7 @@ import { RecipientPreferencesService } from '../../messaging/recipient-preferenc
 import { RecipientTokensService } from '../../messaging/recipient-tokens.service'
 import { CdpUsageReporterService } from '../../usage/cdp-usage-reporter.service'
 import { HogFlowFunctionsService } from '../hogflow-functions.service'
+import { TRANSACTIONAL_EMAIL_EVENT } from '../hogflow-sensitivity'
 import { findActionByType } from '../hogflow-utils'
 import { HogFunctionHandler, messageOptOutsAfterStep } from './hog_function'
 
@@ -133,7 +138,8 @@ describe('HogFunctionHandler', () => {
             mockHogFlowFunctionsService,
             mockRecipientPreferencesService,
             mockEmailValidationService,
-            'fetch'
+            'fetch',
+            null
         )
 
         // Simple hog function that prints the inputs
@@ -557,6 +563,99 @@ describe('HogFunctionHandler', () => {
         }
     )
 
+    describe('transactional email status', () => {
+        const skipOptedOut = (): void => {
+            ;(mockRecipientPreferencesService.shouldSkipAction as jest.Mock).mockResolvedValueOnce('opted_out')
+        }
+        const skipSuppressed = (): void => {
+            ;(mockRecipientPreferencesService.shouldSkipAction as jest.Mock).mockResolvedValueOnce('suppressed')
+        }
+        const skipUnreachable = (): void => {
+            ;(mockEmailValidationService.getSkipReason as jest.Mock).mockResolvedValueOnce(
+                'Skipping send: the domain has no reachable mail servers.'
+            )
+        }
+        const skipFrequencyCapped = (): void => {
+            ;(mockRecipientPreferencesService.isFrequencyCapped as jest.Mock).mockResolvedValueOnce(true)
+        }
+
+        it.each([
+            ['an opt-out', skipOptedOut, TRANSACTIONAL_EMAIL_EVENT, 'function_email', false, 'recipient_opted_out'],
+            [
+                'a suppression',
+                skipSuppressed,
+                TRANSACTIONAL_EMAIL_EVENT,
+                'function_email',
+                false,
+                'recipient_suppressed',
+            ],
+            [
+                'a predicted bounce',
+                skipUnreachable,
+                TRANSACTIONAL_EMAIL_EVENT,
+                'function_email',
+                false,
+                'recipient_unreachable',
+            ],
+            [
+                'the frequency cap',
+                skipFrequencyCapped,
+                TRANSACTIONAL_EMAIL_EVENT,
+                'function_email',
+                false,
+                'frequency_capped',
+            ],
+            ['a suppression of a test send', skipSuppressed, TRANSACTIONAL_EMAIL_EVENT, 'function_email', true, null],
+            ['a suppression of a workflow email', skipSuppressed, '$pageview', 'function_email', false, null],
+            ['a suppression of an SMS step', skipSuppressed, TRANSACTIONAL_EMAIL_EVENT, 'function_sms', false, null],
+        ] as const)(
+            'records the email status of a send skipped by %s',
+            async (_desc, skip, triggerEvent, actionType, isTest, reason) => {
+                const produce = jest.fn().mockResolvedValue(undefined)
+                const emailStatusService = new EmailStatusService(
+                    { produce } as unknown as IngestionOutputs<EmailStatusOutput>,
+                    { EMAIL_STATUS_ENABLED: true }
+                )
+                const handler = new HogFunctionHandler(
+                    mockHogFlowFunctionsService,
+                    mockRecipientPreferencesService,
+                    mockEmailValidationService,
+                    'email',
+                    emailStatusService
+                )
+                invocation.id = new UUIDT().toString()
+                invocation.state.event.event = triggerEvent
+                skip()
+                const invocationResult = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {
+                    queue: 'hog',
+                    queuePriority: 0,
+                })
+
+                await handler.execute({
+                    invocation,
+                    action: { ...action, type: actionType } as Extract<HogFlowAction, { type: typeof actionType }>,
+                    result: invocationResult,
+                    hogExecutorOptions: { isTest },
+                })
+                await emailStatusService.flush()
+
+                const rows = produce.mock.calls.map(([, message]) => parseJSON(message.value.toString()))
+                expect(rows).toEqual(
+                    reason
+                        ? [
+                              expect.objectContaining({
+                                  team_id: team.id,
+                                  message_id: invocation.id,
+                                  status: 'failed',
+                                  reason,
+                              }),
+                          ]
+                        : []
+                )
+            }
+        )
+    })
+
     // The billing kind is the whole point of the per-channel handlers: push bills at its own rate
     // (roughly half of email), so a completed invocation must emit exactly one billable_invocation
     // carrying the handler's billing type — never fall back to another channel's kind.
@@ -575,6 +674,7 @@ describe('HogFunctionHandler', () => {
                 mockRecipientPreferencesService,
                 mockEmailValidationService,
                 billingType,
+                null,
                 usageReporter
             )
 
@@ -618,6 +718,7 @@ describe('HogFunctionHandler', () => {
             mockRecipientPreferencesService,
             mockEmailValidationService,
             'push',
+            null,
             usageReporter
         )
 
@@ -663,7 +764,8 @@ describe('HogFunctionHandler', () => {
                 mockHogFlowFunctionsService,
                 mockRecipientPreferencesService,
                 mockEmailValidationService,
-                billingType
+                billingType,
+                null
             )
             // The run entered on v1; v3 is what is live now and what this step executes under.
             const republished = {
@@ -859,6 +961,7 @@ describe('HogFunctionHandler', () => {
                 mockRecipientPreferencesService,
                 mockEmailValidationService,
                 'fetch',
+                null,
                 undefined,
                 { awaitedStepsEnabled: true }
             )
@@ -925,6 +1028,7 @@ describe('HogFunctionHandler', () => {
                 mockRecipientPreferencesService,
                 mockEmailValidationService,
                 'fetch',
+                null,
                 undefined,
                 { awaitedStepsEnabled }
             )
@@ -1042,6 +1146,7 @@ describe('HogFunctionHandler', () => {
                     mockRecipientPreferencesService,
                     mockEmailValidationService,
                     'fetch',
+                    null,
                     undefined,
                     { awaitedStepsEnabled }
                 )

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

@@ -18,6 +18,7 @@ import {
     MinimalLogEntry,
 } from '../../../types'
 import { HogExecutorExecuteAsyncOptions } from '../../hog-executor-async.service'
+import { EmailSkipReason, EmailStatusService } from '../../messaging/email-status.service'
 import { EmailValidationService } from '../../messaging/email-validation.service'
 import { RecipientPreferencesService } from '../../messaging/recipient-preferences.service'
 import { CdpUsageReporterService } from '../../usage/cdp-usage-reporter.service'
@@ -162,6 +163,9 @@ export class HogFunctionHandler implements ActionHandler {
         private recipientPreferencesService: RecipientPreferencesService,
         private emailValidationService: EmailValidationService,
         private hogFlowActionBillingType: HogFlowActionBillingType,
+        // Required, so that a caller which builds an email handler without it fails to compile. The
+        // handlers for other channels pass null.
+        private emailStatusService: EmailStatusService | null,
         private usageReporter?: Pick<CdpUsageReporterService, 'reportBillableInvocation'>,
         private options: { awaitedStepsEnabled?: boolean } = {}
     ) {}
@@ -402,6 +406,18 @@ export class HogFunctionHandler implements ActionHandler {
         return { scheduledAt: deadline }
     }
 
+    // These checks skip the send before the email service runs, so the email's status is recorded here.
+    private recordSkippedEmail(
+        invocation: CyclotronJobInvocationHogFunction,
+        action: Action,
+        reason: EmailSkipReason,
+        hogExecutorOptions?: HogExecutorExecuteAsyncOptions
+    ): void {
+        if (action.type === 'function_email') {
+            this.emailStatusService?.queueSkippedSend(invocation, reason, hogExecutorOptions?.isTest ?? false)
+        }
+    }
+
     private async executeHogFunction(
         invocation: CyclotronJobInvocationHogFlow,
         action: Action,
@@ -453,6 +469,12 @@ export class HogFunctionHandler implements ActionHandler {
                           },
                       ]
                     : []
+            this.recordSkippedEmail(
+                hogFunctionInvocation,
+                action,
+                skipReason === 'suppressed' ? 'recipient_suppressed' : 'recipient_opted_out',
+                hogExecutorOptions
+            )
             return {
                 finished: true,
                 skipped: true,
@@ -475,6 +497,7 @@ export class HogFunctionHandler implements ActionHandler {
             () => this.emailValidationService.getSkipReason(hogFunctionInvocation, action)
         )
         if (emailSkipReason) {
+            this.recordSkippedEmail(hogFunctionInvocation, action, 'recipient_unreachable', hogExecutorOptions)
             return {
                 finished: true,
                 skipped: true,
@@ -504,6 +527,7 @@ export class HogFunctionHandler implements ActionHandler {
             !hogExecutorOptions?.isTest &&
             (await this.recipientPreferencesService.isFrequencyCapped(hogFunctionInvocation, action))
         ) {
+            this.recordSkippedEmail(hogFunctionInvocation, action, 'frequency_capped', hogExecutorOptions)
             return {
                 finished: true,
                 skipped: true,

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

@@ -184,7 +184,8 @@ describe('Hogflow Executor', () => {
             recipientPreferencesService,
             emailValidationService,
             cohortMembershipRepository,
-            hub.integrationManager
+            hub.integrationManager,
+            null
         )
     })
 

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

@@ -30,6 +30,7 @@ import {
     StoredCohortMembership,
 } from '../cohorts/cohort-membership-repository'
 import { HogExecutorExecuteAsyncOptions } from '../hog-executor-async.service'
+import { EmailStatusService } from '../messaging/email-status.service'
 import { EmailValidationService } from '../messaging/email-validation.service'
 import { RecipientPreferencesService } from '../messaging/recipient-preferences.service'
 import { CdpUsageReporterService } from '../usage/cdp-usage-reporter.service'
@@ -133,6 +134,9 @@ export class HogFlowExecutorService {
         emailValidationService: EmailValidationService,
         cohortMembershipRepository: CohortMembershipRepository,
         integrationManager: SlackAppLookup,
+        // Required, so that a caller which builds the executor without it fails to compile. Pass null
+        // only where no email status is recorded.
+        emailStatusService: EmailStatusService | null,
         duplicateObserver?: HogFlowDuplicateObserverService,
         usageReporter?: CdpUsageReporterService,
         options: { awaitedStepsEnabled?: boolean } = {}
@@ -145,6 +149,7 @@ export class HogFlowExecutorService {
             recipientPreferencesService,
             emailValidationService,
             'fetch',
+            null,
             usageReporter,
             options
         )
@@ -153,13 +158,15 @@ export class HogFlowExecutorService {
             recipientPreferencesService,
             emailValidationService,
             'email',
+            emailStatusService,
             usageReporter
         )
         const hogFunctionSmsHandler = new HogFunctionHandler(
             hogFlowFunctionsService,
             recipientPreferencesService,
             emailValidationService,
             'sms',
+            null,
             usageReporter,
             options
         )
@@ -168,6 +175,7 @@ export class HogFlowExecutorService {
             recipientPreferencesService,
             emailValidationService,
             'push',
+            null,
             usageReporter
         )
 

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

@@ -55,6 +55,7 @@ describe('InvocationResultsService', () => {
         } as unknown as jest.Mocked<ConversionWatchersService>
 
         emailStatusService = {
+            queueInvocationResults: jest.fn(),
             flush: jest.fn().mockResolvedValue(undefined),
         } as unknown as jest.Mocked<EmailStatusService>
 
@@ -83,6 +84,8 @@ describe('InvocationResultsService', () => {
             expect(capturedEventsService.queueInvocationResults).toHaveBeenCalledWith(results)
             expect(messageAssetsService.queueInvocationResults).toHaveBeenCalledTimes(1)
             expect(messageAssetsService.queueInvocationResults).toHaveBeenCalledWith(results)
+            expect(emailStatusService.queueInvocationResults).toHaveBeenCalledTimes(1)
+            expect(emailStatusService.queueInvocationResults).toHaveBeenCalledWith(results)
         })
 
         it('awaits the captured-events service (which is async)', async () => {

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

@@ -33,8 +33,9 @@ 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
+ * - `EmailStatusService`          — status rows of transactional emails: the run
+ *                                    outcome when a run fails or is canceled, and
+ *                                    the rows 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
@@ -72,6 +73,7 @@ export class InvocationResultsService {
             this.warehouseWebhooksService.queueInvocationResults(results)
             this.messageAssetsService.queueInvocationResults(results)
             this.conversionWatchersService.queueInvocationResults(results)
+            this.emailStatusService.queueInvocationResults(results)
             await this.capturedEventsService.queueInvocationResults(results)
         })
     }