Action matching juice

benchmarks/postgres/ingestion.benchmark.ts

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

package.json

@@ -61,6 +61,7 @@
         "@types/lru-cache": "^5.1.0",
         "adm-zip": "^0.4.16",
         "aws-sdk": "^2.884.0",
+        "escape-string-regexp": "^4.0.0",
         "fast-deep-equal": "^3.1.3",
         "generic-pool": "^3.7.1",
         "graphile-worker": "^0.11.1",
@@ -75,6 +76,7 @@
         "posthog-js-lite": "^0.0.5",
         "pretty-bytes": "^5.6.0",
         "protobufjs": "^6.10.2",
+        "re2": "^1.16.0",
         "redlock": "^4.2.0",
         "snowflake-sdk": "^1.6.0",
         "tar-stream": "^2.1.4",

src/types.ts

@@ -13,6 +13,8 @@ import { DB } from './utils/db/db'
 import { KafkaProducerWrapper } from './utils/db/kafka-producer-wrapper'
 import { InternalMetrics } from './utils/internal-metrics'
 import { UUID } from './utils/utils'
+import { ActionManager } from './worker/ingestion/action-manager'
+import { ActionMatcher } from './worker/ingestion/action-matcher'
 import { EventsProcessor } from './worker/ingestion/process-event'
 import { LazyPluginVM } from './worker/vm/lazy'
 
