feat(ingest): disable person processing override ingest changes (#25749)

plugin-server/src/types.ts

@@ -627,6 +627,7 @@ export interface Team {
     api_token: string
     slack_incoming_webhook: string | null
     session_recording_opt_in: boolean
+    person_processing_opt_out?: boolean
     heatmaps_opt_in: boolean | null
     ingested_event: boolean
     person_display_name_properties: string[] | null

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

@@ -29,35 +29,23 @@ export async function populateTeamDataStep(
         })
         .inc()
 
-    // 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.kafkaProducer, 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
-    }
-
+    let team = null
     // Events with no token or team_id are dropped, they should be blocked by capture
-    if (!event.token) {
+    if (!event.token && !event.team_id) {
         eventDroppedCounter
             .labels({
                 event_type: 'analytics',
                 drop_cause: 'no_token',
             })
             .inc()
         return null
+    } else if (event.team_id) {
+        team = await runner.hub.teamManager.fetchTeam(event.team_id)
+    } else if (event.token) {
+        team = await runner.hub.teamManager.getTeamByToken(event.token)
     }
 
-    // Team lookup is cached, but will fail if PG is unavailable and the key expired.
-    // We should retry processing this event.
-    const team = await runner.hub.teamManager.getTeamByToken(event.token)
-
-    // If the token does not resolve to an existing team, drop the events.
+    // If the token or team_id does not resolve to an existing team, drop the events.
     if (!team) {
         eventDroppedCounter
             .labels({
@@ -76,6 +64,18 @@ export async function populateTeamDataStep(
         throw new Error(`Not a valid UUID: "${event.uuid}"`)
     }
 
+    // We allow teams to set the person processing mode on a per-event basis, but override
+    // it with the team-level setting, if it's set to opt-out (since this is billing related,
+    // we go with preferring not to do the processing even if the event says to do it, if the
+    // setting says not to).
+    if (team.person_processing_opt_out) {
+        if (event.properties) {
+            event.properties.$process_person_profile = false
+        } else {
+            event.properties = { $process_person_profile: false }
+        }
+    }
+
     event = {
         ...event,
         team_id: team.id,

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

@@ -143,6 +143,7 @@ export class EventPipelineRunner {
         const kafkaAcks: Promise<void>[] = []
 
         let processPerson = true // The default.
+        // Set either at capture time, or in the populateTeamData step, if team-level opt-out is enabled.
         if (event.properties && '$process_person_profile' in event.properties) {
             const propValue = event.properties.$process_person_profile
             if (propValue === true) {

plugin-server/src/worker/ingestion/team-manager.ts

@@ -161,6 +161,7 @@ export async function fetchTeam(client: PostgresRouter, teamId: Team['id']): Pro
                 api_token,
                 slack_incoming_webhook,
                 session_recording_opt_in,
+                person_processing_opt_out,
                 heatmaps_opt_in,
                 ingested_event,
                 person_display_name_properties,
@@ -187,6 +188,7 @@ export async function fetchTeamByToken(client: PostgresRouter, token: string): P
                 api_token,
                 slack_incoming_webhook,
                 session_recording_opt_in,
+                person_processing_opt_out,
                 heatmaps_opt_in,
                 ingested_event,
                 test_account_filters

plugin-server/tests/main/db.test.ts

@@ -859,6 +859,7 @@ describe('DB', () => {
                 name: 'TEST PROJECT',
                 organization_id: organizationId,
                 session_recording_opt_in: true,
+                person_processing_opt_out: null,
                 heatmaps_opt_in: null,
                 slack_incoming_webhook: null,
                 uuid: expect.any(String),
@@ -887,6 +888,7 @@ describe('DB', () => {
                 name: 'TEST PROJECT',
                 organization_id: organizationId,
                 session_recording_opt_in: true,
+                person_processing_opt_out: null,
                 heatmaps_opt_in: null,
                 slack_incoming_webhook: null,
                 uuid: expect.any(String),

plugin-server/tests/worker/ingestion/event-pipeline/populateTeamDataStep.test.ts

@@ -38,6 +38,16 @@ beforeEach(() => {
                 getTeamByToken: jest.fn((token) => {
                     return token === teamTwoToken ? teamTwo : null
                 }),
+
+                fetchTeam: jest.fn((teamId) => {
+                    if (teamId === 2) {
+                        return teamTwo
+                    }
+                    if (teamId === 3) {
+                        return { ...teamTwo, person_processing_opt_out: true }
+                    }
+                    return null
+                }),
             },
         },
     }
@@ -90,12 +100,17 @@ describe('populateTeamDataStep()', () => {
     })
 
     it('event with a team_id value is returned unchanged', async () => {
-        jest.mocked(runner.hub.teamManager.getTeamByToken).mockRejectedValueOnce(new Error('should not be called'))
-        const input = { ...pipelineEvent, team_id: 43 }
+        const input = { ...pipelineEvent, team_id: 2 }
         const response = await populateTeamDataStep(runner, input)
         expect(response).toEqual(input)
     })
 
+    it('event with a team_id whose team is opted-out from person processing', async () => {
+        const input = { ...pipelineEvent, team_id: 3 }
+        const response = await populateTeamDataStep(runner, input)
+        expect(response.properties.$process_person_profile).toBe(false)
+    })
+
     it('PG errors are propagated up to trigger retries', async () => {
         jest.mocked(runner.hub.teamManager.getTeamByToken).mockRejectedValueOnce(new Error('retry me'))
         await expect(async () => {