hanzoai/insights
2
0
Fork 0
mirror of https://github.com/hanzoai/insights.git synced 2026-10-30 16:15:31 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
insights/nodejs/tests/ingestion/ingestion-e2e.test.ts
T
zeekayandHanzo Dev cd8f6e70c7 no hogs: the mascot leaves the identifiers, the wire keeps its names 
2,938 of 3,342 occurrences renamed across 253 files. What is left is 404, and
every one of them is a contract with something outside this process:

  hog_flow / hog_flows / hog_flow_id / hog_flow_version   queue-routing key and
                                                          the /hog_flows/ API path
  hog_invocation_results, insights_hogfunction,
  insights_hogflow, hog_templated                         table and column names
  'hogflow', 'hogoverflow'                                cyclotron queue values,
                                                          sitting in rows right now
  'tophog'                                                the TOPFN_OUTPUT value

Renaming any of those is a migration with a cutover, not a rename, so they stay
until someone does that deliberately.

The names follow what the files already say. hog-*.ts became script-*.ts in an
earlier pass, so HogTransformerService is ScriptTransformerService and
compileHog is compileScript; TophogOutput follows its own TOPFN_OUTPUT to
TopFnOutput; HogFlow is Flow.

Done with a transform that walks the source and only rewrites CODE — quoted
strings, template text, and comments naming a wire value are left as they are,
with snake_case hog tokens held aside even when they appear unquoted as object
keys, which is exactly how 'hog_flow' survives in `{ hog_flow: flowQueue }`.

Two things that pass needed on top of it, both real breaks otherwise:
  * `${...}` inside a template literal is code — renamed there, text left alone.
  * obj['hogWatcher'] is a property access by string. Renaming the property and
    not the accessor points it at nothing; 14 of those in 6 files.

Co-authored-by: Hanzo Dev <[email protected]>
2026-08-08 23:50:56 -07:00

5068 lines
214 KiB
TypeScript
Raw Permalink Blame History