@@ -115,6 +117,8 @@ export interface Hub extends PluginsServerConfig {
     pluginConfigSecrets: Map<PluginConfigId, string>
     pluginConfigSecretLookup: Map<string, PluginConfigId>
     // tools
+    actionManager: ActionManager
+    actionMatcher: ActionMatcher
     eventsProcessor: EventsProcessor
     jobQueueManager: JobQueueManager
     // diagnostics
@@ -487,44 +491,43 @@ export enum PropertyOperator {
 }
 
 /** Sync with posthog/frontend/src/types.ts */
-interface BasePropertyFilter {
+interface PropertyFilterBase {
     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'
+export interface PropertyFilterWithOperator extends PropertyFilterBase {
     operator: PropertyOperator
 }
 
 /** Sync with posthog/frontend/src/types.ts */
-export interface PersonPropertyFilter extends BasePropertyFilter {
+export interface EventPropertyFilter extends PropertyFilterWithOperator {
+    type: 'event'
+}
+
+/** Sync with posthog/frontend/src/types.ts */
+export interface PersonPropertyFilter extends PropertyFilterWithOperator {
     type: 'person'
-    operator: PropertyOperator
 }
 
 /** Sync with posthog/frontend/src/types.ts */
-export interface ElementPropertyFilter extends BasePropertyFilter {
+export interface ElementPropertyFilter extends PropertyFilterWithOperator {
     type: 'element'
     key: 'tag_name' | 'text' | 'href' | 'selector'
-    operator: PropertyOperator
+    value: string
 }
 
 /** Sync with posthog/frontend/src/types.ts */
-export interface CohortPropertyFilter extends BasePropertyFilter {
+export interface CohortPropertyFilter extends PropertyFilterBase {
     type: 'cohort'
     key: 'id'
-    value: number
+    value: number | string
 }
 
 /** Sync with posthog/frontend/src/types.ts */
-export type ActionStepProperties =
-    | EventPropertyFilter
-    | PersonPropertyFilter
-    | ElementPropertyFilter
-    | CohortPropertyFilter
+export type PropertyFilter = EventPropertyFilter | PersonPropertyFilter | ElementPropertyFilter | CohortPropertyFilter
 
 /** Sync with posthog/frontend/src/types.ts */
 export enum ActionStepUrlMatching {
@@ -544,7 +547,7 @@ export interface ActionStep {
     url_matching: ActionStepUrlMatching | null
     name: string | null
     event: string | null
-    properties: ActionStepProperties[] | null
+    properties: PropertyFilter[] | null
 }
 
 /** Raw Action row from database. */

src/utils/db/db.ts

@@ -536,6 +536,17 @@ export class DB {
         }
     }
 
+    // Cohort & CohortPeople
+
+    public async doesPersonBelongToCohort(cohortId: number, personId: Person['id']): Promise<boolean> {
+        const selectResult = await this.postgresQuery(
+            `SELECT EXISTS (SELECT 1 FROM posthog_cohortpeople WHERE cohort_id = $1 AND person_id = $2);`,
+            [cohortId, personId],
+            'doesPersonBelongToCohort'
+        )
+        return selectResult.rows[0]
+    }
+
     // Organization
 
     public async fetchOrganization(organizationId: string): Promise<RawOrganization | undefined> {
@@ -724,7 +735,7 @@ export class DB {
         ).rows as PropertyDefinitionType[]
     }
 
-    // Action & ActionStep
+    // Action & ActionStep & Action<>Event
 
     public async fetchAllActionsGroupedByTeam(): Promise<Record<Team['id'], Record<Action['id'], Action>>> {
         const rawActions: RawAction[] = (
@@ -772,6 +783,14 @@ export class DB {
         return action
     }
 
+    public async registerActionOccurrence(actionId: Action['id'], eventId: Event['id']): Promise<void> {
+        await this.postgresQuery(
+            `INSERT INTO posthog_action_events_2 (action_id, event_id) VALUES ($1, $2)`, // TODO: remove _2
+            [actionId, eventId],
+            'registerActionOccurrence'
+        )
+    }
+
     // Team Internal Metrics
 
     public async fetchInternalMetricsTeam(): Promise<Team['id'] | null> {

src/utils/db/hub.ts

@@ -13,6 +13,8 @@ import { ConnectionOptions } from 'tls'
 import { defaultConfig } from '../../config/config'
 import { JobQueueManager } from '../../main/job-queues/job-queue-manager'
 import { Hub, PluginsServerConfig } from '../../types'
+import { ActionManager } from '../../worker/ingestion/action-manager'
+import { ActionMatcher } from '../../worker/ingestion/action-matcher'
 import { EventsProcessor } from '../../worker/ingestion/process-event'
 import { InternalMetrics } from '../internal-metrics'
 import { killProcess } from '../kill'
@@ -175,8 +177,10 @@ export async function createHub(
     }
 
     // :TODO: This is only used on worker threads, not main
+    hub.actionManager = new ActionManager(db)
+    await hub.actionManager.prepare()
+    hub.actionMatcher = new ActionMatcher(db, hub.actionManager)
     hub.eventsProcessor = new EventsProcessor(hub as Hub)
-    await hub.eventsProcessor.prepare()
     hub.jobQueueManager = new JobQueueManager(hub as Hub)
 
     if (serverConfig.CAPTURE_INTERNAL_METRICS) {

src/utils/utils.ts

@@ -10,7 +10,7 @@ import { Readable } from 'stream'
 import * as tar from 'tar-stream'
 import * as zlib from 'zlib'
 
-import { LogLevel, Plugin, PluginConfigId, PluginsServerConfig, TimestampFormat } from '../types'
+import { Element, LogLevel, Plugin, PluginConfigId, PluginsServerConfig, TimestampFormat } from '../types'
 import { status } from './status'
 
 /** Time until autoexit (due to error) gives up on graceful exit and kills the process right away. */
@@ -606,3 +606,16 @@ export function stringClamp(value: string, def: number, min: number, max: number
     const nanToNull = (nr: number): null | number => (isNaN(nr) ? null : nr)
     return clamp(nanToNull(parseInt(value)) ?? def, min, max)
 }
+
+export function extractElements(elements: Record<string, any>[]): Element[] {
+    return elements.map((el) => ({
+        text: el['$el_text']?.slice(0, 400),
+        tag_name: el['tag_name'],
+        href: el['attr__href']?.slice(0, 2048),
+        attr_class: el['attr__class']?.split(' '),
+        attr_id: el['attr__id'],
+        nth_child: el['nth_child'],
+        nth_of_type: el['nth_of_type'],
+        attributes: Object.fromEntries(Object.entries(el).filter(([key]) => key.startsWith('attr__'))),
+    }))
+}

src/worker/ingestion/action-manager.ts

@@ -21,11 +21,11 @@ export class ActionManager {
         this.ready = true
     }
 
-    public getTeamActions(teamId: Team['id']): ActionMap | null {
+    public getTeamActions(teamId: Team['id']): ActionMap {
         if (!this.ready) {
             throw new Error('ActionManager is not ready! Run actionManager.prepare() before this')
         }
-        return this.actionCache[teamId] || null
+        return this.actionCache[teamId] || {}
     }
 
     public async reloadAllActions(): Promise<void> {
@@ -35,7 +35,15 @@ export class ActionManager {
 
     public async reloadAction(teamId: Team['id'], actionId: Action['id']): Promise<void> {
         const refetchedAction = await this.db.fetchAction(actionId)
-        const wasCachedAlready = teamId in this.actionCache && actionId in this.actionCache[teamId]
+
+        let wasCachedAlready = true
+        if (!this.actionCache[teamId]) {
+            wasCachedAlready = false
+            this.actionCache[teamId] = {}
+        } else if (!this.actionCache[teamId][actionId]) {
+            wasCachedAlready = false
+        }
+
         if (refetchedAction) {
             status.info(
                 '🍿',
@@ -59,7 +67,8 @@ export class ActionManager {
     }
 
     public dropAction(teamId: Team['id'], actionId: Action['id']): void {
-        const wasCachedAlready = teamId in this.actionCache && actionId in this.actionCache[teamId]
+        const wasCachedAlready = !!this.actionCache?.[teamId]?.[actionId]
+
         if (wasCachedAlready) {
             status.info('🍿', `Deleted action ID ${actionId} (team ID ${teamId}) from cache`)
             delete this.actionCache[teamId][actionId]