feat(workflows): capture missing activation funnel steps

nodejs/src/cdp/cdp-api.ts

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

nodejs/src/cdp/cdp-services.ts

@@ -43,6 +43,7 @@ import { RecipientTokensService } from './services/messaging/recipient-tokens.se
 import { HogFunctionMonitoringService } from './services/monitoring/hog-function-monitoring.service'
 import { HogInvocationResultsService } from './services/monitoring/hog-invocation-results.service'
 import { HogWatcherService } from './services/monitoring/hog-watcher.service'
+import { WorkflowsActivationReporter } from './services/monitoring/workflows-activation-reporter'
 import { NativeDestinationExecutorService } from './services/native-destination-executor.service'
 import { RateLimiterService } from './services/rate-limiter/rate-limiter.service'
 import { SegmentDestinationExecutorService } from './services/segment-destination-executor.service'
@@ -100,6 +101,7 @@ export interface CdpCoreServices {
     cohortMembershipRepository: CohortMembershipRepository
     hogFlowExecutor: HogFlowExecutorService
     hogFunctionMonitoringService: HogFunctionMonitoringService
+    workflowsActivationReporter: WorkflowsActivationReporter
     cdpUsageReporter: CdpUsageReporterService
     capturedEventsService: CapturedEventsService
     /** Per-invocation lifecycle row producer for the new runs/invocations UI + rerun path. */
@@ -432,6 +434,7 @@ export function createCdpCoreServices(
         transientBounceThreshold: config.EMAIL_SUPPRESSION_TRANSIENT_BOUNCE_THRESHOLD,
     })
     const recipientsManager = new RecipientsManagerService(deps.postgres)
+    const workflowsActivationReporter = new WorkflowsActivationReporter(deps.teamManager)
     // Per-workflow send pacing rides the SES Valkey pool, which is only populated on pods that
     // execute email actions — exactly where the limit is enforced. Null elsewhere disables it.
     const workflowEmailRateLimiter = deps.emailValidationValkey
