insights/nodejs/tests/ingestion/person-updates-e2e.test.ts

/**
 * Person Updates E2E Tests
 *
 * This test suite verifies that person creation and updates work correctly
 * across all combinations of configuration flags:
 *
 * - PERSON_BATCH_WRITING_DB_WRITE_MODE: 'NO_ASSERT' | 'ASSERT_VERSION'
 * - PERSON_BATCH_WRITING_USE_BATCH_UPDATES: true | false (only applies to NO_ASSERT)
 * - PERSONS_PREFETCH_ENABLED: true | false
 *
 * The goal is to ensure basic person operations work correctly regardless of
 * which flag combination is used, catching regressions early.
 */
import { DateTime } from 'luxon'
import { Message } from 'node-rdkafka'
import { v4 } from 'uuid'

import { createScriptTransformerService } from '~/cdp/script-transformations/script-transformer.service'
import { DatastoreGroupRepository } from '~/common/groups/repositories/datastore-group-repository'
import { KafkaProducerWrapper } from '~/common/kafka/producer'
import { UUIDT } from '~/common/utils/utils'
import { PersonBatchWritingDbWriteMode } from '~/ingestion/config'
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 { IngestionTestInfra, createIngestionTestInfra } from '~/tests/helpers/ingestion-e2e'
import { createTestIngestionOutputs, createTestMonitoringOutputs } from '~/tests/helpers/ingestion-outputs'
import { TEST_KAFKA_TOPICS, ensureKafkaTopics } from '~/tests/helpers/kafka'
import { createUserTeamAndOrganization, resetTestDatabase } from '~/tests/helpers/sql'
import { PipelineEvent, ProjectId, Team } from '~/types'

jest.mock('~/common/utils/token-bucket', () => {
    const mockConsume = jest.fn().mockReturnValue(true)
    return {
        IngestionWarningLimiter: {
            consume: mockConsume,
        },
    }
})

jest.mock('~/common/utils/logger')

const DEFAULT_TEAM: Team = {
    id: 1,
    project_id: 1 as ProjectId,
    uuid: 'team-uuid',
    organization_id: 'org-id',
    name: 'Test Team',
    anonymize_ips: false,
    api_token: 'test-token',
    secret_api_token: null,
    session_recording_opt_in: false,
    heatmaps_opt_in: null,
    ingested_event: true,
    person_display_name_properties: [],
    minimal_flag_called_events: false,
    person_processing_opt_out: null,
    test_account_filters: [],
    timezone: 'UTC',
    cookieless_server_hash_mode: null,
    available_features: [],
    drop_events_older_than_seconds: null,
    extra_settings: { person_last_seen_at_enabled: true },
}

class EventBuilder {
    private event: Partial<PipelineEvent> = {}

    constructor(team: Team, distinctId: string = new UUIDT().toString()) {
        this.event = {
            event: 'custom event',
            properties: {},
            timestamp: new Date().toISOString(),
            now: new Date().toISOString(),
            ip: null,
            site_url: 'https://example.com',
            uuid: new UUIDT().toString(),
        }
        this.event.distinct_id = distinctId
        this.event.team_id = team.id
    }

    withEvent(event: string) {
        this.event.event = event
        return this
    }

    withProperties(properties: Record<string, any>) {
        this.event.properties = properties
        return this
    }

    withTimestamp(timestamp: number) {
        const date = DateTime.fromMillis(timestamp)
        this.event.timestamp = date.toString()
        this.event.now = date.toString()
        return this
    }

    build(): PipelineEvent {
        return this.event as PipelineEvent
    }
}

let offsetIncrementer = 0
let currentToken: string

const createKafkaMessage = (event: PipelineEvent, timestamp: number = Date.now()): Message => {
    const token = currentToken
    // TRICKY: This is the slightly different format that capture sends
    const captureEvent = {
        uuid: event.uuid,
        distinct_id: event.distinct_id,
        ip: event.ip,
        now: event.now,
        token,
        data: JSON.stringify(event),
    }
    const headers: { [key: string]: Buffer }[] = [
        { token: Buffer.from(token) },
        { distinct_id: Buffer.from(event.distinct_id!) },
    ]
    return {
        key: `${token}:${event.distinct_id}`,
        value: Buffer.from(JSON.stringify(captureEvent)),
        size: 1,
        topic: 'test',
        offset: offsetIncrementer++,
        timestamp: timestamp + offsetIncrementer,
        partition: 1,
        headers,
    }
}

