In-memory action definitions synced with Django

benchmarks/postgres/ingestion.benchmark.ts

@@ -52,6 +52,7 @@ describe('ingestion benchmarks', () => {
             LOG_LEVEL: LogLevel.Log,
         })
         eventsProcessor = new EventsProcessor(hub)
+        await eventsProcessor.prepare()
         team = await getFirstTeam(hub)
         now = DateTime.utc()
 

src/config/config.ts

@@ -47,7 +47,7 @@ export function getDefaultConfig(): PluginsServerConfig {
         WORKER_CONCURRENCY: coreCount,
         TASK_TIMEOUT: 30,
         TASKS_PER_WORKER: 10,
-        LOG_LEVEL: LogLevel.Info,
+        LOG_LEVEL: isTestEnv ? LogLevel.Warn : LogLevel.Info,
         SENTRY_DSN: null,
         STATSD_HOST: null,
         STATSD_PORT: 8125,

src/main/pluginsServer.ts

@@ -2,16 +2,16 @@ import { ReaderModel } from '@maxmind/geoip2-node'
 import Piscina from '@posthog/piscina'
 import * as Sentry from '@sentry/node'
 import { FastifyInstance } from 'fastify'
-import Redis from 'ioredis'
 import net, { AddressInfo } from 'net'
 import * as schedule from 'node-schedule'
 
 import { defaultConfig } from '../config/config'
 import { Hub, JobQueueConsumerControl, PluginsServerConfig, Queue, ScheduleControl } from '../types'
 import { createHub } from '../utils/db/hub'
 import { killProcess } from '../utils/kill'
+import { PubSub } from '../utils/pubsub'
 import { status } from '../utils/status'
-import { createRedis, delay, getPiscinaStats } from '../utils/utils'
+import { delay, getPiscinaStats } from '../utils/utils'
 import { startQueue } from './ingestion-queues/queue'
 import { startJobQueueConsumer } from './job-queues/job-queue-consumer'
 import { createMmdbServer, performMmdbStalenessCheck, prepareMmdb } from './services/mmdb'
@@ -41,7 +41,7 @@ export async function startPluginsServer(
 
     status.info('ℹ️', `${serverConfig.WORKER_CONCURRENCY} workers, ${serverConfig.TASKS_PER_WORKER} tasks per worker`)
 
-    let pubSub: Redis.Redis | undefined
+    let pubSub: PubSub | undefined
     let hub: Hub | undefined
     let fastifyInstance: FastifyInstance | undefined
     let pingJob: schedule.Job | undefined
@@ -73,7 +73,7 @@ export async function startPluginsServer(
         }
         lastActivityCheck && clearInterval(lastActivityCheck)
         await queue?.stop()
-        await pubSub?.quit()
+        await pubSub?.stop()
         pingJob && schedule.cancelJob(pingJob)
         statsJob && schedule.cancelJob(statsJob)
         await jobQueueConsumer?.stop()
@@ -141,23 +141,29 @@ export async function startPluginsServer(
             void jobQueueConsumer?.resume()
         })
 
-        // use one extra connection for redis pubsub
-        pubSub = await createRedis(hub)
-        await pubSub.subscribe(hub.PLUGINS_RELOAD_PUBSUB_CHANNEL)
-        pubSub.on('message', async (channel: string, message) => {
-            if (channel === hub!.PLUGINS_RELOAD_PUBSUB_CHANNEL) {
+        // use one extra Redis connection for pub-sub
+        pubSub = new PubSub(hub, {
+            [hub.PLUGINS_RELOAD_PUBSUB_CHANNEL]: async () => {
                 status.info('⚡', 'Reloading plugins!')
-
                 await piscina?.broadcastTask({ task: 'reloadPlugins' })
                 await scheduleControl?.reloadSchedule()
-            }
+            },
+            'reload-action': async (message) =>
+                await piscina?.broadcastTask({ task: 'reloadAction', args: { actionId: parseInt(message) } }),
+            'drop-action': async (message) =>
+                await piscina?.broadcastTask({ task: 'dropAction', args: { actionId: parseInt(message) } }),
         })
+        await pubSub.start()
 
         if (hub.jobQueueManager) {
             const queueString = hub.jobQueueManager.getJobQueueTypesAsString()
             await hub!.db!.redisSet('@posthog-plugin-server/enabled-job-queues', queueString)
         }
 
+        // every 5 minutes all ActionManager caches are reloaded for eventual consistency
+        pingJob = schedule.scheduleJob('*/5 * * * *', async () => {
+            await piscina?.broadcastTask({ task: 'reloadAllActions' })
+        })
         // every 5 seconds set Redis keys @posthog-plugin-server/ping and @posthog-plugin-server/version
         pingJob = schedule.scheduleJob('*/5 * * * * *', async () => {
             await hub!.db!.redisSet('@posthog-plugin-server/ping', new Date().toISOString(), 60, {

src/types.ts

@@ -16,12 +16,21 @@ import { EventsProcessor } from './worker/ingestion/process-event'
 import { LazyPluginVM } from './worker/vm/lazy'
 
 export enum LogLevel {
+    None = 'none',
     Debug = 'debug',
     Info = 'info',
     Log = 'log',
     Warn = 'warn',
     Error = 'error',
-    None = 'none',
+}
+
+export const logLevelToNumber: Record<LogLevel, number> = {
+    [LogLevel.None]: 0,
+    [LogLevel.Debug]: 10,
+    [LogLevel.Info]: 20,
+    [LogLevel.Log]: 30,
+    [LogLevel.Warn]: 40,
+    [LogLevel.Error]: 50,
 }
 
 export interface PluginsServerConfig extends Record<string, any> {
@@ -318,6 +327,7 @@ export interface RawOrganization {
     name: string
     created_at: string
     updated_at: string
+    available_features: string[]
 }
 
 /** Usable Team model. */
@@ -429,6 +439,101 @@ export interface CohortPeople {
     person_id: number
 }
 
+/** Sync with posthog/frontend/src/types.ts */
+export enum PropertyOperator {
+    Exact = 'exact',
+    IsNot = 'is_not',
+    IContains = 'icontains',
+    NotIContains = 'not_icontains',
+    Regex = 'regex',
+    NotRegex = 'not_regex',
+    GreaterThan = 'gt',
+    LessThan = 'lt',
+    IsSet = 'is_set',
+    IsNotSet = 'is_not_set',
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+interface BasePropertyFilter {
+    key: string
+    value: string | number | Array<string | number> | null
+    label?: string
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+export interface EventPropertyFilter extends BasePropertyFilter {
+    type: 'event'
+    operator: PropertyOperator
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+export interface PersonPropertyFilter extends BasePropertyFilter {
+    type: 'person'
+    operator: PropertyOperator
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+export interface ElementPropertyFilter extends BasePropertyFilter {
+    type: 'element'
+    key: 'tag_name' | 'text' | 'href' | 'selector'
+    operator: PropertyOperator
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+export interface CohortPropertyFilter extends BasePropertyFilter {
+    type: 'cohort'
+    key: 'id'
+    value: number
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+export type ActionStepProperties =
+    | EventPropertyFilter
+    | PersonPropertyFilter
+    | ElementPropertyFilter
+    | CohortPropertyFilter
+
+/** Sync with posthog/frontend/src/types.ts */
+export enum ActionStepUrlMatching {
+    Contains = 'contains',
+    Regex = 'regex',
+    Exact = 'exact',
+}
+
+export interface ActionStep {
+    id: number
+    action_id: number
+    tag_name: string | null
+    text: string | null
+    href: string | null
+    selector: string | null
+    url: string | null
+    url_matching: ActionStepUrlMatching | null
+    name: string | null
+    event: string | null
+    properties: ActionStepProperties[] | null
+}
+
+/** Raw Action row from database. */
+export interface RawAction {
+    id: number
+    team_id: TeamId
+    name: string | null
+    created_at: string
+    created_by_id: number | null
+    deleted: boolean
+    post_to_slack: boolean
+    slack_message_format: string
+    is_calculating: boolean
+    updated_at: string
+    last_calculated_at: string
+}
+
+/** Usable Action model. */
+export interface Action extends RawAction {
+    steps: ActionStep[]
+}
+
 export interface SessionRecordingEvent {
     uuid: string
     timestamp: string

src/utils/db/db.ts

@@ -10,6 +10,8 @@ import { Pool, PoolClient, QueryConfig, QueryResult, QueryResultRow } from 'pg'
 
 import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID, KAFKA_PLUGIN_LOG_ENTRIES } from '../../config/kafka-topics'
 import {
+    Action,
+    ActionStep,
     ClickHouseEvent,
     ClickHousePerson,
     ClickHousePersonDistinctId,
@@ -26,6 +28,7 @@ import {
     PluginLogEntryType,
     PostgresSessionRecordingEvent,
     PropertyDefinitionType,
+    RawAction,
     RawOrganization,
     RawPerson,
     SessionRecordingEvent,
@@ -705,14 +708,63 @@ export class DB {
         return entry
     }
 
+    // EventDefinition
+
     public async fetchEventDefinitions(): Promise<EventDefinitionType[]> {
         return (await this.postgresQuery('SELECT * FROM posthog_eventdefinition', undefined, 'fetchEventDefinitions'))
             .rows as EventDefinitionType[]
     }
 
+    // PropertyDefinition
+
     public async fetchPropertyDefinitions(): Promise<PropertyDefinitionType[]> {
         return (
             await this.postgresQuery('SELECT * FROM posthog_propertydefinition', undefined, 'fetchPropertyDefinitions')
         ).rows as PropertyDefinitionType[]
     }
+
+    // Action & ActionStep
+
+    public async fetchAllActionsMap(): Promise<Record<Action['id'], Action>> {
+        const rawActions: RawAction[] = (
+            await this.postgresQuery(`SELECT * FROM posthog_action WHERE deleted = FALSE`, undefined, 'fetchActions')
+        ).rows
+        const actionSteps: ActionStep[] = (
+            await this.postgresQuery(
+                `SELECT posthog_actionstep.*, posthog_action.deleted FROM posthog_actionstep
+            JOIN posthog_action ON (posthog_action.id = posthog_actionstep.action_id)
+            WHERE posthog_action.deleted = FALSE`,
+                undefined,
+                'fetchActionSteps'
+            )
+        ).rows
+        const actionsMap: Record<Action['id'], Action> = {}
+        for (const rawAction of rawActions) {
+            actionsMap[rawAction.id] = { ...rawAction, steps: [] }
+        }
+        for (const actionStep of actionSteps) {
+            if (actionStep.action_id in actionsMap) {
+                actionsMap[actionStep.action_id].steps.push(actionStep)
+            }
+        }
+        return actionsMap
+    }
+
+    public async fetchAction(id: Action['id']): Promise<Action | null> {
+        const rawActions: RawAction[] = (
+            await this.postgresQuery(
+                `SELECT * FROM posthog_action WHERE id = $1 AND deleted = FALSE`,
+                [id],
+                'fetchActions'
+            )
+        ).rows
+        if (!rawActions.length) {
+            return null
+        }
+        const steps: ActionStep[] = (
+            await this.postgresQuery(`SELECT * FROM posthog_actionstep WHERE action_id = $1`, [id], 'fetchActionSteps')
+        ).rows
+        const action: Action = { ...rawActions[0], steps }
+        return action
+    }
 }

src/utils/db/hub.ts

@@ -175,6 +175,7 @@ export async function createHub(
 
     // :TODO: This is only used on worker threads, not main
     hub.eventsProcessor = new EventsProcessor(hub as Hub)
+    await hub.eventsProcessor.prepare()
     hub.jobQueueManager = new JobQueueManager(hub as Hub)
 
     try {

src/utils/pubsub.ts

@@ -0,0 +1,57 @@
+import { captureException } from '@sentry/node'
+import { Redis } from 'ioredis'
+
+import { PluginsServerConfig } from '../types'
+import { status } from './status'
+import { createRedis } from './utils'
+
+export type PubSubTask = ((message: string) => void) | ((message: string) => Promise<void>)
+
+export interface PubSubTaskMap {
+    [channel: string]: PubSubTask
+}
+
+export class PubSub {
+    private serverConfig: PluginsServerConfig
+    private redis: Redis | null
+    public taskMap: PubSubTaskMap
+
+    constructor(serverConfig: PluginsServerConfig, taskMap: PubSubTaskMap = {}) {
+        this.serverConfig = serverConfig
+        this.redis = null
+        this.taskMap = taskMap
+    }
+
+    public async start(): Promise<void> {
+        if (this.redis) {
+            throw new Error('Started PubSub cannot be started again!')
+        }
+        this.redis = await createRedis(this.serverConfig)
+        const channels = Object.keys(this.taskMap)
+        await this.redis.subscribe(channels)
+        this.redis.on('message', (channel: string, message: string) => {
+            const task: PubSubTask | undefined = this.taskMap[channel]
+            if (!task) {
+                captureException(
+                    new Error(
+                        `Received a pubsub message for unassociated channel ${channel}! Associated channels are: ${Object.keys(
+                            this.taskMap
+                        ).join(', ')}`
+                    )
+                )
+            }
+            void task(message)
+        })
+        status.info('👀', `Pub-sub started for channels: ${channels.join(', ')}`)
+    }
+
+    public async stop(): Promise<void> {
+        if (!this.redis) {
+            throw new Error('Unstarted PubSub cannot be stopped!')
+        }
+        await this.redis.unsubscribe()
+        this.redis.disconnect()
+        this.redis = null
+        status.info('🛑', `Pub-sub stopped for channels: ${Object.keys(this.taskMap).join(', ')}`)
+    }
+}

src/utils/utils.ts

@@ -5,7 +5,7 @@ import AdmZip from 'adm-zip'
 import { randomBytes } from 'crypto'
 import Redis, { RedisOptions } from 'ioredis'
 import { DateTime } from 'luxon'
-import { Pool, PoolClient, PoolConfig } from 'pg'
+import { Pool, PoolConfig } from 'pg'
 import { Readable } from 'stream'
 import * as tar from 'tar-stream'
 import * as zlib from 'zlib'