@@ -461,6 +464,7 @@ export function createCdpCoreServices(
         trackingCodeSigner,
         emailSuppressionService,
         recipientsManager,
+        workflowsActivationReporter,
         messageAssetsService,
         workflowEmailRateLimiter,
         teamEmailRateLimiter
@@ -516,7 +520,11 @@ export function createCdpCoreServices(
         hogExecutorAsync
     )
 
-    const recipientPreferencesService = new RecipientPreferencesService(recipientsManager, emailSuppressionService)
+    const recipientPreferencesService = new RecipientPreferencesService(
+        recipientsManager,
+        emailSuppressionService,
+        workflowsActivationReporter
+    )
     // MX verdicts live on the dedicated SES Valkey (same instance as the SES rate
     // limiter, separate pool). The pool is created by the server only on pods
     // whose capabilities execute email actions; everywhere else this is null
@@ -578,6 +586,7 @@ export function createCdpCoreServices(
         cohortMembershipRepository,
         hogFlowExecutor,
         hogFunctionMonitoringService,
+        workflowsActivationReporter,
         cdpUsageReporter,
         capturedEventsService,
         hogInvocationResultsService,

nodejs/src/cdp/consumers/cdp-base.consumer.ts

@@ -32,6 +32,7 @@ import { RecipientPreferencesService } from '../services/messaging/recipient-pre
 import { HogFunctionMonitoringService } from '../services/monitoring/hog-function-monitoring.service'
 import { HogMaskerService } from '../services/monitoring/hog-masker.service'
 import { HogWatcherService } from '../services/monitoring/hog-watcher.service'
+import { WorkflowsActivationReporter } from '../services/monitoring/workflows-activation-reporter'
 import { NativeDestinationExecutorService } from '../services/native-destination-executor.service'
 import { SegmentDestinationExecutorService } from '../services/segment-destination-executor.service'
 import { CdpUsageReporterService } from '../services/usage/cdp-usage-reporter.service'
@@ -74,6 +75,7 @@ export abstract class CdpConsumerBase<TConfig extends CdpConsumerBaseConfig = Cd
 
     emailService: EmailService
     hogFunctionMonitoringService: HogFunctionMonitoringService
+    workflowsActivationReporter: WorkflowsActivationReporter
     cdpUsageReporter: CdpUsageReporterService
     invocationResultsService: InvocationResultsService
     nativeDestinationExecutorService: NativeDestinationExecutorService
@@ -105,6 +107,7 @@ export abstract class CdpConsumerBase<TConfig extends CdpConsumerBaseConfig = Cd
         this.hogFlowExecutor = services.hogFlowExecutor
         this.emailService = services.emailService
         this.hogFunctionMonitoringService = services.hogFunctionMonitoringService
+        this.workflowsActivationReporter = services.workflowsActivationReporter
         this.cdpUsageReporter = services.cdpUsageReporter
         this.invocationResultsService = services.invocationResultsService
         this.nativeDestinationExecutorService = services.nativeDestinationExecutorService

nodejs/src/cdp/consumers/cdp-data-warehouse-events.consumer.ts

@@ -73,6 +73,7 @@ export class CdpDatawarehouseEventsConsumer extends CdpConsumerBase {
             hogWatcherMirror: this.hogWatcherMirror,
             hogMasker: this.hogMasker,
             hogFunctionMonitoringService: this.hogFunctionMonitoringService,
+            workflowsActivationReporter: this.workflowsActivationReporter,
             quotaLimiting: deps.quotaLimiting,
             redis: this.redis,
             valkeyShadow: this.valkeyShadow,

nodejs/src/cdp/consumers/cdp-dlq-replay.consumer.ts

@@ -151,6 +151,7 @@ export class CdpDlqReplayConsumer extends CdpConsumerBase<PluginsServerConfig> {
             hogWatcherMirror: this.hogWatcherMirror,
             hogMasker: this.hogMasker,
             hogFunctionMonitoringService: this.hogFunctionMonitoringService,
+            workflowsActivationReporter: this.workflowsActivationReporter,
             quotaLimiting: deps.quotaLimiting,
             redis: this.redis,
             valkeyShadow: this.valkeyShadow,

nodejs/src/cdp/consumers/cdp-events.consumer.ts

@@ -71,6 +71,7 @@ export class CdpEventsConsumer<
             hogWatcherMirror: this.hogWatcherMirror,
             hogMasker: this.hogMasker,
             hogFunctionMonitoringService: this.hogFunctionMonitoringService,
+            workflowsActivationReporter: this.workflowsActivationReporter,
             quotaLimiting: deps.quotaLimiting,
             redis: this.redis,
             valkeyShadow: this.valkeyShadow,

nodejs/src/cdp/consumers/cdp-internal-event.consumer.ts

@@ -78,6 +78,7 @@ export class CdpInternalEventsConsumer extends CdpConsumerBase {
             hogWatcherMirror: this.hogWatcherMirror,
             hogMasker: this.hogMasker,
             hogFunctionMonitoringService: this.hogFunctionMonitoringService,
+            workflowsActivationReporter: this.workflowsActivationReporter,
             quotaLimiting: deps.quotaLimiting,
             redis: this.redis,
             valkeyShadow: this.valkeyShadow,

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

@@ -84,7 +84,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),
+            { report: jest.fn() }
         )
         const recipientTokensService = new RecipientTokensService(hub.ENCRYPTION_SALT_KEYS, hub.SITE_URL)
         executor = new HogExecutorAsyncService(

nodejs/src/cdp/services/hog-flow-invocation-pipeline.service.test.ts

@@ -89,6 +89,7 @@ describe('HogFlowInvocationPipeline', () => {
             hogWatcherMirror: hogWatcher,
             hogMasker,
             hogFunctionMonitoringService,
+            workflowsActivationReporter: { report: jest.fn() },
             quotaLimiting,
             redis: {} as any,
             valkeyShadow: { writer: {} as any, reader: {} as any },

nodejs/src/cdp/services/hog-flow-invocation-pipeline.service.ts

@@ -25,6 +25,7 @@ import { shouldBlockHogFlowDueToQuota } from './hogflows/hogflow-quota-limiting'
 import { HogFunctionMonitoringService } from './monitoring/hog-function-monitoring.service'
 import { HogMaskerService } from './monitoring/hog-masker.service'
 import { HogWatcherService, HogWatcherState, sameWatcherStates } from './monitoring/hog-watcher.service'
+import { WorkflowsActivationReporter } from './monitoring/workflows-activation-reporter'
 
 export interface HogFlowInvocationPipelineConfig {
     CDP_RATE_LIMITER_BUCKET_SIZE: number
@@ -39,6 +40,7 @@ export interface HogFlowInvocationPipelineDeps {
     hogWatcherMirror: HogWatcherService
     hogMasker: HogMaskerService
     hogFunctionMonitoringService: HogFunctionMonitoringService
+    workflowsActivationReporter: Pick<WorkflowsActivationReporter, 'report'>
     quotaLimiting: QuotaLimiting
     redis: RedisV2
     valkeyShadow: CdpValkeyShadowPools
@@ -196,6 +198,7 @@ export class HogFlowInvocationPipeline {
                 const isQuotaLimited = await shouldBlockHogFlowDueToQuota(item, {
                     quotaLimiting: this.deps.quotaLimiting,
                     hogFunctionMonitoringService: this.deps.hogFunctionMonitoringService,
+                    workflowsActivationReporter: this.deps.workflowsActivationReporter,
                 })
 
                 if (isQuotaLimited) {

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

@@ -74,7 +74,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),
+            { report: jest.fn() }
         )
         mockHogFunctionExecutor = new HogExecutorAsyncService(
             new HogExecutorService({ executionTimeoutMs: hub.CDP_WATCHER_HOG_COST_TIMING_UPPER_MS }, hogInputsService),
@@ -345,17 +346,23 @@ describe('HogFunctionHandler', () => {
         )
     })
 
-    it('should check recipient preferences before execution', async () => {
+    it.each([false, true])('should check recipient preferences before execution with isTest %s', async (isTest) => {
         const invocationResult = createInvocationResult<CyclotronJobInvocationHogFlow>(invocation, {
             queue: 'hog',
             queuePriority: 0,
         })
 
-        await hogFunctionHandler.execute({ invocation, action, result: invocationResult })
+        await hogFunctionHandler.execute({
+            invocation,
+            action,
+            result: invocationResult,
+            hogExecutorOptions: { isTest },
+        })
 
         const callArgs = (mockRecipientPreferencesService.shouldSkipAction as jest.Mock).mock.calls[0]
         expect(callArgs[0]).toBeTruthy()
         expect(callArgs[1]).toBe(action)
+        expect(callArgs[2]).toBe(isTest)
     })
 
     it('should pass proper inputs to buildHogFunctionInvocation', async () => {

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

@@ -361,7 +361,12 @@ export class HogFunctionHandler implements ActionHandler {
 
         const skipReason = await instrumentFn(
             { key: 'hogFlow.action.hogFunction.recipientPreferences', sendException: false },
-            () => this.recipientPreferencesService.shouldSkipAction(hogFunctionInvocation, action)
+            () =>
+                this.recipientPreferencesService.shouldSkipAction(
+                    hogFunctionInvocation,
+                    action,
+                    hogExecutorOptions?.isTest
+                )
         )
         if (skipReason) {
             // Suppression and opt-out both short-circuit the send, but a customer reading the run

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

@@ -98,7 +98,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),
+            { report: jest.fn() }
         )
         const recipientTokensService = new RecipientTokensService(hub.ENCRYPTION_SALT_KEYS, hub.SITE_URL)
         const hogExecutor = new HogExecutorAsyncService(
@@ -130,7 +131,11 @@ describe('Hogflow Executor', () => {
         const hogFunctionTemplateManager = new HogFunctionTemplateManagerService(hub.postgres)
         hogFlowFunctionsService = new HogFlowFunctionsService(hub.SITE_URL, hogFunctionTemplateManager, hogExecutor)
         const recipientsManager = new RecipientsManagerService(hub.postgres)
-        const recipientPreferencesService = new RecipientPreferencesService(recipientsManager, emailSuppressionService)
+        const recipientPreferencesService = new RecipientPreferencesService(
+            recipientsManager,
+            emailSuppressionService,
+            { report: jest.fn() }
+        )
         // Stubbed to always allow: this suite covers executor routing and flow control,
         // not MX validation (email-validation.service.test.ts does), and the real
         // service would fire live DNS lookups for the fixture recipients here.

nodejs/src/cdp/services/hogflows/hogflow-quota-limiting.test.ts

@@ -167,6 +167,7 @@ describe('HogFlow Quota Limiting', () => {
 
     describe('shouldBlockHogFlowDueToQuota', () => {
         let mockHogFunctionMonitoringService: jest.Mocked<HogFunctionMonitoringService>
+        let mockWorkflowsActivationReporter: { report: jest.Mock }
         const teamId = 123
         const baseItem: CyclotronJobInvocationHogFlow = {
             teamId,
@@ -189,6 +190,7 @@ describe('HogFlow Quota Limiting', () => {
             mockHogFunctionMonitoringService = {
                 queueAppMetric: jest.fn(),
             } as any
+            mockWorkflowsActivationReporter = { report: jest.fn() }
             // Reset the counter spy
             jest.spyOn(counterHogFlowQuotaLimited, 'labels').mockReturnValue({
                 inc: jest.fn(),
@@ -209,29 +211,46 @@ describe('HogFlow Quota Limiting', () => {
             const result = await shouldBlockHogFlowDueToQuota(item, {
                 quotaLimiting: mockQuotaLimiting,
                 hogFunctionMonitoringService: mockHogFunctionMonitoringService,
+                workflowsActivationReporter: mockWorkflowsActivationReporter,
             })
 
             expect(result).toBe(false)
             expect(mockHogFunctionMonitoringService.queueAppMetric).not.toHaveBeenCalled()
             expect(counterHogFlowQuotaLimited.labels).not.toHaveBeenCalled()
+            expect(mockWorkflowsActivationReporter.report).not.toHaveBeenCalled()
         })
 
-        it('should block invocation and emit metrics when quota limited', async () => {
+        it.each([
+            [
+                'an email send on the email quota',
+                'workflow_emails',
+                'function_email',
+                [
+                    [
+                        teamId,
+                        'workflows send blocked',
+                        { reason: 'quota_limited', channel: 'email', workflow_id: 'test-flow-id' },
+                    ],
+                ],
+            ],
+            ['a webhook on the destination quota', 'workflow_destinations_dispatched', 'function', []],
+        ] as const)('should block %s and emit metrics', async (_name, limitedResource, billableActionType, reports) => {
             mockQuotaLimiting.isTeamQuotaLimited.mockImplementation((_teamId, resource) => {
-                return Promise.resolve(resource === 'workflow_emails')
+                return Promise.resolve(resource === limitedResource)
             })
 
             const item: CyclotronJobInvocationHogFlow = {
                 ...baseItem,
                 hogFlow: {
                     ...baseItem.hogFlow,
-                    billable_action_types: ['function_email'],
+                    billable_action_types: [billableActionType],
                 },
             }
 
             const result = await shouldBlockHogFlowDueToQuota(item, {
                 quotaLimiting: mockQuotaLimiting,
                 hogFunctionMonitoringService: mockHogFunctionMonitoringService,
+                workflowsActivationReporter: mockWorkflowsActivationReporter,
             })
 
             expect(result).toBe(true)
@@ -247,6 +266,7 @@ describe('HogFlow Quota Limiting', () => {
                 },
                 'hog_flow'
             )
+            expect(mockWorkflowsActivationReporter.report.mock.calls).toEqual(reports)
         })
 
         it('should handle workflow with no billable actions', async () => {
@@ -263,6 +283,7 @@ describe('HogFlow Quota Limiting', () => {
             const result = await shouldBlockHogFlowDueToQuota(item, {
                 quotaLimiting: mockQuotaLimiting,
                 hogFunctionMonitoringService: mockHogFunctionMonitoringService,
+                workflowsActivationReporter: mockWorkflowsActivationReporter,
             })
 
             expect(result).toBe(false)

nodejs/src/cdp/services/hogflows/hogflow-quota-limiting.ts

@@ -5,6 +5,7 @@ import { HogFlow } from '~/cdp/schema/hogflow'
 import { QuotaLimiting } from '../../../common/services/quota-limiting.service'
 import { CyclotronJobInvocationHogFlow } from '../../types'
 import { HogFunctionMonitoringService } from '../monitoring/hog-function-monitoring.service'
+import { WorkflowsActivationReporter } from '../monitoring/workflows-activation-reporter'
 
 export const counterHogFlowQuotaLimited = new Counter({
     name: 'cdp_hog_flow_quota_limited',
@@ -14,6 +15,7 @@ export const counterHogFlowQuotaLimited = new Counter({
 
 export interface HogFlowQuotaLimitResult {
     isLimited: boolean
+    limitedBy?: 'workflow_emails' | 'workflow_destinations_dispatched'
 }
 
 /**
@@ -41,15 +43,15 @@ export async function checkHogFlowQuotaLimits(
 
     // Check if any billable action type is quota limited
     if (isEmailQuotaLimited && billableActionTypes.includes('function_email')) {
-        return { isLimited: true }
+        return { isLimited: true, limitedBy: 'workflow_emails' }
     }
 
     // Push sends bill as destinations for now, so they fall under the destination quota
     if (
         isDestinationQuotaLimited &&
         (billableActionTypes.includes('function') || billableActionTypes.includes('function_push'))
     ) {
-        return { isLimited: true }
+        return { isLimited: true, limitedBy: 'workflow_destinations_dispatched' }
     }
 
     return { isLimited: false }
@@ -58,6 +60,7 @@ export async function checkHogFlowQuotaLimits(
 export interface HogFlowQuotaLimitingContext {
     quotaLimiting: QuotaLimiting
     hogFunctionMonitoringService: HogFunctionMonitoringService
+    workflowsActivationReporter: Pick<WorkflowsActivationReporter, 'report'>
 }
 
 /**
@@ -84,6 +87,13 @@ export async function shouldBlockHogFlowDueToQuota(
             },
             'hog_flow'
         )
+        if (quotaLimitResult.limitedBy === 'workflow_emails') {
+            void context.workflowsActivationReporter.report(item.teamId, 'workflows send blocked', {
+                reason: 'quota_limited',
+                channel: 'email',
+                workflow_id: item.functionId,
+            })
+        }
         return true
     }