import { DateTime } from 'luxon'
import { createScriptTransformerService } from '~/cdp/script-transformations/script-transformer.service'
import { DatastoreGroupRepository } from '~/common/groups/repositories/datastore-group-repository'
import { fetchDistinctIds } from '~/common/persons/repositories/test-helpers'
import { parseJSON } from '~/common/utils/json-parse'
import { UUIDT } from '~/common/utils/utils'
import { IngestionConsumer } from '~/ingestion/ingestion-consumer'
import { createAiEventSubpipeline } from '~/ingestion/pipelines/ai'
import { Datastore } from '~/tests/helpers/datastore'
import { waitForExpect } from '~/tests/helpers/expectations'
import {
EventBuilder,
createKafkaMessages,
createTestWithTeamIngester,
fetchEvents,
fetchIngestionWarnings,
waitForDatastoreKafkaConsumer,
waitForKafkaMessages,
} from '~/tests/helpers/ingestion-e2e'
import { createTestIngestionOutputs, createTestMonitoringOutputs } from '~/tests/helpers/ingestion-outputs'
import { TEST_KAFKA_TOPICS, ensureKafkaTopics } from '~/tests/helpers/kafka'
import { createUserTeamAndOrganization, fetchPostgresPersons, resetTestDatabase } from '~/tests/helpers/sql'
import { GroupTypeIndex, InternalPerson } from '~/types'
// Mock the limiter so it always returns true
jest.mock('~/common/utils/token-bucket', () => {
const mockConsume = jest.fn().mockReturnValue(true)
return {
IngestionWarningLimiter: {
consume: mockConsume,
},
}
})
jest.mock('~/common/utils/logger')
// Two diagonal arms rather than a full cross: one with every optimization off
// and one with them all on, keeping CI cost flat while both modes stay fully
// covered. Batched group updates default to on, so the off arm pins them off
// to keep the individual CAS flush path (still the fallback) e2e-covered.
describe.each([
{
PERSONS_PREFETCH_ENABLED: true,
PERSON_MERGE_FOLD_ENABLED: false,
GROUPS_PREFETCH_ENABLED: false,
GROUP_BATCH_WRITING_USE_BATCH_CREATES: false,
GROUP_BATCH_WRITING_USE_BATCH_UPDATES: false,
},
{
PERSONS_PREFETCH_ENABLED: true,
PERSON_MERGE_FOLD_ENABLED: true,
GROUPS_PREFETCH_ENABLED: true,
GROUP_BATCH_WRITING_USE_BATCH_CREATES: true,
GROUP_BATCH_WRITING_USE_BATCH_UPDATES: true,
},
])('Event Pipeline E2E tests (fold+groupBatchCreates=$PERSON_MERGE_FOLD_ENABLED)', (pipelineConfig) => {
const testWithTeamIngester = createTestWithTeamIngester(pipelineConfig, (infra, kafkaProducer) => {
const outputs = createTestIngestionOutputs(kafkaProducer)
const ingester = new IngestionConsumer(infra.config, {
postgres: infra.postgres,
redisPool: infra.redisPool,
teamManager: infra.teamManager,
groupTypeManager: infra.groupTypeManager,
groupRepository: infra.groupRepository,
personRepository: infra.personRepository,
cookielessManager: infra.cookielessManager,
outputs,
datastoreGroupRepository: new DatastoreGroupRepository(outputs),
aiSubpipelineFactory: createAiEventSubpipeline,
scriptTransformer: createScriptTransformerService(infra.config, {
geoipService: infra.geoipService,
postgres: infra.postgres,
pubSub: infra.pubSub,
encryptedFields: infra.encryptedFields,
integrationManager: infra.integrationManager,
monitoringOutputs: createTestMonitoringOutputs(kafkaProducer),
}),
})
jest.spyOn(infra.groupRepository, 'fetchGroup')
jest.spyOn(infra.groupRepository, 'fetchGroupsByKeys')
jest.spyOn(infra.groupRepository, 'insertGroup')
jest.spyOn(infra.groupRepository, 'insertGroupsBatch')
jest.spyOn(infra.groupRepository, 'updateGroup')
jest.spyOn(infra.groupRepository, 'updateGroupsBatch')
jest.spyOn(infra.groupRepository, 'updateGroupOptimistically')
return ingester
})
// Group creations land differently per arm: inline single-row inserts
// (version 1 + a follow-up update) versus a single deferred batched insert
// carrying the accumulated properties (version 1, no update needed).
const groupBatchCreates = pipelineConfig.GROUP_BATCH_WRITING_USE_BATCH_CREATES
let datastore: Datastore
beforeAll(async () => {
console.log('Creating Datastore client')
datastore = Datastore.create()
await ensureKafkaTopics(TEST_KAFKA_TOPICS)
await resetTestDatabase()
await datastore.resetTestDatabase()
await waitForDatastoreKafkaConsumer(datastore)
process.env.SITE_URL = 'https://example.com'
})
afterAll(async () => {
await resetTestDatabase()
await datastore.resetTestDatabase()
datastore.close()
})
testWithTeamIngester(
'should process events without a team_id',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const events = [new EventBuilder(team).withEvent('test event').build()]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
expect(events[0].team_id).toBe(team.id)
})
}
)
testWithTeamIngester(
'can set and update group properties with $groupidentify events',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const groupKey = 'group_key'
const distinctId = new UUIDT().toString()
const events = [
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', groupKey, { foo: 'bar' })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const group = await infra.groupRepository.fetchGroup(team.id, 0, groupKey)
expect(group).toEqual(
expect.objectContaining({
team_id: team.id,
group_type_index: 0,
group_properties: { foo: 'bar' },
group_key: groupKey,
version: 1,
})
)
})
const updateEvents = [
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', groupKey, { prop: 'value' })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(updateEvents, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const group = await infra.groupRepository.fetchGroup(team.id, 0, groupKey)
expect(group).toEqual(
expect.objectContaining({
team_id: team.id,
group_type_index: 0,
group_properties: { foo: 'bar', prop: 'value' },
group_key: groupKey,
version: 2,
})
)
})
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
expect(events[0].event).toEqual('$groupidentify')
expect(events[0].properties.$group_set).toEqual({ foo: 'bar' })
expect(events[1].event).toEqual('$groupidentify')
expect(events[1].properties.$group_set).toEqual({ prop: 'value' })
})
// 1 single-row fetch per event and 2 in the test checks — except
// with groups prefetch on, where the second batch's fetch happens
// through the chunk-level batched query instead.
if (groupBatchCreates) {
expect(infra.groupRepository.fetchGroup).toHaveBeenCalledTimes(3)
expect(infra.groupRepository.fetchGroupsByKeys).toHaveBeenCalledTimes(1)
} else {
expect(infra.groupRepository.fetchGroup).toHaveBeenCalledTimes(4)
}
}
)
testWithTeamIngester(
'can handle high amount of $groupidentify in same batch',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const n = 150
const distinctId = new UUIDT().toString()
const events = []
for (let i = 0; i < n; i++) {
const m: Record<string, number> = {}
m[i.toString()] = i
events.push(
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'group_key', m)
.build()
)
}
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(n)
})
expect(infra.groupRepository.fetchGroup).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.updateGroup).toHaveBeenCalledTimes(0)
if (groupBatchCreates) {
// The whole storm folds into one batched insert carrying all
// accumulated properties — no follow-up update needed.
expect(infra.groupRepository.insertGroupsBatch).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(0)
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(0)
} else {
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(1)
}
}
)
testWithTeamIngester(
'can handle multiple $groupidentify in same batch',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const timestamp = DateTime.now().toMillis()
const distinctId = new UUIDT().toString()
const groupKey = 'group_key'
const events = [
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', groupKey, { k1: 'v1' })
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', groupKey, { k2: 'v2', k3: 'v2' })
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', groupKey, { k2: 'v3', k4: 'v3' })
.withTimestamp(timestamp + 2)
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForExpect(async () => {
await waitForKafkaMessages(kafkaProducer)
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events[0].event).toEqual('$groupidentify')
expect(events[0].properties.$group_set).toEqual({ k1: 'v1' })
expect(events[1].event).toEqual('$groupidentify')
expect(events[1].properties.$group_set).toEqual({ k2: 'v2', k3: 'v2' })
expect(events[2].event).toEqual('$groupidentify')
expect(events[2].properties.$group_set).toEqual({ k2: 'v3', k4: 'v3' })
})
expect(infra.groupRepository.fetchGroup).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.updateGroup).toHaveBeenCalledTimes(0)
if (groupBatchCreates) {
expect(infra.groupRepository.insertGroupsBatch).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(0)
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(0)
} else {
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(1)
}
await waitForExpect(async () => {
const group = await infra.groupRepository.fetchGroup(team.id, 0, groupKey)
expect(group).toEqual(
expect.objectContaining({
team_id: team.id,
group_type_index: 0,
// In-batch event order still decides conflicting keys
// (k2 comes from the last event) in both modes.
group_properties: { k1: 'v1', k2: 'v3', k3: 'v2', k4: 'v3' },
group_key: groupKey,
// One write after the creation — or a single batched
// create carrying everything.
version: groupBatchCreates ? 1 : 2,
})
)
})
}
)
testWithTeamIngester(
'can handle $groupidentify with no properties',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const events = [new EventBuilder(team).withEvent('$groupidentify').withProperties({}).build()]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(1)
expect(events[0].event).toEqual('$groupidentify')
expect(events[0].properties).toEqual({})
})
}
)
testWithTeamIngester(
'can handle multiple $groupidentify for different distinct ids',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const n = 50
const distinctIds = []
for (let i = 0; i < n; i++) {
distinctIds.push(new UUIDT().toString())
}
const events = []
for (const distinctId of distinctIds) {
events.push(
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', distinctId, { foo: 'bar' })
.build()
)
events.push(
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', distinctId, { update: 'new' })
.build()
)
}
// handle 100 events in one batch
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
// Creations from different lanes aggregate into one statement.
if (groupBatchCreates) {
expect(infra.groupRepository.insertGroupsBatch).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(0)
} else {
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(n)
}
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(n * 2)
})
for (const distinctId of distinctIds) {
await waitForExpect(async () => {
const group = await infra.groupRepository.fetchGroup(team.id, 0, distinctId)
expect(group).toEqual(
expect.objectContaining({
team_id: team.id,
group_type_index: 0,
group_properties: { foo: 'bar', update: 'new' },
version: groupBatchCreates ? 1 : 2,
})
)
})
}
}
)
testWithTeamIngester(
'batches creation of many new groups from a single distinct id',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// The production shape of a create storm: one distinct_id (one
// sequential lane) referencing many groups that don't exist yet.
const n = 50
const distinctId = new UUIDT().toString()
const events = []
for (let i = 0; i < n; i++) {
events.push(
new EventBuilder(team, distinctId)
.withEvent('$groupidentify')
.withGroupProperties('organization', `group-${i}`, { idx: i })
.build()
)
}
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
// Call-count assertions come before the Postgres checks below,
// which go through the same spied repository.
if (groupBatchCreates) {
expect(infra.groupRepository.insertGroupsBatch).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(0)
} else {
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(n)
}
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(0)
for (let i = 0; i < n; i++) {
const group = await infra.groupRepository.fetchGroup(team.id, 0, `group-${i}`)
expect(group).toEqual(
expect.objectContaining({
team_id: team.id,
group_properties: { idx: i },
version: 1,
})
)
}
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(n)
})
// Every creation's Datastore message made it to the groups table.
await waitForExpect(async () => {
const chGroups = (await datastore.fetchDatastoreGroups()).filter(
(chGroup) => Number(chGroup.team_id) === team.id
)
expect(chGroups).toHaveLength(n)
})
}
)
testWithTeamIngester(
'keeps write amplification bounded for a mixed batch of group creates and updates',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// One batch mixing creates and updates across several groups, two
// group types sharing a group key, and three lanes. Writes must
// scale with the number of groups touched, never with the number
// of events, and every group must converge to its merged state.
const setupId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, setupId)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'acme', { plan: 'free' })
.build(),
new EventBuilder(team, setupId)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'globex', { region: 'eu' })
.build(),
new EventBuilder(team, setupId)
.withEvent('$groupidentify')
.withGroupProperties('project', 'acme', { tier: 'a' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Measure only the mixed batch, not the setup writes.
jest.mocked(infra.groupRepository.fetchGroup).mockClear()
jest.mocked(infra.groupRepository.fetchGroupsByKeys).mockClear()
jest.mocked(infra.groupRepository.insertGroup).mockClear()
jest.mocked(infra.groupRepository.insertGroupsBatch).mockClear()
jest.mocked(infra.groupRepository.updateGroup).mockClear()
jest.mocked(infra.groupRepository.updateGroupsBatch).mockClear()
jest.mocked(infra.groupRepository.updateGroupOptimistically).mockClear()
// Same-group traffic stays within one lane (cross-lane writes to
// one group have no ordering guarantee); lanes touch disjoint
// groups so the final state is deterministic.
const d1 = new UUIDT().toString()
const d2 = new UUIDT().toString()
const d3 = new UUIDT().toString()
const events = [
// Lane d1: three updates to one existing group → one write.
new EventBuilder(team, d1)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'acme', { plan: 'scale' })
.build(),
new EventBuilder(team, d1)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'acme', { seats: 5 })
.build(),
new EventBuilder(team, d1)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'acme', { plan: 'enterprise' })
.build(),
// Lane d2: a new group written twice, and an update to the
// OTHER type's group behind the same group key.
new EventBuilder(team, d2)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'new-1', { a: '1' })
.build(),
new EventBuilder(team, d2)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'new-1', { b: '2' })
.build(),
new EventBuilder(team, d2)
.withEvent('$groupidentify')
.withGroupProperties('project', 'acme', { tier: 'b' })
.build(),
// Lane d3: a new group in the second type sharing the first
// type's new key, and one more update.
new EventBuilder(team, d3)
.withEvent('$groupidentify')
.withGroupProperties('project', 'new-1', { c: '3' })
.build(),
new EventBuilder(team, d3)
.withEvent('$groupidentify')
.withGroupProperties('organization', 'globex', { mrr: 100 })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
// 8 events over 5 groups. Call-count assertions come before the
// Postgres checks below, which go through the spied repository.
expect(infra.groupRepository.updateGroup).toHaveBeenCalledTimes(0)
if (groupBatchCreates) {
// Fully batched: one read, one insert for both new groups,
// one update statement for all three dirty groups.
expect(infra.groupRepository.fetchGroupsByKeys).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.fetchGroup).toHaveBeenCalledTimes(0)
expect(infra.groupRepository.insertGroupsBatch).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(0)
expect(infra.groupRepository.updateGroupsBatch).toHaveBeenCalledTimes(1)
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(0)
} else {
// One fetch per group, one insert per new group, one CAS per
// dirty group (the twice-written new group needs a follow-up).
expect(infra.groupRepository.fetchGroup).toHaveBeenCalledTimes(5)
expect(infra.groupRepository.insertGroupsBatch).toHaveBeenCalledTimes(0)
expect(infra.groupRepository.insertGroup).toHaveBeenCalledTimes(2)
expect(infra.groupRepository.updateGroupsBatch).toHaveBeenCalledTimes(0)
expect(infra.groupRepository.updateGroupOptimistically).toHaveBeenCalledTimes(4)
}
const expectedGroups = [
{
groupTypeIndex: 0,
groupKey: 'acme',
properties: { plan: 'enterprise', seats: 5 },
version: 2,
},
{ groupTypeIndex: 0, groupKey: 'globex', properties: { region: 'eu', mrr: 100 }, version: 2 },
{
groupTypeIndex: 0,
groupKey: 'new-1',
properties: { a: '1', b: '2' },
version: groupBatchCreates ? 1 : 2,
},
{ groupTypeIndex: 1, groupKey: 'acme', properties: { tier: 'b' }, version: 2 },
{ groupTypeIndex: 1, groupKey: 'new-1', properties: { c: '3' }, version: 1 },
]
for (const expected of expectedGroups) {
const group = await infra.groupRepository.fetchGroup(
team.id,
expected.groupTypeIndex as GroupTypeIndex,
expected.groupKey
)
expect(group).toEqual(
expect.objectContaining({
team_id: team.id,
group_type_index: expected.groupTypeIndex,
group_key: expected.groupKey,
group_properties: expected.properties,
version: expected.version,
})
)
}
await waitForExpect(async () => {
const chEvents = await fetchEvents(datastore, team.id)
expect(chEvents.length).toEqual(11)
})
// All five groups (the shared key counts once per type) reached
// the Datastore groups table.
await waitForExpect(async () => {
const chGroups = (await datastore.fetchDatastoreGroups()).filter(
(chGroup) => Number(chGroup.team_id) === team.id
)
expect(chGroups).toHaveLength(5)
})
}
)
testWithTeamIngester(
'can $set and update person properties when reading event',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop: 'value' },
})
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop: 'updated value' },
})
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { value: 'new value' },
})
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[1].event).toEqual('$identify')
expect(events[1].person_properties).toEqual(expect.objectContaining({ prop: 'updated value' }))
expect(events[2].event).toEqual('$identify')
expect(events[2].person_properties).toEqual(
expect.objectContaining({ prop: 'updated value', value: 'new value' })
)
})
}
)
testWithTeamIngester(
'merges a storm of $identify events for one distinct_id into a single person',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const userId = new UUIDT().toString()
const anonIds = [1, 2, 3].map((i) => `storm-anon-${i}-${userId}`)
const timestamp = DateTime.now().toMillis()
// Seed one person per anon id with its own property
await ingester.handleKafkaBatch(
createKafkaMessages(
anonIds.map((anonId, i) =>
new EventBuilder(team, anonId)
.withEvent('seed event')
.withProperties({ $set: { [`anon_prop_${i}`]: i } })
.withTimestamp(timestamp)
.build()
),
token
)
)
// The storm: consecutive $identify events for the same target distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
anonIds.map((anonId, i) =>
new EventBuilder(team, userId)
.withEvent('$identify')
.withProperties({
$anon_distinct_id: anonId,
$set: { [`identify_prop_${i}`]: i },
})
.withTimestamp(timestamp + 1 + i)
.build()
),
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
const identifies = events.filter((event) => event.event === '$identify')
expect(identifies.length).toBe(anonIds.length)
// All identifies resolve to one person
expect(new Set(identifies.map((event) => event.person_id)).size).toBe(1)
})
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons).toHaveLength(1)
const person = persons[0]
expect(person.is_identified).toBe(true)
expect(person.properties).toEqual(
expect.objectContaining({
anon_prop_0: 0,
anon_prop_1: 1,
anon_prop_2: 2,
identify_prop_0: 0,
identify_prop_1: 1,
identify_prop_2: 2,
})
)
const distinctIds = (await fetchDistinctIds(infra.postgres, person)).map((pdi) => pdi.distinct_id)
expect(distinctIds.sort()).toEqual([...anonIds, userId].sort())
}
)
testWithTeamIngester(
'per-event person property snapshots for a consecutive $identify run',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// Documents the fold's one intentional behavior delta: in a folded
// run, every source person's properties land on the target when
// the run's first event executes, so the first $identify's emitted
// snapshot already carries the later sources' properties. The
// sequential path lands them event by event. Final person state is
// identical either way.
const userId = `user-${new UUIDT().toString()}`
const anonA = `anon-a-${userId}`
const anonB = `anon-b-${userId}`
const timestamp = DateTime.now().toMillis()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, anonA)
.withEvent('seed a')
.withProperties({ $set: { from_a: true } })
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, anonB)
.withEvent('seed b')
.withProperties({ $set: { from_b: true } })
.withTimestamp(timestamp)
.build(),
],
token
)
)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, userId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonA, $set: { i1: true } })
.withTimestamp(timestamp + 10)
.build(),
new EventBuilder(team, userId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonB, $set: { i2: true } })
.withTimestamp(timestamp + 20)
.build(),
new EventBuilder(team, userId)
.withEvent('after event')
.withTimestamp(timestamp + 30)
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(5)
})
const events = await fetchEvents(datastore, team.id)
const identify1 = events.find(
(event) => event.event === '$identify' && event.properties.$anon_distinct_id === anonA
)!
const identify2 = events.find(
(event) => event.event === '$identify' && event.properties.$anon_distinct_id === anonB
)!
const after = events.find((event) => event.event === 'after event')!
// The first identify always has its own source's properties and
// $set, and never a later event's $set.
expect(identify1.person_properties).toEqual(expect.objectContaining({ from_a: true, i1: true }))
expect(identify1.person_properties).not.toHaveProperty('i2')
if (pipelineConfig.PERSON_MERGE_FOLD_ENABLED) {
// Folded: B's person properties landed with the fold, before
// identify1's row was emitted.
expect(identify1.person_properties).toEqual(expect.objectContaining({ from_b: true }))
} else {
// Sequential: B's person merges only at identify2.
expect(identify1.person_properties).not.toHaveProperty('from_b')
}
// From the second identify onward the arms are indistinguishable.
expect(identify2.person_properties).toEqual(
expect.objectContaining({ from_a: true, from_b: true, i1: true, i2: true })
)
expect(after.person_properties).toEqual(
expect.objectContaining({ from_a: true, from_b: true, i1: true, i2: true })
)
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons).toHaveLength(1)
expect(persons[0].is_identified).toBe(true)
expect(persons[0].properties).toEqual(
expect.objectContaining({ from_a: true, from_b: true, i1: true, i2: true })
)
}
)
testWithTeamIngester(
'$identify and $set events force person property updates even for filtered properties',
{},
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// Chain of events: normal event with filtered props, then $identify with filtered props, then another normal event
await ingester.handleKafkaBatch(
createKafkaMessages(
[
// Event 1: Normal pageview with filtered properties (should be ignored on its own)
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: { $browser: 'Chrome', utm_source: 'google' },
})
.withTimestamp(timestamp)
.build(),
// Event 2: $identify with ONLY filtered properties (should force update)
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { $browser: 'Safari', $geoip_city_name: 'San Francisco' },
})
.withTimestamp(timestamp + 1)
.build(),
// Event 3: Another normal event with filtered properties
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: { utm_source: 'facebook' },
})
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
// Event 0 (first pageview): Should have properties from first event only
expect(events[0].person_properties).toEqual({
$browser: 'Chrome',
$creator_event_uuid: events[0].uuid,
utm_source: 'google',
})
// Event 1 ($identify): Should have accumulated properties from first two events
expect(events[1].person_properties).toEqual({
$browser: 'Safari', // Updated by $identify
$creator_event_uuid: events[0].uuid,
utm_source: 'google', // Still from first event
$geoip_city_name: 'San Francisco', // Added by $identify
})
// Event 2 (last pageview): Should have all accumulated properties
expect(events[2].person_properties).toEqual({
$browser: 'Safari', // From $identify
$creator_event_uuid: events[0].uuid,
utm_source: 'facebook', // Updated by this event
$geoip_city_name: 'San Francisco', // From $identify
})
// Verify the final state of the person in the database
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual({
$browser: 'Safari',
$creator_event_uuid: events[0].uuid,
utm_source: 'facebook',
$geoip_city_name: 'San Francisco',
})
// last_seen_at should be set
expect(person!.last_seen_at).toBeDefined()
expect(person!.last_seen_at!.minute).toBe(0)
expect(person!.last_seen_at!.second).toBe(0)
})
}
)
testWithTeamIngester(
'$set events force person property updates even for filtered properties',
{},
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// Similar test but with $set event instead of $identify
await ingester.handleKafkaBatch(
createKafkaMessages(
[
// Event 1: Normal event with filtered properties
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: { $browser: 'Chrome' },
})
.withTimestamp(timestamp)
.build(),
// Event 2: $set event with filtered properties (should force update)
new EventBuilder(team, distinctId)
.withEvent('$set')
.withProperties({
$set: { utm_source: 'twitter', $geoip_country_code: 'US' },
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
// Event 0 (first pageview): Should have properties from first event only
expect(events[0].person_properties).toEqual({
$browser: 'Chrome',
$creator_event_uuid: events[0].uuid,
})
// Event 1 ($set): Should have accumulated properties from both events
expect(events[1].person_properties).toEqual({
$browser: 'Chrome', // From first event
$creator_event_uuid: events[0].uuid,
utm_source: 'twitter', // From $set event
$geoip_country_code: 'US', // From $set event
})
// Verify the final state of the person in the database
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual({
$browser: 'Chrome',
$creator_event_uuid: events[0].uuid,
utm_source: 'twitter',
$geoip_country_code: 'US',
})
// last_seen_at should be set to hour-rounded timestamp
expect(person!.last_seen_at).toBeDefined()
expect(person!.last_seen_at!.minute).toBe(0)
expect(person!.last_seen_at!.second).toBe(0)
})
}
)
testWithTeamIngester(
'PERSON_PROPERTIES_UPDATE_ALL flag enables updates for normally filtered properties',
{ pluginServerConfig: { PERSON_PROPERTIES_UPDATE_ALL: true } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// With PERSON_PROPERTIES_UPDATE_ALL=true, even normal pageview events should update
// filtered properties like $browser, $geoip_*, etc.
await ingester.handleKafkaBatch(
createKafkaMessages(
[
// Event 1: Normal pageview that creates the person with initial properties
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: { $browser: 'Chrome', $geoip_city_name: 'New York' },
})
.withTimestamp(timestamp)
.build(),
// Event 2: Another normal pageview with updated filtered properties
// With the flag enabled, these should trigger a person update
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: { $browser: 'Safari', $geoip_city_name: 'San Francisco' },
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
// Event 0 (first pageview): Should have initial properties
expect(events[0].person_properties).toEqual({
$browser: 'Chrome',
$creator_event_uuid: events[0].uuid,
$geoip_city_name: 'New York',
})
// Event 1 (second pageview): Should have UPDATED properties
// This is the key difference - without the flag, these would still show the old values
expect(events[1].person_properties).toEqual({
$browser: 'Safari',
$creator_event_uuid: events[0].uuid,
$geoip_city_name: 'San Francisco',
})
// Verify the final state of the person in the database reflects the updates
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual({
$browser: 'Safari',
$creator_event_uuid: events[0].uuid,
$geoip_city_name: 'San Francisco',
})
})
}
)
testWithTeamIngester(
'allowed geoip properties ($geoip_country_name, $geoip_city_name) trigger person updates alongside blocked geoip properties',
{},
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// When $geoip_country_name or $geoip_city_name changes, all geoip properties in the batch
// should be updated, even the normally-blocked ones like $geoip_latitude
await ingester.handleKafkaBatch(
createKafkaMessages(
[
// Event 1: Create person with initial geoip properties
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: {
$geoip_country_name: 'Canada',
$geoip_city_name: 'Toronto',
$geoip_latitude: 43.6532,
$geoip_longitude: -79.3832,
},
})
.withTimestamp(timestamp)
.build(),
// Event 2: Update geoip properties including allowed ones (country/city)
// Since $geoip_country_name changes, all geoip properties should be updated
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({
$set: {
$geoip_country_name: 'United States',
$geoip_city_name: 'San Francisco',
$geoip_latitude: 37.7749,
$geoip_longitude: -122.4194,
},
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
// Event 0 (first pageview): Should have initial geoip properties
expect(events[0].person_properties).toEqual({
$creator_event_uuid: events[0].uuid,
$geoip_country_name: 'Canada',
$geoip_city_name: 'Toronto',
$geoip_latitude: 43.6532,
$geoip_longitude: -79.3832,
})
// Event 1 (second pageview): Should have UPDATED geoip properties
// Because $geoip_country_name is an allowed property, all geoip properties get updated
expect(events[1].person_properties).toEqual({
$creator_event_uuid: events[0].uuid,
$geoip_country_name: 'United States',
$geoip_city_name: 'San Francisco',
$geoip_latitude: 37.7749,
$geoip_longitude: -122.4194,
})
// Verify the final state of the person in the database reflects the updates
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual({
$creator_event_uuid: events[0].uuid,
$geoip_country_name: 'United States',
$geoip_city_name: 'San Francisco',
$geoip_latitude: 37.7749,
$geoip_longitude: -122.4194,
})
})
}
)
testWithTeamIngester(
'can handle events with $process_person_profile=false',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
distinct_id: distinctId,
$set: { prop: 'value' },
})
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('custom event')
.withProperties({
distinctId: distinctId,
$process_person_profile: false,
$group_0: 'group_key',
$set: {
c: 3,
},
$set_once: {
d: 4,
},
$unset: ['prop'],
})
.withOverrides({
$set: {
a: 1,
},
$set_once: {
b: 2,
},
})
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('custom event')
.withProperties({})
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[1].event).toEqual('custom event')
expect(events[1].person_properties).toEqual({})
expect(events[2].event).toEqual('custom event')
expect(events[2].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
})
}
)
testWithTeamIngester(
'force_upgrade triggers when personless event sent after person creation',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
distinct_id: distinctId,
$set: { prop: 'value' },
})
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('custom event')
.withProperties({
$process_person_profile: false,
})
.withTimestamp(timestamp + 120000) // 2 minutes later (beyond 1-minute grace period)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_mode).toEqual('full')
expect(events[1].event).toEqual('custom event')
expect(events[1].person_mode).toEqual('force_upgrade')
expect(events[1].person_properties).toEqual({})
})
}
)
testWithTeamIngester(
'can handle multiple personless events for different users in same batch',
{},
async ({ ingester, team, token }) => {
const distinctId1 = new UUIDT().toString()
const distinctId2 = new UUIDT().toString()
const distinctId3 = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId1)
.withEvent('pageview')
.withProperties({ $process_person_profile: false })
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId2)
.withEvent('pageview')
.withProperties({ $process_person_profile: false })
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId3)
.withEvent('pageview')
.withProperties({ $process_person_profile: false })
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events.every((e) => e.event === 'pageview')).toBe(true)
expect(events.every((e) => e.person_mode === 'propertyless')).toBe(true)
expect(events.every((e) => Object.keys(e.person_properties).length === 0)).toBe(true)
})
}
)
testWithTeamIngester(
'can handle repeated personless events for same user in same batch',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('pageview')
.withProperties({ $process_person_profile: false, page: '/home' })
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('pageview')
.withProperties({ $process_person_profile: false, page: '/about' })
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('pageview')
.withProperties({ $process_person_profile: false, page: '/contact' })
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events.every((e) => e.person_mode === 'propertyless')).toBe(true)
// All events should have the same deterministic person UUID
const personIds = events.map((e) => e.person_id)
expect(new Set(personIds).size).toBe(1)
})
}
)
testWithTeamIngester(
'can $set and update person properties with top level $set',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
distinct_id: distinctId,
})
.withOverrides({
$set: { prop: 'value' },
})
.build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(1)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
})
}
)
testWithTeamIngester(
'should guarantee that person properties are set in the order of the events',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
const events = [
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({ $set: { prop: 'value' } })
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('custom event')
.withProperties({})
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('custom event')
.withProperties({
$set: {
prop: 'updated value',
new_prop: 'new value',
},
})
.withTimestamp(timestamp + 2)
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[1].event).toEqual('custom event')
expect(events[1].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[2].event).toEqual('custom event')
expect(events[2].person_properties).toEqual(
expect.objectContaining({ prop: 'updated value', new_prop: 'new value' })
)
})
}
)
testWithTeamIngester(
'should be able to $set_once person properties but not update',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const events = [
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({ $set_once: { prop: 'value' } })
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({ $set_once: { prop: 'updated value' } })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[1].event).toEqual('$identify')
expect(events[1].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
})
}
)
testWithTeamIngester(
'should be able to $set_once person properties but not update, at the top level',
{},
async ({ ingester, team, token }) => {
const distinctId = new UUIDT().toString()
const events = [
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({})
.withOverrides({ $set_once: { prop: 'value' } })
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({})
.withOverrides({ $set_once: { prop: 'updated value' } })
.build(),
new EventBuilder(team, distinctId).withEvent('custom event').withProperties({}).build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(events[0].event).toEqual('$identify')
expect(events[0].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[1].event).toEqual('$identify')
expect(events[1].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
expect(events[2].event).toEqual('custom event')
expect(events[2].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
})
}
)
testWithTeamIngester(
'should identify previous events with $anon_distinct_id',
{},
async ({ ingester, team, token }) => {
const initialDistinctId = new UUIDT().toString()
const personIdentifier = '[email protected]'
const events = [
new EventBuilder(team, initialDistinctId).withEvent('custom event').withProperties({}).build(),
new EventBuilder(team, personIdentifier)
.withEvent('$identify')
.withProperties({
distinct_id: personIdentifier,
$anon_distinct_id: initialDistinctId,
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
expect(events[0].person_id).toEqual(events[1].person_id)
})
}
)
testWithTeamIngester(
'should perserve all events if merge fails',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const illegalDistinctId = '0'
const distinctId = new UUIDT().toString()
const events = [
new EventBuilder(team, illegalDistinctId).withEvent('custom event').withProperties({}).build(),
new EventBuilder(team, distinctId).withEvent('custom event 2').withProperties({}).build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPersons(team.id)
expect(persons.length).toEqual(2)
})
const mergeEvents = [
new EventBuilder(team, distinctId)
.withEvent('$merge_dangerously')
.withProperties({
distinct_id: distinctId,
alias: illegalDistinctId,
$set: { prop: 'value' },
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(mergeEvents, token))
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
// Assert that there are 2 different persons in person_id column
const personIds = new Set(events.map((event) => event.person_id))
expect(personIds.size).toEqual(2)
})
}
)
testWithTeamIngester('should preserve properties if merge fails', {}, async ({ ingester, team, token }) => {
const illegalDistinctId = '0'
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$merge_dangerously')
.withProperties({
distinct_id: distinctId,
alias: illegalDistinctId,
$set: { prop: 'value' },
})
.build(),
new EventBuilder(team, distinctId).withEvent('custom event').withProperties({}).build(),
],
token
)
)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
expect(events[1].person_properties).toEqual(expect.objectContaining({ prop: 'value' }))
})
})
testWithTeamIngester(
'should merge all events into same person id',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const initialDistinctId = 'id1'
const secondDistinctId = 'id2'
const personIdentifier = 'person_id'
const event1 = new EventBuilder(team, initialDistinctId)
.withEvent('custom event')
.withProperties({})
.build()
const event2 = new EventBuilder(team, secondDistinctId)
.withEvent('custom event 2')
.withProperties({})
.build()
await ingester.handleKafkaBatch(createKafkaMessages([event1, event2], token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPersons(team.id)
expect(persons).toEqual(
expect.arrayContaining([
expect.objectContaining({
properties: expect.objectContaining({
$creator_event_uuid: event1.uuid,
}),
}),
expect.objectContaining({
properties: expect.objectContaining({
$creator_event_uuid: event2.uuid,
}),
}),
])
)
})
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, personIdentifier)
.withEvent('$identify')
.withProperties({
distinct_id: personIdentifier,
$anon_distinct_id: initialDistinctId,
})
.build(),
new EventBuilder(team, personIdentifier)
.withEvent('$identify')
.withProperties({
distinct_id: personIdentifier,
$anon_distinct_id: secondDistinctId,
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(4)
// assert all events have the same person_id
const personIds = new Set(events.map((event) => event.person_id))
expect(personIds.size).toEqual(1)
})
}
)
testWithTeamIngester(
'should resolve to same person id chained merges',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const initialDistinctId = 'initialId'
const secondDistinctId = 'secondId'
const thirdDistinctId = 'thirdId'
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, initialDistinctId).withEvent('custom event').withProperties({}).build(),
new EventBuilder(team, secondDistinctId).withEvent('custom event 2').withProperties({}).build(),
new EventBuilder(team, thirdDistinctId).withEvent('custom event 3').withProperties({}).build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(3)
expect(new Set(events.map((event) => event.person_id)).size).toEqual(3)
})
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({
distinct_id: initialDistinctId,
$anon_distinct_id: secondDistinctId,
})
.build(),
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({
distinct_id: initialDistinctId,
$anon_distinct_id: thirdDistinctId,
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(5)
expect(new Set(events.map((event) => event.person_id)).size).toEqual(1)
})
}
)
testWithTeamIngester(
'should resolve to same person id even with complex chained merges',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const initialDistinctId = new UUIDT().toString()
const secondDistinctId = new UUIDT().toString()
const thirdDistinctId = new UUIDT().toString()
const forthDistinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, initialDistinctId).withEvent('custom event').withProperties({}).build(),
new EventBuilder(team, secondDistinctId).withEvent('custom event 2').withProperties({}).build(),
new EventBuilder(team, thirdDistinctId).withEvent('custom event 3').withProperties({}).build(),
new EventBuilder(team, forthDistinctId).withEvent('custom event 4').withProperties({}).build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPersons(team.id)
expect(persons.length).toBe(4)
})
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({
distinct_id: initialDistinctId,
$anon_distinct_id: secondDistinctId,
})
.build(),
new EventBuilder(team, thirdDistinctId)
.withEvent('$identify')
.withProperties({
distinct_id: thirdDistinctId,
$anon_distinct_id: forthDistinctId,
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(6)
})
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, secondDistinctId)
.withEvent('$merge_dangerously')
.withProperties({
distinct_id: secondDistinctId,
alias: thirdDistinctId,
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(7)
expect(new Set(events.map((event) => event.person_id)).size).toBe(1)
})
}
)
testWithTeamIngester(
'should produce ingestion warnings for messages over 1MB',
{},
async ({ ingester, team, kafkaProducer, token }) => {
// For this we basically want the processor to try and produce a new
// message larger than 1MB. We do this by creating a person with a lot of
// properties. We will end up denormalizing the person properties onto the
// event, which already has the properties as $set therefore resulting in a
// message that's larger than 1MB. There may also be other attributes that
// are added to the event which pushes it over the limit.
//
// We verify that this is handled by checking that there is a message in the
// appropriate topic.
const distinctId = new UUIDT().toString()
const personProperties = {
distinct_id: distinctId,
$set: {} as Record<string, string>,
}
for (let i = 0; i < 10000; i++) {
personProperties.$set[new UUIDT().toString()] = new UUIDT().toString()
}
const events = [
new EventBuilder(team, distinctId).withEvent('$identify').withProperties(personProperties).build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const ingestionWarnings = await fetchIngestionWarnings(datastore, team.id)
expect(ingestionWarnings.length).toBe(1)
expect(ingestionWarnings[0].details.eventUuid).toBe(events[0].uuid)
})
}
)
const fetchPersons = async (teamId: number) => {
const persons = await datastore.fetchPersons(teamId)
return persons.map((person) => ({
...person,
properties: parseJSON(person.properties),
}))
}
// TODO: Re-enable after re-adding FK constraints to insights_persondistinctid
// testWithTeamIngester('alias events ordering scenario 1: original order', {}, async ({ ingester, infra, team, kafkaProducer, token }) => {
// const testName = DateTime.now().toFormat('yyyy-MM-dd-HH-mm-ss')
// const user1DistinctId = 'user1-distinct-id'
// const user2DistinctId = 'user2-distinct-id'
// const user3DistinctId = 'user3-distinct-id'
// const events = [
// // User 1 creation
// new EventBuilder(team, user1DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// name: 'User 1',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// },
// })
// .build(),
// new EventBuilder(team, user1DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// new_name: 'User 1 - Updated',
// },
// })
// .build(),
// // User 2 creation
// new EventBuilder(team, user2DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// name: 'User 2',
// email: `user2-${user2DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// },
// })
// .build(),
// new EventBuilder(team, user2DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// new_name: 'User 2 - Updated',
// },
// })
// .build(),
// // Merge users: alias user1 -> user2
// new EventBuilder(team, user1DistinctId)
// .withEvent('$create_alias')
// .withProperties({
// distinct_id: user1DistinctId,
// alias: user2DistinctId,
// })
// .build(),
// // Create alias for user2 -> user3
// new EventBuilder(team, user2DistinctId)
// .withEvent('$create_alias')
// .withProperties({
// distinct_id: user2DistinctId,
// alias: user3DistinctId,
// })
// .build(),
// ]
// await ingester.handleKafkaBatch(createKafkaMessages(events, token))
// await waitForKafkaMessages(kafkaProducer)
// await waitForExpect(async () => {
// const events = await fetchEvents(datastore, team.id)
// expect(events.length).toBe(6)
// // TODO: Add specific assertions based on expected behavior
// // All events should be processed without errors
// expect(events).toBeDefined()
// })
// // fetch the person properties
// await waitForExpect(async () => {
// const persons = await fetchPostgresPersons(infra.postgres, team.id)
// expect(persons.length).toBe(1)
// const personsDatastore = await fetchPersons(team.id)
// expect(personsDatastore.length).toBe(1)
// expect(persons[0].properties).toMatchObject(
// expect.objectContaining({
// name: 'User 1',
// new_name: 'User 1 - Updated',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// })
// )
// expect(personsDatastore[0].properties).toMatchObject(
// expect.objectContaining({
// name: 'User 1',
// new_name: 'User 1 - Updated',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// })
// )
// const distinctIdsPersons = await fetchDistinctIds(infra.postgres, {
// id: persons[0].id,
// team_id: team.id,
// } as InternalPerson)
// expect(distinctIdsPersons.length).toBe(3)
// // Except distinctids to match the ids, in any order
// expect(distinctIdsPersons.map((distinctId) => distinctId.distinct_id)).toEqual(
// expect.arrayContaining([user1DistinctId, user2DistinctId, user3DistinctId])
// )
// })
// })
// TODO: Re-enable after re-adding FK constraints to insights_persondistinctid
// testWithTeamIngester('alias events ordering scenario 2: alias first', {}, async ({ ingester, infra, team, kafkaProducer, token }) => {
// const testName = DateTime.now().toFormat('yyyy-MM-dd-HH-mm-ss')
// const user1DistinctId = 'user1-distinct-id'
// const user2DistinctId = 'user2-distinct-id'
// const user3DistinctId = 'user3-distinct-id'
// const events = [
// // User 1 creation
// new EventBuilder(team, user1DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// name: 'User 1',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// },
// })
// .build(),
// new EventBuilder(team, user1DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// new_name: 'User 1 - Updated',
// },
// })
// .build(),
// // User 2 creation
// new EventBuilder(team, user2DistinctId)
// .withProperties({
// anon_distinct_id: user2DistinctId,
// $set: {
// name: 'User 2',
// email: `user2-${user2DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// },
// })
// .build(),
// new EventBuilder(team, user2DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// new_name: 'User 2 - Updated',
// },
// })
// .build(),
// // Create alias for user2 -> user3
// new EventBuilder(team, user2DistinctId)
// .withEvent('$create_alias')
// .withProperties({
// distinct_id: user2DistinctId,
// alias: user3DistinctId,
// })
// .build(),
// // Merge users: alias user1 -> user2
// new EventBuilder(team, user1DistinctId)
// .withEvent('$create_alias')
// .withProperties({
// distinct_id: user1DistinctId,
// alias: user2DistinctId,
// })
// .build(),
// ]
// await ingester.handleKafkaBatch(createKafkaMessages(events, token))
// await waitForKafkaMessages(kafkaProducer)
// await waitForExpect(async () => {
// const events = await fetchEvents(datastore, team.id)
// expect(events.length).toBe(6)
// // TODO: Add specific assertions based on expected behavior
// // All events should be processed without errors
// expect(events).toBeDefined()
// })
// // fetch the person properties
// await waitForExpect(async () => {
// const persons = await fetchPostgresPersons(infra.postgres, team.id)
// expect(persons.length).toBe(1)
// const personsDatastore = await fetchPersons(team.id)
// expect(personsDatastore.length).toBe(1)
// expect(persons[0].properties).toMatchObject(
// expect.objectContaining({
// name: 'User 1',
// new_name: 'User 1 - Updated',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// })
// )
// expect(personsDatastore[0].properties).toMatchObject(
// expect.objectContaining({
// name: 'User 1',
// new_name: 'User 1 - Updated',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// })
// )
// const distinctIdsPersons = await fetchDistinctIds(infra.postgres, {
// id: persons[0].id,
// team_id: team.id,
// } as InternalPerson)
// expect(distinctIdsPersons.length).toBe(3)
// // Except distinctids to match the ids, in any order
// expect(distinctIdsPersons.map((distinctId) => distinctId.distinct_id)).toEqual(
// expect.arrayContaining([user1DistinctId, user2DistinctId, user3DistinctId])
// )
// })
// })
// TODO: Re-enable after re-adding FK constraints to insights_persondistinctid
// testWithTeamIngester('alias events ordering scenario 2: user 2 first', {}, async ({ ingester, infra, team, kafkaProducer, token }) => {
// const testName = DateTime.now().toFormat('yyyy-MM-dd-HH-mm-ss')
// const user1DistinctId = 'user1-distinct-id'
// const user2DistinctId = 'user2-distinct-id'
// const user3DistinctId = 'user3-distinct-id'
// const events = [
// // User 2 creation
// new EventBuilder(team, user2DistinctId)
// .withProperties({
// anon_distinct_id: user2DistinctId,
// $set: {
// name: 'User 2',
// email: `user2-${user2DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// },
// })
// .build(),
// new EventBuilder(team, user2DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// new_name: 'User 2 - Updated',
// },
// })
// .build(),
// // Create alias for user2 -> user3
// new EventBuilder(team, user2DistinctId)
// .withEvent('$create_alias')
// .withProperties({
// distinct_id: user2DistinctId,
// alias: user3DistinctId,
// })
// .build(),
// // User 1 creation
// new EventBuilder(team, user1DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// name: 'User 1',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// },
// })
// .build(),
// new EventBuilder(team, user1DistinctId)
// .withEvent('$identify')
// .withProperties({
// $set: {
// new_name: 'User 1 - Updated',
// },
// })
// .build(),
// // Merge users: alias user1 -> user2
// new EventBuilder(team, user1DistinctId)
// .withEvent('$create_alias')
// .withProperties({
// distinct_id: user1DistinctId,
// alias: user2DistinctId,
// })
// .build(),
// ]
// await ingester.handleKafkaBatch(createKafkaMessages(events, token))
// await waitForKafkaMessages(kafkaProducer)
// await waitForExpect(async () => {
// const events = await fetchEvents(datastore, team.id)
// expect(events.length).toBe(6)
// // TODO: Add specific assertions based on expected behavior
// // All events should be processed without errors
// expect(events).toBeDefined()
// })
// // fetch the person properties
// await waitForExpect(async () => {
// const persons = await fetchPostgresPersons(infra.postgres, team.id)
// expect(persons.length).toBe(1)
// const personsDatastore = await fetchPersons(team.id)
// expect(personsDatastore.length).toBe(1)
// expect(persons[0].properties).toMatchObject(
// expect.objectContaining({
// name: 'User 1',
// new_name: 'User 1 - Updated',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// })
// )
// expect(personsDatastore[0].properties).toMatchObject(
// expect.objectContaining({
// name: 'User 1',
// new_name: 'User 1 - Updated',
// email: `user1-${user1DistinctId}@example.com`,
// age: 30,
// test_name: testName,
// })
// )
// const distinctIdsPersons = await fetchDistinctIds(infra.postgres, {
// id: persons[0].id,
// team_id: team.id,
// } as InternalPerson)
// expect(distinctIdsPersons.length).toBe(3)
// // Except distinctids to match the ids, in any order
// expect(distinctIdsPersons.map((distinctId) => distinctId.distinct_id)).toEqual(
// expect.arrayContaining([user1DistinctId, user2DistinctId, user3DistinctId])
// )
// })
// })
testWithTeamIngester(
'alias events ordering scenario 2: user 2 first, separate batch',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const testName = DateTime.now().toFormat('yyyy-MM-dd-HH-mm-ss')
const user1DistinctId = 'user1-distinct-id'
const user2DistinctId = 'user2-distinct-id'
const user3DistinctId = 'user3-distinct-id'
const events = [
// User 2 creation
new EventBuilder(team, user2DistinctId)
.withProperties({
anon_distinct_id: user2DistinctId,
$set: {
name: 'User 2',
email: `user2-${user2DistinctId}@example.com`,
age: 30,
test_name: testName,
},
})
.build(),
new EventBuilder(team, user2DistinctId)
.withEvent('$identify')
.withProperties({
$set: {
new_name: 'User 2 - Updated',
},
})
.build(),
// Create alias for user2 -> user3
new EventBuilder(team, user2DistinctId)
.withEvent('$create_alias')
.withProperties({
distinct_id: user2DistinctId,
alias: user3DistinctId,
})
.build(),
]
const events2 = [
// User 1 creation
new EventBuilder(team, user1DistinctId)
.withEvent('$identify')
.withProperties({
$set: {
name: 'User 1',
email: `user1-${user1DistinctId}@example.com`,
age: 30,
test_name: testName,
},
})
.build(),
new EventBuilder(team, user1DistinctId)
.withEvent('$identify')
.withProperties({
$set: {
new_name: 'User 1 - Updated',
},
})
.build(),
// Merge users: alias user1 -> user2
new EventBuilder(team, user1DistinctId)
.withEvent('$create_alias')
.withProperties({
distinct_id: user1DistinctId,
alias: user2DistinctId,
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await ingester.handleKafkaBatch(createKafkaMessages(events2, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(6)
// TODO: Add specific assertions based on expected behavior
// All events should be processed without errors
expect(events).toBeDefined()
})
// fetch the person properties
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
const personsDatastore = await fetchPersons(team.id)
expect(personsDatastore.length).toBe(2)
expect(persons.map((person) => person.properties)).toEqual(
expect.arrayContaining([
expect.objectContaining({
name: 'User 1',
new_name: 'User 1 - Updated',
email: `user1-${user1DistinctId}@example.com`,
age: 30,
test_name: testName,
}),
expect.objectContaining({
name: 'User 2',
new_name: 'User 2 - Updated',
email: `user2-${user2DistinctId}@example.com`,
age: 30,
test_name: testName,
}),
])
)
expect(personsDatastore.map((person) => person.properties)).toEqual(
expect.arrayContaining([
expect.objectContaining({
name: 'User 1',
new_name: 'User 1 - Updated',
email: `user1-${user1DistinctId}@example.com`,
age: 30,
test_name: testName,
}),
expect.objectContaining({
name: 'User 2',
new_name: 'User 2 - Updated',
email: `user2-${user2DistinctId}@example.com`,
age: 30,
test_name: testName,
}),
])
)
const person1 = persons.find((person) => person.properties.name === 'User 1')!
const person2 = persons.find((person) => person.properties.name === 'User 2')!
const distinctIdsPersons1 = await fetchDistinctIds(infra.postgres, {
id: person1.id,
team_id: team.id,
} as InternalPerson)
expect(distinctIdsPersons1.length).toBe(1)
// Except distinctids to match the ids, in any order
expect(distinctIdsPersons1.map((distinctId) => distinctId.distinct_id)).toEqual(
expect.arrayContaining([user1DistinctId])
)
const distinctIdsPersons2 = await fetchDistinctIds(infra.postgres, {
id: person2.id,
team_id: team.id,
} as InternalPerson)
expect(distinctIdsPersons2.length).toBe(2)
// Except distinctids to match the ids, in any order
expect(distinctIdsPersons2.map((distinctId) => distinctId.distinct_id)).toEqual(
expect.arrayContaining([user2DistinctId, user3DistinctId])
)
})
}
)
testWithTeamIngester(
'Should set and $unset person properties, different batches',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const user1DistinctId = 'user1-distinct-id'
const events = [
new EventBuilder(team, user1DistinctId)
.withEvent('$identify')
.withProperties({
$set: {
name: 'User 1',
property_to_unset: 'value',
},
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const personsDatastore = await fetchPersons(team.id)
expect(personsDatastore.length).toBe(1)
expect(persons[0].properties).toMatchObject(
expect.objectContaining({
name: 'User 1',
property_to_unset: 'value',
})
)
expect(personsDatastore[0].properties).toMatchObject(
expect.objectContaining({
name: 'User 1',
property_to_unset: 'value',
})
)
})
const events2 = [
new EventBuilder(team, user1DistinctId)
.withEvent('$identify')
.withProperties({
$unset: ['property_to_unset'],
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events2, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const personsDatastore = await fetchPersons(team.id)
expect(personsDatastore.length).toBe(1)
expect(persons[0].properties).toMatchObject(
expect.objectContaining({
name: 'User 1',
})
)
expect(persons[0].properties).not.toHaveProperty('property_to_unset')
expect(personsDatastore[0].properties).toMatchObject(
expect.objectContaining({
name: 'User 1',
})
)
expect(personsDatastore[0].properties).not.toHaveProperty('property_to_unset')
})
}
)
testWithTeamIngester(
'Should set and $unset person properties, same batch',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const user1DistinctId = 'user1-distinct-id'
const events = [
new EventBuilder(team, user1DistinctId)
.withEvent('$identify')
.withProperties({
$set: {
name: 'User 1',
property_to_unset: 'value',
},
})
.build(),
new EventBuilder(team, user1DistinctId)
.withEvent('$identify')
.withProperties({
$unset: ['property_to_unset'],
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const personsDatastore = await fetchPersons(team.id)
expect(personsDatastore.length).toBe(1)
expect(persons[0].properties).toMatchObject(
expect.objectContaining({
name: 'User 1',
})
)
expect(persons[0].properties).not.toHaveProperty('property_to_unset')
expect(personsDatastore[0].properties).toMatchObject(
expect.objectContaining({
name: 'User 1',
})
)
expect(personsDatastore[0].properties).not.toHaveProperty('property_to_unset')
})
}
)
testWithTeamIngester(
'should process events with various timestamps when dropping is disabled',
{ teamOverrides: { drop_events_older_than_seconds: null } },
async ({ ingester, team, kafkaProducer, token }) => {
const currentTime = DateTime.now().toMillis()
const events = [
// Current time event
new EventBuilder(team)
.withEvent('current_time_event')
.withTimestamp(currentTime)
.withNow(currentTime)
.build(),
// Recent event (30 minutes ago)
new EventBuilder(team)
.withEvent('recent_event')
.withTimestamp(currentTime - 30 * 60 * 1000)
.withNow(currentTime)
.build(),
// Old event (2 hours ago)
new EventBuilder(team)
.withEvent('old_event')
.withTimestamp(currentTime - 2 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Very old event (1 day ago)
new EventBuilder(team)
.withEvent('very_old_event')
.withTimestamp(currentTime - 24 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Extremely old event (1 month ago)
new EventBuilder(team)
.withEvent('extremely_old_event')
.withTimestamp(currentTime - 30 * 24 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Future event (1 hour from now)
new EventBuilder(team)
.withEvent('future_event')
.withTimestamp(currentTime + 60 * 60 * 1000)
.withNow(currentTime)
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
// All events should be processed when dropping is disabled
expect(events.length).toBe(6)
// Verify all events are present
const eventTypes = events.map((e) => e.event)
expect(eventTypes).toContain('current_time_event')
expect(eventTypes).toContain('recent_event')
expect(eventTypes).toContain('old_event')
expect(eventTypes).toContain('very_old_event')
expect(eventTypes).toContain('extremely_old_event')
expect(eventTypes).toContain('future_event')
})
}
)
testWithTeamIngester(
'should drop old events when dropping is enabled',
{ teamOverrides: { drop_events_older_than_seconds: 3600 } },
async ({ ingester, team, kafkaProducer, token }) => {
const currentTime = DateTime.now().toMillis()
const events = [
// Current time event - should pass
new EventBuilder(team)
.withEvent('current_time_event')
.withTimestamp(currentTime)
.withNow(currentTime)
.build(),
// Recent event (30 minutes ago) - should pass
new EventBuilder(team)
.withEvent('recent_event')
.withTimestamp(currentTime - 30 * 60 * 1000)
.withNow(currentTime)
.build(),
// Just under threshold (59 minutes ago) - should pass
new EventBuilder(team)
.withEvent('just_under_threshold')
.withTimestamp(currentTime - 59 * 60 * 1000)
.withNow(currentTime)
.build(),
// Just over threshold (61 minutes ago) - should be dropped
new EventBuilder(team)
.withEvent('just_over_threshold')
.withTimestamp(currentTime - 61 * 60 * 1000)
.withNow(currentTime)
.build(),
// Old event (2 hours ago) - should be dropped
new EventBuilder(team)
.withEvent('old_event')
.withTimestamp(currentTime - 2 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Very old event (1 day ago) - should be dropped
new EventBuilder(team)
.withEvent('very_old_event')
.withTimestamp(currentTime - 24 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Future event (1 hour from now) - should pass
new EventBuilder(team)
.withEvent('future_event')
.withTimestamp(currentTime + 60 * 60 * 1000)
.withNow(currentTime)
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
// Only events under 1 hour old should be processed
expect(events.length).toBe(4)
// Verify only expected events are present
const eventTypes = events.map((e) => e.event)
expect(eventTypes).toContain('current_time_event')
expect(eventTypes).toContain('recent_event')
expect(eventTypes).toContain('just_under_threshold')
expect(eventTypes).toContain('future_event')
// Verify dropped events are not present
expect(eventTypes).not.toContain('just_over_threshold')
expect(eventTypes).not.toContain('old_event')
expect(eventTypes).not.toContain('very_old_event')
})
// Check that ingestion warnings were created for dropped events
await waitForExpect(async () => {
const warnings = await fetchIngestionWarnings(datastore, team.id)
expect(warnings.length).toBe(3)
const warningTypes = warnings.map((w) => w.type)
expect(warningTypes).toEqual(
expect.arrayContaining(['event_dropped_too_old', 'event_dropped_too_old', 'event_dropped_too_old'])
)
})
}
)
testWithTeamIngester(
'should process all events when drop threshold is set to 0',
{ teamOverrides: { drop_events_older_than_seconds: 0 } },
async ({ ingester, team, kafkaProducer, token }) => {
const currentTime = DateTime.now().toMillis()
const events = [
// Current time event - should pass
new EventBuilder(team)
.withEvent('current_time_event')
.withTimestamp(currentTime)
.withNow(currentTime)
.build(),
// Recent event (30 minutes ago) - should pass
new EventBuilder(team)
.withEvent('recent_event')
.withTimestamp(currentTime - 30 * 60 * 1000)
.withNow(currentTime)
.build(),
// Old event (2 hours ago) - should pass (0 threshold means no dropping)
new EventBuilder(team)
.withEvent('old_event')
.withTimestamp(currentTime - 2 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Very old event (1 day ago) - should pass (0 threshold means no dropping)
new EventBuilder(team)
.withEvent('very_old_event')
.withTimestamp(currentTime - 24 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Extremely old event (1 month ago) - should pass (0 threshold means no dropping)
new EventBuilder(team)
.withEvent('extremely_old_event')
.withTimestamp(currentTime - 30 * 24 * 60 * 60 * 1000)
.withNow(currentTime)
.build(),
// Future event (1 hour from now) - should pass
new EventBuilder(team)
.withEvent('future_event')
.withTimestamp(currentTime + 60 * 60 * 1000)
.withNow(currentTime)
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
// All events should be processed when threshold is 0 (no dropping)
expect(events.length).toBe(6)
// Verify all events are present
const eventTypes = events.map((e) => e.event)
expect(eventTypes).toContain('current_time_event')
expect(eventTypes).toContain('recent_event')
expect(eventTypes).toContain('old_event')
expect(eventTypes).toContain('very_old_event')
expect(eventTypes).toContain('extremely_old_event')
expect(eventTypes).toContain('future_event')
})
// Check that no ingestion warnings were created (no events should be dropped)
await waitForExpect(async () => {
const warnings = await fetchIngestionWarnings(datastore, team.id)
expect(warnings.length).toBe(0)
})
}
)
testWithTeamIngester(
'we do not alias users if distinct id changes but we are already identified, with no anonymous event',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// This test is similar to the previous one, except it does not include an initial anonymous event.
const anonymousId = 'anonymous_id'
const initialDistinctId = 'initial_distinct_id'
const p2DistinctId = 'p2_distinct_id'
const p2NewDistinctId = 'new_distinct_id'
// Play out a sequence of events that should result in two users being
// identified, with the first to events associated with one user, and
// the third with another.
const events = [
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
new EventBuilder(team, initialDistinctId).withEvent('event 2').build(),
new EventBuilder(team, p2DistinctId).withEvent('event 3').build(),
new EventBuilder(team, p2NewDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: p2DistinctId })
.build(),
new EventBuilder(team, p2NewDistinctId).withEvent('event 4').build(),
// Let's also make sure that we do not alias when switching back to initialDistictId
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: p2NewDistinctId })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
// Both persons should be identified
expect(persons.map((person) => person.is_identified)).toEqual([true, true])
// Check that events are grouped correctly by person
const events = await fetchEvents(datastore, team.id)
const eventsByPersonId = new Map<string, string[]>()
for (const event of events) {
const personId = event.person_id!
if (!eventsByPersonId.has(personId)) {
eventsByPersonId.set(personId, [])
}
eventsByPersonId.get(personId)!.push(event.event)
}
expect(eventsByPersonId.size).toBe(2)
const eventGroups = Array.from(eventsByPersonId.values()).sort((a, b) => a.length - b.length)
expect(eventGroups).toEqual(
expect.arrayContaining([
expect.arrayContaining(['event 3', '$identify', 'event 4']),
expect.arrayContaining(['$identify', 'event 2', '$identify']),
])
)
})
}
)
testWithTeamIngester(
'we do not leave things in inconsistent state if $identify is run concurrently',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// There are a few places where we have the pattern of:
//
// 1. fetch from postgres
// 2. check rows match condition
// 3. perform update
//
// This test is designed to check the specific case where, in
// handling we are creating an unidentified user, then updating this
// user to have is_identified = true. Since we are using the
// is_identified to decide on if we will merge persons, we want to
// make sure we guard against this race condition.
const anonymousId = 'anonymous_id'
const initialDistinctId = 'initial-distinct-id'
const newDistinctId = 'new-distinct-id'
const events = [
new EventBuilder(team, newDistinctId).withEvent('some event').build(),
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
new EventBuilder(team, newDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
expect(persons.map((person) => person.is_identified)).toEqual([true, true])
})
}
)
testWithTeamIngester(
'we can alias an anonymous person to an anonymous person',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const anonymous1 = 'anonymous-1'
const anonymous2 = 'anonymous-2'
const events = [
new EventBuilder(team, anonymous1).withEvent('anonymous event 1').build(),
new EventBuilder(team, anonymous2).withEvent('anonymous event 2').build(),
// Then try to alias them
new EventBuilder(team, anonymous2)
.withEvent('$create_alias')
.withProperties({ alias: anonymous1 })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
// Make sure there is one identified person
expect(persons.map((person) => person.is_identified)).toEqual([true])
// Check that events are grouped correctly by person
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(3)
// All events should belong to the same person
const personIds = new Set(events.map((e) => e.person_id))
expect(personIds.size).toBe(1)
const eventNames = events.map((e) => e.event).sort()
expect(eventNames).toEqual(['$create_alias', 'anonymous event 1', 'anonymous event 2'])
})
}
)
testWithTeamIngester(
'we can alias two non-existent persons',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const anonymous1 = 'anonymous-1'
const anonymous2 = 'anonymous-2'
const events = [
new EventBuilder(team, anonymous1)
.withEvent('$create_alias')
.withProperties({ alias: anonymous2 })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].is_identified).toBe(true)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([anonymous1, anonymous2])
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
expect(events[0].event).toBe('$create_alias')
})
}
)
testWithTeamIngester(
'$merge_dangerously merges two persons into one',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const oldDistinctId = 'old_distinct_id'
const newDistinctId = 'new_distinct_id'
// Create a person for old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, oldDistinctId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
})
// Send $merge_dangerously from new_distinct_id with alias: old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, newDistinctId)
.withEvent('$merge_dangerously')
.withProperties({ alias: oldDistinctId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([newDistinctId, oldDistinctId])
})
}
)
testWithTeamIngester(
'$create_alias links existing person to new distinct id',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const oldDistinctId = 'old_distinct_id'
const newDistinctId = 'new_distinct_id'
// Create a person for old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, oldDistinctId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
})
// Send $create_alias from new_distinct_id with alias: old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, newDistinctId)
.withEvent('$create_alias')
.withProperties({ alias: oldDistinctId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([newDistinctId, oldDistinctId])
})
}
)
testWithTeamIngester(
'alias works in reverse direction',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const oldDistinctId = 'old_distinct_id'
const newDistinctId = 'new_distinct_id'
// Create a person for old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, oldDistinctId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Send $create_alias from old_distinct_id with alias: new_distinct_id (reversed)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, oldDistinctId)
.withEvent('$create_alias')
.withProperties({ alias: newDistinctId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([newDistinctId, oldDistinctId])
})
}
)
testWithTeamIngester(
'chained alias merges three persons into one',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const oldDistinctId = 'old_distinct_id'
const newDistinctId = 'new_distinct_id'
const oldDistinctId2 = 'old_distinct_id_2'
// Create person for old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, oldDistinctId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// First alias: link new_distinct_id to old_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, newDistinctId)
.withEvent('$create_alias')
.withProperties({ alias: oldDistinctId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
})
// Create a second person
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, oldDistinctId2).withEvent('custom event 2').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
})
// Second alias: merge old_distinct_id_2 into new_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, newDistinctId)
.withEvent('$create_alias')
.withProperties({ alias: oldDistinctId2 })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([
newDistinctId,
oldDistinctId,
oldDistinctId2,
])
})
}
)
testWithTeamIngester(
'$create_alias before any person exists creates one person with both distinct ids',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$create_alias')
.withProperties({ alias: 'old_distinct_id' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual(['new_distinct_id', 'old_distinct_id'])
})
}
)
testWithTeamIngester(
'$create_alias merges two existing persons into one',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// Create two separate persons
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'old_distinct_id').withEvent('event 1').build(),
new EventBuilder(team, 'new_distinct_id').withEvent('event 2').build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
})
// Alias them together
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$create_alias')
.withProperties({ alias: 'old_distinct_id' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual(['new_distinct_id', 'old_distinct_id'])
})
}
)
testWithTeamIngester(
'alias merges person properties correctly',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const newDistinctId = 'new_distinct_id'
const oldDistinctId = 'old_distinct_id'
// Create two persons with different properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, newDistinctId)
.withEvent('$pageview')
.withProperties({
$set: { key_on_both: 'new value both', key_on_new: 'new value' },
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, oldDistinctId)
.withEvent('$pageview')
.withProperties({
$set: { key_on_both: 'old value both', key_on_old: 'old value' },
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
expect(persons.map((p) => p.is_identified)).toEqual([false, false])
})
// Alias them: new_distinct_id's values should win for shared keys
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, newDistinctId)
.withEvent('$create_alias')
.withProperties({ alias: oldDistinctId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([newDistinctId, oldDistinctId])
expect(persons[0].properties).toEqual(
expect.objectContaining({
key_on_both: 'new value both',
key_on_new: 'new value',
key_on_old: 'old value',
})
)
})
}
)
testWithTeamIngester(
'$identify with illegal distinct ids does not merge persons',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const anonymousId = 'im-an-anonymous-id'
// Create a person for the anonymous ID
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, anonymousId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
})
// Try to $identify with various illegal IDs — each should fail to merge
const illegalIds = ['0', 'None', 'undefined', 'NaN', ' ', 'distinctId']
for (const illegalId of illegalIds) {
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, illegalId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
}
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
// 1 original + 5 illegal IDs each created their own person (merges failed)
expect(persons.length).toBe(1 + illegalIds.length)
})
// 'Nan' is a legal ID (only 'NaN' is illegal), so the merge should work
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'Nan')
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
// No extra person created because 'Nan' merged with the anonymous person
expect(persons.length).toBe(1 + illegalIds.length)
})
}
)
testWithTeamIngester(
'$create_alias with illegal distinct id does not merge persons',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const legalId = 'user123'
const illegalId = 'null'
// Create a person for the legal ID
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, legalId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Try $create_alias with an illegal alias
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, illegalId)
.withEvent('$create_alias')
.withProperties({ alias: legalId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
// Person with illegal ID got created but not merged
expect(persons.length).toBe(2)
})
}
)
// This case is likely to happen after signup, for example:
// 1. User browses website with anonymous_id
// 2. User signs up, triggers event with their new_distinct_id (creating a new Person)
// 3. In the frontend, try to alias anonymous_id with new_distinct_id
// Result should be that we end up with one Person with both ID's
testWithTeamIngester(
'$identify merges anonymous person into existing person preserving properties',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// Create two persons: one anonymous, one with properties
await ingester.handleKafkaBatch(
createKafkaMessages([new EventBuilder(team, 'anonymous_id').withEvent('anon event').build()], token)
)
await waitForKafkaMessages(kafkaProducer)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$identify')
.withProperties({ $set: { email: '[email protected]' } })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
})
// $identify merges anonymous_id into new_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$identify')
.withProperties({ $anon_distinct_id: 'anonymous_id' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual(['anonymous_id', 'new_distinct_id'])
expect(persons[0].properties).toEqual(expect.objectContaining({ email: '[email protected]' }))
expect(persons[0].is_identified).toBe(true)
})
}
)
testWithTeamIngester(
'$identify where distinct_id equals $anon_distinct_id is a no-op',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const anonymousId = 'anonymous_id'
// Create a person for anonymous_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[new EventBuilder(team, anonymousId).withEvent('custom event').withProperties({}).build()],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Send $identify where distinct_id == $anon_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, anonymousId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id)).toEqual([anonymousId])
// Person should NOT be marked as identified (self-identify is a no-op)
expect(persons[0].is_identified).toBe(false)
})
}
)
testWithTeamIngester(
'$identify merges multiple anonymous persons sequentially preserving properties',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// Create anonymous person and identified person with properties
await ingester.handleKafkaBatch(
createKafkaMessages([new EventBuilder(team, 'anonymous_id').withEvent('anon event').build()], token)
)
await waitForKafkaMessages(kafkaProducer)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$identify')
.withProperties({ $set: { email: '[email protected]' } })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// First $identify: merge anonymous_id into new_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$identify')
.withProperties({ $anon_distinct_id: 'anonymous_id' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].properties).toEqual(expect.objectContaining({ email: '[email protected]' }))
expect(persons[0].is_identified).toBe(true)
})
// Create a second anonymous person
await ingester.handleKafkaBatch(
createKafkaMessages([new EventBuilder(team, 'anonymous_id_2').withEvent('anon event 2').build()], token)
)
await waitForKafkaMessages(kafkaProducer)
// Second $identify: merge anonymous_id_2 into new_distinct_id
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'new_distinct_id')
.withEvent('$identify')
.withProperties({ $anon_distinct_id: 'anonymous_id_2' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual([
'anonymous_id',
'anonymous_id_2',
'new_distinct_id',
])
expect(persons[0].properties).toEqual(expect.objectContaining({ email: '[email protected]' }))
expect(persons[0].is_identified).toBe(true)
})
}
)
testWithTeamIngester(
'$identify does not merge persons across teams',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// Create a second team
const team2Id = Math.floor((Date.now() % 1000000000) + Math.random() * 1000000)
await createUserTeamAndOrganization(
infra.postgres,
team2Id,
team2Id,
new UUIDT().toString(),
new UUIDT().toString(),
new UUIDT().toString()
)
const team2 = (await infra.teamManager.getTeam(team2Id))!
// Create a person for team2 with distinct_id '2'
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team2, '2')
.withEvent('$identify')
.withProperties({ $set: { email: '[email protected]' } })
.build(),
],
team2.api_token
)
)
await waitForKafkaMessages(kafkaProducer)
// Create persons for team1 with distinct_ids '1' and '2'
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, '1').withEvent('custom event').withProperties({}).build(),
new EventBuilder(team, '2').withEvent('custom event').withProperties({}).build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Merge '1' into '2' on team1
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, '2')
.withEvent('$identify')
.withProperties({ $anon_distinct_id: '1' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Verify team1 merged correctly
await waitForExpect(async () => {
const team1Persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(team1Persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, team1Persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual(['1', '2'])
})
// Verify team2's person is untouched
await waitForExpect(async () => {
const team2Persons = await fetchPostgresPersons(infra.postgres, team2Id)
expect(team2Persons.length).toBe(1)
expect(team2Persons[0].properties).toEqual(expect.objectContaining({ email: '[email protected]' }))
const distinctIds = await fetchDistinctIds(infra.postgres, team2Persons[0])
expect(distinctIds.map((d) => d.distinct_id)).toEqual(['2'])
})
}
)
testWithTeamIngester(
'we do not alias users if distinct id changes but we are already identified',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// This test is in reference to
// https://github.com/Insights/insights/issues/5527 , where we were
// correctly identifying that an anonymous user before login should be
// aliased to the user they subsequently login as, but incorrectly
// aliasing on subsequent $identify events. The anonymous case is
// special as we want to alias to a known user, but otherwise we
// shouldn't be doing so.
const anonymousId = 'anonymous_id'
const initialDistinctId = 'initial_distinct_id'
const p2DistinctId = 'p2_distinct_id'
const p2NewDistinctId = 'new_distinct_id'
// Batch 1: anonymous event, identify, event, second person event, second identify, event
const events = [
new EventBuilder(team, anonymousId).withEvent('event 1').build(),
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
new EventBuilder(team, initialDistinctId).withEvent('event 2').build(),
new EventBuilder(team, p2DistinctId).withEvent('event 3').build(),
new EventBuilder(team, p2NewDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: p2DistinctId })
.build(),
new EventBuilder(team, p2NewDistinctId).withEvent('event 4').build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
expect(persons.every((p) => p.is_identified)).toBe(true)
})
// Batch 2: try to alias back to initialDistinctId (should fail — both already identified)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: p2NewDistinctId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
// Still 2 separate persons — merge was rejected because both are identified
expect(persons.length).toBe(2)
expect(persons.every((p) => p.is_identified)).toBe(true)
})
}
)
testWithTeamIngester(
'we do not alias already identified users with no anonymous event',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// This test is in reference to
// https://github.com/Insights/insights/issues/5527 , where we were
// correctly identifying that an anonymous user before login should be
// aliased to the user they subsequently login as, but incorrectly
// aliasing on subsequent $identify events. The anonymous case is
// special as we want to alias to a known user, but otherwise we
// shouldn't be doing so. This test is similar to the previous one,
// except it does not include an initial anonymous event.
const anonymousId = 'anonymous_id'
const initialDistinctId = 'initial_distinct_id'
const p2DistinctId = 'p2_distinct_id'
const p2NewDistinctId = 'new_distinct_id'
const events = [
// First user: identify (no prior anonymous event)
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: anonymousId })
.build(),
new EventBuilder(team, initialDistinctId).withEvent('event 2').build(),
// Second user
new EventBuilder(team, p2DistinctId).withEvent('event 3').build(),
new EventBuilder(team, p2NewDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: p2DistinctId })
.build(),
new EventBuilder(team, p2NewDistinctId).withEvent('event 4').build(),
// Try to alias back to initialDistinctId (should fail — both already identified)
new EventBuilder(team, initialDistinctId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: p2NewDistinctId })
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
expect(persons.map((p) => p.is_identified)).toEqual([true, true])
const allEvents = await fetchEvents(datastore, team.id)
const eventsByPersonId = new Map<string, string[]>()
for (const event of allEvents) {
const personId = event.person_id!
if (!eventsByPersonId.has(personId)) {
eventsByPersonId.set(personId, [])
}
eventsByPersonId.get(personId)!.push(event.event)
}
expect(eventsByPersonId.size).toBe(2)
const eventGroups = Array.from(eventsByPersonId.values()).sort((a, b) => a.length - b.length)
expect(eventGroups).toEqual(
expect.arrayContaining([
expect.arrayContaining(['event 3', '$identify', 'event 4']),
expect.arrayContaining(['$identify', 'event 2', '$identify']),
])
)
})
}
)
testWithTeamIngester(
'$create_alias merges anonymous person into identified person',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const someAnonId = 'some_anon_id'
const identifiedId = 'identified_id'
const anonymousId = 'anonymous_id'
// Create a truly identified person (via real $identify merge)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, identifiedId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: someAnonId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].is_identified).toBe(true)
})
// Create an anonymous person
await ingester.handleKafkaBatch(
createKafkaMessages([new EventBuilder(team, anonymousId).withEvent('anonymous event').build()], token)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
})
// $create_alias: alias person (anonymous) is not identified → merge allowed
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, identifiedId)
.withEvent('$create_alias')
.withProperties({ alias: anonymousId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
const distinctIds = await fetchDistinctIds(infra.postgres, persons[0])
expect(distinctIds.map((d) => d.distinct_id).sort()).toEqual(
[someAnonId, anonymousId, identifiedId].sort()
)
expect(persons[0].is_identified).toBe(true)
})
}
)
testWithTeamIngester(
'$create_alias does not merge when alias person is already identified',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const someAnonId = 'some_anon_id'
const identifiedId = 'identified_id'
const anonymousId = 'anonymous_id'
// Create a truly identified person (via real $identify merge)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, identifiedId)
.withEvent('$identify')
.withProperties({ $anon_distinct_id: someAnonId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Create an anonymous person
await ingester.handleKafkaBatch(
createKafkaMessages([new EventBuilder(team, anonymousId).withEvent('anonymous event').build()], token)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(2)
})
// $create_alias: alias person (identifiedId) IS identified → merge blocked
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, anonymousId)
.withEvent('$create_alias')
.withProperties({ alias: identifiedId })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
// Merge blocked: alias person (identifiedId) has is_identified=true
expect(persons.length).toBe(2)
})
}
)
testWithTeamIngester(
'person and group properties are set on events',
{},
async ({ ingester, team, kafkaProducer, token }) => {
// Create person with properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('$whatever')
.withProperties({ $set: { pineapple: 'on', pizza: 1 } })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Set up group types
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('$groupidentify')
.withProperties({
$group_type: 'organization',
$group_key: 'org:5',
$group_set: { foo: 'bar' },
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('$groupidentify')
.withProperties({
$group_type: 'second',
$group_key: 'second_key',
$group_set: { pineapple: 'yummy' },
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
// Send a regular event with group references and $set
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('test event')
.withProperties({
$set: { new: 5 },
$group_0: 'org:5',
$group_1: 'second_key',
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
const testEvent = events.find((e) => e.event === 'test event')
expect(testEvent).toBeDefined()
expect(testEvent!.person_properties).toEqual(
expect.objectContaining({ pineapple: 'on', pizza: 1, new: 5 })
)
expect(testEvent!.properties.$group_0).toEqual('org:5')
expect(testEvent!.properties.$group_1).toEqual('second_key')
})
}
)
testWithTeamIngester(
'$set wins over $set_once on the same key',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('some_event')
.withProperties({
$set: { a_prop: 'test-set' },
$set_once: { a_prop: 'test-set_once' },
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
expect(events[0].properties['$set']).toEqual({ a_prop: 'test-set' })
expect(events[0].properties['$set_once']).toEqual({ a_prop: 'test-set_once' })
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].properties).toEqual(expect.objectContaining({ a_prop: 'test-set' }))
})
}
)
testWithTeamIngester(
'$unset removes person properties',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// Create person with properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('$identify')
.withProperties({ $set: { a: 1, b: 2, c: 3 } })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].properties).toEqual(expect.objectContaining({ a: 1, b: 2, c: 3 }))
})
// $unset a and c
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, 'distinct_id1')
.withEvent('some_event')
.withProperties({ $unset: ['a', 'c'] })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
const unsetEvent = events.find((e) => e.event === 'some_event')
expect(unsetEvent).toBeDefined()
expect(unsetEvent!.properties['$unset']).toEqual(['a', 'c'])
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].properties).toEqual(expect.objectContaining({ b: 2 }))
expect(persons[0].properties).not.toHaveProperty('a')
expect(persons[0].properties).not.toHaveProperty('c')
})
}
)
testWithTeamIngester(
'$set and $set_once apply correctly across sequential events',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
// key encodes when the value is updated, e.g. s0 means only set call for the 0th event
// s03o23 means via a set in events 0 and 3 plus via set_once on 2nd and 3rd event
const set0 = { s0123o0123: 's0a', s02o13: 's0b', s013: 's0e' }
const setOnce0 = { s0123o0123: 'o0a', s13o02: 'o0g', o023: 'o0f' }
const set1 = { s0123o0123: 's1a', s13o02: 's1g', s1: 's1c', s013: 's1e' }
const setOnce1 = { s0123o0123: 'o1a', s02o13: 'o1b', o1: 'o1d' }
const set2 = { s0123o0123: 's2a', s02o13: 's2b' }
const setOnce2 = { s0123o0123: 'o2a', s13o02: 'o2g', o023: 'o2f' }
const set3 = { s0123o0123: 's3a', s13o02: 's3g', s013: 's3e' }
const setOnce3 = { s0123o0123: 'o3a', s02o13: 'o3b', o023: 'o3f' }
const distinctId = 'distinct_id1'
// Send 4 events sequentially, each with different $set/$set_once
for (const [set, setOnce] of [
[set0, setOnce0],
[set1, setOnce1],
[set2, setOnce2],
[set3, setOnce3],
]) {
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('some_event')
.withProperties({ $set: set, $set_once: setOnce })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
}
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
// $set wins: last write for each key
// $set_once wins: first write for each key (only if not already set by $set)
expect(persons[0].properties).toEqual(
expect.objectContaining({
s0123o0123: 's3a',
s02o13: 's2b',
s1: 's1c',
o1: 'o1d',
s013: 's3e',
o023: 'o0f',
s13o02: 's3g',
})
)
})
}
)
testWithTeamIngester(
'Should not share cached person data between batches',
{},
async ({ ingester, infra, team, kafkaProducer, token }) => {
const distinctId = 'user-across-batches'
// First batch: Create a person with initial properties
const batch1Events = [
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { batch1_prop: 'value1', shared_prop: 'batch1' },
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(batch1Events, token))
await waitForKafkaMessages(kafkaProducer)
// Verify batch 1 wrote correctly
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
expect(persons[0].properties).toMatchObject({
batch1_prop: 'value1',
shared_prop: 'batch1',
})
})
// Second batch: Update the same person with new properties
// If the store wasn't reset, it might use stale cached data
const batch2Events = [
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { batch2_prop: 'value2', shared_prop: 'batch2' },
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(batch2Events, token))
await waitForKafkaMessages(kafkaProducer)
// Verify batch 2 wrote correctly and merged with existing properties
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
// Should have properties from both batches, with batch2 overwriting shared_prop
expect(persons[0].properties).toMatchObject({
batch1_prop: 'value1',
batch2_prop: 'value2',
shared_prop: 'batch2',
})
})
// Third batch: Verify a third batch also works correctly
const batch3Events = [
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { batch3_prop: 'value3' },
})
.build(),
]
await ingester.handleKafkaBatch(createKafkaMessages(batch3Events, token))
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const persons = await fetchPostgresPersons(infra.postgres, team.id)
expect(persons.length).toBe(1)
// Should have properties from all three batches
expect(persons[0].properties).toMatchObject({
batch1_prop: 'value1',
batch2_prop: 'value2',
batch3_prop: 'value3',
shared_prop: 'batch2',
})
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should correctly apply properties_to_set when updating person',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First event: Create a person with initial properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { initial_prop: 'initial_value' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
// Wait for person to be created
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
initial_prop: 'initial_value',
})
)
})
// Second event: Update the person with new properties via $identify
// This is where the bug manifests - properties_to_set is not applied in ASSERT_VERSION mode
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { email: '[email protected]', name: 'Test User' },
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
// Verify the properties were updated in the database
// BUG: With ASSERT_VERSION mode, properties_to_set is not merged into properties
// before writing to database, so only the original properties are written
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
// This assertion will FAIL with the bug - email and name won't be set
expect(person!.properties).toEqual(
expect.objectContaining({
initial_prop: 'initial_value',
email: '[email protected]',
name: 'Test User',
})
)
})
// Also verify the event has the correct person properties attached
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toEqual(2)
// The second event should have accumulated properties
expect(events[1].person_properties).toEqual(
expect.objectContaining({
initial_prop: 'initial_value',
email: '[email protected]',
name: 'Test User',
})
)
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should correctly apply $set_once for new properties',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First event: Create a person
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { name: 'Test User' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
name: 'Test User',
})
)
})
// Second event: Use $set_once for a new property
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: { first_seen: '2024-01-01' },
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
// Verify the $set_once property was added
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
name: 'Test User',
first_seen: '2024-01-01',
})
)
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should preserve $set_once semantics across multiple events',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First event: Create person with $set_once for prop1
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: { prop1: 'first_value' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop1: 'first_value',
})
)
})
// Second event: Try to $set_once prop1 again (should NOT overwrite) and add prop2
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: { prop1: 'second_value_should_be_ignored', prop2: 'new_property' },
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
// Verify prop1 retains first value, prop2 is added
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop1: 'first_value', // Should NOT be overwritten
prop2: 'new_property', // New property should be added
})
)
})
// Third event: Another $set_once with prop1 (still should not change) and prop3
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: { prop1: 'third_value_also_ignored', prop3: 'another_new' },
})
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
// Final verification
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop1: 'first_value', // Still the original value
prop2: 'new_property',
prop3: 'another_new',
})
)
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should correctly apply $unset operations',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First event: Create a person with properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: {
name: 'Test User',
email: '[email protected]',
to_remove: 'will be removed',
},
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
name: 'Test User',
email: '[email protected]',
to_remove: 'will be removed',
})
)
})
// Second event: Unset a property
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$unset: ['to_remove'],
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
// Verify the property was removed
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
name: 'Test User',
email: '[email protected]',
})
)
expect(person!.properties).not.toHaveProperty('to_remove')
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should correctly apply properties after person merge',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const anonDistinctId = new UUIDT().toString()
const identifiedDistinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First event: Create anonymous person with properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, anonDistinctId)
.withEvent('pageview')
.withProperties({
$set: { anon_prop: 'anon_value' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, anonDistinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
anon_prop: 'anon_value',
})
)
})
// Second event: Identify with merge and set new properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, identifiedDistinctId)
.withEvent('$identify')
.withProperties({
$anon_distinct_id: anonDistinctId,
$set: { email: '[email protected]', name: 'Identified User' },
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
// Verify merged person has all properties
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, identifiedDistinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
anon_prop: 'anon_value',
email: '[email protected]',
name: 'Identified User',
})
)
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should correctly apply properties from multiple events in same batch',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First: Create person
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { initial: 'value' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
})
// Send multiple events in same batch that each set different properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop1: 'value1' },
})
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop2: 'value2' },
})
.withTimestamp(timestamp + 2)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop3: 'value3' },
})
.withTimestamp(timestamp + 3)
.build(),
],
token
)
)
// Verify all properties from the batch were applied
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
initial: 'value',
prop1: 'value1',
prop2: 'value2',
prop3: 'value3',
})
)
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should handle combined $set and $unset in same event',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First event: Create a person with properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { keep_prop: 'keep', remove_prop: 'remove' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
keep_prop: 'keep',
remove_prop: 'remove',
})
)
})
// Second event: Set new property AND unset existing property in same event
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { new_prop: 'new_value' },
$unset: ['remove_prop'],
})
.withTimestamp(timestamp + 1)
.build(),
],
token
)
)
// Verify the new property was added and the old one was removed
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
keep_prop: 'keep',
new_prop: 'new_value',
})
)
expect(person!.properties).not.toHaveProperty('remove_prop')
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should handle $unset overlaps within a batch',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// First: Create person with multiple properties
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop1: 'value1', prop2: 'value2', prop3: 'value3', prop4: 'value4' },
})
.withTimestamp(timestamp)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop1: 'value1',
prop2: 'value2',
prop3: 'value3',
prop4: 'value4',
})
)
})
// Send batch with overlapping $unset operations
await ingester.handleKafkaBatch(
createKafkaMessages(
[
// First event unsets prop1 and prop2
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$unset: ['prop1', 'prop2'],
})
.withTimestamp(timestamp + 1)
.build(),
// Second event also tries to unset prop2 (overlap) and unsets prop3
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$unset: ['prop2', 'prop3'],
})
.withTimestamp(timestamp + 2)
.build(),
// Third event sets a new property and unsets prop2 again (already unset)
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { new_prop: 'new_value' },
$unset: ['prop2'],
})
.withTimestamp(timestamp + 3)
.build(),
],
token
)
)
// Verify all unsets were applied correctly despite overlaps
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop4: 'value4', // Only prop4 should remain from original
new_prop: 'new_value', // New property added
})
)
expect(person!.properties).not.toHaveProperty('prop1')
expect(person!.properties).not.toHaveProperty('prop2')
expect(person!.properties).not.toHaveProperty('prop3')
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should preserve $set_once semantics within same batch',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// Send all events in same batch - $set_once should only apply first value
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: { prop1: 'first_value', prop2: 'first_prop2' },
})
.withTimestamp(timestamp)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: { prop1: 'second_value_ignored', prop3: 'first_prop3' },
})
.withTimestamp(timestamp + 1)
.build(),
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set_once: {
prop1: 'third_value_ignored',
prop2: 'second_prop2_ignored',
prop4: 'first_prop4',
},
})
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop1: 'first_value', // First $set_once wins
prop2: 'first_prop2', // First $set_once wins
prop3: 'first_prop3', // New property from second event
prop4: 'first_prop4', // New property from third event
})
)
})
}
)
testWithTeamIngester(
'ASSERT_VERSION mode should handle combined $set and $unset within same batch',
{ pluginServerConfig: { PERSON_BATCH_WRITING_DB_WRITE_MODE: 'ASSERT_VERSION' } },
async ({ ingester, infra, team, token }) => {
const distinctId = new UUIDT().toString()
const timestamp = DateTime.now().toMillis()
// All operations in same batch including person creation
await ingester.handleKafkaBatch(
createKafkaMessages(
[
// First event creates person with initial properties
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop1: 'value1', prop2: 'value2', prop3: 'value3' },
})
.withTimestamp(timestamp)
.build(),
// Second event sets new property and unsets prop1
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop4: 'value4' },
$unset: ['prop1'],
})
.withTimestamp(timestamp + 1)
.build(),
// Third event updates prop2 and unsets prop3
new EventBuilder(team, distinctId)
.withEvent('$identify')
.withProperties({
$set: { prop2: 'updated_value2' },
$unset: ['prop3'],
})
.withTimestamp(timestamp + 2)
.build(),
],
token
)
)
await waitForExpect(async () => {
const person = await infra.personRepository.fetchPerson(team.id, distinctId)
expect(person).toBeDefined()
expect(person!.properties).toEqual(
expect.objectContaining({
prop2: 'updated_value2', // Updated by third event
prop4: 'value4', // Added by second event
})
)
expect(person!.properties).not.toHaveProperty('prop1') // Unset by second event
expect(person!.properties).not.toHaveProperty('prop3') // Unset by third event
})
}
)
testWithTeamIngester(
'should scrub IPs when team.anonymize_ips=true',
{ teamOverrides: { anonymize_ips: true } },
async ({ ingester, team, kafkaProducer, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$pageview')
.withProperties({ $ip: '11.12.13.14' })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
expect(events[0].properties.$ip).toBeUndefined()
})
}
)
testWithTeamIngester(
'should use $elements_chain string directly as elements_chain',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$autocapture')
.withProperties({ $elements_chain: 'button:nth-child="1"', a: 1 })
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
expect(events[0].elements_chain).toBeDefined()
expect(events[0].properties.$elements_chain).toBeUndefined()
expect(events[0].properties.a).toBe(1)
})
}
)
testWithTeamIngester(
'should convert $elements array to elements_chain',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$autocapture')
.withProperties({
$elements: [{ tag_name: 'div', nth_child: 1, nth_of_type: 2, $el_text: 'text' }],
a: 1,
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
expect(events[0].elements_chain).toBeDefined()
expect(events[0].properties.$elements).toBeUndefined()
expect(events[0].properties.a).toBe(1)
})
}
)
testWithTeamIngester(
'should prefer $elements_chain over $elements when both are present',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$autocapture')
.withProperties({
$elements_chain: 'span:nth-child="1"',
$elements: [{ tag_name: 'div', nth_child: 1, nth_of_type: 2, $el_text: 'text' }],
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
// elements_chain is parsed back into element objects by parseRawDatastoreEvent
// so we check the first element's tag_name to verify $elements_chain was used
expect(events[0].elements_chain).toBeDefined()
expect(events[0].elements_chain![0].tag_name).toBe('span')
expect(events[0].properties.$elements_chain).toBeUndefined()
expect(events[0].properties.$elements).toBeUndefined()
})
}
)
testWithTeamIngester(
'should route $ai_generation events through AI subpipeline with full enrichment',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$ai_generation')
.withProperties({
$ai_model: 'gpt-4',
$ai_provider: 'openai',
$ai_input_tokens: 100,
$ai_output_tokens: 50,
$ai_trace_id: 12345,
$ai_parent_id: 67890,
$ai_model_parameters: {
temperature: 0.7,
stream: true,
max_tokens: 1024,
},
$ai_output_choices: JSON.stringify([
{
role: 'assistant',
message: {
tool_calls: [
{ function: { name: 'get_weather' } },
{ function: { name: 'search_docs' } },
],
},
},
]),
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
const props = events[0].properties
// Cost calculation: input/output/total costs from token counts + model pricing
expect(props.$ai_input_cost_usd).toBeGreaterThan(0)
expect(props.$ai_output_cost_usd).toBeGreaterThan(0)
expect(props.$ai_total_cost_usd).toBeGreaterThan(0)
expect(props.$ai_total_cost_usd).toBe(props.$ai_input_cost_usd + props.$ai_output_cost_usd)
// Cost model metadata
expect(props.$ai_model_cost_used).toBeDefined()
expect(props.$ai_cost_model_provider).toBeDefined()
// Trace property normalization: numeric IDs converted to strings
expect(props.$ai_trace_id).toBe('12345')
expect(props.$ai_parent_id).toBe('67890')
// Model parameter extraction: promoted to top-level properties
expect(props.$ai_temperature).toBe(0.7)
expect(props.$ai_stream).toBe(true)
expect(props.$ai_max_tokens).toBe(1024)
// Tool call extraction: parsed from $ai_output_choices
expect(props.$ai_tools_called).toBe('get_weather,search_docs')
expect(props.$ai_tool_call_count).toBe(2)
})
}
)
testWithTeamIngester(
'should route $ai_trace events through AI subpipeline with trace normalization',
{},
async ({ ingester, team, kafkaProducer, token }) => {
const distinctId = new UUIDT().toString()
await ingester.handleKafkaBatch(
createKafkaMessages(
[
new EventBuilder(team, distinctId)
.withEvent('$ai_trace')
.withProperties({
$ai_trace_id: 'trace-abc',
$ai_span_id: 99999,
$ai_session_id: true,
})
.build(),
],
token
)
)
await waitForKafkaMessages(kafkaProducer)
await waitForExpect(async () => {
const events = await fetchEvents(datastore, team.id)
expect(events.length).toBe(1)
const props = events[0].properties
// Trace normalization applied to all AI events
expect(props.$ai_trace_id).toBe('trace-abc')
expect(props.$ai_span_id).toBe('99999')
expect(props.$ai_session_id).toBe('true')
// No cost enrichment for $ai_trace events
expect(props.$ai_input_cost_usd).toBeUndefined()
expect(props.$ai_output_cost_usd).toBeUndefined()
expect(props.$ai_total_cost_usd).toBeUndefined()
})
}
)
})
Reference in New Issue View Git Blame Copy Permalink