const createKafkaMessages = (events: PipelineEvent[]): Message[] => {
    return events.map(createKafkaMessage)
}

const waitForKafkaMessages = async (kafkaProducer: KafkaProducerWrapper) => {
    await kafkaProducer.flush()
}

// All possible values for each flag
const DB_WRITE_MODES: PersonBatchWritingDbWriteMode[] = ['NO_ASSERT', 'ASSERT_VERSION']
const USE_BATCH_UPDATES_OPTIONS = [true, false]
const PREFETCH_OPTIONS = [true, false]

interface PersonUpdateConfig {
    PERSON_BATCH_WRITING_DB_WRITE_MODE: PersonBatchWritingDbWriteMode
    PERSON_BATCH_WRITING_USE_BATCH_UPDATES: boolean
    PERSONS_PREFETCH_ENABLED: boolean
}

// Generate all combinations of all flags
const FLAG_COMBINATIONS: PersonUpdateConfig[] = DB_WRITE_MODES.flatMap((dbWriteMode) =>
    USE_BATCH_UPDATES_OPTIONS.flatMap((useBatchUpdates) =>
        PREFETCH_OPTIONS.map((prefetch) => ({
            PERSON_BATCH_WRITING_DB_WRITE_MODE: dbWriteMode,
            PERSON_BATCH_WRITING_USE_BATCH_UPDATES: useBatchUpdates,
            PERSONS_PREFETCH_ENABLED: prefetch,
        }))
    )
)

const formatConfigName = (config: PersonUpdateConfig): string => {
    const mode = config.PERSON_BATCH_WRITING_DB_WRITE_MODE
    const batch = config.PERSON_BATCH_WRITING_USE_BATCH_UPDATES ? 'batch' : 'individual'
    const prefetch = config.PERSONS_PREFETCH_ENABLED ? 'prefetch' : 'no-prefetch'
    return `${mode}, ${batch}, ${prefetch}`
}

