chore: Remove runner abstraction from onEvent

plugin-server/src/main/ingestion-queues/batch-processing/each-batch-onevent.ts

@@ -4,6 +4,7 @@ import { EachBatchPayload, KafkaMessage } from 'kafkajs'
 import { RawClickHouseEvent } from '../../../types'
 import { convertToIngestionEvent } from '../../../utils/event'
 import { status } from '../../../utils/status'
+import { processOnEventStep } from '../../../worker/ingestion/event-pipeline/runAsyncHandlersStep'
 import { runInstrumentedFunction } from '../../utils'
 import { KafkaJSIngestionConsumer } from '../kafka-queue'
 import { eventDroppedCounter, latestOffsetTimestampGauge } from '../metrics'
@@ -28,7 +29,7 @@ export async function eachMessageAppsOnEventHandlers(
 
         const event = convertToIngestionEvent(clickHouseEvent, skipElementsChain)
         await runInstrumentedFunction({
-            func: () => queue.workerMethods.runAppsOnEventPipeline(event),
+            func: () => processOnEventStep(queue.pluginsServer, event),
             statsKey: `kafka_queue.process_async_handlers_on_event`,
             timeoutMessage: 'After 30 seconds still running runAppsOnEventPipeline',
             timeoutContext: () => ({

plugin-server/src/main/ingestion-queues/kafka-queue.ts

@@ -6,7 +6,7 @@ import { Counter } from 'prom-client'
 
 import { BatchConsumer, startBatchConsumer } from '../../kafka/batch-consumer'
 import { createRdConnectionConfigFromEnvVars } from '../../kafka/config'
-import { Hub, PipelineEvent, PostIngestionEvent, WorkerMethods } from '../../types'
+import { Hub, PipelineEvent, WorkerMethods } from '../../types'
 import { KafkaConfig } from '../../utils/db/hub'
 import { timeoutGuard } from '../../utils/db/utils'
 import { status } from '../../utils/status'
@@ -59,11 +59,6 @@ export class KafkaJSIngestionConsumer {
         // references to queue.workerMethods buried deep in the codebase
         // #onestepatatime
         this.workerMethods = {
-            runAppsOnEventPipeline: (event: PostIngestionEvent) => {
-                this.pluginsServer.lastActivity = new Date().valueOf()
-                this.pluginsServer.lastActivityType = 'runAppsOnEventPipeline'
-                return piscina.run({ task: 'runAppsOnEventPipeline', args: { event } })
-            },
             runEventPipeline: (event: PipelineEvent) => {
                 this.pluginsServer.lastActivity = new Date().valueOf()
                 this.pluginsServer.lastActivityType = 'runEventPipeline'
@@ -226,11 +221,6 @@ export class IngestionConsumer {
         // references to queue.workerMethods buried deep in the codebase
         // #onestepatatime
         this.workerMethods = {
-            runAppsOnEventPipeline: (event: PostIngestionEvent) => {
-                this.pluginsServer.lastActivity = new Date().valueOf()
-                this.pluginsServer.lastActivityType = 'runAppsOnEventPipeline'
-                return piscina.run({ task: 'runAppsOnEventPipeline', args: { event } })
-            },
             runEventPipeline: (event: PipelineEvent) => {
                 this.pluginsServer.lastActivity = new Date().valueOf()
                 this.pluginsServer.lastActivityType = 'runEventPipeline'

plugin-server/src/types.ts

@@ -476,7 +476,6 @@ export interface PluginTask {
 }
 
 export type WorkerMethods = {
-    runAppsOnEventPipeline: (event: PostIngestionEvent) => Promise<void>
     runEventPipeline: (event: PipelineEvent) => Promise<EventPipelineResult>
 }
 

plugin-server/src/worker/ingestion/event-pipeline/runAsyncHandlersStep.ts

@@ -1,20 +1,19 @@
 import { runInstrumentedFunction } from '../../../main/utils'
-import { PostIngestionEvent } from '../../../types'
+import { Hub, PostIngestionEvent } from '../../../types'
 import { convertToProcessedPluginEvent } from '../../../utils/event'
 import { runOnEvent } from '../../plugins/run'
 import { ActionMatcher } from '../action-matcher'
 import { HookCommander, instrumentWebhookStep } from '../hooks'
-import { EventPipelineRunner } from './runner'
 
-export async function processOnEventStep(runner: EventPipelineRunner, event: PostIngestionEvent) {
+export async function processOnEventStep(hub: Hub, event: PostIngestionEvent) {
     const processedPluginEvent = convertToProcessedPluginEvent(event)
 
     await runInstrumentedFunction({
         timeoutContext: () => ({
             team_id: event.teamId,
             event_uuid: event.eventUuid,
         }),
-        func: () => runOnEvent(runner.hub, processedPluginEvent),
+        func: () => runOnEvent(hub, processedPluginEvent),
         statsKey: `kafka_queue.single_on_event`,
         timeoutMessage: `After 30 seconds still running onEvent`,
         teamId: event.teamId,

plugin-server/src/worker/ingestion/event-pipeline/runner.ts

@@ -4,7 +4,7 @@ import { Counter } from 'prom-client'
 
 import { eventDroppedCounter } from '../../../main/ingestion-queues/metrics'
 import { runInSpan } from '../../../sentry'
-import { Hub, PipelineEvent, PostIngestionEvent } from '../../../types'
+import { Hub, PipelineEvent } from '../../../types'
 import { DependencyUnavailableError } from '../../../utils/db/error'
 import { timeoutGuard } from '../../../utils/db/utils'
 import { stringToBoolean } from '../../../utils/env-utils'
@@ -15,7 +15,6 @@ import { pluginsProcessEventStep } from './pluginsProcessEventStep'
 import { populateTeamDataStep } from './populateTeamDataStep'
 import { prepareEventStep } from './prepareEventStep'
 import { processPersonsStep } from './processPersonsStep'
-import { processOnEventStep } from './runAsyncHandlersStep'
 
 export const silentFailuresAsyncHandlers = new Counter({
     name: 'async_handlers_silent_failure',
@@ -166,26 +165,6 @@ export class EventPipelineRunner {
         }
     }
 
-    async runAppsOnEventPipeline(event: PostIngestionEvent): Promise<EventPipelineResult> {
-        try {
-            this.hub.statsd?.increment('kafka_queue.event_pipeline.start', { pipeline: 'onEvent' })
-            await this.runStep(processOnEventStep, [this, event], event.teamId, false)
-            this.hub.statsd?.increment('kafka_queue.onevent.processed')
-            return this.registerLastStep('processOnEventStep', event.teamId, [event])
-        } catch (error) {
-            if (error instanceof DependencyUnavailableError) {
-                // If this is an error with a dependency that we control, we want to
-                // ensure that the caller knows that the event was not processed,
-                // for a reason that we control and that is transient.
-                throw error
-            }
-
-            silentFailuresAsyncHandlers.inc()
-
-            return { lastStep: error.step, args: [], error: error.message }
-        }
-    }
-
     registerLastStep(
         stepName: string,
         teamId: number | null,

plugin-server/src/worker/tasks.ts

@@ -1,7 +1,6 @@
 import { PluginEvent } from '@posthog/plugin-scaffold/src/types'
 
-import { EnqueuedPluginJob, Hub, PipelineEvent, PluginTaskType, PostIngestionEvent } from '../types'
-import { convertToProcessedPluginEvent } from '../utils/event'
+import { EnqueuedPluginJob, Hub, PipelineEvent, PluginTaskType } from '../types'
 import { EventPipelineRunner } from './ingestion/event-pipeline/runner'
 import { loadSchedule } from './plugins/loadSchedule'
 import { runPluginTask, runProcessEvent } from './plugins/run'
@@ -33,10 +32,6 @@ export const workerTasks: Record<string, TaskRunner> = {
         const runner = new EventPipelineRunner(hub, args.event)
         return await runner.runEventPipeline(args.event)
     },
-    runAppsOnEventPipeline: async (hub, args: { event: PostIngestionEvent }) => {
-        const runner = new EventPipelineRunner(hub, convertToProcessedPluginEvent(args.event))
-        return await runner.runAppsOnEventPipeline(args.event)
-    },
     reloadPlugins: async (hub) => {
         await setupPlugins(hub)
     },

plugin-server/src/worker/worker.ts

@@ -87,11 +87,7 @@ export const createTaskRunner =
                 return response
             },
             (transactionDuration: number) => {
-                if (
-                    task === 'runEventPipeline' ||
-                    task === 'runWebhooksHandlersEventPipeline' ||
-                    task === 'runAppsOnEventPipeline'
-                ) {
+                if (task === 'runEventPipeline') {
                     return transactionDuration > 0.5 ? 1 : 0.01
                 } else {
                     return 1

plugin-server/tests/main/ingestion-queues/analytics-events-ingestion-consumer.test.ts

@@ -55,8 +55,6 @@ describe('eachBatchParallelIngestion with overflow reroute', () => {
                 db: 'database',
             },
             workerMethods: {
-                runAppsOnEventPipeline: jest.fn(),
-                runWebhooksHandlersEventPipeline: jest.fn(),
                 runEventPipeline: jest.fn(() => Promise.resolve({})),
             },
         }
@@ -174,8 +172,6 @@ describe('eachBatchLegacyIngestion with overflow reroute', () => {
                 db: 'database',
             },
             workerMethods: {
-                runAppsOnEventPipeline: jest.fn(),
-                runWebhooksHandlersEventPipeline: jest.fn(),
                 runEventPipeline: jest.fn(() => Promise.resolve({})),
             },
         }

plugin-server/tests/main/ingestion-queues/analytics-events-ingestion-overflow-consumer.test.ts

@@ -56,8 +56,6 @@ describe('eachBatchParallelIngestion with overflow consume', () => {
                 db: 'database',
             },
             workerMethods: {
-                runAppsOnEventPipeline: jest.fn(),
-                runWebhooksHandlersEventPipeline: jest.fn(),
                 runEventPipeline: jest.fn(() => Promise.resolve({})),
             },
         }
@@ -150,8 +148,6 @@ describe('eachBatchLegacyIngestion with overflow consume', () => {
                 db: 'database',
             },
             workerMethods: {
-                runAppsOnEventPipeline: jest.fn(),
-                runWebhooksHandlersEventPipeline: jest.fn(),
                 runEventPipeline: jest.fn(() => Promise.resolve({})),
             },
         }

plugin-server/tests/main/ingestion-queues/each-batch.test.ts

@@ -24,8 +24,11 @@ import {
 import { ActionManager } from '../../../src/worker/ingestion/action-manager'
 import { ActionMatcher } from '../../../src/worker/ingestion/action-matcher'
 import { HookCommander } from '../../../src/worker/ingestion/hooks'
+import { runOnEvent } from '../../../src/worker/plugins/run'
 import { pluginConfig39 } from '../../helpers/plugins'
 
+jest.mock('../../../src/worker/plugins/run')
+
 jest.mock('../../../src/worker/ingestion/event-pipeline/runAsyncHandlersStep', () => {
     const originalModule = jest.requireActual('../../../src/worker/ingestion/event-pipeline/runAsyncHandlersStep')
     return {
@@ -138,32 +141,33 @@ describe('eachBatchX', () => {
                 pluginConfigsPerTeam: new Map(),
             },
             workerMethods: {
-                runAppsOnEventPipeline: jest.fn(),
-                runWebhooksHandlersEventPipeline: jest.fn(),
                 runEventPipeline: jest.fn(() => Promise.resolve({})),
             },
         }
     })
 
     describe('eachBatchAppsOnEventHandlers', () => {
-        it('calls runAppsOnEventPipeline when useful', async () => {
+        it('calls runOnEvent when useful', async () => {
             queue.pluginsServer.pluginConfigsPerTeam.set(2, [pluginConfig39])
             await eachBatchAppsOnEventHandlers(createKafkaJSBatch(clickhouseEvent), queue)
-            expect(queue.workerMethods.runAppsOnEventPipeline).toHaveBeenCalledWith({
-                ...event,
-                properties: {
-                    $ip: '127.0.0.1',
-                },
-            })
+            // TODO fix to jest spy on the actual function
+            expect(runOnEvent).toHaveBeenCalledWith(
+                expect.anything(),
+                expect.objectContaining({
+                    uuid: 'uuid1',
+                    team_id: 2,
+                    distinct_id: 'my_id',
+                })
+            )
             expect(queue.pluginsServer.statsd.timing).toHaveBeenCalledWith(
                 'kafka_queue.each_batch_async_handlers_on_event',
                 expect.any(Date)
             )
         })
-        it('skip runAppsOnEventPipeline when no pluginconfig for team', async () => {
+        it('skip runOnEvent when no pluginconfig for team', async () => {
             queue.pluginsServer.pluginConfigsPerTeam.clear()
             await eachBatchAppsOnEventHandlers(createKafkaJSBatch(clickhouseEvent), queue)
-            expect(queue.workerMethods.runAppsOnEventPipeline).not.toHaveBeenCalled()
+            expect(runOnEvent).not.toHaveBeenCalled()
             expect(queue.pluginsServer.statsd.timing).toHaveBeenCalledWith(
                 'kafka_queue.each_batch_async_handlers_on_event',
                 expect.any(Date)
@@ -179,13 +183,15 @@ describe('eachBatchX', () => {
                 createKafkaJSBatch({ ...clickhouseEvent, elements_chain: 'random' }),
                 queue
             )
-            expect(queue.workerMethods.runAppsOnEventPipeline).toHaveBeenCalledWith({
-                ...event,
-                elementsList: [{ attributes: {}, order: 0, tag_name: 'random' }],
-                properties: {
-                    $ip: '127.0.0.1',
-                },
-            })
+            expect(runOnEvent).toHaveBeenCalledWith(
+                expect.anything(),
+                expect.objectContaining({
+                    uuid: 'uuid1',
+                    team_id: 2,
+                    distinct_id: 'my_id',
+                    elements: [{ attributes: {}, order: 0, tag_name: 'random' }],
+                })
+            )
         })
         it('skips elements parsing when not useful', async () => {
             queue.pluginsServer.pluginConfigsPerTeam.set(2, [
@@ -197,12 +203,15 @@ describe('eachBatchX', () => {
                 createKafkaJSBatch({ ...clickhouseEvent, elements_chain: 'random' }),
                 queue
             )
-            expect(queue.workerMethods.runAppsOnEventPipeline).toHaveBeenCalledWith({
-                ...event,
-                properties: {
-                    $ip: '127.0.0.1',
-                },
-            })
+            expect(runOnEvent).toHaveBeenCalledWith(
+                expect.anything(),
+                expect.objectContaining({
+                    uuid: 'uuid1',
+                    team_id: 2,
+                    distinct_id: 'my_id',
+                    elements: [],
+                })
+            )
         })
     })