diff --git a/package.json b/package.json index a76030d2..bed5959b 100644 --- a/package.json +++ b/package.json @@ -6,8 +6,8 @@ "main": "dist/src/index.js", "scripts": { "test": "jest --runInBand tests/**/*.test.ts", - "test:postgres": "yarn test --testPathIgnorePatterns '.*/clickhouse'", - "test:clickhouse": "yarn test --testPathIgnorePatterns '.*/postgres'", + "test:postgres": "jest --runInBand tests/postgres/*.test.ts tests/*.test.ts", + "test:clickhouse": "jest --runInBand tests/clickhouse/*.test.ts", "benchmark": "node --expose-gc node_modules/.bin/jest --runInBand benchmarks/", "start": "yarn start:dev", "start:dist": "node dist/src/index.js --base-dir ../posthog", @@ -23,7 +23,9 @@ "prettier": "prettier --write .", "prettier:check": "prettier --check .", "prepare": "yarn compile:protobuf", - "prepublishOnly": "yarn build" + "prepublishOnly": "yarn build", + "setup:dev": "cd ../posthog && (dropdb test_posthog || echo 'no db to drop') && createdb test_posthog && source env/bin/activate && DATABASE_URL=postgres://localhost:5432/test_posthog DEBUG=1 python manage.py migrate", + "setup:dev:ee": "export DEBUG=1 PRIMARY_DB=clickhouse DATABASE_URL=postgres://posthog:posthog@localhost:5439/test_posthog PGPASSWORD=posthog && cd ../posthog && (dropdb -p 5439 -h localhost -U posthog test_posthog || echo 'no db to drop') && createdb -p 5439 -h localhost -U posthog test_posthog && source env/bin/activate && python manage.py migrate && python manage.py migrate_clickhouse" }, "bin": { "posthog-plugin-server": "bin/posthog-plugin-server" diff --git a/src/db.ts b/src/db.ts index 5a64697b..7b4f430c 100644 --- a/src/db.ts +++ b/src/db.ts @@ -5,8 +5,16 @@ import { DateTime } from 'luxon' import { Pool, QueryConfig, QueryResult, QueryResultRow } from 'pg' import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' import { unparsePersonPartial } from './ingestion/utils' -import { Person, PersonDistinctId, RawPerson, RawOrganization } from './types' -import { castTimestampOrNow, sanitizeSqlIdentifier } from './utils' +import { + Person, + PersonDistinctId, + RawPerson, + RawOrganization, + Team, + PostgresSessionRecordingEvent, + Event, +} from './types' +import { castTimestampOrNow, sanitizeSqlIdentifier, UUIDT } from './utils' /** The recommended way of accessing the database. */ export class DB { @@ -30,6 +38,13 @@ export class DB { return this.postgres.query(queryTextOrConfig, values) } + // Person + + public async fetchPersons(): Promise { + const result = await this.postgresQuery('SELECT * FROM posthog_person') + return result.rows as Person[] + } + public async fetchPerson(teamId: number, distinctId: string): Promise { const selectResult = await this.postgresQuery( `SELECT @@ -45,8 +60,10 @@ export class DB { AND posthog_persondistinctid.distinct_id = $2`, [teamId, distinctId] ) - const rawPerson: RawPerson = selectResult.rows[0] - return { ...rawPerson, created_at: DateTime.fromISO(rawPerson.created_at) } + if (selectResult.rows.length > 0) { + const rawPerson: RawPerson = selectResult.rows[0] + return { ...rawPerson, created_at: DateTime.fromISO(rawPerson.created_at) } + } } public async createPerson( @@ -55,7 +72,8 @@ export class DB { teamId: number, isUserId: number | null, isIdentified: boolean, - uuid: string + uuid: string, + distinctIds?: string[] ): Promise { const insertResult = await this.postgresQuery( 'INSERT INTO posthog_person (created_at, properties, team_id, is_user_id, is_identified, uuid) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *', @@ -75,16 +93,22 @@ export class DB { messages: [{ value: Buffer.from(JSON.stringify(data)) }], }) } + + for (const distinctId of distinctIds || []) { + await this.addDistinctId(personCreated, distinctId) + } + return personCreated } public async updatePerson(person: Person, update: Partial): Promise { const updatedPerson: Person = { ...person, ...update } + const values = [...Object.values(unparsePersonPartial(update)), person.id] await this.postgresQuery( `UPDATE posthog_person SET ${Object.keys(update).map( (field, index) => sanitizeSqlIdentifier(field) + ' = $' + (index + 1) )} WHERE id = $${Object.values(update).length + 1}`, - [...Object.values(unparsePersonPartial(update)), person.id] + values ) if (this.kafkaProducer) { const data = { @@ -103,7 +127,7 @@ export class DB { } public async deletePerson(personId: number): Promise { - await this.postgresQuery('DELETE FROM person_distinct_id WHERE person_id = $1', [personId]) + await this.postgresQuery('DELETE FROM posthog_persondistinctid WHERE person_id = $1', [personId]) await this.postgresQuery('DELETE FROM posthog_person WHERE id = $1', [personId]) if (this.clickhouse) { await this.clickhouse.query(`ALTER TABLE person DELETE WHERE id = ${personId}`).toPromise() @@ -113,6 +137,16 @@ export class DB { } } + // PersonDistinctId + + public async fetchDistinctIdValues(person: Person): Promise { + const result = await this.postgresQuery( + 'SELECT * FROM posthog_persondistinctid WHERE person_id=$1 and team_id=$2 ORDER BY id', + [person.id, person.team_id] + ) + return (result.rows as PersonDistinctId[]).map((pdi) => pdi.distinct_id) + } + public async addDistinctId(person: Person, distinctId: string): Promise { const insertResult = await this.postgresQuery( 'INSERT INTO posthog_persondistinctid (distinct_id, person_id, team_id) VALUES ($1, $2, $3) RETURNING *', @@ -146,6 +180,8 @@ export class DB { } } + // Organization + public async fetchOrganization(organizationId: string): Promise { const selectResult = await this.postgresQuery(`SELECT * FROM posthog_organization WHERE id $1`, [ organizationId, @@ -153,4 +189,24 @@ export class DB { const rawOrganization: RawOrganization = selectResult.rows[0] return rawOrganization } + + // Event + + public async fetchEvents(): Promise { + const result = await this.postgresQuery('SELECT * FROM posthog_event') + return result.rows as Event[] + } + + // SessionRecordingEvent + + public async fetchSessionRecordingEvents(): Promise { + const result = await this.postgresQuery('SELECT * FROM posthog_sessionrecordingevent') + return result.rows as PostgresSessionRecordingEvent[] + } + + // Element + + public async fetchElements(): Promise { + return (await this.postgresQuery('SELECT * FROM posthog_element')).rows + } } diff --git a/src/ingestion/process-event.ts b/src/ingestion/process-event.ts index decdb1f4..73a8e0db 100644 --- a/src/ingestion/process-event.ts +++ b/src/ingestion/process-event.ts @@ -1,11 +1,22 @@ import { PluginEvent, Properties } from '@posthog/plugin-scaffold' import { DateTime, Duration } from 'luxon' -import { PluginsServer, Element, Team, Person, PersonDistinctId, CohortPeople, SessionRecordingEvent } from '../types' +import { + CohortPeople, + Element, + ElementGroup, + Person, + PersonDistinctId, + PluginsServer, + PostgresSessionRecordingEvent, + SessionRecordingEvent, + Team, + TimestampFormat, +} from '../types' import { castTimestampOrNow, UUIDT } from '../utils' import { Event as EventProto, IEvent } from '../idl/protos' import { Producer } from 'kafkajs' import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from './topics' -import { sanitizeEventName, elementsToString } from './utils' +import { elementsToString, hashElements, sanitizeEventName } from './utils' import { ClickHouse } from 'clickhouse' import { DB } from '../db' import { status } from '../status' @@ -26,8 +37,11 @@ export class EventsProcessor { this.db = pluginsServer.db this.clickhouse = pluginsServer.clickhouse! this.kafkaProducer = pluginsServer.kafkaProducer! - this.celery = new Client(pluginsServer.redis) + this.celery = new Client(pluginsServer.redis, pluginsServer.CELERY_DEFAULT_QUEUE) this.posthog = nodePostHog('sTMFPsFhdP1Ssg') + if (process.env.NODE_ENV === 'test') { + this.posthog.optOut() + } } public async processEvent( @@ -53,7 +67,7 @@ export class EventsProcessor { const personUuid = new UUIDT().toString() const ts = this.handleTimestamp(data, now, sentAt) - this.handleIdentifyOrAlias(data['event'], properties, distinctId, teamId) + await this.handleIdentifyOrAlias(data['event'], properties, distinctId, teamId) let result: IEvent | SessionRecordingEvent @@ -121,9 +135,14 @@ export class EventsProcessor { await this.alias(properties['$anon_distinct_id'], distinctId, teamId) } if (properties['$set'] || properties['$set_once']) { - this.updatePersonProperties(teamId, distinctId, properties['$set'] || {}, properties['$set_once'] || {}) + await this.updatePersonProperties( + teamId, + distinctId, + properties['$set'] || {}, + properties['$set_once'] || {} + ) } - this.setIsIdentified(teamId, distinctId) + await this.setIsIdentified(teamId, distinctId) } } @@ -139,7 +158,7 @@ export class EventsProcessor { true, new UUIDT().toString() ) - this.db.addDistinctId(personCreated, distinctId) + await this.db.addDistinctId(personCreated, distinctId) } catch { // Catch race condition where in between getting and creating, // another request already created this person @@ -190,13 +209,13 @@ export class EventsProcessor { if (oldPerson && !newPerson) { try { - this.db.addDistinctId(oldPerson, distinctId) + await this.db.addDistinctId(oldPerson, distinctId) // Catch race case when somebody already added this distinct_id between .get and .addDistinctId } catch { // integrity error if (retryIfFailed) { // run everything again to merge the users if needed - this.alias(previousDistinctId, distinctId, teamId, false) + await this.alias(previousDistinctId, distinctId, teamId, false) } } return @@ -204,13 +223,13 @@ export class EventsProcessor { if (!oldPerson && newPerson) { try { - this.db.addDistinctId(newPerson, previousDistinctId) + await this.db.addDistinctId(newPerson, previousDistinctId) // Catch race case when somebody already added this distinct_id between .get and .addDistinctId } catch { // integrity error if (retryIfFailed) { // run everything again to merge the users if needed - this.alias(previousDistinctId, distinctId, teamId, false) + await this.alias(previousDistinctId, distinctId, teamId, false) } } return @@ -226,21 +245,21 @@ export class EventsProcessor { false, new UUIDT().toString() ) - this.db.addDistinctId(personCreated, distinctId) - this.db.addDistinctId(personCreated, previousDistinctId) + await this.db.addDistinctId(personCreated, distinctId) + await this.db.addDistinctId(personCreated, previousDistinctId) } catch { // Catch race condition where in between getting and creating, // another request already created this person if (retryIfFailed) { // Try once more, probably one of the two persons exists now - this.alias(previousDistinctId, distinctId, teamId, false) + await this.alias(previousDistinctId, distinctId, teamId, false) } } return } if (oldPerson && newPerson && oldPerson.id !== newPerson.id) { - this.mergePeople(newPerson, [oldPerson]) + await this.mergePeople(newPerson, [oldPerson]) } } @@ -256,14 +275,14 @@ export class EventsProcessor { } } - await this.db.updatePerson(mergeInto, { created_at: firstSeen }) + await this.db.updatePerson(mergeInto, { created_at: firstSeen, properties: mergeInto.properties }) // merge the distinct_ids for (const otherPerson of peopleToMerge) { const otherPersonDistinctIds: PersonDistinctId[] = ( await this.db.postgresQuery( 'SELECT * FROM posthog_persondistinctid WHERE person_id = $1 AND team_id = $2', - [otherPerson, mergeInto.team_id] + [otherPerson.id, mergeInto.team_id] ) ).rows for (const personDistinctId of otherPersonDistinctIds) { @@ -321,7 +340,7 @@ export class EventsProcessor { properties['$ip'] = ip } - this.storeNamesAndProperties(team, event, properties) + await this.storeNamesAndProperties(team, event, properties) const pdiSelectResult = await this.db.postgresQuery( 'SELECT COUNT(*) AS pdicount FROM posthog_persondistinctid WHERE team_id = $1 AND distinct_id = $2', @@ -364,13 +383,13 @@ export class EventsProcessor { team.ingested_event = true save = true } - if (team.event_names && !(event in team.event_names)) { + if (team.event_names && !team.event_names.includes(event)) { save = true team.event_names.push(event) team.event_names_with_usage.push({ event: event, usage_count: null, volume: null }) } for (const [key, value] of Object.entries(properties)) { - if (team.event_properties && !(key in team.event_properties)) { + if (team.event_properties && !team.event_properties.includes(key)) { team.event_properties.push(key) team.event_properties_with_usage.push({ key: key, usage_count: null, volume: null }) save = true @@ -378,7 +397,7 @@ export class EventsProcessor { if ( typeof value === 'number' && team.event_properties_numerical && - !(key in team.event_properties_numerical) + !team.event_properties_numerical.includes(key) ) { team.event_properties_numerical.push(key) save = true @@ -395,7 +414,7 @@ export class EventsProcessor { JSON.stringify(team.event_names), JSON.stringify(team.event_names_with_usage), JSON.stringify(team.event_properties), - JSON.stringify(team.event_names_with_usage), + JSON.stringify(team.event_properties_with_usage), JSON.stringify(team.event_properties_numerical), team.id, ] @@ -413,7 +432,10 @@ export class EventsProcessor { elements?: Element[], siteUrl?: string ): Promise { - const timestampString = castTimestampOrNow(timestamp) + const timestampString = castTimestampOrNow( + timestamp, + this.kafkaProducer ? TimestampFormat.ClickHouse : TimestampFormat.ISO + ) const elementsChain = elements && elements.length ? elementsToString(elements) : '' const data: IEvent = { @@ -427,10 +449,36 @@ export class EventsProcessor { createdAt: timestampString, } - await this.kafkaProducer.send({ - topic: KAFKA_EVENTS, - messages: [{ key: uuid, value: EventProto.encodeDelimited(EventProto.create(data)).finish() as Buffer }], - }) + if (this.kafkaProducer) { + await this.kafkaProducer.send({ + topic: KAFKA_EVENTS, + messages: [ + { + key: uuid, + value: EventProto.encodeDelimited(EventProto.create(data)).finish() as Buffer, + }, + ], + }) + } else { + let elementsHash = '' + if (elements && elements.length > 0) { + elementsHash = await this.createElementGroup(elements, team.id) + } + const insertResult = await this.db.postgresQuery( + 'INSERT INTO posthog_event (created_at, event, distinct_id, properties, team_id, timestamp, elements, elements_hash) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING *', + [ + data.createdAt, + data.event, + distinctId, + data.properties, + data.teamId, + data.timestamp, + JSON.stringify(elements || []), + elementsHash, + ] + ) + const eventCreated = insertResult.rows[0] as Event + } this.celery.sendTask('ee.tasks.webhooks_ee.post_event_to_webhook_ee', [ { @@ -447,6 +495,45 @@ export class EventsProcessor { return data } + private async createElementGroup(elements: Element[], teamId: number): Promise { + const cleanedElements = elements.map((element, index) => ({ ...element, order: index })) + const hash = hashElements(cleanedElements) + + try { + const insertResult = await this.db.postgresQuery( + 'INSERT INTO posthog_elementgroup (hash, team_id) VALUES ($1, $2) RETURNING *', + [hash, teamId] + ) + const elementGroup = insertResult.rows[0] as ElementGroup + for (const element of cleanedElements) { + await this.db.postgresQuery( + 'INSERT INTO posthog_element (text, tag_name, href, attr_id, nth_child, nth_of_type, attributes, "order", event_id, attr_class, group_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)', + [ + element.text, + element.tag_name, + element.href, + element.attr_id, + element.nth_child, + element.nth_of_type, + element.attributes, + element.order, + element.event_id, + element.attr_class, + elementGroup.id, + ] + ) + } + } catch (error) { + // Throw further if not postgres error nr "23505" == "unique_violation" + // https://www.postgresql.org/docs/12/errcodes-appendix.html + if (error.code !== '23505') { + throw error + } + } + + return hash + } + private async createSessionRecordingEvent( uuid: string, team_id: number, @@ -454,7 +541,7 @@ export class EventsProcessor { session_id: string, timestamp: DateTime | string, snapshot_data: Record - ): Promise { + ): Promise { const timestampString = castTimestampOrNow(timestamp) const data: SessionRecordingEvent = { @@ -467,11 +554,19 @@ export class EventsProcessor { created_at: timestampString, } - await this.kafkaProducer.send({ - topic: KAFKA_SESSION_RECORDING_EVENTS, - messages: [{ key: uuid, value: Buffer.from(JSON.stringify(data)) }], - }) - + if (this.kafkaProducer) { + await this.kafkaProducer.send({ + topic: KAFKA_SESSION_RECORDING_EVENTS, + messages: [{ key: uuid, value: Buffer.from(JSON.stringify(data)) }], + }) + } else { + const insertResult = await this.db.postgresQuery( + 'INSERT INTO posthog_sessionrecordingevent (created_at, team_id, distinct_id, session_id, timestamp, snapshot_data) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *', + [data.created_at, data.team_id, data.distinct_id, data.session_id, data.timestamp, data.snapshot_data] + ) + const eventCreated = insertResult.rows[0] as PostgresSessionRecordingEvent + return eventCreated + } return data } } diff --git a/src/ingestion/utils.ts b/src/ingestion/utils.ts index 8098cd1a..fdab780b 100644 --- a/src/ingestion/utils.ts +++ b/src/ingestion/utils.ts @@ -1,5 +1,6 @@ import { DateTime } from 'luxon' import { Element, BaseEventMessage, RawEventMessage, EventMessage, BasePerson, RawPerson, Person } from '../types' +import crypto from 'crypto' export function parseRawEventMessage(message: RawEventMessage): EventMessage { return { @@ -67,3 +68,44 @@ export function sanitizeEventName(eventName: any): string { } return eventName.substr(0, 200) } + +/** Escape UTF-8 characters into `\u1234`. */ +function jsonEscapeUtf8(s: string): string { + return s.replace(/[^\x20-\x7F]/g, (x) => '\\u' + ('000' + x.codePointAt(0)?.toString(16)).slice(-4)) +} + +/** Produce output compatible with that of Python's `json.dumps`. */ +function jsonDumps(obj: any): string { + if (typeof obj === 'object' && obj !== null) { + if (Array.isArray(obj)) { + return `[${obj.map(jsonDumps).join(', ')}]` // space after comma + } else { + return `{${Object.keys(obj) // no space after '{' or before '}' + .sort() // must sort the keys of the object! + .map((k) => `${jsonDumps(k)}: ${jsonDumps(obj[k])}`) // space after ':' + .join(', ')}}` // space after ',' + } + } else if (typeof obj === 'string') { + return jsonEscapeUtf8(JSON.stringify(obj)) + } else { + return JSON.stringify(obj) + } +} + +export function hashElements(elements: Element[]): string { + const elementsList = elements.map((element) => ({ + attributes: element.attributes ?? null, + text: element.text ?? null, + tag_name: element.tag_name ?? null, + href: element.href ?? null, + attr_id: element.attr_id ?? null, + attr_class: element.attr_class ?? null, + nth_child: element.nth_child ?? null, + nth_of_type: element.nth_of_type ?? null, + order: element.order ?? null, + })) + + const serializedString = jsonDumps(elementsList) + + return crypto.createHash('md5').update(serializedString).digest('hex') +} diff --git a/src/server.ts b/src/server.ts index 781b2341..8deedd40 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,4 +1,4 @@ -import { Pool } from 'pg' +import { Pool, types as pgTypes } from 'pg' import * as schedule from 'node-schedule' import Redis from 'ioredis' import { Kafka, logLevel, Producer } from 'kafkajs' @@ -19,6 +19,7 @@ import { status } from './status' import { startSchedule } from './services/schedule' import { ConnectionOptions } from 'tls' import { DB } from './db' +import { DateTime } from 'luxon' export async function createServer( config: Partial = {}, @@ -88,6 +89,19 @@ export async function createServer( kafkaProducer = kafka.producer() } + // `node-postgres` will return dates as plain JS Date objects, which will use the local timezone. + // This converts all date fields to a proper luxon UTC DateTime + // Unfortunately this must be done on a global object before initializing the `Pool` + pgTypes.setTypeParser(1083 /* types.TypeId.TIME */, (timeStr) => + timeStr ? DateTime.fromSQL(timeStr, { zone: 'utc' }).toISO() : null + ) + pgTypes.setTypeParser(1114 /* types.TypeId.TIMESTAMP */, (timeStr) => + timeStr ? DateTime.fromSQL(timeStr, { zone: 'utc' }).toISO() : null + ) + pgTypes.setTypeParser(1184 /* types.TypeId.TIMESTAMPTZ */, (timeStr) => + timeStr ? DateTime.fromSQL(timeStr, { zone: 'utc' }).toISO() : null + ) + const postgres = new Pool({ connectionString: serverConfig.DATABASE_URL, ssl: process.env.DEPLOYMENT?.startsWith('Heroku') diff --git a/src/types.ts b/src/types.ts index 067feeec..e457f674 100644 --- a/src/types.ts +++ b/src/types.ts @@ -242,6 +242,25 @@ export interface Element { group_id?: number } +export interface ElementGroup { + id: number + hash: string + team_id: number +} + +/** Usable Event model. */ +export interface Event { + id: number + event?: string + properties: Record + elements?: Element[] + timestamp: string + team_id: number + distinct_id: string + elements_hash: string + created_at: string +} + /** Properties shared by RawPerson and Person. */ export interface BasePerson { id: number @@ -276,6 +295,7 @@ export interface CohortPeople { cohort_id: number person_id: number } + export interface SessionRecordingEvent { uuid: string timestamp: string @@ -285,3 +305,12 @@ export interface SessionRecordingEvent { snapshot_data: string created_at: string } + +export interface PostgresSessionRecordingEvent extends Omit { + id: string +} + +export enum TimestampFormat { + ClickHouse = 'clickhouse', + ISO = 'iso', +} diff --git a/src/utils.ts b/src/utils.ts index e91ac52a..8a619069 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -2,7 +2,7 @@ import { Readable } from 'stream' import * as tar from 'tar-stream' import AdmZip from 'adm-zip' import * as zlib from 'zlib' -import { LogLevel } from './types' +import { LogLevel, TimestampFormat } from './types' import { randomBytes } from 'crypto' import { DateTime } from 'luxon' import { status } from './status' @@ -280,13 +280,23 @@ export class UUIDT extends UUID { } /** Format timestamp for ClickHouse. */ -export function castTimestampOrNow(timestamp?: DateTime | string | null): string { +export function castTimestampOrNow( + timestamp?: DateTime | string | null, + timestampFormat: TimestampFormat = TimestampFormat.ISO +): string { if (!timestamp) { timestamp = DateTime.utc() } else if (typeof timestamp === 'string') { timestamp = DateTime.fromISO(timestamp) } - return timestamp.toUTC().toFormat('yyyy-MM-dd HH:mm:ss.u') + timestamp = timestamp.toUTC() + if (timestampFormat === TimestampFormat.ClickHouse) { + return timestamp.toFormat('yyyy-MM-dd HH:mm:ss.u') + } else if (timestampFormat === TimestampFormat.ISO) { + return timestamp.toUTC().toISO() + } else { + throw new Error(`Unrecognized timestamp format ${timestampFormat}!`) + } } export function delay(ms: number): Promise { diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index 3ee2d62f..b86ed4be 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -1,54 +1,74 @@ -import { PluginsServer } from '../../src/types' -import { createServer } from '../../src/server' -import { resetTestDatabase } from '../helpers/sql' +import { + Element, + Event, + Person, + PersonDistinctId, + PluginsServer, + PluginsServerConfig, + PostgresSessionRecordingEvent, +} from '../../src/types' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { KafkaCollector, KafkaObserver } from '../helpers/kafka' import { UUIDT } from '../../src/utils' import { DateTime } from 'luxon' +import { createProcessEventTests } from '../shared/process-event' jest.setTimeout(180_000) // 3 minute timeout -let server: PluginsServer -let closeServer: () => Promise -const kafkaObserver = new KafkaObserver() +const extraServerConfig: Partial = { + KAFKA_ENABLED: true, + KAFKA_HOSTS: 'kafka:9092', + DATABASE_URL: 'postgres://posthog:posthog@localhost:5439/test_posthog', +} -beforeEach(async () => { - ;[server, closeServer] = await createServer() - await resetTestDatabase(`const processEvent = event => event`) - await resetTestDatabaseClickhouse() -}) -afterEach(() => { - closeServer() -}) +describe('process event (clickhouse)', () => { + const kafkaObserver = new KafkaObserver(extraServerConfig) -test('event is passed through', async () => { - const uuid = new UUIDT().toString() - const now = DateTime.utc() - console.log('starting kafka observer') - await kafkaObserver.start() - console.log('sending message') - const kafkaCollector = new KafkaCollector(kafkaObserver) - await kafkaObserver.handOffMessage({ - distinct_id: 'abcd', - ip: '1.1.1.1', - site_url: 'x.com', - team_id: 1, - uuid, - data: { + beforeEach(async () => { + await resetTestDatabaseClickhouse(extraServerConfig) + }) + + const server = createProcessEventTests( + 'clickhouse', + { + getSessionRecordingEvents: (server) => server.db.fetchSessionRecordingEvents(), + getEvents: (server) => server.db.fetchEvents(), + getPersons: (server) => server.db.fetchPersons(), + getDistinctIds: (server, person) => server.db.fetchDistinctIdValues(person), + getElements: (server) => server.db.fetchElements(), + }, + extraServerConfig + ) + + test('event is passed through', async () => { + const uuid = new UUIDT().toString() + const now = DateTime.utc() + console.log('starting kafka observer') + await kafkaObserver.start() + console.log('sending message') + const kafkaCollector = new KafkaCollector(kafkaObserver) + await kafkaObserver.handOffMessage({ distinct_id: 'abcd', ip: '1.1.1.1', site_url: 'x.com', team_id: 1, - now: now.toString(), - event: 'test', uuid, - }, - now, - sent_at: null, - }) - console.log('waiting for messages') - const processedMessages = await kafkaCollector.collect(1) + data: { + distinct_id: 'abcd', + ip: '1.1.1.1', + site_url: 'x.com', + team_id: 1, + now: now.toString(), + event: 'test', + uuid, + }, + now, + sent_at: null, + }) + console.log('waiting for messages') + const processedMessages = await kafkaCollector.collect(1) - console.log(processedMessages) - expect(1).toEqual(1) + console.log(processedMessages) + expect(1).toEqual(1) + }) }) diff --git a/tests/helpers/clickhouse.ts b/tests/helpers/clickhouse.ts index aef8f751..9a006dd2 100644 --- a/tests/helpers/clickhouse.ts +++ b/tests/helpers/clickhouse.ts @@ -1,12 +1,14 @@ import { defaultConfig } from '../../src/config' import { ClickHouse } from 'clickhouse' +import { PluginsServerConfig } from '../../src/types' -export async function resetTestDatabaseClickhouse(): Promise { +export async function resetTestDatabaseClickhouse(extraServerConfig: Partial): Promise { + const config = { ...defaultConfig, ...extraServerConfig } const clickhouse = new ClickHouse({ - url: `http://$${defaultConfig.CLICKHOUSE_HOST}`, + url: `http://$${config.CLICKHOUSE_HOST}`, port: 8123, config: { - database: defaultConfig.CLICKHOUSE_DATABASE, + database: config.CLICKHOUSE_DATABASE, }, }) await clickhouse.query('TRUNCATE events').toPromise() diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 37b7dbe7..15613922 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -2,8 +2,9 @@ import { EventEmitter } from 'events' import { Kafka, Consumer, logLevel, EachMessagePayload, Producer } from 'kafkajs' import { KAFKA_EVENTS, KAFKA_EVENTS_INGESTION_HANDOFF } from '../../src/ingestion/topics' import { parseRawEventMessage } from '../../src/ingestion/utils' -import { EventMessage } from '../../src/types' +import { EventMessage, PluginsServerConfig } from '../../src/types' import { UUIDT } from '../../src/utils' +import { defaultConfig, overrideWithEnv } from '../../src/config' export class KafkaObserver extends EventEmitter { public kafka: Kafka @@ -12,11 +13,12 @@ export class KafkaObserver extends EventEmitter { private isStarted: boolean - constructor() { + constructor(extraServerConfig: Partial) { super() + const config = { ...overrideWithEnv(defaultConfig, process.env), ...extraServerConfig } this.kafka = new Kafka({ clientId: `plugin-server-test-${new UUIDT()}`, - brokers: process.env.KAFKA_HOSTS!.split(','), + brokers: (config.KAFKA_HOSTS || '').split(','), logLevel: logLevel.NOTHING, }) this.producer = this.kafka.producer() diff --git a/tests/helpers/plugins.ts b/tests/helpers/plugins.ts index 75f89e9b..224f3840 100644 --- a/tests/helpers/plugins.ts +++ b/tests/helpers/plugins.ts @@ -4,6 +4,8 @@ import path from 'path' import os from 'os' import AdmZip from 'adm-zip' +export const commonUserId = 1001 +export const commonOrganizationMembershipId = '0177364a-fc7b-0000-511c-137090b9e4e1' export const commonOrganizationId = 'ca30f2ec-e9a4-4001-bf27-3ef194086068' export const plugin60: Plugin = { diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 6e4463e0..92e39a95 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -1,49 +1,34 @@ -import { makePluginObjects, commonOrganizationId } from './plugins' +import { makePluginObjects, commonOrganizationId, commonUserId, commonOrganizationMembershipId } from './plugins' import { defaultConfig } from '../../src/config' import { Pool } from 'pg' import { delay, UUIDT } from '../../src/utils' +import { PluginsServerConfig } from '../../src/types' -export async function resetTestDatabase(code: string): Promise { - const db = new Pool({ connectionString: defaultConfig.DATABASE_URL }) +export async function resetTestDatabase( + code: string, + extraServerConfig: Partial = {} +): Promise { + const config = { ...defaultConfig, ...extraServerConfig } + const db = new Pool({ connectionString: config.DATABASE_URL }) const mocks = makePluginObjects(code) + await db.query('DELETE FROM posthog_element') + await db.query('DELETE FROM posthog_elementgroup') + await db.query('DELETE FROM posthog_sessionrecordingevent') + await db.query('DELETE FROM posthog_persondistinctid') + await db.query('DELETE FROM posthog_person') + await db.query('DELETE FROM posthog_event') await db.query('DELETE FROM posthog_pluginstorage') await db.query('DELETE FROM posthog_pluginattachment') await db.query('DELETE FROM posthog_pluginconfig') await db.query('DELETE FROM posthog_plugin') await db.query('DELETE FROM posthog_team') + await db.query('DELETE FROM posthog_organizationmembership') await db.query('DELETE FROM posthog_organization') + await db.query('DELETE FROM posthog_user') const teamIds = mocks.pluginConfigRows.map((c) => c.team_id) - await insertRow(db, 'posthog_organization', { - id: commonOrganizationId, - name: 'TEST ORG', - created_at: new Date().toISOString(), - updated_at: new Date().toISOString(), - personalization: '{}', - }) - for (const teamId of teamIds) { - await insertRow(db, 'posthog_team', { - id: teamId, - organization_id: commonOrganizationId, - app_urls: [], - name: 'TEST PROJECT', - event_names: [], - event_names_with_usage: [], - event_properties: [], - event_properties_with_usage: [], - event_properties_numerical: [], - created_at: new Date().toISOString(), - updated_at: new Date().toISOString(), - anonymize_ips: false, - completed_snippet_onboarding: true, - ingested_event: true, - uuid: new UUIDT().toString(), - session_recording_opt_in: true, - plugins_opt_in: true, - opt_out_capture: false, - is_demo: false, - }) - } + await createUserTeamAndOrganization(db, teamIds[0]) + for (const plugin of mocks.pluginRows) { await insertRow(db, 'posthog_plugin', plugin) } @@ -71,3 +56,59 @@ async function insertRow(db: Pool, table: string, object: Record): throw error } } + +export async function createUserTeamAndOrganization( + db: Pool, + teamId: number, + userId: number = commonUserId, + organizationId: string = commonOrganizationId, + organizationMembershipId: string = commonOrganizationMembershipId +): Promise { + await insertRow(db, 'posthog_user', { + id: userId, + password: 'gibberish', + first_name: 'PluginTest', + last_name: 'User', + email: `test${userId}@posthog.com`, + distinct_id: `plugin_test_user_distinct_id_${userId}`, + is_staff: false, + is_active: false, + date_joined: new Date().toISOString(), + }) + await insertRow(db, 'posthog_organization', { + id: organizationId, + name: 'TEST ORG', + created_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + personalization: '{}', + }) + await insertRow(db, 'posthog_organizationmembership', { + id: organizationMembershipId, + organization_id: organizationId, + user_id: userId, + level: 15, + joined_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + }) + await insertRow(db, 'posthog_team', { + id: teamId, + organization_id: organizationId, + app_urls: [], + name: 'TEST PROJECT', + event_names: JSON.stringify([]), + event_names_with_usage: JSON.stringify([]), + event_properties: JSON.stringify([]), + event_properties_with_usage: JSON.stringify([]), + event_properties_numerical: JSON.stringify([]), + created_at: new Date().toISOString(), + updated_at: new Date().toISOString(), + anonymize_ips: false, + completed_snippet_onboarding: true, + ingested_event: true, + uuid: new UUIDT().toString(), + session_recording_opt_in: true, + plugins_opt_in: true, + opt_out_capture: false, + is_demo: false, + }) +} diff --git a/tests/postgres/process-event.test.ts b/tests/postgres/process-event.test.ts new file mode 100644 index 00000000..d404d826 --- /dev/null +++ b/tests/postgres/process-event.test.ts @@ -0,0 +1,14 @@ +import { PluginsServer, Event, Person, PersonDistinctId, Element, PostgresSessionRecordingEvent } from '../../src/types' +import { createProcessEventTests } from '../shared/process-event' + +jest.setTimeout(600000) // 600 sec timeout + +describe('process event (postgresql)', () => { + createProcessEventTests('postgresql', { + getSessionRecordingEvents: (server) => server.db.fetchSessionRecordingEvents(), + getEvents: (server) => server.db.fetchEvents(), + getPersons: (server) => server.db.fetchPersons(), + getDistinctIds: (server, person) => server.db.fetchDistinctIdValues(person), + getElements: (server) => server.db.fetchElements(), + }) +}) diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index ef459fc1..dfa6ff41 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -714,9 +714,9 @@ test('posthog in runEvery', async () => { expect(Client).toHaveBeenCalledTimes(2) expect((Client as any).mock.calls[0][1]).toEqual(mockServer.CELERY_DEFAULT_QUEUE) // webhook to celery queue - expect((Client as any).mock.calls[1][1]).toEqual(mockServer.PLUGINS_CELERY_QUEUE) // events out to start of plugin + expect((Client as any).mock.calls[1][1]).toEqual(mockServer.PLUGINS_CELERY_QUEUE) // events out to start of plugin queue - const mockClientInstance = (Client as any).mock.instances[0] + const mockClientInstance = (Client as any).mock.instances[1] const mockSendTask = mockClientInstance.sendTask expect(mockSendTask.mock.calls[0][0]).toEqual('posthog.tasks.process_event.process_event_with_plugins') @@ -753,9 +753,9 @@ test('posthog in runEvery with timestamp', async () => { expect(Client).toHaveBeenCalledTimes(2) expect((Client as any).mock.calls[0][1]).toEqual(mockServer.CELERY_DEFAULT_QUEUE) // webhook to celery queue - expect((Client as any).mock.calls[1][1]).toEqual(mockServer.PLUGINS_CELERY_QUEUE) // events out to start of plugin + expect((Client as any).mock.calls[1][1]).toEqual(mockServer.PLUGINS_CELERY_QUEUE) // events out to start of plugin queue - const mockClientInstance = (Client as any).mock.instances[0] + const mockClientInstance = (Client as any).mock.instances[1] const mockSendTask = mockClientInstance.sendTask expect(mockSendTask.mock.calls[0][0]).toEqual('posthog.tasks.process_event.process_event_with_plugins') diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts new file mode 100644 index 00000000..1417dc52 --- /dev/null +++ b/tests/shared/process-event.ts @@ -0,0 +1,1045 @@ +import { PluginEvent } from '@posthog/plugin-scaffold/src/types' +import { createServer } from '../../src/server' +import { + LogLevel, + PluginsServer, + Team, + Event, + Person, + Element, + PostgresSessionRecordingEvent, + PluginsServerConfig, +} from '../../src/types' +import { createUserTeamAndOrganization, resetTestDatabase } from '../helpers/sql' +import { EventsProcessor } from '../../src/ingestion/process-event' +import { DateTime } from 'luxon' +import { UUIDT } from '../../src/utils' + +jest.setTimeout(600000) // 600 sec timeout + +async function getTeams(server: PluginsServer): Promise { + return (await server.db.postgresQuery('SELECT * FROM posthog_team ORDER BY id')).rows +} + +async function getFirstTeam(server: PluginsServer): Promise { + return (await getTeams(server))[0] +} + +async function createPerson( + server: PluginsServer, + team: Team, + distinctIds: string[], + properties: Record = {} +): Promise { + return server.db.createPerson(DateTime.utc(), properties, team.id, null, false, new UUIDT().toString(), distinctIds) +} + +export const createProcessEventTests = ( + database: 'postgresql' | 'clickhouse', + { + getSessionRecordingEvents, + getEvents, + getPersons, + getDistinctIds, + getElements, + }: { + getSessionRecordingEvents: (server: PluginsServer) => Promise + getEvents: (server: PluginsServer) => Promise + getPersons: (server: PluginsServer) => Promise + getDistinctIds: (server: PluginsServer, person: Person) => Promise + getElements: (server: PluginsServer, event: Event) => Promise + }, + extraServerConfig?: Partial +): PluginsServer => { + let queryCounter = 0 + let team: Team + let server: PluginsServer + let stopServer: () => Promise + let eventsProcessor: EventsProcessor + let now = DateTime.utc() + + async function getServer(): Promise<[PluginsServer, () => Promise]> { + const [server, stopServer] = await createServer({ + PLUGINS_CELERY_QUEUE: 'test-plugins-celery-queue', + CELERY_DEFAULT_QUEUE: 'test-celery-default-queue', + LOG_LEVEL: LogLevel.Log, + ...(extraServerConfig ?? {}), + }) + + await server.redis.del(server.PLUGINS_CELERY_QUEUE) + await server.redis.del(server.CELERY_DEFAULT_QUEUE) + + const query = server.postgres.query.bind(server.postgres) + server.postgres.query = (queryText: any, values?: any, callback?: any): any => { + queryCounter++ + return query(queryText, values, callback) + } + + return [server, stopServer] + } + + beforeEach(async () => { + const testCode = ` + function processEvent (event, meta) { + event.properties["somewhere"] = "over the rainbow"; + return event + } + ` + await resetTestDatabase(testCode, extraServerConfig) + ;[server, stopServer] = await getServer() + eventsProcessor = new EventsProcessor(server) + queryCounter = 0 + team = await getFirstTeam(server) + now = DateTime.utc() + }) + + afterEach(async () => { + await stopServer?.() + }) + + test('capture new person', async () => { + await server.db.postgresQuery(`UPDATE posthog_team SET ingested_event = $1 WHERE id = $2`, [true, team.id]) + team = await getFirstTeam(server) + + expect(team.event_names).toEqual([]) + + await eventsProcessor.processEvent( + '2', + '', + '', + ({ + event: '$autocapture', + properties: { + distinct_id: 2, + token: team.api_token, + $elements: [ + { tag_name: 'a', nth_child: 1, nth_of_type: 2, attr__class: 'btn btn-sm' }, + { tag_name: 'div', nth_child: 1, nth_of_type: 2, $el_text: '💻' }, + ], + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect(queryCounter).toBe(12) + + // TODO: 12 vs 28 is a big difference. Why so? + // num_queries = 28 + // if settings.EE_AVAILABLE: # extra queries to check for hooks + // num_queries += 4 + // if settings.MULTI_TENANCY: # extra query to check for billing plan + // num_queries += 1 + // with self.assertNumQueries(num_queries): + + // capture a second time to verify e.g. event_names is not ['$autocapture', '$autocapture'] + await eventsProcessor.processEvent( + '2', + '', + '', + ({ + event: '$autocapture', + properties: { + distinct_id: 2, + token: team.api_token, + $elements: [ + { tag_name: 'a', nth_child: 1, nth_of_type: 2, attr__class: 'btn btn-sm' }, + { tag_name: 'div', nth_child: 1, nth_of_type: 2, $el_text: '💻' }, + ], + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const events = await getEvents(server) + const persons = await getPersons(server) + expect(events.length).toEqual(2) + expect(persons.length).toEqual(1) + + const [event] = events + const [person] = persons + const distinctIds = await getDistinctIds(server, person) + + expect(event.distinct_id).toEqual('2') + expect(distinctIds).toEqual(['2']) + expect(event.event).toEqual('$autocapture') + expect(event.elements_hash).toEqual('0679137c0cd2408a2906839143e7a71f') + + const elements = await getElements(server, event) + expect(elements[0].tag_name).toEqual('a') + expect(elements[0].attr_class).toEqual(['btn', 'btn-sm']) + expect(elements[1].order).toEqual(1) + expect(elements[1].text).toEqual('💻') + + team = await getFirstTeam(server) + expect(team.event_names).toEqual(['$autocapture']) + expect(team.event_names_with_usage).toEqual([{ event: '$autocapture', volume: null, usage_count: null }]) + expect(team.event_properties).toEqual(['distinct_id', 'token', '$ip']) + expect(team.event_properties_with_usage).toEqual([ + { key: 'distinct_id', usage_count: null, volume: null }, + { key: 'token', usage_count: null, volume: null }, + { key: '$ip', usage_count: null, volume: null }, + ]) + }) + + test('capture no element', async () => { + await createPerson(server, team, ['asdfasdfasdf']) + + await eventsProcessor.processEvent( + 'asdfasdfasdf', + '', + '', + ({ + event: '$pageview', + properties: { distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual(['asdfasdfasdf']) + const [event] = await getEvents(server) + expect(event.event).toBe('$pageview') + }) + + test('capture sent_at', async () => { + await createPerson(server, team, ['asdfasdfasdf']) + + const rightNow = DateTime.utc() + const tomorrow = rightNow.plus({ days: 1, hours: 2 }) + const tomorrowSentAt = rightNow.plus({ days: 1, hours: 2, minutes: 10 }) + + await eventsProcessor.processEvent( + 'movie played', + '', + '', + ({ + event: '$pageview', + timestamp: tomorrow.toISO(), + properties: { distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + rightNow, + tomorrowSentAt, + new UUIDT().toString() + ) + + const [event] = await getEvents(server) + const eventSecondsBeforeNow = rightNow.diff(DateTime.fromISO(event.timestamp), 'seconds').seconds + + expect(eventSecondsBeforeNow).toBeGreaterThan(590) + expect(eventSecondsBeforeNow).toBeLessThan(610) + }) + + test('capture sent_at no timezones', async () => { + await createPerson(server, team, ['asdfasdfasdf']) + + const rightNow = DateTime.utc() + const tomorrow = rightNow.plus({ days: 1, hours: 2 }).setZone('UTC+4') + const tomorrowSentAt = rightNow.plus({ days: 1, hours: 2, minutes: 10 }).setZone('UTC+4') + + // TODO: not sure if this is correct? + // tomorrow = tomorrow.replace(tzinfo=None) + // tomorrow_sent_at = tomorrow_sent_at.replace(tzinfo=None) + + await eventsProcessor.processEvent( + 'movie played', + '', + '', + ({ + event: '$pageview', + timestamp: tomorrow, + properties: { distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + rightNow, + tomorrowSentAt, + new UUIDT().toString() + ) + + const [event] = await getEvents(server) + const eventSecondsBeforeNow = rightNow.diff(DateTime.fromISO(event.timestamp), 'seconds').seconds + + expect(eventSecondsBeforeNow).toBeGreaterThan(590) + expect(eventSecondsBeforeNow).toBeLessThan(610) + }) + + test('capture no sent_at', async () => { + await createPerson(server, team, ['asdfasdfasdf']) + + const rightNow = DateTime.utc() + const tomorrow = rightNow.plus({ days: 1, hours: 2 }) + + await eventsProcessor.processEvent( + 'movie played', + '', + '', + ({ + event: '$pageview', + timestamp: tomorrow.toISO(), + properties: { distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + rightNow, + null, + new UUIDT().toString() + ) + + const [event] = await getEvents(server) + const difference = tomorrow.diff(DateTime.fromISO(event.timestamp), 'seconds').seconds + expect(difference).toBeLessThan(1) + }) + + test('ip capture', async () => { + await createPerson(server, team, ['asdfasdfasdf']) + + await eventsProcessor.processEvent( + 'asdfasdfasdf', + '11.12.13.14', + '', + ({ + event: '$pageview', + properties: { distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + const [event] = await getEvents(server) + expect(event.properties['$ip']).toBe('11.12.13.14') + }) + + test('ip override', async () => { + await createPerson(server, team, ['asdfasdfasdf']) + + await eventsProcessor.processEvent( + 'asdfasdfasdf', + '11.12.13.14', + '', + ({ + event: '$pageview', + properties: { $ip: '1.0.0.1', distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + const [event] = await getEvents(server) + expect(event.properties['$ip']).toBe('1.0.0.1') + }) + + test('anonymized ip capture', async () => { + await server.db.postgresQuery('update posthog_team set anonymize_ips = $1', [true]) + await createPerson(server, team, ['asdfasdfasdf']) + + await eventsProcessor.processEvent( + 'asdfasdfasdf', + '11.12.13.14', + '', + ({ + event: '$pageview', + properties: { distinct_id: 'asdfasdfasdf', token: team.api_token }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + const [event] = await getEvents(server) + expect(event.properties['$ip']).not.toBeDefined() + }) + + test('alias', async () => { + await createPerson(server, team, ['old_distinct_id']) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'new_distinct_id', token: team.api_token, alias: 'old_distinct_id' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + 'old_distinct_id', + 'new_distinct_id', + ]) + }) + + test('alias reverse', async () => { + await createPerson(server, team, ['old_distinct_id']) + + await eventsProcessor.processEvent( + 'old_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'old_distinct_id', token: team.api_token, alias: 'new_distinct_id' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + 'old_distinct_id', + 'new_distinct_id', + ]) + }) + + test('alias twice', async () => { + await createPerson(server, team, ['old_distinct_id']) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'new_distinct_id', token: team.api_token, alias: 'old_distinct_id' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getPersons(server)).length).toBe(1) + expect((await getEvents(server)).length).toBe(1) + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + 'old_distinct_id', + 'new_distinct_id', + ]) + + await createPerson(server, team, ['old_distinct_id_2']) + expect((await getPersons(server)).length).toBe(2) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'new_distinct_id', token: team.api_token, alias: 'old_distinct_id_2' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + expect((await getEvents(server)).length).toBe(2) + expect((await getPersons(server)).length).toBe(1) + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + 'old_distinct_id', + 'new_distinct_id', + 'old_distinct_id_2', + ]) + }) + + test('alias before person', async () => { + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'new_distinct_id', token: team.api_token, alias: 'old_distinct_id' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + expect((await getPersons(server)).length).toBe(1) + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + 'new_distinct_id', + 'old_distinct_id', + ]) + }) + + test('alias both existing', async () => { + await createPerson(server, team, ['old_distinct_id']) + await createPerson(server, team, ['new_distinct_id']) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'new_distinct_id', token: team.api_token, alias: 'old_distinct_id' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + 'old_distinct_id', + 'new_distinct_id', + ]) + }) + + test('offset timestamp', async () => { + now = DateTime.fromISO('2020-01-01T12:00:05.200Z') + + await eventsProcessor.processEvent( + 'distinct_id', + '', + '', + ({ offset: 150, event: '$autocapture', distinct_id: 'distinct_id' } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + expect((await getEvents(server)).length).toBe(1) + + const [event] = await getEvents(server) + expect(event.timestamp).toEqual('2020-01-01T12:00:05.050Z') + }) + + test('offset timestamp no sent_at', async () => { + now = DateTime.fromISO('2020-01-01T12:00:05.200Z') + + await eventsProcessor.processEvent( + 'distinct_id', + '', + '', + ({ offset: 150, event: '$autocapture', distinct_id: 'distinct_id' } as any) as PluginEvent, + team.id, + now, + null, + new UUIDT().toString() + ) + expect((await getEvents(server)).length).toBe(1) + + const [event] = await getEvents(server) + expect(event.timestamp).toEqual('2020-01-01T12:00:05.050Z') + }) + + test('alias merge properties', async () => { + await createPerson(server, team, ['old_distinct_id'], { + key_on_both: 'old value both', + key_on_old: 'old value', + }) + await createPerson(server, team, ['new_distinct_id'], { + key_on_both: 'new value both', + key_on_new: 'new value', + }) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$create_alias', + properties: { distinct_id: 'new_distinct_id', token: team.api_token, alias: 'old_distinct_id' }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + expect((await getPersons(server)).length).toBe(1) + const [person] = await getPersons(server) + expect(await getDistinctIds(server, person)).toEqual(['old_distinct_id', 'new_distinct_id']) + expect(person.properties).toEqual({ + key_on_both: 'new value both', + key_on_new: 'new value', + key_on_old: 'old value', + }) + }) + + test('long htext', async () => { + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$autocapture', + properties: { + distinct_id: 'new_distinct_id', + token: team.api_token, + $elements: [ + { + tag_name: 'a', + $el_text: 'a'.repeat(2050), + attr__href: 'a'.repeat(2050), + nth_child: 1, + nth_of_type: 2, + attr__class: 'btn btn-sm', + }, + ], + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const [event] = await getEvents(server) + expect(event.elements_hash).toEqual('c2659b28e72835706835764cf7f63c2a') + const [element] = await getElements(server, event) + expect(element.href?.length).toEqual(2048) + expect(element.text?.length).toEqual(400) + }) + + test('capture first team event', async () => { + await server.db.postgresQuery(`UPDATE posthog_team SET ingested_event = $1 WHERE id = $2`, [false, team.id]) + + eventsProcessor.posthog = { + identify: jest.fn((distinctId) => true), + capture: jest.fn((event, properties) => true), + } as any + + await eventsProcessor.processEvent( + '2', + '', + '', + ({ + event: '$autocapture', + properties: { + distinct_id: 1, + token: team.api_token, + $elements: [{ tag_name: 'a', nth_child: 1, nth_of_type: 2, attr__class: 'btn btn-sm' }], + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect(eventsProcessor.posthog.identify).toHaveBeenCalledWith('plugin_test_user_distinct_id_1001') + expect(eventsProcessor.posthog.capture).toHaveBeenCalledWith('first team event ingested', { + team: team.uuid, + }) + + team = await getFirstTeam(server) + expect(team.ingested_event).toEqual(true) + + const [event] = await getEvents(server) + expect(event.elements_hash).toEqual('a89021a60b3497d24e93ae181fba01aa') + }) + + test('snapshot event stored as session_recording_event', async () => { + await eventsProcessor.processEvent( + 'some-id', + '', + '', + ({ + event: '$snapshot', + properties: { $session_id: 'abcf-efg', $snapshot_data: { timestamp: 123 } }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const events = await getEvents(server) + expect(events.length).toEqual(0) + + const sessionRecordingEvents = await getSessionRecordingEvents(server) + expect(sessionRecordingEvents.length).toBe(1) + + const [event] = sessionRecordingEvents + expect(event.session_id).toEqual('abcf-efg') + expect(event.distinct_id).toEqual('some-id') + expect(event.snapshot_data).toEqual({ timestamp: 123 }) + }) + + test('identify set', async () => { + await createPerson(server, team, ['distinct_id']) + + await eventsProcessor.processEvent( + 'distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + token: team.api_token, + distinct_id: 'distinct_id', + $set: { a_prop: 'test-1', c_prop: 'test-1' }, + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + + const [event] = await getEvents(server) + expect(event.properties['$set']).toEqual({ a_prop: 'test-1', c_prop: 'test-1' }) + + const [person] = await getPersons(server) + expect(await getDistinctIds(server, person)).toEqual(['distinct_id']) + expect(person.properties).toEqual({ a_prop: 'test-1', c_prop: 'test-1' }) + expect(person.is_identified).toEqual(true) + + await eventsProcessor.processEvent( + 'distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + token: team.api_token, + distinct_id: 'distinct_id', + $set: { a_prop: 'test-2', b_prop: 'test-2b' }, + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + expect((await getEvents(server)).length).toBe(2) + const [person2] = await getPersons(server) + expect(person2.properties).toEqual({ a_prop: 'test-2', b_prop: 'test-2b', c_prop: 'test-1' }) + }) + + test('identify set_once', async () => { + await createPerson(server, team, ['distinct_id']) + + await eventsProcessor.processEvent( + 'distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + token: team.api_token, + distinct_id: 'distinct_id', + $set_once: { a_prop: 'test-1', c_prop: 'test-1' }, + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + + const [event] = await getEvents(server) + expect(event.properties['$set_once']).toEqual({ a_prop: 'test-1', c_prop: 'test-1' }) + + const [person] = await getPersons(server) + expect(await getDistinctIds(server, person)).toEqual(['distinct_id']) + expect(person.properties).toEqual({ a_prop: 'test-1', c_prop: 'test-1' }) + expect(person.is_identified).toEqual(true) + + await eventsProcessor.processEvent( + 'distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + token: team.api_token, + distinct_id: 'distinct_id', + $set_once: { a_prop: 'test-2', b_prop: 'test-2b' }, + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + expect((await getEvents(server)).length).toBe(2) + const [person2] = await getPersons(server) + expect(person2.properties).toEqual({ a_prop: 'test-1', b_prop: 'test-2b', c_prop: 'test-1' }) + }) + + test('distinct with anonymous_id', async () => { + await createPerson(server, team, ['anonymous_id']) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + $anon_distinct_id: 'anonymous_id', + token: team.api_token, + distinct_id: 'new_distinct_id', + $set: { a_prop: 'test' }, + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + expect((await getEvents(server)).length).toBe(1) + const [event] = await getEvents(server) + expect(event.properties['$set']).toEqual({ a_prop: 'test' }) + const [person] = await getPersons(server) + expect(await getDistinctIds(server, person)).toEqual(['anonymous_id', 'new_distinct_id']) + expect(person.properties).toEqual({ a_prop: 'test' }) + + // check no errors as this call can happen multiple times + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + $anon_distinct_id: 'anonymous_id', + token: team.api_token, + distinct_id: 'new_distinct_id', + $set: { a_prop: 'test' }, + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + }) + + // This case is likely to happen after signup, for example: + // 1. User browses website with anonymous_id + // 2. User signs up, triggers event with their new_distinct_id (creating a new Person) + // 3. In the frontend, try to alias anonymous_id with new_distinct_id + // Result should be that we end up with one Person with both ID's + test('distinct with anonymous_id which was already created', async () => { + await createPerson(server, team, ['anonymous_id']) + await createPerson(server, team, ['new_distinct_id'], { email: 'someone@gmail.com' }) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + $anon_distinct_id: 'anonymous_id', + token: team.api_token, + distinct_id: 'new_distinct_id', + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const [person] = await getPersons(server) + expect(await getDistinctIds(server, person)).toEqual(['anonymous_id', 'new_distinct_id']) + expect(person.properties['email']).toEqual('someone@gmail.com') + }) + + test('distinct with multiple anonymous_ids which were already created', async () => { + await createPerson(server, team, ['anonymous_id']) + await createPerson(server, team, ['new_distinct_id'], { email: 'someone@gmail.com' }) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + $anon_distinct_id: 'anonymous_id', + token: team.api_token, + distinct_id: 'new_distinct_id', + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const persons1 = await getPersons(server) + expect(persons1.length).toBe(1) + expect(await getDistinctIds(server, persons1[0])).toEqual(['anonymous_id', 'new_distinct_id']) + expect(persons1[0].properties['email']).toEqual('someone@gmail.com') + + await createPerson(server, team, ['anonymous_id_2']) + + await eventsProcessor.processEvent( + 'new_distinct_id', + '', + '', + ({ + event: '$identify', + properties: { + $anon_distinct_id: 'anonymous_id_2', + token: team.api_token, + distinct_id: 'new_distinct_id', + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const persons2 = await getPersons(server) + expect(persons2.length).toBe(1) + expect(await getDistinctIds(server, persons2[0])).toEqual(['anonymous_id', 'new_distinct_id', 'anonymous_id_2']) + expect(persons2[0].properties['email']).toEqual('someone@gmail.com') + }) + + test('distinct team leakage', async () => { + await createUserTeamAndOrganization( + server.postgres, + 3, + 1002, + '01774e2f-0d01-0000-ee94-9a238640c6ee', + '0174f81e-36f5-0000-7ef8-cc26c1fbab1c' + ) + const team2 = (await getTeams(server))[1] + await createPerson(server, team2, ['2'], { email: 'team2@gmail.com' }) + await createPerson(server, team, ['1', '2']) + + await eventsProcessor.processEvent( + '2', + '', + '', + ({ + event: '$identify', + properties: { + $anon_distinct_id: '1', + token: team.api_token, + distinct_id: '2', + }, + } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const people = await getPersons(server) + expect(people.length).toEqual(2) + expect(people[1].team_id).toEqual(team.id) + expect(people[1].properties).toEqual({}) + expect(await getDistinctIds(server, people[1])).toEqual(['1', '2']) + expect(people[0].team_id).toEqual(team2.id) + expect(await getDistinctIds(server, people[0])).toEqual(['2']) + }) + + test('set is_identified', async () => { + const distinct_id = '777' + const person1 = await createPerson(server, team, [distinct_id]) + expect(person1.is_identified).toBe(false) + + await eventsProcessor.processEvent( + distinct_id, + '', + '', + ({ event: '$identify', properties: {} } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + const [person2] = await getPersons(server) + expect(person2.is_identified).toBe(true) + }) + + test('team event_properties', async () => { + expect(team.event_properties_numerical).toEqual([]) + + await eventsProcessor.processEvent( + 'xxx', + '', + '', + ({ event: 'purchase', properties: { price: 299.99, name: 'AirPods Pro' } } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + + team = await getFirstTeam(server) + expect(team.event_properties).toEqual(['price', 'name', '$ip']) + expect(team.event_properties_numerical).toEqual(['price']) + }) + + test('event name object json', async () => { + await eventsProcessor.processEvent( + 'xxx', + '', + '', + ({ event: { 'event name': 'as object' }, properties: {} } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + const [event] = await getEvents(server) + expect(event.event).toEqual('{"event name":"as object"}') + }) + + test('event name array json', async () => { + await eventsProcessor.processEvent( + 'xxx', + '', + '', + ({ event: ['event name', 'a list'], properties: {} } as any) as PluginEvent, + team.id, + now, + now, + new UUIDT().toString() + ) + const [event] = await getEvents(server) + expect(event.event).toEqual('["event name","a list"]') + }) + + test('long event name substr', async () => { + await eventsProcessor.processEvent( + 'xxx', + '', + '', + ({ event: 'E'.repeat(300), properties: { price: 299.99, name: 'AirPods Pro' } } as any) as PluginEvent, + team.id, + DateTime.utc(), + DateTime.utc(), + 'uuid' + ) + + const [event] = await getEvents(server) + expect(event.event?.length).toBe(200) + }) + + return server! +}