describe.each(FLAG_COMBINATIONS)('Person Updates E2E ($#)', (config) => {
    const configName = formatConfigName(config)
    let datastore: Datastore
    let infra: IngestionTestInfra
    let kafkaProducer: KafkaProducerWrapper
    let ingester: IngestionConsumer
    let team: Team

    beforeAll(async () => {
        datastore = Datastore.create()
        await ensureKafkaTopics(TEST_KAFKA_TOPICS)
        await resetTestDatabase()
        await datastore.resetTestDatabase()
    })

    afterAll(async () => {
        await resetTestDatabase()
        await datastore.resetTestDatabase()
        datastore.close()
    })

    beforeEach(async () => {
        infra = await createIngestionTestInfra({
            ...config,
        })
        kafkaProducer = await KafkaProducerWrapper.create(infra.config.KAFKA_CLIENT_RACK)

        const teamId = Math.floor((Date.now() % 1000000000) + Math.random() * 1000000)
        const userId = teamId
        const organizationId = new UUIDT().toString()
        const userUuid = new UUIDT().toString()
        const organizationMembershipId = new UUIDT().toString()

        team = {
            ...DEFAULT_TEAM,
            id: teamId,
            project_id: teamId as ProjectId,
            organization_id: organizationId,
            uuid: v4(),
            name: teamId.toString(),
        }

        await createUserTeamAndOrganization(
            infra.postgres,
            team.id,
            userId,
            userUuid,
            team.organization_id,
            organizationMembershipId,
            { extra_settings: { person_last_seen_at_enabled: true } }
        )

        const fetchedTeam = await infra.teamManager.getTeam(team.id)
        if (!fetchedTeam) {
            throw new Error(`Failed to fetch team ${team.id} from database`)
        }
        team = fetchedTeam
        currentToken = team.api_token

        const outputs = createTestIngestionOutputs(kafkaProducer)
        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,
            aiSubpipelineFactory: createAiEventSubpipeline,
            scriptTransformer: createScriptTransformerService(infra.config, {
                geoipService: infra.geoipService,
                postgres: infra.postgres,
                pubSub: infra.pubSub,
                encryptedFields: infra.encryptedFields,
                integrationManager: infra.integrationManager,
                monitoringOutputs: createTestMonitoringOutputs(kafkaProducer),
            }),
            outputs,
            datastoreGroupRepository: new DatastoreGroupRepository(outputs),
        })
        ingester['kafkaConsumer'] = {
            connect: jest.fn(),
            disconnect: jest.fn(),
            isHealthy: jest.fn(),
        } as any

        await ingester.start()
    })

    afterEach(async () => {
        await ingester.stop()
        await kafkaProducer.disconnect()
        await infra.close()
    })

    describe(configName, () => {
        it('should create a new person on first event', async () => {
            const distinctId = new UUIDT().toString()

            await ingester.handleKafkaBatch(
                createKafkaMessages([new EventBuilder(team, distinctId).withEvent('test_event').build()])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.team_id).toBe(team.id)
                // last_seen_at should be set to an 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)
                expect(person!.last_seen_at!.millisecond).toBe(0)
            })
        })

        it('should set person properties with $identify and $set', async () => {
            const distinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()
            const expectedLastSeenAt = DateTime.fromMillis(timestamp).startOf('hour')

            // Create person with initial properties
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { name: 'Initial Name', email: '[email protected]' },
                        })
                        .withTimestamp(timestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.properties).toEqual(
                    expect.objectContaining({
                        name: 'Initial Name',
                        email: '[email protected]',
                    })
                )
                // last_seen_at should be set to the hour-rounded event timestamp
                expect(person!.last_seen_at).toBeDefined()
                expect(person!.last_seen_at!.toMillis()).toBe(expectedLastSeenAt.toMillis())
            })
        })

        it('should update person properties across multiple events in same batch', async () => {
            const distinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()

            // Send multiple events in a single batch
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { prop1: 'value1' },
                        })
                        .withTimestamp(timestamp)
                        .build(),
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { prop2: 'value2' },
                        })
                        .withTimestamp(timestamp + 1)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            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',
                    })
                )
            })
        })

        it('should update person properties across multiple batches', async () => {
            const distinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()

            // First batch: create person
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { initial_prop: 'initial_value' },
                        })
                        .withTimestamp(timestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            // 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 batch: update person
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { new_prop: 'new_value', updated_prop: 'updated' },
                        })
                        .withTimestamp(timestamp + 1000)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            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',
                        new_prop: 'new_value',
                        updated_prop: 'updated',
                    })
                )
            })
        })

        it('should handle $set_once correctly', async () => {
            const distinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()

            // First event sets initial value
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set_once: { first_seen: 'original_value' },
                        })
                        .withTimestamp(timestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.properties).toEqual(
                    expect.objectContaining({
                        first_seen: 'original_value',
                    })
                )
            })

            // Second event tries to overwrite with $set_once - should be ignored
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set_once: { first_seen: 'should_be_ignored' },
                        })
                        .withTimestamp(timestamp + 1000)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                // Value should remain unchanged
                expect(person!.properties.first_seen).toBe('original_value')
            })
        })

        it('should handle $unset correctly', async () => {
            const distinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()

            // Create person with properties
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { keep_prop: 'keep', remove_prop: 'remove' },
                        })
                        .withTimestamp(timestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.properties).toHaveProperty('remove_prop')
            })

            // Unset a property
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $unset: ['remove_prop'],
                        })
                        .withTimestamp(timestamp + 1000)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.properties).toEqual(
                    expect.objectContaining({
                        keep_prop: 'keep',
                    })
                )
                expect(person!.properties).not.toHaveProperty('remove_prop')
            })
        })

        it('should handle combined $set and $unset in same event', async () => {
            const distinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()

            // Create person with initial properties
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { prop1: 'value1', prop2: 'value2', prop3: 'value3' },
                        })
                        .withTimestamp(timestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            // Update with combined $set and $unset
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $set: { prop1: 'updated_value1', prop4: 'value4' },
                            $unset: ['prop2'],
                        })
                        .withTimestamp(timestamp + 1000)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.properties).toEqual(
                    expect.objectContaining({
                        prop1: 'updated_value1',
                        prop3: 'value3',
                        prop4: 'value4',
                    })
                )
                expect(person!.properties).not.toHaveProperty('prop2')
            })
        })

        it('should update last_seen_at when event timestamp is in a newer hour', async () => {
            const distinctId = new UUIDT().toString()
            // Start at the beginning of an hour to make assertions clearer
            const baseTime = DateTime.now().startOf('hour')
            const firstTimestamp = baseTime.toMillis()
            const secondTimestamp = baseTime.plus({ hours: 2 }).toMillis()

            // First event creates the person
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({ $set: { initial: true } })
                        .withTimestamp(firstTimestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.last_seen_at).toBeDefined()
                expect(person!.last_seen_at!.toMillis()).toBe(baseTime.toMillis())
            })

            // Second event 2 hours later should update last_seen_at
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId).withEvent('pageview').withTimestamp(secondTimestamp).build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.last_seen_at).toBeDefined()
                expect(person!.last_seen_at!.toMillis()).toBe(baseTime.plus({ hours: 2 }).toMillis())
            })
        })

        it('should not update last_seen_at when $update_person_last_seen_at is false', async () => {
            const distinctId = new UUIDT().toString()
            const baseTime = DateTime.now().startOf('hour')
            const firstTimestamp = baseTime.toMillis()
            const secondTimestamp = baseTime.plus({ hours: 2 }).toMillis()

            // First event creates the person
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({ $set: { initial: true } })
                        .withTimestamp(firstTimestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.last_seen_at!.toMillis()).toBe(baseTime.toMillis())
            })

            // Second event 2 hours later with $update_person_last_seen_at=false should NOT update last_seen_at
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('pageview')
                        .withProperties({ $update_person_last_seen_at: false })
                        .withTimestamp(secondTimestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                // last_seen_at should remain unchanged
                expect(person!.last_seen_at!.toMillis()).toBe(baseTime.toMillis())
            })
        })

        it('should set is_identified to true when merging via $identify with $anon_distinct_id', async () => {
            const anonDistinctId = new UUIDT().toString()
            const identifiedDistinctId = new UUIDT().toString()
            const timestamp = DateTime.now().toMillis()

            // First, create an anonymous person
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, anonDistinctId).withEvent('pageview').withTimestamp(timestamp).build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, anonDistinctId)
                expect(person).toBeDefined()
                expect(person!.is_identified).toBe(false)
            })

            // Then identify and merge via $anon_distinct_id
            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, identifiedDistinctId)
                        .withEvent('$identify')
                        .withProperties({
                            $anon_distinct_id: anonDistinctId,
                            $set: { email: '[email protected]' },
                        })
                        .withTimestamp(timestamp + 1000)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                // After merge, the person should be identified and accessible via the identified distinct ID
                const person = await infra.personRepository.fetchPerson(team.id, identifiedDistinctId)
                expect(person).toBeDefined()
                expect(person!.is_identified).toBe(true)
            })
        })
    })

    describe(`${configName} - person_last_seen_at_enabled disabled`, () => {
        beforeEach(async () => {
            const disabledTeamId = Math.floor((Date.now() % 1000000000) + Math.random() * 1000000)
            const disabledUserId = disabledTeamId
            const disabledUserUuid = new UUIDT().toString()
            const disabledOrgId = new UUIDT().toString()
            const disabledOrgMembershipId = new UUIDT().toString()

            await createUserTeamAndOrganization(
                infra.postgres,
                disabledTeamId,
                disabledUserId,
                disabledUserUuid,
                disabledOrgId,
                disabledOrgMembershipId
            )

            const fetchedTeam = await infra.teamManager.getTeam(disabledTeamId)
            if (!fetchedTeam) {
                throw new Error(`Failed to fetch team ${disabledTeamId} from database`)
            }
            team = fetchedTeam
            currentToken = team.api_token
        })

        it('should not update last_seen_at when person_last_seen_at_enabled is not set', async () => {
            const distinctId = new UUIDT().toString()
            const baseTime = DateTime.now().startOf('hour')
            const firstTimestamp = baseTime.toMillis()
            const secondTimestamp = baseTime.plus({ hours: 2 }).toMillis()

            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId)
                        .withEvent('$identify')
                        .withProperties({ $set: { initial: true } })
                        .withTimestamp(firstTimestamp)
                        .build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            let initialLastSeenAt: number | undefined
            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                initialLastSeenAt = person!.last_seen_at?.toMillis()
            })

            await ingester.handleKafkaBatch(
                createKafkaMessages([
                    new EventBuilder(team, distinctId).withEvent('pageview').withTimestamp(secondTimestamp).build(),
                ])
            )

            await waitForKafkaMessages(kafkaProducer)

            await waitForExpect(async () => {
                const person = await infra.personRepository.fetchPerson(team.id, distinctId)
                expect(person).toBeDefined()
                expect(person!.last_seen_at?.toMillis()).toBe(initialLastSeenAt)
            })
        })
    })
})