Mostly remove processEventBatch

.github/workflows/ci.yml

@@ -105,33 +105,45 @@ jobs:
                   DATABASE_URL: 'postgres://postgres:postgres@localhost:5432/test_posthog'
                   REDIS_URL: 'redis://localhost'
               run: |
+                  echo "🔗 Linking this plugin server into posthog(master)/plugins/"
                   cd posthog/plugins
                   yarn link "@posthog/plugin-server"
                   yarn
 
-                  # start this in the background
-                  yarn start &> tmplog.txt &
+                  echo "🏁 Starting the plugin server dist/ (compiled JS) via posthog(master)/plugins/"
+                  mkdir -p tmp
+                  yarn start &> tmp/plugin-server.log &
                   pid=$!
 
+                  echo "⏳ Waiting 10 seconds to see if it runs"
                   sleep 10
 
-                  # check if it's still running, as it should be
+                  echo "🤔 Checking if it's still running, as it should be"
                   if ! kill $pid > /dev/null 2>&1; then
-                      echo "Could not send SIGTERM to process $pid" >&2
+                      echo "😵 It was not running!"
+                      echo "🪵 Here's the log:"
                       echo ""
-                      cat tmplog.txt
+                      cat tmp/plugin-server.log
                       exit 1
                   fi
 
-                  str=`cat tmplog.txt | grep "All systems go"`
+                  echo '🤔 Checking if it logged "All systems go"'
+                  str=`cat tmp/plugin-server.log | grep "All systems go"`
                   if [ ! "$str" ];then
-                      echo 'Did not find "All systems go" in plugin server log output!'
+                      sleep 5
+                      echo '😵 Did not find "All systems go" in plugin server log output!'
+                      echo "🪵 Here's the log:"
                       echo ""
-                      cat tmplog.txt
+                      cat tmp/plugin-server.log
                       exit 1
                   fi
 
-                  rm tmplog.txt
+                  sleep 5
+                  echo '✅ All systems went!'
+                  echo "🪵 Here's the complete log:"
+                  echo ""
+                  cat tmp/plugin-server.log
+                  rm -f tmp/plugin-server.log
 
     tests-postgres-1:
         name: Tests / Postgres + Redis (1)

benchmarks/clickhouse/e2e.kafka.benchmark.ts

