diff --git a/plugin-server/src/worker/ingestion/person-state.ts b/plugin-server/src/worker/ingestion/person-state.ts index 53daf093ef934..61b0c42ce363f 100644 --- a/plugin-server/src/worker/ingestion/person-state.ts +++ b/plugin-server/src/worker/ingestion/person-state.ts @@ -323,19 +323,31 @@ export class PersonState { return undefined } if (isDistinctIdIllegal(mergeIntoDistinctId)) { - await captureIngestionWarning(this.db.kafkaProducer, teamId, 'cannot_merge_with_illegal_distinct_id', { - illegalDistinctId: mergeIntoDistinctId, - otherDistinctId: otherPersonDistinctId, - eventUuid: this.event.uuid, - }) + await captureIngestionWarning( + this.db.kafkaProducer, + teamId, + 'cannot_merge_with_illegal_distinct_id', + { + illegalDistinctId: mergeIntoDistinctId, + otherDistinctId: otherPersonDistinctId, + eventUuid: this.event.uuid, + }, + (teamId) => `${teamId}:mergeIntoDistinctId` + ) return undefined } if (isDistinctIdIllegal(otherPersonDistinctId)) { - await captureIngestionWarning(this.db.kafkaProducer, teamId, 'cannot_merge_with_illegal_distinct_id', { - illegalDistinctId: otherPersonDistinctId, - otherDistinctId: mergeIntoDistinctId, - eventUuid: this.event.uuid, - }) + await captureIngestionWarning( + this.db.kafkaProducer, + teamId, + 'cannot_merge_with_illegal_distinct_id', + { + illegalDistinctId: otherPersonDistinctId, + otherDistinctId: mergeIntoDistinctId, + eventUuid: this.event.uuid, + }, + (teamId) => `${teamId}:mergeIntoDistinctId` + ) return undefined } return promiseRetry( @@ -403,12 +415,17 @@ export class PersonState { // If merge isn't allowed, we will ignore it, log an ingestion warning and exit if (!mergeAllowed) { - // TODO: add event UUID to the ingestion warning - await captureIngestionWarning(this.db.kafkaProducer, this.teamId, 'cannot_merge_already_identified', { - sourcePersonDistinctId: otherPersonDistinctId, - targetPersonDistinctId: mergeIntoDistinctId, - eventUuid: this.event.uuid, - }) + await captureIngestionWarning( + this.db.kafkaProducer, + this.teamId, + 'cannot_merge_already_identified', + { + sourcePersonDistinctId: otherPersonDistinctId, + targetPersonDistinctId: mergeIntoDistinctId, + eventUuid: this.event.uuid, + }, + (teamId) => `${teamId}:mergeIntoDistinctId` + ) status.warn('🤔', 'refused to merge an already identified user via an $identify or $create_alias call') return mergeInto // We're returning the original person tied to distinct_id used for the event } diff --git a/plugin-server/src/worker/ingestion/utils.ts b/plugin-server/src/worker/ingestion/utils.ts index 0ac139dfb137d..66b7935400bec 100644 --- a/plugin-server/src/worker/ingestion/utils.ts +++ b/plugin-server/src/worker/ingestion/utils.ts @@ -67,10 +67,17 @@ export async function captureIngestionWarning( teamId: TeamId, type: string, details: Record, - debounce_key?: string + /** + * Optional key to debounce the warning. If not provided, the teamId and type will be used as the key. + * If a function is provided, it will be called with the teamId and type and should return a string key. + * If a string is provided it is combined with the teamId and type. + */ + debounceKey?: string | ((teamId: number, type: string) => string) ) { - const limiter_key = `${teamId}:${type}:${debounce_key}` - if (IngestionWarningLimiter.consume(limiter_key, 1)) { + const limiterKey = + typeof debounceKey === 'function' ? debounceKey(teamId, type) : `${teamId}:${type}:${debounceKey}` + + if (IngestionWarningLimiter.consume(limiterKey, 1)) { await kafkaProducer.queueMessage({ topic: KAFKA_INGESTION_WARNINGS, messages: [