perf: move UUID validation up in pipeline

plugin-server/.vscode/launch.json

@@ -40,13 +40,15 @@
             "request": "launch",
             "console": "integratedTerminal",
             "internalConsoleOptions": "neverOpen",
-            "disableOptimisticBPs": true,
-            "program": "${workspaceFolder}/node_modules/.bin/jest",
+            "program": "${workspaceFolder}/node_modules/jest/bin/jest.js",
             "cwd": "${workspaceFolder}",
-            "args": ["--runInBand", "--forceExit", "--watchAll=false"],
-            "env": {
-                "DATABASE_URL": "postgres://localhost:5432/test_posthog"
-            }
+            "args": [
+                "${workspaceRoot}/node_modules/.bin/jest",
+                "--runInBand",
+                "--watchAll=false",
+                "./tests/main/process-event.test.ts",
+                "--testNamePattern='validates UUID passed in eventUUID'"
+            ]
         }
     ]
 }

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

@@ -3,6 +3,8 @@ import { Counter } from 'prom-client'
 
 import { eventDroppedCounter } from '../../../main/ingestion-queues/metrics'
 import { PipelineEvent } from '../../../types'
+import { UUID } from '../../../utils/utils'
+import { captureIngestionWarning } from '../utils'
 import { EventPipelineRunner } from './runner'
 
 export const tokenOrTeamPresentCounter = new Counter({
@@ -23,13 +25,16 @@ export async function populateTeamDataStep(
      * For these, we trust the team_id field value.
      */
 
+    const { db } = runner.hub
+
     // Collect statistics on the shape of events we are ingesting.
     tokenOrTeamPresentCounter
         .labels({
             team_id_present: event.team_id ? 'true' : 'false',
             token_present: event.token ? 'true' : 'false',
         })
         .inc()
+
     // statsd copy as prometheus is currently not supported in worker threads.
     runner.hub.statsd?.increment('ingestion_event_hasauthinfo', {
         team_id_present: event.team_id ? 'true' : 'false',
@@ -38,6 +43,14 @@ export async function populateTeamDataStep(
 
     // If a team_id is present (event captured from an app), trust it and return the event as is.
     if (event.team_id) {
+        // Check for an invalid UUID, which should be blocked by capture, when team_id is present
+        if (!UUID.validateString(event.uuid, false)) {
+            await captureIngestionWarning(db, event.team_id, 'skipping_event_invalid_uuid', {
+                eventUuid: JSON.stringify(event.uuid),
+            })
+            throw new Error(`Not a valid UUID: "${event.uuid}"`)
+        }
+
         return event as PluginEvent
     }
 
@@ -69,6 +82,14 @@ export async function populateTeamDataStep(
         return null
     }
 
+    // Check for an invalid UUID, which should be blocked by capture, when team_id is present
+    if (!UUID.validateString(event.uuid, false)) {
+        await captureIngestionWarning(db, team.id, 'skipping_event_invalid_uuid', {
+            eventUuid: JSON.stringify(event.uuid),
+        })
+        throw new Error(`Not a valid UUID: "${event.uuid}"`)
+    }
+
     event = {
         ...event,
         team_id: team.id,

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

@@ -1,21 +1,12 @@
 import { PluginEvent } from '@posthog/plugin-scaffold'
 import { PreIngestionEvent } from 'types'
 
-import { UUID } from '../../../utils/utils'
 import { parseEventTimestamp } from '../timestamps'
 import { captureIngestionWarning } from '../utils'
 import { EventPipelineRunner } from './runner'
 
 export async function prepareEventStep(runner: EventPipelineRunner, event: PluginEvent): Promise<PreIngestionEvent> {
     const { team_id, uuid } = event
-    const { db } = runner.hub
-
-    if (!UUID.validateString(uuid, false)) {
-        await captureIngestionWarning(db, team_id, 'skipping_event_invalid_uuid', {
-            eventUuid: JSON.stringify(uuid),
-        })
-        throw new Error(`Not a valid UUID: "${uuid}"`)
-    }
 
     const tsParsingIngestionWarnings: Promise<void>[] = []
     const invalidTimestampCallback = function (type: string, details: Record<string, any>) {

plugin-server/tests/main/process-event.test.ts

@@ -2459,26 +2459,40 @@ test('long event name substr', async () => {
     expect(event.event?.length).toBe(200)
 })
 
-test('throws with bad uuid', async () => {
-    await expect(
-        eventsProcessor.processEvent(
-            'xxx',
-            { event: 'E', properties: { price: 299.99, name: 'AirPods Pro' } } as any as PluginEvent,
+describe('validates UUID passed in eventUuid', () => {
+    test('throws with invalid uuid string', async () => {
+        await processEvent(
+            'i_am_a_distinct_id',
+            '',
+            '',
+            {
+                event: 'E',
+                properties: { price: 299.99, name: 'AirPods Pro', distinct_id: 'distinct_id1' },
+            } as any as PluginEvent,
             team.id,
-            DateTime.utc(),
+            now,
             'this is not an uuid'
         )
-    ).rejects.toEqual(new Error('Not a valid UUID: "this is not an uuid"'))
 
-    await expect(
-        eventsProcessor.processEvent(
+        // expect((await hub.db.fetchEvents()).length).toBe(1)
+        const [event] = await hub.db.fetchEvents()
+        expect(event.event).toEqual('Not a valid UUID: "this is not an uuid"')
+    })
+    test('throws with null value in eventUUID', async () => {
+        await processEvent(
             'xxx',
+            '',
+            '',
             { event: 'E', properties: { price: 299.99, name: 'AirPods Pro' } } as any as PluginEvent,
             team.id,
             DateTime.utc(),
             null as any
         )
-    ).rejects.toEqual(new Error('Not a valid UUID: "null"'))
+
+        expect((await hub.db.fetchEvents()).length).toBe(1)
+        const [event] = await hub.db.fetchEvents()
+        expect(event.event).toEqual('Not a valid UUID: "null"')
+    })
 })
 
 test('any event can do $set on props (user exists)', async () => {