fix(ingestion): skip $experiment_exposure duplication for v2 flag rules calls

nodejs/src/ingestion/common/steps/event-processing/create-event-step.test.ts

@@ -1,5 +1,6 @@
 import { DateTime } from 'luxon'
 import { Message } from 'node-rdkafka'
+import { v5 as uuidv5 } from 'uuid'
 
 import { EVENTS_OUTPUT } from '~/common/outputs'
 import { castTimestampOrNow } from '~/common/utils/utils'
@@ -350,6 +351,7 @@ describe('create-event-step', () => {
                 ['an empty response', 2, '$feature_flag_called', ''],
                 ['a missing response', 2, '$feature_flag_called', undefined],
                 ['a different event', 2, '$pageview', 'test'],
+                ['a direct $experiment_exposure', 2, '$experiment_exposure', 'test'],
             ])('does not duplicate %s', async (_, teamId, event, response) => {
                 const step = createCreateEventStep(EVENTS_OUTPUT, '2')
                 const result = await step({
@@ -374,6 +376,78 @@ describe('create-event-step', () => {
                 }
             })
 
+            it.each([
+                ['the number 2', 2],
+                ['the string "2"', '2'],
+            ])(
+                'passes a $feature_flag_called through unchanged when $feature_flag_config_version is %s',
+                async (_, configVersion) => {
+                    const properties = {
+                        $feature_flag: 'checkout-experiment',
+                        $feature_flag_response: 'test',
+                        $feature_flag_config_version: configVersion,
+                    }
+                    const step = createCreateEventStep(EVENTS_OUTPUT, '*')
+                    const result = await step({
+                        team: mockTeam,
+                        person: mockPerson,
+                        preparedEvent: {
+                            ...mockPreparedEvent,
+                            teamId: 2,
+                            projectId: 2 as ProjectId,
+                            event: '$feature_flag_called',
+                            properties,
+                        },
+                        processPerson: true,
+                        historicalMigration: false,
+                        headers: createTestEventHeaders(),
+                        message: mockMessage,
+                    })
+
+                    expect(isOkResult(result)).toBe(true)
+                    if (isOkResult(result)) {
+                        expect(result.value.eventsToEmit).toHaveLength(1)
+                        const [flagCalled] = result.value.eventsToEmit
+                        expect(flagCalled.output).toBe(EVENTS_OUTPUT)
+                        expect(flagCalled.event.event).toBe('$feature_flag_called')
+                        expect(flagCalled.event.uuid).toBe('event-uuid-456')
+                        expect(flagCalled.event.properties).toEqual(properties)
+                    }
+                }
+            )
+
+            it.each([
+                ['1', 1],
+                ['missing', undefined],
+                ['the string "1"', '1'],
+            ])('duplicates as before when $feature_flag_config_version is %s', async (_, configVersion) => {
+                const step = createCreateEventStep(EVENTS_OUTPUT, '2')
+                const result = await step({
+                    team: mockTeam,
+                    person: mockPerson,
+                    preparedEvent: {
+                        ...mockPreparedEvent,
+                        teamId: 2,
+                        projectId: 2 as ProjectId,
+                        event: '$feature_flag_called',
+                        properties: { $feature_flag_response: 'test', $feature_flag_config_version: configVersion },
+                    },
+                    processPerson: true,
+                    historicalMigration: false,
+                    headers: createTestEventHeaders(),
+                    message: mockMessage,
+                })
+
+                expect(isOkResult(result)).toBe(true)
+                if (isOkResult(result)) {
+                    expect(result.value.eventsToEmit).toHaveLength(2)
+                    const exposure = result.value.eventsToEmit[1].event
+                    expect(exposure.event).toBe('$experiment_exposure')
+                    // Pins the derived uuid so a redelivered message keeps collapsing in ClickHouse.
+                    expect(exposure.uuid).toBe(uuidv5('event-uuid-456', '1b7c9119-5953-4668-97b7-ab0ef8a6bb48'))
+                }
+            })
+
             it('does not duplicate when no teams are configured', async () => {
                 const step = createCreateEventStep(EVENTS_OUTPUT)
                 const result = await step({

nodejs/src/ingestion/common/steps/event-processing/create-event-step.ts

@@ -18,6 +18,14 @@ const EXPERIMENT_EXPOSURE_EVENT = '$experiment_exposure'
 // event's uuid from its source event's uuid.
 const EXPERIMENT_EXPOSURE_UUIDV5_NAMESPACE = parseUuid('1b7c9119-5953-4668-97b7-ab0ef8a6bb48')
 
+// SDKs mark calls to flags on v2 rules with this property and send $experiment_exposure
+// themselves, so duplicating those calls would count each exposure twice. SDKs send the
+// number; the string is matched too because a duplicate costs more than a skipped one.
+function isConfigVersion2FeatureFlagCalledEvent(event: PreIngestionEvent): boolean {
+    const configVersion = event.properties['$feature_flag_config_version']
+    return configVersion === 2 || configVersion === '2'
+}
+
 // Flag definitions aren't available in ingestion, so multivariate is inferred from the
 // response shape; a variant literally keyed "true" or "false" is indistinguishable from
 // a boolean flag response and knowingly skipped.
@@ -26,6 +34,7 @@ function isMultivariateFeatureFlagCalledEvent(event: PreIngestionEvent): boolean
 
     return (
         event.event === FEATURE_FLAG_CALLED_EVENT &&
+        !isConfigVersion2FeatureFlagCalledEvent(event) &&
         typeof response === 'string' &&
         response !== '' &&
         response !== 'true' &&