perf: move UUID validation up in pipeline

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

@@ -1,12 +1,22 @@
 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>) {
         // TODO: make that metric name more generic when transitionning to prometheus

plugin-server/src/worker/ingestion/process-event.ts

@@ -26,13 +26,12 @@ import { elementsToString, extractElements } from '../../utils/db/elements-chain
 import { KafkaProducerWrapper } from '../../utils/db/kafka-producer-wrapper'
 import { safeClickhouseString, sanitizeEventName, timeoutGuard } from '../../utils/db/utils'
 import { status } from '../../utils/status'
-import { castTimestampOrNow, UUID } from '../../utils/utils'
+import { castTimestampOrNow } from '../../utils/utils'
 import { GroupTypeManager } from './group-type-manager'
 import { addGroupProperties } from './groups'
 import { upsertGroup } from './properties-updater'
 import { PropertyDefinitionsManager } from './property-definitions-manager'
 import { TeamManager } from './team-manager'
-import { captureIngestionWarning } from './utils'
 
 export class EventsProcessor {
     pluginsServer: Hub
@@ -66,12 +65,6 @@ export class EventsProcessor {
         timestamp: DateTime,
         eventUuid: string
     ): Promise<PreIngestionEvent> {
-        if (!UUID.validateString(eventUuid, false)) {
-            await captureIngestionWarning(this.db, teamId, 'skipping_event_invalid_uuid', {
-                eventUuid: JSON.stringify(eventUuid),
-            })
-            throw new Error(`Not a valid UUID: "${eventUuid}"`)
-        }
         const singleSaveTimer = new Date()
         const timeout = timeoutGuard('Still inside "EventsProcessor.processEvent". Timeout warning after 30 sec!', {
             event: JSON.stringify(data),