@@ -31,13 +31,10 @@ describe('e2e kafka & clickhouse benchmark', () => {
 
     beforeEach(async () => {
         await resetTestDatabase(`
-            async function processEventBatch (batch) {
-                // console.log(\`Received batch of \${batch.length} events\`)
-                return batch.map(event => {
-                    event.properties.processed = 'hell yes'
-                    event.properties.upperUuid = event.properties.uuid?.toUpperCase()
-                    return event
-                })
+            async function processEvent (event) {
+                event.properties.processed = 'hell yes'
+                event.properties.upperUuid = event.properties.uuid?.toUpperCase()
+                return event
             }
         `)
         await resetKafka(extraServerConfig)

benchmarks/vm/worker.benchmark.ts

@@ -27,36 +27,16 @@ function processOneEvent(
     return processEvent(defaultEvent)
 }
 
-function processOneBatch(
-    processEventBatch: (batch: PluginEvent[]) => Promise<PluginEvent[]>,
-    batchSize: number,
-    batchIndex: number
-): Promise<PluginEvent[]> {
-    const events = [...Array(batchSize)].map((_, i) => ({
-        distinct_id: 'my_id',
-        ip: '127.0.0.1',
-        site_url: 'http://localhost',
-        team_id: 2,
-        now: new Date().toISOString(),
-        event: 'default event',
-        properties: { key: 'value', batchIndex, indexInBatch: i },
-    }))
-
-    return processEventBatch(events)
-}
-
 async function processCountEvents(piscina: ReturnType<typeof makePiscina>, count: number, batchSize = 1) {
     const maxPromises = 1000
     const promises = Array(maxPromises)
     const processEvent = (event: PluginEvent) => piscina.runTask({ task: 'processEvent', args: { event } })
-    const processEventBatch = (batch: PluginEvent[]) => piscina.runTask({ task: 'processEventBatch', args: { batch } })
 
-    const groups = Math.ceil(count / maxPromises)
+    const groups = Math.ceil((count * batchSize) / maxPromises)
     for (let j = 0; j < groups; j++) {
-        const groupCount = groups === 1 ? count : j === groups - 1 ? count % maxPromises : maxPromises
+        const groupCount = groups === 1 ? count : j === groups - 1 ? (count * batchSize) % maxPromises : maxPromises
         for (let i = 0; i < groupCount; i++) {
-            promises[i] =
-                batchSize === 1 ? processOneEvent(processEvent, i) : processOneBatch(processEventBatch, batchSize, i)
+            promises[i] = processOneEvent(processEvent, i)
         }
         await Promise.all(promises)
     }

src/main/ingestion-queues/queue.ts

@@ -40,11 +40,6 @@ export async function startQueue(
             server.lastActivityType = 'processEvent'
             return piscina.runTask({ task: 'processEvent', args: { event } })
         },
-        processEventBatch: (batch: PluginEvent[]) => {
-            server.lastActivity = new Date().valueOf()
-            server.lastActivityType = 'processEventBatch'
-            return piscina.runTask({ task: 'processEventBatch', args: { batch } })
-        },
         ingestEvent: (event: PluginEvent) => {
             server.lastActivity = new Date().valueOf()
             server.lastActivityType = 'ingestEvent'

src/types.ts

@@ -290,7 +290,6 @@ export type WorkerMethods = {
     onEvent: (event: PluginEvent) => Promise<void>
     onSnapshot: (event: PluginEvent) => Promise<void>
     processEvent: (event: PluginEvent) => Promise<PluginEvent | null>
-    processEventBatch: (batch: PluginEvent[]) => Promise<(PluginEvent | null)[]>
     ingestEvent: (event: PluginEvent) => Promise<IngestEventResponse>
 }
 
@@ -301,8 +300,6 @@ export type VMMethods = {
     onSnapshot?: (event: PluginEvent) => Promise<void>
     exportEvents?: (events: PluginEvent[]) => Promise<void>
     processEvent?: (event: PluginEvent) => Promise<PluginEvent>
-    // DEPRECATED
-    processEventBatch?: (batch: PluginEvent[]) => Promise<PluginEvent[]>
 }
 
 export interface PluginConfigVMResponse {

src/worker/plugins/run.ts

@@ -98,71 +98,6 @@ export async function runProcessEvent(server: Hub, event: PluginEvent): Promise<
     return returnedEvent
 }
 
-export async function runProcessEventBatch(server: Hub, batch: PluginEvent[]): Promise<PluginEvent[]> {
-    const eventsByTeam = new Map<number, PluginEvent[]>()
-
-    for (const event of batch) {
-        if (eventsByTeam.has(event.team_id)) {
-            eventsByTeam.get(event.team_id)!.push(event)
-        } else {
-            eventsByTeam.set(event.team_id, [event])
-        }
-    }
-
-    let allReturnedEvents: PluginEvent[] = []
-
-    for (const [teamId, teamEvents] of eventsByTeam.entries()) {
-        const pluginsToRun = getPluginsForTeam(server, teamId)
-
-        let returnedEvents: PluginEvent[] = teamEvents
-        const pluginsSucceeded = []
-        const pluginsFailed = []
-        for (const pluginConfig of pluginsToRun) {
-            const timer = new Date()
-            const processEventBatch = await pluginConfig.vm?.getProcessEventBatch()
-            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])
-                    server.statsd?.increment(`plugin.${pluginConfig.plugin?.name}.process_event_batch.ERROR`)
-                    pluginsFailed.push(`${pluginConfig.plugin?.name} (${pluginConfig.id})`)
-                }
-                server.statsd?.timing(`plugin.${pluginConfig.plugin?.name}.process_event_batch`, timer)
-                server.statsd?.timing('plugin.process_event_batch', timer, {
-                    plugin: pluginConfig.plugin?.name ?? '?',
-                    teamId: teamId.toString(),
-                })
-            }
-        }
-
-        for (const event of returnedEvents) {
-            if (event && (pluginsSucceeded.length > 0 || pluginsFailed.length > 0)) {
-                event.properties = {
-                    ...event.properties,
-                    $plugins_succeeded: pluginsSucceeded,
-                    $plugins_failed: pluginsFailed,
-                }
-            }
-        }
-
-        allReturnedEvents = allReturnedEvents.concat(returnedEvents)
-    }
-
-    return allReturnedEvents.filter(Boolean)
-}
-
 export async function runPluginTask(
     server: Hub,
     taskName: string,

src/worker/tasks.ts

@@ -2,7 +2,7 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types'
 
 import { Action, EnqueuedJob, Hub, PluginTaskType, Team } from '../types'
 import { ingestEvent } from './ingestion/ingest-event'
-import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent, runProcessEventBatch } from './plugins/run'
+import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent } from './plugins/run'
 import { loadSchedule, setupPlugins } from './plugins/setup'
 import { teardownPlugins } from './plugins/teardown'
 
@@ -18,9 +18,6 @@ export const workerTasks: Record<string, TaskRunner> = {
     processEvent: (hub, args: { event: PluginEvent }) => {
         return runProcessEvent(hub, args.event)
     },
-    processEventBatch: (hub, args: { batch: PluginEvent[] }) => {
-        return runProcessEventBatch(hub, args.batch)
-    },
     runJob: (hub, { job }: { job: EnqueuedJob }) => {
         return runPluginTask(hub, job.type, PluginTaskType.Job, job.pluginConfigId, job.payload)
     },

src/worker/vm/lazy.ts

@@ -72,10 +72,6 @@ export class LazyPluginVM {
         return (await this.resolveInternalVm)?.methods.processEvent || null
     }
 
-    async getProcessEventBatch(): Promise<PluginConfigVMResponse['methods']['processEventBatch'] | null> {
-        return (await this.resolveInternalVm)?.methods.processEventBatch || null
-    }
-
     async getTeardownPlugin(): Promise<PluginConfigVMResponse['methods']['teardownPlugin'] | null> {
         return (await this.resolveInternalVm)?.methods.teardownPlugin || null
     }

src/worker/vm/vm.ts

@@ -140,23 +140,8 @@ export async function createPluginConfigVM(
                 if (func) return func(...args);
             }
 
-            // we have processEvent, but not processEventBatch
-            if (!__getExported('processEventBatch') && __getExported('processEvent')) {
-                exports.processEventBatch = async function __processEventBatch${pluginConfigIdentifier} (batch, meta) {
-                    const processEvent = __getExported('processEvent');
-                    let waitFor = false
-                    const processedEvents = batch.map(function __eventBatchToEvent${pluginConfigIdentifier} (event) {
-                        const e = processEvent(event, meta)
-                        if (e && typeof e.then !== 'undefined') {
-                            waitFor = true
-                        }
-                        return e
-                    })
-                    const response = waitFor ? (await Promise.all(processedEvents)) : processedEvents;
-                    return response.filter(r => r)
-                }
             // we have processEventBatch, but not processEvent
-            } else if (!__getExported('processEvent') && __getExported('processEventBatch')) {
+            if (!__getExported('processEvent') && __getExported('processEventBatch')) {
                 exports.processEvent = async function __processEvent${pluginConfigIdentifier} (event, meta) {
                     return (await (__getExported('processEventBatch'))([event], meta))?.[0]
                 }
@@ -170,7 +155,6 @@ export async function createPluginConfigVM(
                 onEvent: __asyncFunctionGuard(__bindMeta('onEvent')),
                 onSnapshot: __asyncFunctionGuard(__bindMeta('onSnapshot')),
                 processEvent: __asyncFunctionGuard(__bindMeta('processEvent')),
-                processEventBatch: __asyncFunctionGuard(__bindMeta('processEventBatch')),
             };
 
             const __tasks = {