diff --git a/src/db.ts b/src/db.ts index 98f635dc..4458298b 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,5 +1,5 @@ import { Properties } from '@posthog/plugin-scaffold' -import { ClickHouse, QueryCursor } from 'clickhouse' +import { ClickHouse } from 'clickhouse' import { Producer } from 'kafkajs' import { DateTime } from 'luxon' import { Pool, QueryConfig, QueryResult, QueryResultRow } from 'pg' @@ -7,18 +7,22 @@ import { string } from 'yargs' import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' import { chainToElements, hashElements, unparsePersonPartial } from './ingestion/utils' import { + ClickHouseEvent, + ClickHousePerson, + ClickHousePersonDistinctId, + Database, + Element, + ElementGroup, + Event, Person, PersonDistinctId, - RawPerson, - RawOrganization, PostgresSessionRecordingEvent, - Event, - ClickHouseEvent, - Element, + RawOrganization, + RawPerson, SessionRecordingEvent, - ElementGroup, + TimestampFormat, } from './types' -import { castTimestampOrNow, clickHouseTimestampToISO, sanitizeSqlIdentifier } from './utils' +import { castTimestampOrNow, clickHouseTimestampToISO, escapeClickHouseString, sanitizeSqlIdentifier } from './utils' /** The recommended way of accessing the database. */ export class DB { @@ -53,9 +57,22 @@ export class DB { // Person - public async fetchPersons(): Promise { - const result = await this.postgresQuery('SELECT * FROM posthog_person') - return result.rows as Person[] + public async fetchPersons(database?: Database.Postgres): Promise + public async fetchPersons(database: Database.ClickHouse): Promise + public async fetchPersons(database: Database = Database.Postgres): Promise { + if (database === Database.ClickHouse) { + return (await this.clickhouseQuery('SELECT * FROM person')) as ClickHousePerson[] + } else if (database === Database.Postgres) { + return ((await this.postgresQuery('SELECT * FROM posthog_person')).rows as RawPerson[]).map( + (rawPerson: RawPerson) => + ({ + ...rawPerson, + created_at: DateTime.fromISO(rawPerson.created_at).toUTC(), + } as Person) + ) + } else { + throw new Error(`Can't fetch persons for database: ${database}`) + } } public async fetchPerson(teamId: number, distinctId: string): Promise { @@ -75,7 +92,7 @@ export class DB { ) if (selectResult.rows.length > 0) { const rawPerson: RawPerson = selectResult.rows[0] - return { ...rawPerson, created_at: DateTime.fromISO(rawPerson.created_at) } + return { ...rawPerson, created_at: DateTime.fromISO(rawPerson.created_at).toUTC() } } } @@ -92,10 +109,11 @@ export class DB { 'INSERT INTO posthog_person (created_at, properties, team_id, is_user_id, is_identified, uuid) VALUES ($1, $2, $3, $4, $5, $6) RETURNING *', [createdAt.toISO(), JSON.stringify(properties), teamId, isUserId, isIdentified, uuid] ) - const personCreated = insertResult.rows[0] as Person + const personCreated = insertResult.rows[0] as RawPerson + const person = { ...personCreated, created_at: DateTime.fromISO(personCreated.created_at).toUTC() } as Person if (this.kafkaProducer) { const data = { - created_at: castTimestampOrNow(createdAt), + created_at: castTimestampOrNow(createdAt, TimestampFormat.ClickHouse), properties: JSON.stringify(properties), team_id: teamId, is_identified: isIdentified, @@ -108,10 +126,9 @@ export class DB { } for (const distinctId of distinctIds || []) { - await this.addDistinctId(personCreated, distinctId) + await this.addDistinctId(person, distinctId) } - - return personCreated + return person } public async updatePerson(person: Person, update: Partial): Promise { @@ -123,39 +140,71 @@ export class DB { )} WHERE id = $${Object.values(update).length + 1}`, values ) - if (this.kafkaProducer) { - const data = { - created_at: castTimestampOrNow(updatedPerson.created_at), - properties: JSON.stringify(updatedPerson.properties), - team_id: updatedPerson.team_id, - is_identified: updatedPerson.is_identified, - id: updatedPerson.uuid.toString(), + if (this.clickhouse) { + const { is_user_id, id, uuid, ...validUpdates } = update + const updateString = Object.entries(validUpdates) + .map(([key, value]) => { + let clickhouseValue: string + if (typeof value === 'string') { + clickhouseValue = value + } else if (typeof value === 'boolean') { + clickhouseValue = value ? '1' : '0' + } else if (DateTime.isDateTime(value)) { + clickhouseValue = castTimestampOrNow(value, TimestampFormat.ClickHouse) + } else { + clickhouseValue = JSON.stringify(value) + } + return `${sanitizeSqlIdentifier(key)} = '${escapeClickHouseString(clickhouseValue)}'` + }) + .join(', ') + if (updateString.length > 0) { + await this.clickhouseQuery( + `ALTER TABLE person UPDATE ${updateString} WHERE id = '${escapeClickHouseString(person.uuid)}'` + ) } - await this.kafkaProducer.send({ - topic: KAFKA_PERSON, - messages: [{ value: Buffer.from(JSON.stringify(data)) }], - }) } return updatedPerson } - public async deletePerson(personId: number): Promise { - await this.postgresQuery('DELETE FROM posthog_persondistinctid WHERE person_id = $1', [personId]) - await this.postgresQuery('DELETE FROM posthog_person WHERE id = $1', [personId]) + public async deletePerson(person: Person): Promise { + await this.postgresQuery('DELETE FROM posthog_persondistinctid WHERE person_id = $1', [person.id]) + await this.postgresQuery('DELETE FROM posthog_person WHERE id = $1', [person.id]) if (this.clickhouse) { - await this.clickhouseQuery(`ALTER TABLE person DELETE WHERE id = ${personId}`) - await this.clickhouseQuery(`ALTER TABLE person_distinct_id DELETE WHERE person_id = ${personId}`) + await this.clickhouseQuery(`ALTER TABLE person DELETE WHERE id = '${escapeClickHouseString(person.uuid)}'`) + await this.clickhouseQuery( + `ALTER TABLE person_distinct_id DELETE WHERE person_id = '${escapeClickHouseString(person.uuid)}'` + ) } } // 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 fetchDistinctIds(person: Person, database?: Database.Postgres): Promise + public async fetchDistinctIds(person: Person, database: Database.ClickHouse): Promise + public async fetchDistinctIds( + person: Person, + database: Database = Database.Postgres + ): Promise { + if (database === Database.ClickHouse) { + return (await this.clickhouseQuery( + `SELECT * FROM person_distinct_id WHERE person_id='${escapeClickHouseString( + person.uuid + )}' and team_id='${person.team_id}' ORDER BY id` + )) as ClickHousePersonDistinctId[] + } else if (database === Database.Postgres) { + 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[] + } else { + throw new Error(`Can't fetch persons for database: ${database}`) + } + } + + public async fetchDistinctIdValues(person: Person, database: Database = Database.Postgres): Promise { + const personDistinctIds = await this.fetchDistinctIds(person, database as any) + return personDistinctIds.map((pdi) => pdi.distinct_id) } public async addDistinctId(person: Person, distinctId: string): Promise { @@ -167,26 +216,27 @@ export class DB { if (this.kafkaProducer) { await this.kafkaProducer.send({ topic: KAFKA_PERSON_UNIQUE_ID, - messages: [{ value: Buffer.from(JSON.stringify(personDistinctIdCreated)) }], + messages: [ + { value: Buffer.from(JSON.stringify({ ...personDistinctIdCreated, person_id: person.uuid })) }, + ], }) } } - public async updateDistinctId( + public async moveDistinctId( + person: Person, personDistinctId: PersonDistinctId, - update: Partial + moveToPerson: Person ): Promise { - const updatedPersonDistinctId: PersonDistinctId = { ...personDistinctId, ...update } - await this.postgresQuery( - `UPDATE posthog_persondistinctid SET ${Object.keys(update).map( - (field, index) => `"${sanitizeSqlIdentifier(field)}" = $${index + 1}` - )} WHERE id = $${Object.values(update).length + 1}`, - [...Object.values(update), personDistinctId.id] - ) + await this.postgresQuery(`UPDATE posthog_persondistinctid SET person_id = $1 WHERE id = $2`, [ + moveToPerson.id, + personDistinctId.id, + ]) if (this.kafkaProducer) { + const clickhouseModel: ClickHousePersonDistinctId = { ...personDistinctId, person_id: moveToPerson.uuid } await this.kafkaProducer.send({ topic: KAFKA_PERSON_UNIQUE_ID, - messages: [{ value: Buffer.from(JSON.stringify(updatedPersonDistinctId)) }], + messages: [{ value: Buffer.from(JSON.stringify(clickhouseModel)) }], }) } } @@ -248,7 +298,7 @@ export class DB { 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)}'` + `SELECT elements_chain FROM events WHERE uuid='${escapeClickHouseString((event as any).uuid)}'` )) as ClickHouseEvent[] const chain = events?.[0]?.elements_chain return chainToElements(chain) diff --git a/src/ingestion/process-event.ts b/src/ingestion/process-event.ts index 304fd9d2..acc584e6 100644 --- a/src/ingestion/process-event.ts +++ b/src/ingestion/process-event.ts @@ -288,7 +288,7 @@ export class EventsProcessor { ) ).rows for (const personDistinctId of otherPersonDistinctIds) { - await this.db.updateDistinctId(personDistinctId, { person_id: mergeInto.id }) + await this.db.moveDistinctId(otherPerson, personDistinctId, mergeInto) } const otherCohortPeople: CohortPeople[] = ( @@ -301,7 +301,7 @@ export class EventsProcessor { ]) } - await this.db.deletePerson(otherPerson.id) + await this.db.deletePerson(otherPerson) } } @@ -359,9 +359,9 @@ export class EventsProcessor { teamId, null, false, - personUuid.toString() + personUuid.toString(), + [distinctId] ) - await this.db.addDistinctId(personCreated, distinctId) } catch {} } diff --git a/src/types.ts b/src/types.ts index 5e644da9..4e70ef5c 100644 --- a/src/types.ts +++ b/src/types.ts @@ -290,6 +290,16 @@ export interface Person extends BasePerson { created_at: DateTime } +/** Clickhouse Person model. */ +export interface ClickHousePerson { + id: string + created_at: string + team_id: number + properties: string + is_identified: number + timestamp: string +} + /** Usable PersonDistinctId model. */ export interface PersonDistinctId { id: number @@ -298,6 +308,14 @@ export interface PersonDistinctId { distinct_id: string } +/** ClickHouse PersonDistinctId model. */ +export interface ClickHousePersonDistinctId { + id: number + team_id: number + person_id: string + distinct_id: string +} + /** Usable CohortPeople model. */ export interface CohortPeople { id: number @@ -323,3 +341,8 @@ export enum TimestampFormat { ClickHouse = 'clickhouse', ISO = 'iso', } + +export enum Database { + ClickHouse = 'clickhouse', + Postgres = 'postgres', +} diff --git a/src/utils.ts b/src/utils.ts index eccf8e5e..bd6227a4 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -325,3 +325,10 @@ export function delay(ms: number): Promise { export function sanitizeSqlIdentifier(unquotedIdentifier: string): string { return unquotedIdentifier.replace(/[^\w\d_]+/g, '') } + +/** Escape single quotes and slashes */ +export function escapeClickHouseString(string: string): string { + // In string literals, you need to escape at least `'` and `\`. + // https://clickhouse.tech/docs/en/sql-reference/syntax/ + return string.replace(/\\/g, '\\\\').replace(/'/g, "\\'") +} diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts new file mode 100644 index 00000000..a964faf1 --- /dev/null +++ b/tests/clickhouse/postgres-parity.test.ts @@ -0,0 +1,298 @@ +import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team, TimestampFormat } from '../../src/types' +import { getFirstTeam, resetTestDatabase } from '../helpers/sql' +import { startPluginsServer } from '../../src/server' +import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/extensions/posthog' +import { pluginConfig39 } from '../helpers/plugins' +import { castTimestampOrNow, UUIDT } from '../../src/utils' +import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' +import { resetKafka } from '../helpers/kafka' +import { delayUntilEventIngested } from '../shared/process-event' +import { DateTime } from 'luxon' + +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('postgres parity', () => { + let server: PluginsServer + let stopServer: () => Promise + let posthog: DummyPostHog + let team: Team + + 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) + team = await getFirstTeam(server) + }) + + afterEach(async () => { + await stopServer() + }) + + test('createPerson', async () => { + const uuid = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + true, + uuid, + ['distinct1', 'distinct2'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(person, Database.ClickHouse), 2) + + const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) + expect(clickHousePersons).toEqual([ + { + id: uuid, + created_at: expect.any(String), // '2021-02-04 00:18:26.472', + team_id: team.id, + properties: '{"userProp":"propValue"}', + is_identified: 1, + _timestamp: expect.any(String), + _offset: expect.any(Number), + }, + ]) + const clickHouseDistinctIds = await server.db.fetchDistinctIdValues(person, Database.ClickHouse) + expect(clickHouseDistinctIds).toEqual(['distinct1', 'distinct2']) + + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + expect(postgresPersons).toEqual([ + { + id: expect.any(Number), + created_at: expect.any(DateTime), + properties: { + userProp: 'propValue', + }, + team_id: 2, + is_user_id: null, + is_identified: true, + uuid: uuid, + }, + ]) + const postgresDistinctIds = await server.db.fetchDistinctIdValues(person, Database.Postgres) + expect(postgresDistinctIds).toEqual(['distinct1', 'distinct2']) + + expect(person).toEqual(postgresPersons[0]) + }) + + test('updatePerson', async () => { + const uuid = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + false, + uuid, + ['distinct1', 'distinct2'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(person, Database.ClickHouse), 2) + + // update JSON and boolean to true + + await server.db.updatePerson(person, { properties: { replacedUserProp: 'propValue' }, is_identified: true }) + + await delayUntilEventIngested(async () => + (await server.db.fetchPersons(Database.ClickHouse)).filter((p) => p.is_identified) + ) + + const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons.length).toEqual(1) + expect(postgresPersons.length).toEqual(1) + + expect(postgresPersons[0].is_identified).toEqual(true) + expect(postgresPersons[0].properties).toEqual({ replacedUserProp: 'propValue' }) + + expect(clickHousePersons[0].is_identified).toEqual(1) + expect(clickHousePersons[0].properties).toEqual('{"replacedUserProp":"propValue"}') + + // update date and boolean to false + + const randomDate = DateTime.utc().minus(100000).setZone('UTC') + await server.db.updatePerson(person, { created_at: randomDate, is_identified: false }) + + await delayUntilEventIngested(async () => + (await server.db.fetchPersons(Database.ClickHouse)).filter((p) => p.is_identified) + ) + + const clickHousePersons2 = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons2 = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons2.length).toEqual(1) + expect(postgresPersons2.length).toEqual(1) + + expect(postgresPersons2[0].is_identified).toEqual(false) + expect(postgresPersons2[0].created_at.toISO()).toEqual(randomDate.toISO()) + + expect(clickHousePersons2[0].is_identified).toEqual(0) + expect(clickHousePersons2[0].created_at).toEqual(castTimestampOrNow(randomDate, TimestampFormat.ClickHouse)) + }) + + test('deletePerson', async () => { + const uuid = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + false, + uuid, + ['distinct1', 'distinct2'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(person, Database.ClickHouse), 2) + + await server.db.deletePerson(person) + + await delayUntilEventIngested(async () => + (await server.db.fetchPersons(Database.ClickHouse)).length === 0 ? ['deleted!'] : [] + ) + + const clickHousePersons = await server.db.fetchPersons(Database.ClickHouse) + const postgresPersons = await server.db.fetchPersons(Database.Postgres) + + expect(clickHousePersons.length).toEqual(0) + expect(postgresPersons.length).toEqual(0) + + const clickHouseDistinctIdValues = await server.db.fetchDistinctIdValues(person, Database.ClickHouse) + const postgresDistinctIdValues = await server.db.fetchDistinctIdValues(person, Database.Postgres) + expect(clickHouseDistinctIdValues.length).toEqual(0) + expect(postgresDistinctIdValues.length).toEqual(0) + }) + + test('addDistinctId & moveDistinctId', async () => { + const uuid = new UUIDT().toString() + const uuid2 = new UUIDT().toString() + const person = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + true, + uuid, + ['distinct1'] + ) + const anotherPerson = await server.db.createPerson( + DateTime.utc(), + { userProp: 'propValue' }, + team.id, + null, + true, + uuid2, + ['another_distinct_id'] + ) + await delayUntilEventIngested(() => server.db.fetchPersons(Database.ClickHouse)) + const [postgresPerson] = await server.db.fetchPersons(Database.Postgres) + + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse), 1) + const clickHouseDistinctIdValues = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) + const postgresDistinctIdValues = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + + // check that all is in the right format + + expect(clickHouseDistinctIdValues).toEqual(['distinct1']) + expect(postgresDistinctIdValues).toEqual(['distinct1']) + + const clickHouseDistinctIds = await server.db.fetchDistinctIds(postgresPerson, Database.ClickHouse) + const postgresDistinctIds = await server.db.fetchDistinctIds(postgresPerson, Database.Postgres) + + expect(clickHouseDistinctIds).toEqual([ + { + id: expect.any(Number), + distinct_id: 'distinct1', + person_id: person.uuid, + team_id: team.id, + _timestamp: expect.any(String), + _offset: expect.any(Number), + }, + ]) + expect(postgresDistinctIds).toEqual([ + { + id: expect.any(Number), + distinct_id: 'distinct1', + person_id: person.id, + team_id: team.id, + }, + ]) + expect(clickHouseDistinctIds[0].id).toEqual(postgresDistinctIds[0].id) + + // add 'anotherOne' to person + + await server.db.addDistinctId(postgresPerson, 'anotherOne') + + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse), 2) + + const clickHouseDistinctIdValues2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.ClickHouse) + const postgresDistinctIdValues2 = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + + expect(clickHouseDistinctIdValues2).toEqual(['distinct1', 'anotherOne']) + expect(postgresDistinctIdValues2).toEqual(['distinct1', 'anotherOne']) + + // check anotherPerson for their initial distinct id + + const clickHouseDistinctIdValuesOther = await server.db.fetchDistinctIdValues( + anotherPerson, + Database.ClickHouse + ) + const postgresDistinctIdValuesOther = await server.db.fetchDistinctIdValues(anotherPerson, Database.Postgres) + + expect(clickHouseDistinctIdValuesOther).toEqual(['another_distinct_id']) + expect(postgresDistinctIdValuesOther).toEqual(['another_distinct_id']) + + // move 'distinct1' from person to to anotherPerson + + await server.db.moveDistinctId(postgresPerson, postgresDistinctIds[0], anotherPerson) + await delayUntilEventIngested(() => server.db.fetchDistinctIdValues(anotherPerson, Database.ClickHouse), 2) + + // it got added + + const clickHouseDistinctIdValuesMoved = await server.db.fetchDistinctIdValues( + anotherPerson, + Database.ClickHouse + ) + const postgresDistinctIdValuesMoved = await server.db.fetchDistinctIdValues(anotherPerson, Database.Postgres) + + expect(clickHouseDistinctIdValuesMoved).toEqual(['distinct1', 'another_distinct_id']) + expect(postgresDistinctIdValuesMoved).toEqual(['distinct1', 'another_distinct_id']) + + // it got removed + + const clickHouseDistinctIdValuesRemoved = await server.db.fetchDistinctIdValues( + postgresPerson, + Database.ClickHouse + ) + const postgresDistinctIdValuesRemoved = await server.db.fetchDistinctIdValues(postgresPerson, Database.Postgres) + + // The `distinct1` key is still there in clickhouse, yet ALSO there for the new person. + // Eventually this should be compacted away but it's not right now. + expect(clickHouseDistinctIdValuesRemoved).toEqual(['distinct1', 'anotherOne']) + expect(postgresDistinctIdValuesRemoved).toEqual(['anotherOne']) + }) +}) diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 7d1b535a..6c4ca51b 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -2,7 +2,14 @@ import { Kafka, logLevel } from 'kafkajs' import { PluginsServerConfig } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { defaultConfig, overrideWithEnv } from '../../src/config' -import { KAFKA_EVENTS_INGESTION_HANDOFF, KAFKA_SESSION_RECORDING_EVENTS } from '../../src/ingestion/topics' +import { + KAFKA_EVENTS, + KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_EVENTS_WAL, + KAFKA_PERSON, + KAFKA_PERSON_UNIQUE_ID, + KAFKA_SESSION_RECORDING_EVENTS, +} from '../../src/ingestion/topics' /** Clear the kafka queue */ export async function resetKafka(extraServerConfig: Partial, delayMs = 2000) { @@ -19,25 +26,14 @@ export async function resetKafka(extraServerConfig: Partial }) 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) + await createTopics(kafka, [ + KAFKA_EVENTS, + KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_EVENTS_WAL, + KAFKA_SESSION_RECORDING_EVENTS, + KAFKA_PERSON, + KAFKA_PERSON_UNIQUE_ID, + ]) const connected = await new Promise(async (resolve, reject) => { console.info('setting group join and crash listeners') @@ -76,3 +72,13 @@ export async function resetKafka(extraServerConfig: Partial return true } + +async function createTopics(kafka: Kafka, topics: string[]) { + const admin = kafka.admin() + await admin.connect() + await admin.createTopics({ + waitForLeaders: true, + topics: topics.map((topic) => ({ topic })), + }) + await admin.disconnect() +} diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 4c1aef36..35464c0e 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -2,7 +2,7 @@ import { makePluginObjects, commonOrganizationId, commonUserId, commonOrganizati import { defaultConfig } from '../../src/config' import { Pool } from 'pg' import { delay, UUIDT } from '../../src/utils' -import { PluginsServerConfig } from '../../src/types' +import { PluginsServer, PluginsServerConfig, Team } from '../../src/types' export async function resetTestDatabase( code: string, @@ -113,3 +113,11 @@ export async function createUserTeamAndOrganization( is_demo: false, }) } + +export async function getTeams(server: PluginsServer): Promise { + return (await server.db.postgresQuery('SELECT * FROM posthog_team ORDER BY id')).rows +} + +export async function getFirstTeam(server: PluginsServer): Promise { + return (await getTeams(server))[0] +} diff --git a/tests/postgres/process-event.test.ts b/tests/postgres/process-event.test.ts index 2c12142b..3288aea1 100644 --- a/tests/postgres/process-event.test.ts +++ b/tests/postgres/process-event.test.ts @@ -13,6 +13,8 @@ describe('process event (postgresql)', () => { const elementsHash = server!.db.createElementGroup(elements, 2) const elementGroup = await server!.db.fetchElements() + console.log(elementGroup) + expect(elementGroup[0].tag_name).toEqual('button') expect(elementGroup[1].tag_name).toEqual('div') expect(elementGroup.length).toEqual(2) diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index b9a4c058..249b67ab 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -12,7 +12,7 @@ import { ClickHouseEvent, SessionRecordingEvent, } from '../../src/types' -import { createUserTeamAndOrganization, resetTestDatabase } from '../helpers/sql' +import { createUserTeamAndOrganization, getFirstTeam, getTeams, resetTestDatabase } from '../helpers/sql' import { EventsProcessor } from '../../src/ingestion/process-event' import { DateTime } from 'luxon' import { delay, UUIDT } from '../../src/utils' @@ -21,14 +21,6 @@ import { hashElements } from '../../src/ingestion/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] -} - export async function delayUntilEventIngested(fetchEvents: () => Promise, minCount = 1): Promise { for (let i = 0; i < 30; i++) { if ((await fetchEvents()).length >= minCount) { diff --git a/tests/utils.test.ts b/tests/utils.test.ts index 27e1de0a..6cdef39c 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -8,6 +8,7 @@ import { UUID, UUIDT, sanitizeSqlIdentifier, + escapeClickHouseString, } from '../src/utils' import { randomBytes } from 'crypto' import { LogLevel } from '../src/types' @@ -306,3 +307,13 @@ describe('sanitizeSqlIdentifier', () => { expect(sanitizedIdentifier).toStrictEqual('some_fieldDROPTABLEactually_an_injection9') }) }) + +describe('escapeClickHouseString', () => { + it('escapes single quotes and slashes', () => { + const rawString = "insert'escape \\" + + const sanitizedString = escapeClickHouseString(rawString) + + expect(sanitizedString).toStrictEqual("insert\\'escape \\\\") + }) +})