Slightly change the way changing team_id by plugins is illegal

src/worker/plugins/run.ts

@@ -3,6 +3,14 @@ import { PluginEvent } from '@posthog/plugin-scaffold'
 import { PluginConfig, PluginsServer, PluginTaskType } from '../../types'
 import { processError } from '../../utils/db/error'
 
+export class IllegalOperationError extends Error {
+    name = 'IllegalOperationError'
+
+    constructor(operation: string) {
+        super(operation)
+    }
+}
+
 export async function runOnEvent(server: PluginsServer, event: PluginEvent): Promise<void> {
     const pluginsToRun = getPluginsForTeam(server, event.team_id)
 
@@ -59,8 +67,8 @@ export async function runProcessEvent(server: PluginsServer, event: PluginEvent)
             try {
                 returnedEvent = (await processEvent(returnedEvent)) || null
                 if (returnedEvent.team_id != teamId) {
-                    returnedEvent = null // don't try to ingest events with modified teamIDs
-                    throw new Error('Illegal Operation: Plugin tried to change teamID')
+                    returnedEvent.team_id = teamId
+                    throw new IllegalOperationError('Plugin tried to change event.team_id')
                 }
                 pluginsSucceeded.push(`${pluginConfig.plugin?.name} (${pluginConfig.id})`)
             } catch (error) {
@@ -115,6 +123,16 @@ export async function runProcessEventBatch(server: PluginsServer, batch: PluginE
             if (processEventBatch && returnedEvents.length > 0) {
                 try {
                     returnedEvents = (await processEventBatch(returnedEvents)) || []
+                    let wasChangedTeamIdFound = false
+                    for (const returnedEvent of returnedEvents) {
+                        if (returnedEvent.team_id != teamId) {
+                            returnedEvent.team_id = teamId
+                            wasChangedTeamIdFound = true
+                        }
+                    }
+                    if (wasChangedTeamIdFound) {
+                        throw new IllegalOperationError('Plugin tried to change event.team_id')
+                    }
                     pluginsSucceeded.push(`${pluginConfig.plugin?.name} (${pluginConfig.id})`)
                 } catch (error) {
                     await processError(server, pluginConfig, error, returnedEvents[0])

tests/plugins.test.ts

@@ -5,7 +5,7 @@ import { LogLevel, PluginsServer, PluginTaskType } from '../src/types'
 import { clearError, processError } from '../src/utils/db/error'
 import { createServer } from '../src/utils/db/server'
 import { loadPlugin } from '../src/worker/plugins/loadPlugin'
-import { runProcessEvent } from '../src/worker/plugins/run'
+import { IllegalOperationError, runProcessEvent, runProcessEventBatch } from '../src/worker/plugins/run'
 import { loadSchedule, setupPlugins } from '../src/worker/plugins/setup'
 import {
     commonOrganizationId,
@@ -196,12 +196,13 @@ test('local plugin with broken index.js does not do much', async () => {
     unlink()
 })
 
-test('plugin changing teamID throws error', async () => {
+test('plugin changing event.team_id throws error (single)', async () => {
     getPluginRows.mockReturnValueOnce([
         mockPluginWithArchive(`
-            function processEvent (event, meta) { 
+            function processEvent (event, meta) {
                 event.team_id = 400
-                return event }
+                return event
+            }
         `),
     ])
 
@@ -214,34 +215,61 @@ test('plugin changing teamID throws error', async () => {
     const event = { event: '$test', properties: {}, team_id: 2 } as PluginEvent
     const returnedEvent = await runProcessEvent(mockServer, event)
 
-    expect(returnedEvent).toEqual(null)
+    const expectedReturnedEvent = {
+        event: '$test',
+        properties: {
+            $plugins_failed: ['test-maxmind-plugin (39)'],
+            $plugins_succeeded: [],
+        },
+        team_id: 2,
+    }
+    expect(returnedEvent).toEqual(expectedReturnedEvent)
 
     expect(processError).toHaveBeenCalledWith(
         mockServer,
         pluginConfigs.get(39)!,
-        Error('Illegal Operation: Plugin tried to change teamID'),
-        null
+        new IllegalOperationError('Plugin tried to change event.team_id'),
+        expectedReturnedEvent
     )
 })
 
-test('plugin changing teamID prevents ingestion', async () => {
+test('plugin changing event.team_id throws error (batch)', async () => {
     getPluginRows.mockReturnValueOnce([
         mockPluginWithArchive(`
-            function processEvent (event, meta) { 
-                event.team_id = 400
-                return event }
+            function processEventBatch (events, meta) {
+                for (const event of events) {
+                    event.team_id = 400
+                }
+                return events
+            }
         `),
     ])
 
     getPluginConfigRows.mockReturnValueOnce([pluginConfig39])
     getPluginAttachmentRows.mockReturnValueOnce([])
 
     await setupPlugins(mockServer)
+    const { pluginConfigs } = mockServer
 
-    const event = { event: '$test', properties: {}, team_id: 2 } as PluginEvent
-    const returnedEvent = await runProcessEvent(mockServer, event)
+    const events = [{ event: '$test', properties: {}, team_id: 2 } as PluginEvent]
+    const returnedEvent = await runProcessEventBatch(mockServer, events)
 
-    expect(returnedEvent).toEqual(null)
+    const expectedReturnedEvent = {
+        event: '$test',
+        properties: {
+            $plugins_failed: ['test-maxmind-plugin (39)'],
+            $plugins_succeeded: [],
+        },
+        team_id: 2,
+    }
+    expect(returnedEvent).toEqual([expectedReturnedEvent])
+
+    expect(processError).toHaveBeenCalledWith(
+        mockServer,
+        pluginConfigs.get(39)!,
+        new IllegalOperationError('Plugin tried to change event.team_id'),
+        expectedReturnedEvent
+    )
 })
 
 test('plugin throwing error does not prevent ingestion and failure is noted in event', async () => {