Merge pull request #576 from elbwalker/560-duplicate-events-in-api-destination

alexanderkirtzel · web-flow · commit e2bfa09565b8 · 2025-12-16T18:12:58.000+01:00
batch with wildcard mapping
diff --git a/packages/collector/src/__tests__/destination.test.ts b/packages/collector/src/__tests__/destination.test.ts
@@ -422,4 +422,187 @@ describe('Destination', () => {
       );
     });
   });
+
+  describe('batch with wildcard mapping', () => {
+    beforeEach(() => {
+      jest.useFakeTimers();
+    });
+
+    afterEach(() => {
+      jest.useRealTimers();
+    });
+
+    it('should not duplicate events when using wildcard batch config', async () => {
+      // Capture events at call time (since batched.events gets cleared after pushBatch)
+      const capturedBatches: { events: WalkerOS.Events; key: string }[] = [];
+      const mockPushBatch = jest.fn((batch) => {
+        // Clone the events array to capture at call time
+        capturedBatches.push({
+          events: [...batch.events],
+          key: batch.key,
+        });
+      });
+      const mockPush = jest.fn();
+
+      const destinationWithBatch: Destination.Instance = {
+        push: mockPush,
+        pushBatch: mockPushBatch,
+        config: {
+          init: true,
+          mapping: {
+            '*': {
+              '*': { batch: 50 }, // 50ms debounce
+            },
+          },
+        },
+      };
+
+      const { elb } = await startFlow({
+        destinations: { batchDest: { code: destinationWithBatch } },
+      });
+
+      // Send different event types (all should match wildcard)
+      await elb('page view');
+      await elb('product click');
+      await elb('button press');
+
+      // Advance timers to trigger debounce
+      jest.advanceTimersByTime(100);
+
+      // pushBatch should be called exactly once with all 3 events
+      expect(mockPushBatch).toHaveBeenCalledTimes(1);
+      expect(capturedBatches[0].events).toHaveLength(3);
+
+      // Verify all events are present (not duplicated)
+      const eventNames = capturedBatches[0].events.map((e) => e.name);
+      expect(eventNames).toContain('page view');
+      expect(eventNames).toContain('product click');
+      expect(eventNames).toContain('button press');
+
+      // Individual push should NOT be called (batch handles it)
+      expect(mockPush).not.toHaveBeenCalled();
+    });
+
+    it('should batch events separately per mapping key', async () => {
+      // Capture events at call time
+      const capturedBatches: { events: WalkerOS.Events; key: string }[] = [];
+      const mockPushBatch = jest.fn((batch) => {
+        capturedBatches.push({
+          events: [...batch.events],
+          key: batch.key,
+        });
+      });
+      const mockPush = jest.fn();
+
+      const destinationWithBatch: Destination.Instance = {
+        push: mockPush,
+        pushBatch: mockPushBatch,
+        config: {
+          init: true,
+          mapping: {
+            page: {
+              '*': { batch: 50 }, // page events batch together
+            },
+            product: {
+              '*': { batch: 50 }, // product events batch together
+            },
+          },
+        },
+      };
+
+      const { elb } = await startFlow({
+        destinations: { batchDest: { code: destinationWithBatch } },
+      });
+
+      // Send events that match different mapping keys
+      await elb('page view');
+      await elb('page scroll');
+      await elb('product click');
+      await elb('product view');
+
+      // Advance timers to trigger debounce
+      jest.advanceTimersByTime(100);
+
+      // pushBatch should be called twice (once per mapping key)
+      expect(mockPushBatch).toHaveBeenCalledTimes(2);
+
+      // Each batch should have 2 events
+      expect(capturedBatches[0].events).toHaveLength(2);
+      expect(capturedBatches[1].events).toHaveLength(2);
+    });
+
+    it('should isolate batch state when multiple destinations share same mapping config', async () => {
+      // This test reproduces the bug where shared mapping config causes duplicate events
+      // Two destinations with the SAME mapping config object (shared reference)
+      const sharedMapping = {
+        '*': {
+          '*': { batch: 50 },
+        },
+      };
+
+      const capturedBatches: {
+        destination: string;
+        events: WalkerOS.Events;
+        key: string;
+      }[] = [];
+
+      const mockPushBatch1 = jest.fn((batch) => {
+        capturedBatches.push({
+          destination: 'dest1',
+          events: [...batch.events],
+          key: batch.key,
+        });
+      });
+      const mockPushBatch2 = jest.fn((batch) => {
+        capturedBatches.push({
+          destination: 'dest2',
+          events: [...batch.events],
+          key: batch.key,
+        });
+      });
+
+      const destination1: Destination.Instance = {
+        push: jest.fn(),
+        pushBatch: mockPushBatch1,
+        config: {
+          init: true,
+          mapping: sharedMapping, // Shared reference!
+        },
+      };
+
+      const destination2: Destination.Instance = {
+        push: jest.fn(),
+        pushBatch: mockPushBatch2,
+        config: {
+          init: true,
+          mapping: sharedMapping, // Same shared reference!
+        },
+      };
+
+      const { elb } = await startFlow({
+        destinations: {
+          dest1: { code: destination1 },
+          dest2: { code: destination2 },
+        },
+      });
+
+      // Send events
+      await elb('page view');
+      await elb('product click');
+
+      // Advance timers to trigger debounce
+      jest.advanceTimersByTime(100);
+
+      // Each destination should receive its own batch
+      // BUG: Currently shared mapping causes only last destination to receive events
+      // or events get duplicated across destinations
+      const totalPushBatchCalls =
+        mockPushBatch1.mock.calls.length + mockPushBatch2.mock.calls.length;
+      expect(totalPushBatchCalls).toBe(2); // Should be 2 (one per destination)
+
+      // Each destination should receive exactly 2 events
+      expect(mockPushBatch1).toHaveBeenCalledTimes(1);
+      expect(mockPushBatch2).toHaveBeenCalledTimes(1);
+    });
+  });
 });
