refactor(plugin-server): refactor the event pipeline

plugin-server/benchmarks/postgres/helpers/piscina.ts

@@ -35,7 +35,7 @@ export function ingestOneEvent(
 export async function ingestCountEvents(piscina: ReturnType<typeof makePiscina>, count: number): Promise<void> {
     const maxPromises = 500
     const promises = Array(maxPromises)
-    const ingestEvent = (event: PluginEvent) => piscina.run({ task: 'ingestEvent', args: { event } })
+    const ingestEvent = (event: PluginEvent) => piscina.run({ task: 'runEventPipeline', args: { event } })
 
     const groups = Math.ceil(count / maxPromises)
     for (let j = 0; j < groups; j++) {

plugin-server/benchmarks/vm/worker.benchmark.ts

@@ -30,7 +30,7 @@ function processOneEvent(
 async function processCountEvents(piscina: ReturnType<typeof makePiscina>, count: number, batchSize = 1) {
     const maxPromises = 1000
     const promises = Array(maxPromises)
-    const processEvent = (event: PluginEvent) => piscina.run({ task: 'processEvent', args: { event } })
+    const processEvent = (event: PluginEvent) => piscina.run({ task: '_testsRunProcessEvent', args: { event } })
 
     const groups = Math.ceil((count * batchSize) / maxPromises)
     for (let j = 0; j < groups; j++) {

plugin-server/package.json

@@ -6,7 +6,7 @@
     "main": "dist/index.js",
     "scripts": {
         "test": "jest --runInBand --forceExit tests/**/*.test.ts",
-        "test:postgres:1": "jest --runInBand --forceExit tests/ --testPathIgnorePatterns tests/clickhouse/ --testPathIgnorePatterns tests/postgres/ --testPathIgnorePatterns tests/shared/process-event.test.ts",
+        "test:postgres:1": "jest --runInBand --forceExit tests/ --testPathIgnorePatterns tests/clickhouse/ --testPathIgnorePatterns tests/postgres/ --testPathIgnorePatterns tests/shared/process-event.test.ts --testPathIgnorePatterns tests/worker/ingestion/event-pipeline",
         "test:postgres:2": "jest --runInBand --forceExit tests/postgres/",
         "test:clickhouse:1": "jest --runInBand --forceExit tests/clickhouse/",
         "test:clickhouse:2": "jest --runInBand --forceExit tests/shared/process-event.test.ts",

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

@@ -60,9 +60,9 @@ export class KafkaQueue implements Queue {
         await runInstrumentedFunction({
             server: this.pluginsServer,
             event: bufferEvent,
-            func: (_) => this.workerMethods.ingestBufferEvent(bufferEvent),
+            func: (_) => this.workerMethods.runBufferEventPipeline(bufferEvent),
             statsKey: `kafka_queue.ingest_buffer_event`,
-            timeoutMessage: 'After 30 seconds still running ingestBufferEvent',
+            timeoutMessage: 'After 30 seconds still running runBufferEventPipeline',
         })
         resolveOffset(message.offset)
     }

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

@@ -39,35 +39,10 @@ export async function startQueues(
     workerMethods: Partial<WorkerMethods> = {}
 ): Promise<Queues> {
     const mergedWorkerMethods = {
-        onEvent: (event: ProcessedPluginEvent) => {
+        runBufferEventPipeline: (event: PreIngestionEvent) => {
             server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'onEvent'
-            return piscina.run({ task: 'onEvent', args: { event } })
-        },
-        onAction: (action: Action, event: ProcessedPluginEvent) => {
-            server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'onAction'
-            return piscina.run({ task: 'onAction', args: { event, action } })
-        },
-        onSnapshot: (event: ProcessedPluginEvent) => {
-            server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'onSnapshot'
-            return piscina.run({ task: 'onSnapshot', args: { event } })
-        },
-        processEvent: (event: PluginEvent) => {
-            server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'processEvent'
-            return piscina.run({ task: 'processEvent', args: { event } })
-        },
-        ingestEvent: (event: PluginEvent) => {
-            server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'ingestEvent'
-            return piscina.run({ task: 'ingestEvent', args: { event } })
-        },
-        ingestBufferEvent: (event: PreIngestionEvent) => {
-            server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'ingestBufferEvent'
-            return piscina.run({ task: 'ingestBufferEvent', args: { event } })
+            server.lastActivityType = 'runBufferEventPipeline'
+            return piscina.run({ task: 'runBufferEventPipeline', args: { event } })
         },
         runEventPipeline: (event: PluginEvent) => {
             server.lastActivity = new Date().valueOf()

plugin-server/src/types.ts

@@ -386,12 +386,7 @@ export interface PluginTask {
 }
 
 export type WorkerMethods = {
-    onEvent: (event: ProcessedPluginEvent) => Promise<void>
-    onAction: (action: Action, event: ProcessedPluginEvent) => Promise<void>
-    onSnapshot: (event: ProcessedPluginEvent) => Promise<void>
-    processEvent: (event: PluginEvent) => Promise<PluginEvent | null>
-    ingestEvent: (event: PluginEvent) => Promise<IngestEventResponse>
-    ingestBufferEvent: (event: PreIngestionEvent) => Promise<IngestEventResponse>
+    runBufferEventPipeline: (event: PreIngestionEvent) => Promise<IngestEventResponse>
     runEventPipeline: (event: PluginEvent) => Promise<void>
 }
 

plugin-server/src/utils/internal-metrics.ts

@@ -43,7 +43,7 @@ export class InternalMetrics {
                 uuid: new UUIDT().toString(),
             }
 
-            promises.push(piscina.run({ task: 'ingestEvent', args: { event } }))
+            promises.push(piscina.run({ task: 'runEventPipeline', args: { event } }))
         }
 
         await Promise.all(promises)

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

@@ -0,0 +1,11 @@
+import { Person, PreIngestionEvent } from '../../../types'
+import { EventPipelineRunner, StepResult } from './runner'
+
+export async function createEventStep(
+    runner: EventPipelineRunner,
+    event: PreIngestionEvent,
+    person: Person | undefined
+): Promise<StepResult> {
+    const [, , elements] = await runner.hub.eventsProcessor.createEvent(event)
+    return runner.nextStep('runAsyncHandlersStep', event, person, elements)
+}

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

@@ -0,0 +1,52 @@
+import { DateTime } from 'luxon'
+
+import { Hub, Person, PreIngestionEvent, TeamId } from '../../../types'
+import { EventPipelineRunner, StepResult } from './runner'
+
+export async function emitToBufferStep(
+    runner: EventPipelineRunner,
+    event: PreIngestionEvent,
+    shouldBuffer: (
+        hub: Hub,
+        event: PreIngestionEvent,
+        person: Person | undefined,
+        teamId: TeamId
+    ) => boolean = shouldSendEventToBuffer
+): Promise<StepResult> {
+    const person = await runner.hub.db.fetchPerson(event.teamId, event.distinctId)
+
+    if (shouldBuffer(runner.hub, event, person, event.teamId)) {
+        await runner.hub.eventsProcessor.produceEventToBuffer(event)
+        return null
+    } else {
+        return runner.nextStep('createEventStep', event, person)
+    }
+}
+
+// context: https://github.com/PostHog/posthog/issues/9182
+// TL;DR: events from a recently created non-anonymous person are sent to a buffer
+// because their person_id might change. We merge based on the person_id of the anonymous user
+// so ingestion is delayed for those events to increase our chances of getting person_id correctly
+export function shouldSendEventToBuffer(
+    hub: Hub,
+    event: PreIngestionEvent,
+    person: Person | undefined,
+    teamId: TeamId
+): boolean {
+    const isAnonymousEvent =
+        event.properties && event.properties['$device_id'] && event.distinctId === event.properties['$device_id']
+    const isRecentPerson =
+        !person || DateTime.now().diff(person.created_at).as('seconds') < hub.BUFFER_CONVERSION_SECONDS
+    const ingestEventDirectly = isAnonymousEvent || event.event === '$identify' || !isRecentPerson
+    const sendToBuffer = !ingestEventDirectly
+
+    if (sendToBuffer) {
+        hub.statsd?.increment('conversion_events_buffer_size', { teamId: event.teamId.toString() })
+    }
+
+    if (!hub.CONVERSION_BUFFER_ENABLED && !hub.conversionBufferEnabledTeams.has(teamId)) {
+        return false
+    }
+
+    return sendToBuffer
+}

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

@@ -0,0 +1,30 @@
+import { PluginEvent } from '@posthog/plugin-scaffold'
+
+import { runInstrumentedFunction } from '../../../main/utils'
+import { runProcessEvent } from '../../plugins/run'
+import { EventPipelineRunner, StepResult } from './runner'
+
+export async function pluginsProcessEventStep(runner: EventPipelineRunner, event: PluginEvent): Promise<StepResult> {
+    let processedEvent: PluginEvent | null = event
+
+    // run processEvent on all events that are not $snapshot
+    if (event.event !== '$snapshot') {
+        processedEvent = await runInstrumentedFunction({
+            server: runner.hub,
+            event,
+            func: (event) => runProcessEvent(runner.hub, event),
+            statsKey: 'kafka_queue.single_event',
+            timeoutMessage: 'Still running plugins on event. Timeout warning after 30 sec!',
+        })
+    }
+
+    if (processedEvent) {
+        return runner.nextStep('prepareEventStep', processedEvent)
+    } else {
+        // processEvent might not return an event. This is expected and plugins, e.g. downsample plugin uses it.
+        runner.hub.statsd?.increment('kafka_queue.dropped_event', {
+            teamID: String(event.team_id),
+        })
+        return null
+    }
+}

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

@@ -0,0 +1,27 @@
+import { PluginEvent } from '@posthog/plugin-scaffold'
+import { DateTime } from 'luxon'
+
+import { EventPipelineRunner, StepResult } from './runner'
+
+export async function prepareEventStep(runner: EventPipelineRunner, event: PluginEvent): Promise<StepResult> {
+    const { ip, site_url, team_id, now, sent_at, uuid } = event
+    const distinctId = String(event.distinct_id)
+    const preIngestionEvent = await runner.hub.eventsProcessor.processEvent(
+        distinctId,
+        ip,
+        event,
+        team_id,
+        DateTime.fromISO(now),
+        sent_at ? DateTime.fromISO(sent_at) : null,
+        uuid!, // it will throw if it's undefined,
+        site_url
+    )
+
+    if (preIngestionEvent && preIngestionEvent.event !== '$snapshot') {
+        return runner.nextStep('emitToBufferStep', preIngestionEvent)
+    } else if (preIngestionEvent && preIngestionEvent.event === '$snapshot') {
+        return runner.nextStep('runAsyncHandlersStep', preIngestionEvent, undefined, undefined)
+    } else {
+        return null
+    }
+}

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

@@ -0,0 +1,47 @@
+import { runInstrumentedFunction } from '../../../main/utils'
+import { Action, Element, Person, PreIngestionEvent } from '../../../types'
+import { convertToProcessedPluginEvent } from '../../../utils/event'
+import { runOnAction, runOnEvent, runOnSnapshot } from '../../plugins/run'
+import { EventPipelineRunner, StepResult } from './runner'
+
+export async function runAsyncHandlersStep(
+    runner: EventPipelineRunner,
+    event: PreIngestionEvent,
+    person: Person | undefined,
+    elements: Element[] | undefined
+): Promise<StepResult> {
+    const promises = []
+    let actionMatches: Action[] = []
+    if (event.event !== '$snapshot') {
+        actionMatches = await runner.hub.actionMatcher.match(event, person, elements)
+        promises.push(runner.hub.hookCannon.findAndFireHooks(event, person, event.siteUrl, actionMatches))
+    }
+
+    const processedPluginEvent = convertToProcessedPluginEvent(event)
+    const isSnapshot = event.event === '$snapshot'
+    const method = isSnapshot ? runOnSnapshot : runOnEvent
+    promises.push(
+        runInstrumentedFunction({
+            server: runner.hub,
+            event: processedPluginEvent,
+            func: (event) => method(runner.hub, event),
+            statsKey: `kafka_queue.single_${isSnapshot ? 'on_snapshot' : 'on_event'}`,
+            timeoutMessage: `After 30 seconds still running ${isSnapshot ? 'onSnapshot' : 'onEvent'}`,
+        })
+    )
+    for (const actionMatch of actionMatches) {
+        promises.push(
+            runInstrumentedFunction({
+                server: runner.hub,
+                event: processedPluginEvent,
+                func: (event) => runOnAction(runner.hub, actionMatch, event),
+                statsKey: `kafka_queue.on_action`,
+                timeoutMessage: 'After 30 seconds still running onAction',
+            })
+        )
+    }
+
+    await Promise.all(promises)
+
+    return null
+}