Ingestion postgres e2e test

src/config.ts

@@ -20,6 +20,7 @@ export function getDefaultConfig(): PluginsServerConfig {
         KAFKA_CLIENT_CERT_B64: null,
         KAFKA_CLIENT_CERT_KEY_B64: null,
         KAFKA_TRUSTED_CERT_B64: null,
+        PLUGIN_SERVER_INGESTION: false,
         PLUGINS_CELERY_QUEUE: 'posthog-plugins',
         REDIS_URL: 'redis://127.0.0.1',
         BASE_DIR: '.',
@@ -40,6 +41,7 @@ export function getDefaultConfig(): PluginsServerConfig {
 
 export function getConfigHelp(): Record<keyof PluginsServerConfig, string> {
     return {
+        PLUGIN_SERVER_INGESTION: 'Ingest events via plugin-server',
         CELERY_DEFAULT_QUEUE: 'Celery outgoing queue',
         PLUGINS_CELERY_QUEUE: 'Celery incoming queue',
         DATABASE_URL: 'Postgres database URL',

src/types.ts

@@ -42,6 +42,7 @@ export interface PluginsServerConfig extends Record<string, any> {
     WEB_PORT: number
     WEB_HOSTNAME: string
     LOG_LEVEL: LogLevel
+    PLUGIN_SERVER_INGESTION: boolean
     SENTRY_DSN: string | null
     STATSD_HOST: string | null
     STATSD_PORT: number

src/worker/queue.ts

@@ -6,6 +6,7 @@ import Client from '../celery/client'
 import { PluginsServer, Queue } from '../types'
 import { KafkaQueue } from '../ingestion/kafka-queue'
 import { status } from '../status'
+import { UUIDT } from '../utils'
 
 export async function startQueue(
     server: PluginsServer,
@@ -45,15 +46,29 @@ async function startQueueRedis(
                 const processedEvent = await processEvent(event)
                 if (processedEvent) {
                     const { distinct_id, ip, site_url, team_id, now, sent_at, ...data } = processedEvent
-                    client.sendTask('posthog.tasks.process_event.process_event', [], {
-                        distinct_id,
-                        ip,
-                        site_url,
-                        data,
-                        team_id,
-                        now,
-                        sent_at,
-                    })
+
+                    if (server.PLUGIN_SERVER_INGESTION) {
+                        await server.eventsProcessor.processEvent(
+                            distinct_id,
+                            ip,
+                            site_url,
+                            processedEvent,
+                            team_id,
+                            DateTime.fromISO(now),
+                            sent_at ? DateTime.fromISO(sent_at) : null,
+                            new UUIDT().toString()
+                        )
+                    } else {
+                        client.sendTask('posthog.tasks.process_event.process_event', [], {
+                            distinct_id,
+                            ip,
+                            site_url,
+                            data,
+                            team_id,
+                            now,
+                            sent_at,
+                        })
+                    }
                 }
             } catch (e) {
                 Sentry.captureException(e)

tests/clickhouse/process-event.test.ts

@@ -1,12 +1,4 @@
-import {
-    Element,
-    Event,
-    Person,
-    PersonDistinctId,
-    PluginsServer,
-    PluginsServerConfig,
-    PostgresSessionRecordingEvent,
-} from '../../src/types'
+import { PluginsServerConfig } from '../../src/types'
 import { resetTestDatabaseClickhouse } from '../helpers/clickhouse'
 import { KafkaCollector, KafkaObserver } from '../helpers/kafka'
 import { UUIDT } from '../../src/utils'

tests/postgres/e2e.test.ts

@@ -0,0 +1,56 @@
+import { LogLevel } from '../../src/types'
+import { resetTestDatabase } from '../helpers/sql'
+import { startPluginsServer } from '../../src/server'
+import { makePiscina } from '../../src/worker/piscina'
+import { PluginsServer } from '../../src/types'
+import { createPosthog, DummyPostHog } from '../../src/extensions/posthog'
+import { pluginConfig39 } from '../helpers/plugins'
+import { delay } from '../../src/utils'
+
+jest.setTimeout(60000) // 60 sec timeout
+
+describe('e2e postgres ingestion', () => {
+    let server: PluginsServer
+    let stopServer: () => Promise<void>
+    let posthog: DummyPostHog
+
+    beforeEach(async () => {
+        await resetTestDatabase(`
+            async function processEvent (event) {
+                event.properties.processed = 'hell yes'
+                return event
+            }
+        `)
+        const startResponse = await startPluginsServer(
+            {
+                WORKER_CONCURRENCY: 2,
+                PLUGINS_CELERY_QUEUE: 'test-plugins-celery-queue',
+                CELERY_DEFAULT_QUEUE: 'test-celery-default-queue',
+                PLUGIN_SERVER_INGESTION: true,
+                LOG_LEVEL: LogLevel.Log,
+                KAFKA_ENABLED: false,
+            },
+            makePiscina
+        )
+        server = startResponse.server
+        stopServer = startResponse.stop
+
+        await server.redis.del(server.PLUGINS_CELERY_QUEUE)
+        await server.redis.del(server.CELERY_DEFAULT_QUEUE)
+
+        posthog = createPosthog(server, pluginConfig39)
+    })
+
+    afterEach(async () => {
+        await stopServer()
+    })
+
+    test('event captured, processed, ingested', async () => {
+        expect((await server.db.fetchEvents()).length).toBe(0)
+        posthog.capture('custom event', { name: 'haha' })
+        await delay(2000)
+        const events = await server.db.fetchEvents()
+        expect(events.length).toBe(1)
+        expect(events[0].properties.processed).toEqual('hell yes')
+    })
+})