diff --git a/packages/collector/src/destination.ts b/packages/collector/src/destination.ts
@@ -290,48 +290,62 @@ export async function destinationPush<Destination extends Destination.Instance>(
   };
 
   const eventMapping = processed.mapping;
+  const mappingKey = processed.mappingKey || '* *';
+
   if (eventMapping?.batch && destination.pushBatch) {
-    const batched = eventMapping.batched || {
-      key: processed.mappingKey || '',
-      events: [],
-      data: [],
-    };
-    batched.events.push(processed.event);
-    if (isDefined(processed.data)) batched.data.push(processed.data);
-
-    eventMapping.batchFn =
-      eventMapping.batchFn ||
-      debounce((destination, collector) => {
-        const batchDestType = destination.type || 'unknown';
-        const batchLogger = collector.logger.scope(batchDestType);
-
-        const batchContext: Destination.PushBatchContext = {
-          collector,
-          config,
-          data: processed.data,
-          mapping: eventMapping,
-          env: mergeEnvironments(destination.env, config.env),
-          logger: batchLogger,
-        };
-
-        batchLogger.debug('push batch', {
-          events: batched.events.length,
-        });
-
-        useHooks(
-          destination.pushBatch!,
-          'DestinationPushBatch',
-          (collector as Collector.Instance).hooks,
-        )(batched, batchContext);
-
-        batchLogger.debug('push batch done');
-
-        batched.events = [];
-        batched.data = [];
-      }, eventMapping.batch);
-
-    eventMapping.batched = batched;
-    eventMapping.batchFn?.(destination, collector);
+    // Initialize batch registry on destination (not on shared mapping config)
+    destination.batches = destination.batches || {};
+
+    // Get or create batch state for this mapping key
+    if (!destination.batches[mappingKey]) {
+      const batched: Destination.Batch<unknown> = {
+        key: mappingKey,
+        events: [],
+        data: [],
+      };
+
+      destination.batches[mappingKey] = {
+        batched,
+        batchFn: debounce(() => {
+          const batchState = destination.batches![mappingKey];
+          const currentBatched = batchState.batched;
+
+          const batchContext: Destination.PushBatchContext = {
+            collector,
+            config,
+            // Note: batch.data contains all transformed data; context.data is for single events
+            data: undefined,
+            mapping: eventMapping,
+            env: mergeEnvironments(destination.env, config.env),
+            logger: destLogger,
+          };
+
+          destLogger.debug('push batch', {
+            events: currentBatched.events.length,
+          });
+
+          useHooks(
+            destination.pushBatch!,
+            'DestinationPushBatch',
+            collector.hooks,
+          )(currentBatched, batchContext);
+
+          destLogger.debug('push batch done');
+
+          // Reset batch
+          currentBatched.events = [];
+          currentBatched.data = [];
+        }, eventMapping.batch),
+      };
+    }
+
+    // Add event to batch
+    const batchState = destination.batches[mappingKey];
+    batchState.batched.events.push(processed.event);
+    if (isDefined(processed.data)) batchState.batched.data.push(processed.data);
+
+    // Trigger debounced batch
+    batchState.batchFn();
   } else {
     destLogger.debug('push', { event: processed.event.name });
 
diff --git a/packages/core/src/types/destination.ts b/packages/core/src/types/destination.ts
@@ -60,6 +60,7 @@ export interface Instance<T extends TypesGeneric = Types> {
   config: Config<T>;
   queue?: WalkerOS.Events;
   dlq?: DLQ;
+  batches?: BatchRegistry<Mapping<T>>;
   type?: string;
   env?: Env<T>;
   init?: InitFn<T>;
@@ -126,13 +127,15 @@ export interface InitContext<T extends TypesGeneric = Types> {
   logger: Logger.Instance;
 }
 
-export interface PushContext<T extends TypesGeneric = Types>
-  extends Context<T> {
+export interface PushContext<
+  T extends TypesGeneric = Types,
+> extends Context<T> {
   mapping?: WalkerOSMapping.Rule<Mapping<T>>;
 }
 
-export interface PushBatchContext<T extends TypesGeneric = Types>
-  extends Context<T> {
+export interface PushBatchContext<
+  T extends TypesGeneric = Types,
+> extends Context<T> {
   mapping?: WalkerOSMapping.Rule<Mapping<T>>;
 }
 
@@ -163,6 +166,13 @@ export interface Batch<Mapping> {
   mapping?: WalkerOSMapping.Rule<Mapping>;
 }
 
+export interface BatchRegistry<Mapping> {
+  [mappingKey: string]: {
+    batched: Batch<Mapping>;
+    batchFn: () => void;
+  };
+}
+
 export type Data =
   | WalkerOS.Property
   | undefined
