diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index faeda4cf..d0030629 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -122,29 +122,13 @@ jobs: --health-interval 10s --health-timeout 5s --health-retries 5 - clickhouse: - image: yandex/clickhouse-server - ports: - - '8123:8123' - - '9000:9000' - - '9440:9440' - - '9009:9009' - zookeeper: - image: wurstmeister/zookeeper - kafka: - image: wurstmeister/kafka - ports: - - '9092:9092' - env: - KAFKA_ADVERTISED_HOST_NAME: localhost - KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 env: REDIS_URL: 'redis://localhost' CLICKHOUSE_HOST: 'localhost' CLICKHOUSE_DATABASE: 'posthog_test' KAFKA_ENABLED: 'true' - KAFKA_HOSTS: 'localhost:9092' + KAFKA_HOSTS: 'kafka:9092' steps: - name: Check out Django server for database setup @@ -158,6 +142,17 @@ jobs: with: path: 'plugin-server' + - name: Fix Kafka Hostname + run: | + sudo bash -c 'echo "127.0.0.1 kafka zookeeper" >> /etc/hosts' + ping -c 1 kafka + ping -c 1 zookeeper + + - name: Start Kafka, Clickhouse, Zookeeper + run: | + cd posthog/ee + docker-compose -f docker-compose.ch.yml up -d zookeeper kafka clickhouse + - name: Set up Python uses: actions/setup-python@v2 with: diff --git a/package.json b/package.json index fbd50b6e..d314c9cf 100644 --- a/package.json +++ b/package.json @@ -12,7 +12,7 @@ "start": "yarn start:dev", "start:dist": "node dist/src/index.js --base-dir ../posthog", "start:dev": "ts-node-dev --exit-child src/index.ts --base-dir ../posthog", - "start:dev:ee": "DATABASE_URL=postgres://posthog:posthog@localhost:5439/posthog KAFKA_ENABLED=true KAFKA_HOSTS=localhost:9092 yarn start:dev", + "start:dev:ee": "KAFKA_ENABLED=true KAFKA_HOSTS=localhost:9092 yarn start:dev", "build": "yarn clean && yarn compile", "clean": "rimraf dist/*", "compile:protobuf": "cd src/idl/ && rimraf protos.* && pbjs -t static-module -w commonjs -o protos.js *.proto && pbts -o protos.d.ts protos.js && eslint --fix . && prettier --write .", @@ -24,8 +24,10 @@ "prettier:check": "prettier --check .", "prepare": "yarn compile:protobuf", "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" + "setup:dev:clickhouse": "cd ../posthog && export DEBUG=1 PRIMARY_DB=clickhouse && source env/bin/activate && python manage.py migrate_clickhouse", + "setup:test:ee": "yarn setup:test:postgres && yarn setup:test:clickhouse", + "setup:test:postgres": "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:test:clickhouse": "cd ../posthog && export TEST=1 PRIMARY_DB=clickhouse CLICKHOUSE_DATABASE=posthog_test && source env/bin/activate && python manage.py migrate_clickhouse" }, "bin": { "posthog-plugin-server": "bin/posthog-plugin-server" diff --git a/src/celery/worker.ts b/src/celery/worker.ts index 19d96f7f..107dad04 100644 --- a/src/celery/worker.ts +++ b/src/celery/worker.ts @@ -171,14 +171,14 @@ export class Worker extends Base implements Queue { throw new Error(`Missing process handler for task ${taskName}`) } - console.info( + console.debug( `celery.node Received task: ${taskName}[${taskId}], args: ${args}, kwargs: ${JSON.stringify(kwargs)}` ) const timeStart = process.hrtime() const taskPromise = handler(...args, kwargs).then((result) => { const diff = process.hrtime(timeStart) - console.info( + console.debug( `celery.node Task ${taskName}[${taskId}] succeeded in ${diff[0] + diff[1] / 1e9}s: ${result}` ) this.activeTasks.delete(taskPromise) diff --git a/src/config.ts b/src/config.ts index d3e08dee..a1758dc4 100644 --- a/src/config.ts +++ b/src/config.ts @@ -10,7 +10,7 @@ export function getDefaultConfig(): PluginsServerConfig { CELERY_DEFAULT_QUEUE: 'celery', DATABASE_URL: isTestEnv ? 'postgres://localhost:5432/test_posthog' : 'postgres://localhost:5432/posthog', CLICKHOUSE_HOST: 'localhost', - CLICKHOUSE_DATABASE: 'default', + CLICKHOUSE_DATABASE: isTestEnv ? 'posthog_test' : 'default', CLICKHOUSE_USERNAME: 'default', CLICKHOUSE_PASSWORD: null, CLICKHOUSE_CA: null, diff --git a/src/db.ts b/src/db.ts index 7b4f430c..98f635dc 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,20 +1,24 @@ import { Properties } from '@posthog/plugin-scaffold' -import { ClickHouse } from 'clickhouse' +import { ClickHouse, QueryCursor } from 'clickhouse' import { Producer } from 'kafkajs' import { DateTime } from 'luxon' import { Pool, QueryConfig, QueryResult, QueryResultRow } from 'pg' +import { string } from 'yargs' import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' -import { unparsePersonPartial } from './ingestion/utils' +import { chainToElements, hashElements, unparsePersonPartial } from './ingestion/utils' import { Person, PersonDistinctId, RawPerson, RawOrganization, - Team, PostgresSessionRecordingEvent, Event, + ClickHouseEvent, + Element, + SessionRecordingEvent, + ElementGroup, } from './types' -import { castTimestampOrNow, sanitizeSqlIdentifier, UUIDT } from './utils' +import { castTimestampOrNow, clickHouseTimestampToISO, sanitizeSqlIdentifier } from './utils' /** The recommended way of accessing the database. */ export class DB { @@ -31,6 +35,8 @@ export class DB { this.clickhouse = clickhouse } + // Direct queries + public async postgresQuery( queryTextOrConfig: string | QueryConfig, values?: I @@ -38,6 +44,13 @@ export class DB { return this.postgres.query(queryTextOrConfig, values) } + public async clickhouseQuery(query: string, reqParams?: Record): Promise> { + if (!this.clickhouse) { + throw new Error('ClickHouse connection has not been provided to this DB instance!') + } + return this.clickhouse.query(query, reqParams).toPromise() + } + // Person public async fetchPersons(): Promise { @@ -106,7 +119,7 @@ export class DB { 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) + (field, index) => `"${sanitizeSqlIdentifier(field)}" = $${index + 1}` )} WHERE id = $${Object.values(update).length + 1}`, values ) @@ -130,10 +143,8 @@ export class DB { 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() - await this.clickhouse - .query(`ALTER TABLE person_distinct_id DELETE WHERE person_id = ${personId}`) - .toPromise() + await this.clickhouseQuery(`ALTER TABLE person DELETE WHERE id = ${personId}`) + await this.clickhouseQuery(`ALTER TABLE person_distinct_id DELETE WHERE person_id = ${personId}`) } } @@ -168,7 +179,7 @@ export class DB { const updatedPersonDistinctId: PersonDistinctId = { ...personDistinctId, ...update } await this.postgresQuery( `UPDATE posthog_persondistinctid SET ${Object.keys(update).map( - (field, index) => sanitizeSqlIdentifier(field) + ' = $' + (index + 1) + (field, index) => `"${sanitizeSqlIdentifier(field)}" = $${index + 1}` )} WHERE id = $${Object.values(update).length + 1}`, [...Object.values(update), personDistinctId.id] ) @@ -192,21 +203,96 @@ export class DB { // Event - public async fetchEvents(): Promise { - const result = await this.postgresQuery('SELECT * FROM posthog_event') - return result.rows as Event[] + public async fetchEvents(): Promise { + if (this.kafkaProducer) { + const events = (await this.clickhouseQuery(`SELECT * FROM events`)) as ClickHouseEvent[] + return ( + events?.map( + (event) => + ({ + ...event, + ...(typeof event['properties'] === 'string' + ? { properties: JSON.parse(event.properties) } + : {}), + timestamp: clickHouseTimestampToISO(event.timestamp), + } as ClickHouseEvent) + ) || [] + ) + } else { + 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[] + public async fetchSessionRecordingEvents(): Promise { + if (this.kafkaProducer) { + const events = ((await this.clickhouseQuery( + `SELECT * FROM session_recording_events` + )) as SessionRecordingEvent[]).map((event) => { + return { + ...event, + snapshot_data: event.snapshot_data ? JSON.parse(event.snapshot_data) : null, + } + }) + return events + } else { + 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 + public async fetchElements(event?: Event): Promise { + if (this.kafkaProducer) { + const events = (await this.clickhouseQuery( + `SELECT elements_chain FROM events WHERE uuid='${sanitizeSqlIdentifier((event as any).uuid)}'` + )) as ClickHouseEvent[] + const chain = events?.[0]?.elements_chain + return chainToElements(chain) + } else { + return (await this.postgresQuery('SELECT * FROM posthog_element')).rows + } + } + + public 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.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.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 } } diff --git a/src/extensions/posthog.ts b/src/extensions/posthog.ts index 51a09b04..94af08ca 100644 --- a/src/extensions/posthog.ts +++ b/src/extensions/posthog.ts @@ -26,10 +26,11 @@ export function createPosthog(server: PluginsServer, pluginConfig: PluginConfig) if (server.KAFKA_ENABLED) { // Sending event to our Kafka>ClickHouse pipeline - const producer = server.kafka!.producer() sendEvent = async (data) => { - await producer.connect() - producer!.send({ + if (!server.kafkaProducer) { + throw new Error('kafkaProducer not configured!') + } + server.kafkaProducer.send({ topic: KAFKA_EVENTS_INGESTION_HANDOFF, messages: [ { diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index e5d68be3..26d1cc69 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -47,11 +47,16 @@ export class KafkaQueue implements Queue { ...rawEvent, data: JSON.parse(rawEvent.data), })) - const pluginEvents: PluginEvent[] = parsedEvents.map((parsedEvent) => ({ - ...parsedEvent, - event: parsedEvent.data.event, - properties: parsedEvent.data.properties, - })) + const pluginEvents: PluginEvent[] = rawEvents.map((rawEvent) => { + const { data: dataStr, ...restOfRawEvent } = rawEvent + const event = { ...restOfRawEvent, ...JSON.parse(dataStr) } + return { + ...event, + kafka_offset: restOfRawEvent.kafka_offset, + site_url: event.site_url || null, + ip: event.ip || null, + } + }) const processedEvents: PluginEvent[] = ( await this.processEventBatch(pluginEvents) ).filter((event: PluginEvent[] | false | null | undefined) => Boolean(event)) diff --git a/src/ingestion/process-event.ts b/src/ingestion/process-event.ts index 73a8e0db..304fd9d2 100644 --- a/src/ingestion/process-event.ts +++ b/src/ingestion/process-event.ts @@ -3,7 +3,6 @@ import { DateTime, Duration } from 'luxon' import { CohortPeople, Element, - ElementGroup, Person, PersonDistinctId, PluginsServer, @@ -12,11 +11,11 @@ import { Team, TimestampFormat, } from '../types' -import { castTimestampOrNow, UUIDT } from '../utils' +import { castTimestampOrNow, UUID, 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 { elementsToString, hashElements, sanitizeEventName } from './utils' +import { elementsToString, sanitizeEventName } from './utils' import { ClickHouse } from 'clickhouse' import { DB } from '../db' import { status } from '../status' @@ -54,6 +53,9 @@ export class EventsProcessor { sentAt: DateTime | null, eventUuid: string ): Promise { + if (!UUID.validateString(eventUuid, false)) { + throw new Error(`Not a valid UUID: "${eventUuid}"`) + } const singleSaveTimer = new Date() const properties: Properties = data.properties ?? {} @@ -462,7 +464,7 @@ export class EventsProcessor { } else { let elementsHash = '' if (elements && elements.length > 0) { - elementsHash = await this.createElementGroup(elements, team.id) + elementsHash = await this.db.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 *', @@ -495,45 +497,6 @@ 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, @@ -542,7 +505,10 @@ export class EventsProcessor { timestamp: DateTime | string, snapshot_data: Record ): Promise { - const timestampString = castTimestampOrNow(timestamp) + const timestampString = castTimestampOrNow( + timestamp, + this.kafkaProducer ? TimestampFormat.ClickHouse : TimestampFormat.ISO + ) const data: SessionRecordingEvent = { uuid, diff --git a/src/ingestion/utils.ts b/src/ingestion/utils.ts index fdab780b..5b4cbb2a 100644 --- a/src/ingestion/utils.ts +++ b/src/ingestion/utils.ts @@ -50,7 +50,7 @@ export function elementsToString(elements: Element[]): string { ) el_string += ':' el_string += Object.entries(attributes) - .map(([key, value]) => `${key}=${value}`) + .map(([key, value]) => `${key}="${value}"`) .join('') return el_string }) @@ -109,3 +109,61 @@ export function hashElements(elements: Element[]): string { return crypto.createHash('md5').update(serializedString).digest('hex') } + +export function chainToElements(chain: string): Element[] { + const elements: Element[] = [] + + // Below splits all elements by ;, while ignoring escaped quotes and semicolons within quotes + const splitChainRegex = /(?:[^\s;"]|"(?:\\.|[^"])*")+/g + + // Below splits the tag/classes from attributes + // Needs a regex because classes can have : too + const splitClassAttributes = /(.*?)($|:([a-zA-Z\-_0-9]*=.*))/g + const parseAttributesRegex = /((.*?)="(.*?[^\\])")/gm + + Array.from(chain.matchAll(splitChainRegex)) + .map((r) => r[0]) + .forEach((elString, index) => { + const elStringSplit = Array.from(elString.matchAll(splitClassAttributes))[0] + const attributes = + elStringSplit.length > 3 + ? Array.from(elStringSplit[3].matchAll(parseAttributesRegex)).map((a) => [a[2], a[3]]) + : [] + + const element: Element = { + attributes: {}, + order: index, + } + + if (elStringSplit[1]) { + const tagAndClass = elStringSplit[1].split('.') + element.tag_name = tagAndClass[0] + if (tagAndClass.length > 1) { + const [_, ...rest] = tagAndClass + element.attr_class = rest.filter((t) => t) + } + } + + for (const [key, value] of attributes) { + if (key == 'href') { + element.href = value + } else if (key == 'nth-child') { + element.nth_child = parseInt(value) + } else if (key == 'nth-of-type') { + element.nth_of_type = parseInt(value) + } else if (key == 'text') { + element.text = value + } else if (key == 'attr_id') { + element.attr_id = value + } else if (key) { + if (!element.attributes) { + element.attributes = {} + } + element.attributes[key] = value + } + } + elements.push(element) + }) + + return elements +} diff --git a/src/server.ts b/src/server.ts index 8deedd40..4c63b893 100644 --- a/src/server.ts +++ b/src/server.ts @@ -83,10 +83,11 @@ export async function createServer( kafka = new Kafka({ clientId: `plugin-server-v${version}-${new UUIDT()}`, brokers: serverConfig.KAFKA_HOSTS.split(','), - logLevel: logLevel.NOTHING, + logLevel: logLevel.WARN, ssl: kafkaSsl, }) kafkaProducer = kafka.producer() + await kafkaProducer?.connect() } // `node-postgres` will return dates as plain JS Date objects, which will use the local timezone. @@ -149,6 +150,7 @@ export async function createServer( server.eventsProcessor = new EventsProcessor(server as PluginsServer) const closeServer = async () => { + await kafkaProducer?.disconnect() await server.redis.quit() await server.postgres.end() } @@ -177,7 +179,6 @@ export async function startPluginsServer( let pingJob: schedule.Job | undefined let statsJob: schedule.Job | undefined let piscina: Piscina | undefined - let kafkaProducer: Producer | undefined let queue: Queue | undefined let closeServer: () => Promise | undefined let stopSchedule: () => Promise | undefined @@ -200,7 +201,6 @@ export async function startPluginsServer( await stopFastifyInstance(fastifyInstance!) } await queue?.stop() - await kafkaProducer?.disconnect() await pubSub?.quit() pingJob && schedule.cancelJob(pingJob) statsJob && schedule.cancelJob(statsJob) @@ -225,8 +225,6 @@ export async function startPluginsServer( } ;[server, closeServer] = await createServer(serverConfig, null) - await server.kafkaProducer?.connect() - piscina = makePiscina(serverConfig) const processEvent = (event: PluginEvent) => { if ((piscina?.queueSize || 0) > (server?.WORKER_CONCURRENCY || 4) * (server?.WORKER_CONCURRENCY || 4)) { diff --git a/src/types.ts b/src/types.ts index 5aeb9416..66453552 100644 --- a/src/types.ts +++ b/src/types.ts @@ -189,6 +189,8 @@ export interface RawEventMessage extends BaseEventMessage { now: string /** ISO-formatted datetime. May be empty! */ sent_at: string + /** JSON-encoded number. */ + kafka_offset: string } /** Usable event message. */ @@ -237,7 +239,7 @@ export interface Element { attr_class?: string[] nth_child?: number nth_of_type?: number - attributes: Record + attributes?: Record event_id?: number order?: number group_id?: number @@ -262,6 +264,11 @@ export interface Event { created_at: string } +export interface ClickHouseEvent extends Omit { + uuid: string + elements_chain: string +} + /** Properties shared by RawPerson and Person. */ export interface BasePerson { id: number diff --git a/src/utils.ts b/src/utils.ts index 8a619069..eccf8e5e 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -147,12 +147,24 @@ for (let i = 0; i < 256; i++) { } export class UUID { - static validateString(candidate: string): void { - if (!candidate.match(/^[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}$/i)) { + /** + * Check whether str + * + * This does not care about RFC4122, since neither does UUIDT above. + * https://stackoverflow.com/questions/7905929/how-to-test-valid-uuid-guid + */ + static validateString(candidate: any, throwOnInvalid = true): boolean { + const isValid = Boolean( + candidate && + typeof candidate === 'string' && + candidate.match(/^[0-9A-F]{8}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{4}-[0-9A-F]{12}$/i) + ) + if (!isValid && throwOnInvalid) { throw new Error( 'String does not match format XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX (where each X is a hexadecimal character)!' ) } + return isValid } array: Uint8Array @@ -299,6 +311,10 @@ export function castTimestampOrNow( } } +export function clickHouseTimestampToISO(timestamp: string): string { + return DateTime.fromFormat(timestamp, 'yyyy-MM-dd HH:mm:ss.u', { zone: 'UTC' }).toISO() +} + export function delay(ms: number): Promise { return new Promise((resolve) => { setTimeout(resolve, ms) @@ -307,5 +323,5 @@ export function delay(ms: number): Promise { /** Remove all quotes from the provided identifier to prevent SQL injection. */ export function sanitizeSqlIdentifier(unquotedIdentifier: string): string { - return '"' + unquotedIdentifier.replace(/[^\w\d_]+/g, '') + '"' + return unquotedIdentifier.replace(/[^\w\d_]+/g, '') } diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts new file mode 100644 index 00000000..1bd7e7fc --- /dev/null +++ b/tests/clickhouse/e2e.test.ts @@ -0,0 +1,61 @@ +import { LogLevel, PluginsServerConfig } 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, UUIDT } from '../../src/utils' +import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' +import { resetKafka } from '../helpers/kafka' +import { delayUntilEventIngested } from '../shared/process-event' + +jest.setTimeout(60000) // 60 sec timeout + +const extraServerConfig: Partial = { + KAFKA_ENABLED: true, + KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', + WORKER_CONCURRENCY: 2, + PLUGIN_SERVER_INGESTION: true, + LOG_LEVEL: LogLevel.Log, +} + +describe('e2e clickhouse ingestion', () => { + let server: PluginsServer + let stopServer: () => Promise + let posthog: DummyPostHog + + beforeAll(async () => { + await resetKafka(extraServerConfig) + }) + + beforeEach(async () => { + await resetTestDatabase(` + async function processEvent (event) { + event.properties.processed = 'hell yes' + event.properties.upperUuid = event.properties.uuid?.toUpperCase() + return event + } + `) + await resetTestDatabaseClickhouse(extraServerConfig) + const startResponse = await startPluginsServer(extraServerConfig, makePiscina) + server = startResponse.server + stopServer = startResponse.stop + posthog = createPosthog(server, pluginConfig39) + }) + + afterEach(async () => { + await stopServer() + }) + + test('event captured, processed, ingested', async () => { + expect((await server.db.fetchEvents()).length).toBe(0) + const uuid = new UUIDT().toString() + posthog.capture('custom event', { name: 'haha', uuid }) + await delayUntilEventIngested(() => server.db.fetchEvents()) + const events = await server.db.fetchEvents() + expect(events.length).toBe(1) + expect(events[0].properties.processed).toEqual('hell yes') + expect(events[0].properties.upperUuid).toEqual(uuid.toUpperCase()) + }) +}) diff --git a/tests/clickhouse/ingestion-utils.test.ts b/tests/clickhouse/ingestion-utils.test.ts new file mode 100644 index 00000000..ee2ea40f --- /dev/null +++ b/tests/clickhouse/ingestion-utils.test.ts @@ -0,0 +1,50 @@ +import { chainToElements, elementsToString } from '../../src/ingestion/utils' + +test('elementsToString and chainToElements', async () => { + const elementsString = elementsToString([ + { + tag_name: 'a', + href: '/a-url', + attr_class: ['small'], + text: 'bla bla', + attributes: { + prop: 'value', + number: 33, + 'data-attr': 'something " that; could mess up', + style: 'min-height: 100vh;', + }, + nth_child: 1, + nth_of_type: 0, + }, + { tag_name: 'button', attr_class: ['btn', 'btn-primary'], nth_child: 0, nth_of_type: 0 }, + { tag_name: 'div', nth_child: 0, nth_of_type: 0 }, + { tag_name: 'div', nth_child: 0, nth_of_type: 0, attr_id: 'nested' }, + ]) + + expect(elementsString).toEqual( + [ + 'a.small:data-attr="something \\" that; could mess up"href="/a-url"nth-child="1"nth-of-type="0"number="33"prop="value"style="min-height: 100vh;"text="bla bla"', + 'button.btn.btn-primary:nth-child="0"nth-of-type="0"', + 'div:nth-child="0"nth-of-type="0"', + 'div:attr_id="nested"nth-child="0"nth-of-type="0"', + ].join(';') + ) + + const elements = chainToElements(elementsString) + expect(elements.length).toBe(4) + expect(elements[0].tag_name).toEqual('a') + expect(elements[0].href).toEqual('/a-url') + expect(elements[0].attr_class).toEqual(['small']) + expect(elements[0].attributes).toEqual({ + prop: 'value', + number: '33', + // NB! The original Python code also does not unescape `\"` -> `"` + // Could be fixed later, but keeping as is for parity. + 'data-attr': 'something \\" that; could mess up', + style: 'min-height: 100vh;', + }) + expect(elements[0].nth_child).toEqual(1) + expect(elements[0].nth_of_type).toEqual(0) + expect(elements[1].attr_class).toEqual(['btn', 'btn-primary']) + expect(elements[3].attr_id).toEqual('nested') +}) diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index 96214920..7604e29d 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -1,66 +1,23 @@ -import { PluginsServerConfig } from '../../src/types' +import { PluginsServerConfig, Event } from '../../src/types' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' -import { KafkaCollector, KafkaObserver } from '../helpers/kafka' -import { UUIDT } from '../../src/utils' -import { DateTime } from 'luxon' +import { resetKafka } from '../helpers/kafka' import { createProcessEventTests } from '../shared/process-event' jest.setTimeout(180_000) // 3 minute timeout const extraServerConfig: Partial = { KAFKA_ENABLED: true, - KAFKA_HOSTS: 'kafka:9092', - DATABASE_URL: 'postgres://posthog:posthog@localhost:5439/test_posthog', + KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', } describe('process event (clickhouse)', () => { - const kafkaObserver = new KafkaObserver(extraServerConfig) + beforeAll(async () => { + await resetKafka(extraServerConfig) + }) 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, - uuid, - 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) - }) + createProcessEventTests('clickhouse', extraServerConfig) }) diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 15613922..5195e85c 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -1,9 +1,13 @@ import { EventEmitter } from 'events' import { Kafka, Consumer, logLevel, EachMessagePayload, Producer } from 'kafkajs' -import { KAFKA_EVENTS, KAFKA_EVENTS_INGESTION_HANDOFF } from '../../src/ingestion/topics' +import { + KAFKA_EVENTS, + KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_SESSION_RECORDING_EVENTS, +} from '../../src/ingestion/topics' import { parseRawEventMessage } from '../../src/ingestion/utils' import { EventMessage, PluginsServerConfig } from '../../src/types' -import { UUIDT } from '../../src/utils' +import { delay, UUIDT } from '../../src/utils' import { defaultConfig, overrideWithEnv } from '../../src/config' export class KafkaObserver extends EventEmitter { @@ -19,7 +23,7 @@ export class KafkaObserver extends EventEmitter { this.kafka = new Kafka({ clientId: `plugin-server-test-${new UUIDT()}`, brokers: (config.KAFKA_HOSTS || '').split(','), - logLevel: logLevel.NOTHING, + logLevel: logLevel.WARN, }) this.producer = this.kafka.producer() this.consumer = this.kafka.consumer({ @@ -107,3 +111,76 @@ export class KafkaCollector extends EventEmitter { }) } } + +/** Clear the kafka queue */ +export async function resetKafka(extraServerConfig: Partial, delayMs = 2000) { + console.log('Resetting Kafka!') + const config = { ...overrideWithEnv(defaultConfig, process.env), ...extraServerConfig } + const kafka = new Kafka({ + clientId: `plugin-server-test-${new UUIDT()}`, + brokers: (config.KAFKA_HOSTS || '').split(','), + logLevel: logLevel.WARN, + }) + const producer = kafka.producer() + const consumer = kafka.consumer({ + groupId: 'clickhouse-ingestion-test', + }) + const messages = [] + + async function createTopic(topic: string) { + try { + const admin = kafka.admin() + await admin.connect() + await admin.createTopics({ + waitForLeaders: true, + topics: [{ topic }], + }) + await admin.disconnect() + } catch (e) { + if (!e?.error?.includes('Topic with this name already exists')) { + console.error(`Error creating kafka topic "${topic}". This might be fine!`) + console.error(e) + } + } + } + + await createTopic(KAFKA_EVENTS_INGESTION_HANDOFF) + await createTopic(KAFKA_SESSION_RECORDING_EVENTS) + + const connected = await new Promise(async (resolve, reject) => { + console.info('setting group join and crash listeners') + const { CONNECT, GROUP_JOIN, CRASH } = consumer.events + consumer.on(CONNECT, () => { + console.log('consumer connected to kafka') + }) + consumer.on(GROUP_JOIN, () => { + console.log('joined group') + resolve() + }) + consumer.on(CRASH, ({ payload: { error } }) => reject(error)) + console.info('connecting producer') + await producer.connect() + console.info('subscribing consumer') + + await consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) + console.info('running consumer') + await consumer.run({ + eachMessage: async (payload) => { + console.info('message received!') + messages.push(payload) + }, + }) + }) + + console.info(`awaiting ${delayMs} ms before disconnecting`) + await delay(delayMs) + + console.info('disconnecting producer') + await producer.disconnect() + console.info('stopping consumer') + await consumer.stop() + console.info('disconnecting consumer') + await consumer.disconnect() + + return true +} diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 92e39a95..4c1aef36 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -81,6 +81,7 @@ export async function createUserTeamAndOrganization( created_at: new Date().toISOString(), updated_at: new Date().toISOString(), personalization: '{}', + setup_section_2_completed: true, }) await insertRow(db, 'posthog_organizationmembership', { id: organizationMembershipId, diff --git a/tests/postgres/e2e.test.ts b/tests/postgres/e2e.test.ts index 208eef5a..dc3a2b18 100644 --- a/tests/postgres/e2e.test.ts +++ b/tests/postgres/e2e.test.ts @@ -5,7 +5,8 @@ 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' +import { delay, UUIDT } from '../../src/utils' +import { delayUntilEventIngested } from '../shared/process-event' jest.setTimeout(60000) // 60 sec timeout @@ -18,6 +19,7 @@ describe('e2e postgres ingestion', () => { await resetTestDatabase(` async function processEvent (event) { event.properties.processed = 'hell yes' + event.properties.upperUuid = event.properties.uuid?.toUpperCase() return event } `) @@ -47,10 +49,12 @@ describe('e2e postgres ingestion', () => { test('event captured, processed, ingested', async () => { expect((await server.db.fetchEvents()).length).toBe(0) - posthog.capture('custom event', { name: 'haha' }) - await delay(2000) + const uuid = new UUIDT().toString() + posthog.capture('custom event', { name: 'haha', uuid, randomProperty: 'lololo' }) + await delayUntilEventIngested(() => server.db.fetchEvents()) const events = await server.db.fetchEvents() expect(events.length).toBe(1) expect(events[0].properties.processed).toEqual('hell yes') + expect(events[0].properties.upperUuid).toEqual(uuid.toUpperCase()) }) }) diff --git a/tests/postgres/process-event.test.ts b/tests/postgres/process-event.test.ts index d404d826..2c12142b 100644 --- a/tests/postgres/process-event.test.ts +++ b/tests/postgres/process-event.test.ts @@ -1,14 +1,54 @@ -import { PluginsServer, Event, Person, PersonDistinctId, Element, PostgresSessionRecordingEvent } from '../../src/types' import { createProcessEventTests } from '../shared/process-event' +import { createUserTeamAndOrganization } from '../helpers/sql' +import { Team } from '../../src/types' 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(), + createProcessEventTests('postgresql', {}, (response) => { + test('element group', async () => { + const { server } = response + const elements = [{ tag_name: 'button', text: 'Sign up!' }, { tag_name: 'div' }] + + const elementsHash = server!.db.createElementGroup(elements, 2) + const elementGroup = await server!.db.fetchElements() + + expect(elementGroup[0].tag_name).toEqual('button') + expect(elementGroup[1].tag_name).toEqual('div') + expect(elementGroup.length).toEqual(2) + + const elements2 = [ + { tag_name: 'button', text: 'Sign up!' }, + // make sure we remove events if we can + { tag_name: 'div', event: { id: 'blabla' } }, + ] + + const elementsHash2 = server!.db.createElementGroup(elements2, 2) + const elementGroup2 = await server!.db.fetchElements() + // we are fetching all the elements, so expect there to be no new ones + expect(elementGroup2.length).toEqual(2) + expect(elementsHash).toEqual(elementsHash2) + + await createUserTeamAndOrganization( + server!.postgres, + 3, + 1002, + '01774e2f-0d01-0000-ee94-9a238640c6ee', + '0174f81e-36f5-0000-7ef8-cc26c1fbab1c' + ) + + const teams = (await server!.db.postgresQuery('SELECT * FROM posthog_team ORDER BY id')).rows as Team[] + + // # Test no team leakage + const team2 = teams[1] + + const elementsHash3 = server!.db.createElementGroup(elements2, 3) + const elementGroup3 = await server!.db.fetchElements() + console.log(elementGroup3) + // created new elements as it's different team even if the hash is the same + expect(elementGroup3.length).toEqual(4) + expect(elementsHash).toEqual(elementsHash2) + expect(elementsHash).toEqual(elementsHash3) + }) }) }) diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index 1417dc52..b9a4c058 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -9,11 +9,15 @@ import { Element, PostgresSessionRecordingEvent, PluginsServerConfig, + ClickHouseEvent, + SessionRecordingEvent, } 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' +import { delay, UUIDT } from '../../src/utils' +import { IEvent } from '../../src/idl/protos' +import { hashElements } from '../../src/ingestion/utils' jest.setTimeout(600000) // 600 sec timeout @@ -25,6 +29,15 @@ async function getFirstTeam(server: PluginsServer): Promise { return (await getTeams(server))[0] } +export async function delayUntilEventIngested(fetchEvents: () => Promise, minCount = 1): Promise { + for (let i = 0; i < 30; i++) { + if ((await fetchEvents()).length >= minCount) { + return + } + await delay(500) + } +} + async function createPerson( server: PluginsServer, team: Team, @@ -34,29 +47,21 @@ async function createPerson( return server.db.createPerson(DateTime.utc(), properties, team.id, null, false, new UUIDT().toString(), distinctIds) } +type ReturnWithServer = { server?: PluginsServer; stopServer?: () => Promise } + 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 => { + extraServerConfig?: Partial, + createTests?: (response: ReturnWithServer) => void +): ReturnWithServer => { let queryCounter = 0 + let processEventCounter = 0 let team: Team let server: PluginsServer let stopServer: () => Promise let eventsProcessor: EventsProcessor let now = DateTime.utc() + const returned: ReturnWithServer = {} async function getServer(): Promise<[PluginsServer, () => Promise]> { const [server, stopServer] = await createServer({ @@ -78,6 +83,32 @@ export const createProcessEventTests = ( return [server, stopServer] } + async function processEvent( + distinctId: string, + ip: string, + siteUrl: string, + data: PluginEvent, + teamId: number, + now: DateTime, + sentAt: DateTime | null, + eventUuid: string + ): Promise { + const response = await eventsProcessor.processEvent( + distinctId, + ip, + siteUrl, + data, + teamId, + now, + sentAt, + eventUuid + ) + if (database === 'clickhouse') { + await delayUntilEventIngested(() => server.db.fetchEvents(), ++processEventCounter) + } + return response + } + beforeEach(async () => { const testCode = ` function processEvent (event, meta) { @@ -87,8 +118,11 @@ export const createProcessEventTests = ( ` await resetTestDatabase(testCode, extraServerConfig) ;[server, stopServer] = await getServer() + returned.server = server + returned.stopServer = stopServer eventsProcessor = new EventsProcessor(server) queryCounter = 0 + processEventCounter = 0 team = await getFirstTeam(server) now = DateTime.utc() }) @@ -97,13 +131,15 @@ export const createProcessEventTests = ( await stopServer?.() }) + createTests?.(returned) + 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( + await processEvent( '2', '', '', @@ -124,18 +160,14 @@ export const createProcessEventTests = ( 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): + if (database === 'clickhouse') { + expect(queryCounter).toBe(8) + } else if (database === 'postgresql') { + expect(queryCounter).toBe(12) + } // capture a second time to verify e.g. event_names is not ['$autocapture', '$autocapture'] - await eventsProcessor.processEvent( + await processEvent( '2', '', '', @@ -156,26 +188,31 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const events = await getEvents(server) - const persons = await getPersons(server) + const events = await server.db.fetchEvents() + const persons = await server.db.fetchPersons() expect(events.length).toEqual(2) expect(persons.length).toEqual(1) - const [event] = events const [person] = persons - const distinctIds = await getDistinctIds(server, person) + const distinctIds = await server.db.fetchDistinctIdValues(person) + const [event] = events as Event[] 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) + const elements = await server.db.fetchElements(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('💻') + if (database === 'clickhouse') { + expect(hashElements(elements)).toEqual('0679137c0cd2408a2906839143e7a71f') + } else if (database === 'postgresql') { + expect(event.elements_hash).toEqual('0679137c0cd2408a2906839143e7a71f') + } + team = await getFirstTeam(server) expect(team.event_names).toEqual(['$autocapture']) expect(team.event_names_with_usage).toEqual([{ event: '$autocapture', volume: null, usage_count: null }]) @@ -190,7 +227,7 @@ export const createProcessEventTests = ( test('capture no element', async () => { await createPerson(server, team, ['asdfasdfasdf']) - await eventsProcessor.processEvent( + await processEvent( 'asdfasdfasdf', '', '', @@ -204,8 +241,8 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual(['asdfasdfasdf']) - const [event] = await getEvents(server) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual(['asdfasdfasdf']) + const [event] = await server.db.fetchEvents() expect(event.event).toBe('$pageview') }) @@ -216,7 +253,7 @@ export const createProcessEventTests = ( const tomorrow = rightNow.plus({ days: 1, hours: 2 }) const tomorrowSentAt = rightNow.plus({ days: 1, hours: 2, minutes: 10 }) - await eventsProcessor.processEvent( + await processEvent( 'movie played', '', '', @@ -231,7 +268,7 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() const eventSecondsBeforeNow = rightNow.diff(DateTime.fromISO(event.timestamp), 'seconds').seconds expect(eventSecondsBeforeNow).toBeGreaterThan(590) @@ -249,7 +286,7 @@ export const createProcessEventTests = ( // tomorrow = tomorrow.replace(tzinfo=None) // tomorrow_sent_at = tomorrow_sent_at.replace(tzinfo=None) - await eventsProcessor.processEvent( + await processEvent( 'movie played', '', '', @@ -264,7 +301,7 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() const eventSecondsBeforeNow = rightNow.diff(DateTime.fromISO(event.timestamp), 'seconds').seconds expect(eventSecondsBeforeNow).toBeGreaterThan(590) @@ -277,7 +314,7 @@ export const createProcessEventTests = ( const rightNow = DateTime.utc() const tomorrow = rightNow.plus({ days: 1, hours: 2 }) - await eventsProcessor.processEvent( + await processEvent( 'movie played', '', '', @@ -292,7 +329,7 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() const difference = tomorrow.diff(DateTime.fromISO(event.timestamp), 'seconds').seconds expect(difference).toBeLessThan(1) }) @@ -300,7 +337,7 @@ export const createProcessEventTests = ( test('ip capture', async () => { await createPerson(server, team, ['asdfasdfasdf']) - await eventsProcessor.processEvent( + await processEvent( 'asdfasdfasdf', '11.12.13.14', '', @@ -313,14 +350,14 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() expect(event.properties['$ip']).toBe('11.12.13.14') }) test('ip override', async () => { await createPerson(server, team, ['asdfasdfasdf']) - await eventsProcessor.processEvent( + await processEvent( 'asdfasdfasdf', '11.12.13.14', '', @@ -333,7 +370,8 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - const [event] = await getEvents(server) + + const [event] = await server.db.fetchEvents() expect(event.properties['$ip']).toBe('1.0.0.1') }) @@ -341,7 +379,7 @@ export const createProcessEventTests = ( await server.db.postgresQuery('update posthog_team set anonymize_ips = $1', [true]) await createPerson(server, team, ['asdfasdfasdf']) - await eventsProcessor.processEvent( + await processEvent( 'asdfasdfasdf', '11.12.13.14', '', @@ -354,14 +392,15 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - const [event] = await getEvents(server) + + const [event] = await server.db.fetchEvents() expect(event.properties['$ip']).not.toBeDefined() }) test('alias', async () => { await createPerson(server, team, ['old_distinct_id']) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -375,8 +414,8 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) - expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + expect((await server.db.fetchEvents()).length).toBe(1) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual([ 'old_distinct_id', 'new_distinct_id', ]) @@ -385,7 +424,7 @@ export const createProcessEventTests = ( test('alias reverse', async () => { await createPerson(server, team, ['old_distinct_id']) - await eventsProcessor.processEvent( + await processEvent( 'old_distinct_id', '', '', @@ -399,8 +438,8 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) - expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + expect((await server.db.fetchEvents()).length).toBe(1) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual([ 'old_distinct_id', 'new_distinct_id', ]) @@ -409,7 +448,7 @@ export const createProcessEventTests = ( test('alias twice', async () => { await createPerson(server, team, ['old_distinct_id']) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -423,17 +462,17 @@ export const createProcessEventTests = ( 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([ + expect((await server.db.fetchPersons()).length).toBe(1) + expect((await server.db.fetchEvents()).length).toBe(1) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual([ 'old_distinct_id', 'new_distinct_id', ]) await createPerson(server, team, ['old_distinct_id_2']) - expect((await getPersons(server)).length).toBe(2) + expect((await server.db.fetchPersons()).length).toBe(2) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -446,9 +485,9 @@ export const createProcessEventTests = ( 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([ + expect((await server.db.fetchEvents()).length).toBe(2) + expect((await server.db.fetchPersons()).length).toBe(1) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual([ 'old_distinct_id', 'new_distinct_id', 'old_distinct_id_2', @@ -456,7 +495,7 @@ export const createProcessEventTests = ( }) test('alias before person', async () => { - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -470,9 +509,9 @@ export const createProcessEventTests = ( 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([ + expect((await server.db.fetchEvents()).length).toBe(1) + expect((await server.db.fetchPersons()).length).toBe(1) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual([ 'new_distinct_id', 'old_distinct_id', ]) @@ -482,7 +521,7 @@ export const createProcessEventTests = ( await createPerson(server, team, ['old_distinct_id']) await createPerson(server, team, ['new_distinct_id']) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -496,8 +535,8 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) - expect(await getDistinctIds(server, (await getPersons(server))[0])).toEqual([ + expect((await server.db.fetchEvents()).length).toBe(1) + expect(await server.db.fetchDistinctIdValues((await server.db.fetchPersons())[0])).toEqual([ 'old_distinct_id', 'new_distinct_id', ]) @@ -506,7 +545,7 @@ export const createProcessEventTests = ( test('offset timestamp', async () => { now = DateTime.fromISO('2020-01-01T12:00:05.200Z') - await eventsProcessor.processEvent( + await processEvent( 'distinct_id', '', '', @@ -516,16 +555,16 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) + expect((await server.db.fetchEvents()).length).toBe(1) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() 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( + await processEvent( 'distinct_id', '', '', @@ -535,9 +574,9 @@ export const createProcessEventTests = ( null, new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) + expect((await server.db.fetchEvents()).length).toBe(1) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() expect(event.timestamp).toEqual('2020-01-01T12:00:05.050Z') }) @@ -551,7 +590,7 @@ export const createProcessEventTests = ( key_on_new: 'new value', }) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -565,10 +604,10 @@ export const createProcessEventTests = ( 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((await server.db.fetchEvents()).length).toBe(1) + expect((await server.db.fetchPersons()).length).toBe(1) + const [person] = await server.db.fetchPersons() + expect(await server.db.fetchDistinctIdValues(person)).toEqual(['old_distinct_id', 'new_distinct_id']) expect(person.properties).toEqual({ key_on_both: 'new value both', key_on_new: 'new value', @@ -577,7 +616,7 @@ export const createProcessEventTests = ( }) test('long htext', async () => { - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -604,11 +643,15 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const [event] = await getEvents(server) - expect(event.elements_hash).toEqual('c2659b28e72835706835764cf7f63c2a') - const [element] = await getElements(server, event) + const [event] = (await server.db.fetchEvents()) as Event[] + const [element] = await server.db.fetchElements(event) expect(element.href?.length).toEqual(2048) expect(element.text?.length).toEqual(400) + if (database === 'postgresql') { + expect(event.elements_hash).toEqual('c2659b28e72835706835764cf7f63c2a') + } else if (database === 'clickhouse') { + expect(hashElements([element])).toEqual('c2659b28e72835706835764cf7f63c2a') + } }) test('capture first team event', async () => { @@ -619,7 +662,7 @@ export const createProcessEventTests = ( capture: jest.fn((event, properties) => true), } as any - await eventsProcessor.processEvent( + await processEvent( '2', '', '', @@ -645,8 +688,13 @@ export const createProcessEventTests = ( team = await getFirstTeam(server) expect(team.ingested_event).toEqual(true) - const [event] = await getEvents(server) - expect(event.elements_hash).toEqual('a89021a60b3497d24e93ae181fba01aa') + const [event] = (await server.db.fetchEvents()) as Event[] + if (database === 'postgresql') { + expect(event.elements_hash).toEqual('a89021a60b3497d24e93ae181fba01aa') + } else if (database === 'clickhouse') { + const elements = await server.db.fetchElements(event) + expect(hashElements(elements)).toEqual('a89021a60b3497d24e93ae181fba01aa') + } }) test('snapshot event stored as session_recording_event', async () => { @@ -663,11 +711,12 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) + await delayUntilEventIngested(() => server.db.fetchSessionRecordingEvents()) - const events = await getEvents(server) + const events = await server.db.fetchEvents() expect(events.length).toEqual(0) - const sessionRecordingEvents = await getSessionRecordingEvents(server) + const sessionRecordingEvents = await server.db.fetchSessionRecordingEvents() expect(sessionRecordingEvents.length).toBe(1) const [event] = sessionRecordingEvents @@ -679,7 +728,7 @@ export const createProcessEventTests = ( test('identify set', async () => { await createPerson(server, team, ['distinct_id']) - await eventsProcessor.processEvent( + await processEvent( 'distinct_id', '', '', @@ -697,17 +746,17 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) + expect((await server.db.fetchEvents()).length).toBe(1) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() 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']) + const [person] = await server.db.fetchPersons() + expect(await server.db.fetchDistinctIdValues(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( + await processEvent( 'distinct_id', '', '', @@ -724,15 +773,15 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(2) - const [person2] = await getPersons(server) + expect((await server.db.fetchEvents()).length).toBe(2) + const [person2] = await server.db.fetchPersons() 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( + await processEvent( 'distinct_id', '', '', @@ -750,17 +799,17 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) + expect((await server.db.fetchEvents()).length).toBe(1) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() 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']) + const [person] = await server.db.fetchPersons() + expect(await server.db.fetchDistinctIdValues(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( + await processEvent( 'distinct_id', '', '', @@ -777,15 +826,15 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(2) - const [person2] = await getPersons(server) + expect((await server.db.fetchEvents()).length).toBe(2) + const [person2] = await server.db.fetchPersons() 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( + await processEvent( 'new_distinct_id', '', '', @@ -804,15 +853,15 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - expect((await getEvents(server)).length).toBe(1) - const [event] = await getEvents(server) + expect((await server.db.fetchEvents()).length).toBe(1) + const [event] = await server.db.fetchEvents() expect(event.properties['$set']).toEqual({ a_prop: 'test' }) - const [person] = await getPersons(server) - expect(await getDistinctIds(server, person)).toEqual(['anonymous_id', 'new_distinct_id']) + const [person] = await server.db.fetchPersons() + expect(await server.db.fetchDistinctIdValues(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( + await processEvent( 'new_distinct_id', '', '', @@ -841,7 +890,7 @@ export const createProcessEventTests = ( await createPerson(server, team, ['anonymous_id']) await createPerson(server, team, ['new_distinct_id'], { email: 'someone@gmail.com' }) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -859,8 +908,8 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const [person] = await getPersons(server) - expect(await getDistinctIds(server, person)).toEqual(['anonymous_id', 'new_distinct_id']) + const [person] = await server.db.fetchPersons() + expect(await server.db.fetchDistinctIdValues(person)).toEqual(['anonymous_id', 'new_distinct_id']) expect(person.properties['email']).toEqual('someone@gmail.com') }) @@ -868,7 +917,7 @@ export const createProcessEventTests = ( await createPerson(server, team, ['anonymous_id']) await createPerson(server, team, ['new_distinct_id'], { email: 'someone@gmail.com' }) - await eventsProcessor.processEvent( + await processEvent( 'new_distinct_id', '', '', @@ -886,14 +935,14 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const persons1 = await getPersons(server) + const persons1 = await server.db.fetchPersons() expect(persons1.length).toBe(1) - expect(await getDistinctIds(server, persons1[0])).toEqual(['anonymous_id', 'new_distinct_id']) + expect(await server.db.fetchDistinctIdValues(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( + await processEvent( 'new_distinct_id', '', '', @@ -911,9 +960,13 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const persons2 = await getPersons(server) + const persons2 = await server.db.fetchPersons() expect(persons2.length).toBe(1) - expect(await getDistinctIds(server, persons2[0])).toEqual(['anonymous_id', 'new_distinct_id', 'anonymous_id_2']) + expect(await server.db.fetchDistinctIdValues(persons2[0])).toEqual([ + 'anonymous_id', + 'new_distinct_id', + 'anonymous_id_2', + ]) expect(persons2[0].properties['email']).toEqual('someone@gmail.com') }) @@ -929,7 +982,7 @@ export const createProcessEventTests = ( await createPerson(server, team2, ['2'], { email: 'team2@gmail.com' }) await createPerson(server, team, ['1', '2']) - await eventsProcessor.processEvent( + await processEvent( '2', '', '', @@ -947,13 +1000,13 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const people = await getPersons(server) + const people = await server.db.fetchPersons() 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(await server.db.fetchDistinctIdValues(people[1])).toEqual(['1', '2']) expect(people[0].team_id).toEqual(team2.id) - expect(await getDistinctIds(server, people[0])).toEqual(['2']) + expect(await server.db.fetchDistinctIdValues(people[0])).toEqual(['2']) }) test('set is_identified', async () => { @@ -961,7 +1014,7 @@ export const createProcessEventTests = ( const person1 = await createPerson(server, team, [distinct_id]) expect(person1.is_identified).toBe(false) - await eventsProcessor.processEvent( + await processEvent( distinct_id, '', '', @@ -972,14 +1025,14 @@ export const createProcessEventTests = ( new UUIDT().toString() ) - const [person2] = await getPersons(server) + const [person2] = await server.db.fetchPersons() expect(person2.is_identified).toBe(true) }) test('team event_properties', async () => { expect(team.event_properties_numerical).toEqual([]) - await eventsProcessor.processEvent( + await processEvent( 'xxx', '', '', @@ -996,7 +1049,7 @@ export const createProcessEventTests = ( }) test('event name object json', async () => { - await eventsProcessor.processEvent( + await processEvent( 'xxx', '', '', @@ -1006,12 +1059,12 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() expect(event.event).toEqual('{"event name":"as object"}') }) test('event name array json', async () => { - await eventsProcessor.processEvent( + await processEvent( 'xxx', '', '', @@ -1021,12 +1074,12 @@ export const createProcessEventTests = ( now, new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() expect(event.event).toEqual('["event name","a list"]') }) test('long event name substr', async () => { - await eventsProcessor.processEvent( + await processEvent( 'xxx', '', '', @@ -1034,12 +1087,40 @@ export const createProcessEventTests = ( team.id, DateTime.utc(), DateTime.utc(), - 'uuid' + new UUIDT().toString() ) - const [event] = await getEvents(server) + const [event] = await server.db.fetchEvents() expect(event.event?.length).toBe(200) }) - return server! + test('throws with bad uuid', async () => { + await expect( + processEvent( + 'xxx', + '', + '', + ({ event: 'E', properties: { price: 299.99, name: 'AirPods Pro' } } as any) as PluginEvent, + team.id, + DateTime.utc(), + DateTime.utc(), + 'this is not an uuid' + ) + ).rejects.toEqual(new Error('Not a valid UUID: "this is not an uuid"')) + + await expect( + processEvent( + 'xxx', + '', + '', + ({ event: 'E', properties: { price: 299.99, name: 'AirPods Pro' } } as any) as PluginEvent, + team.id, + DateTime.utc(), + DateTime.utc(), + null as any + ) + ).rejects.toEqual(new Error('Not a valid UUID: "null"')) + }) + + return returned } diff --git a/tests/utils.test.ts b/tests/utils.test.ts index 0331b53a..27e1de0a 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -303,6 +303,6 @@ describe('sanitizeSqlIdentifier', () => { const sanitizedIdentifier = sanitizeSqlIdentifier(rawIdentifier) - expect(sanitizedIdentifier).toStrictEqual('"some_fieldDROPTABLEactually_an_injection9"') + expect(sanitizedIdentifier).toStrictEqual('some_fieldDROPTABLEactually_an_injection9') }) })