fix: fix event flush process (#673)

Author: yuhao900914

packages/analytics-core/src/plugins/destination.ts

@@ -59,7 +59,6 @@ export class Destination implements DestinationPlugin {
 
     this.storageKey = `${STORAGE_PREFIX}_${this.config.apiKey.substring(0, 10)}`;
     const unsent = await this.config.storageProvider?.get(this.storageKey);
-    this.saveEvents(); // sets storage to '[]'
     if (unsent && unsent.length > 0) {
       void Promise.all(unsent.map((event) => this.execute(event))).catch();
     }
@@ -83,6 +82,7 @@ export class Destination implements DestinationPlugin {
     const tryable = list.filter((context) => {
       if (context.attempts < this.config.flushMaxRetries) {
         context.attempts += 1;
+
         return true;
       }
       void this.fulfillRequest([context], 500, MAX_RETRIES_EXCEEDED_MESSAGE);
@@ -102,7 +102,7 @@ export class Destination implements DestinationPlugin {
       }, context.timeout);
     });
 
-    this.saveEvents();
+    void this.updateEventStorage([], this.queue);
   }
 
   schedule(timeout: number) {
@@ -186,19 +186,19 @@ export class Destination implements DestinationPlugin {
 
     switch (status) {
       case Status.Success: {
-        this.handleSuccessResponse(res, list);
+        void this.handleSuccessResponse(res, list);
         break;
       }
       case Status.Invalid: {
-        this.handleInvalidResponse(res, list);
+        void this.handleInvalidResponse(res, list);
         break;
       }
       case Status.PayloadTooLarge: {
-        this.handlePayloadTooLargeResponse(res, list);
+        void this.handlePayloadTooLargeResponse(res, list);
         break;
       }
       case Status.RateLimit: {
-        this.handleRateLimitResponse(res, list);
+        void this.handleRateLimitResponse(res, list);
         break;
       }
       default: {
@@ -241,6 +241,7 @@ export class Destination implements DestinationPlugin {
       // log intermediate event status before retry
       this.config.loggerProvider.warn(getResponseBodyString(res));
     }
+
     this.addToQueue(...retry);
   }
 
@@ -297,21 +298,37 @@ export class Destination implements DestinationPlugin {
   }
 
   fulfillRequest(list: Context[], code: number, message: string) {
-    this.saveEvents();
     list.forEach((context) => context.callback(buildResult(context.event, code, message)));
+    void this.updateEventStorage(list);
   }
 
   /**
-   * Saves events to storage
    * This is called on
    * 1) new events are added to queue; or
    * 2) response comes back for a request
+   *
+   * update the event storage
    */
-  saveEvents() {
+  async updateEventStorage(eventsToRemove: Context[], eventsToAdd?: Context[]) {
     if (!this.config.storageProvider) {
       return;
     }
-    const events = Array.from(this.queue.map((context) => context.event));
-    void this.config.storageProvider.set(this.storageKey, events);
+
+    const filterEventInsertIdSet = eventsToRemove.reduce((filtered, context) => {
+      if (context.event.insert_id) {
+        filtered.add(context.event.insert_id);
+      }
+      return filtered;
+    }, new Set<string>());
+
+    const savedEvents = await this.config.storageProvider.get(this.storageKey);
+    const updatedEvents: Event[] = eventsToAdd?.map((context) => context.event) || [];
+
+    savedEvents?.forEach((event) => {
+      if (event.insert_id && !filterEventInsertIdSet.has(event.insert_id)) {
+        updatedEvents.push(event);
+      }
+    });
+    await this.config.storageProvider.set(this.storageKey, updatedEvents);
   }
 }

packages/analytics-core/test/plugins/destination.test.ts

@@ -548,8 +548,15 @@ describe('destination', () => {
     });
   });
 
-  describe('saveEvents', () => {
-    test('should save to storage provider', () => {
+  describe('filterEvent', () => {
+    test('should be ok with no storage provider', async () => {
+      const destination = new Destination();
+      destination.config = useDefaultConfig();
+      destination.config.storageProvider = undefined;
+      expect(await destination.updateEventStorage([])).toBe(undefined);
+    });
+
+    test('should filter dropped event and update the storage provider', async () => {
       const destination = new Destination();
       destination.config = useDefaultConfig();
       destination.config.storageProvider = {
@@ -560,16 +567,46 @@ describe('destination', () => {
         reset: async () => undefined,
         getRaw: async () => undefined,
       };
+      const event1 = { event_type: 'event', insert_id: '1' };
+      const event2 = { event_type: 'filtered_event', insert_id: '2' };
+      const get = jest.spyOn(destination.config.storageProvider, 'get').mockResolvedValueOnce([event1, event2]);
       const set = jest.spyOn(destination.config.storageProvider, 'set').mockResolvedValueOnce(undefined);
-      destination.saveEvents();
+      const context = {
+        event: event2,
+        attempts: 0,
+        callback: () => undefined,
+        timeout: 0,
+      };
+      await destination.updateEventStorage([context]);
+      expect(get).toHaveBeenCalledTimes(1);
       expect(set).toHaveBeenCalledTimes(1);
+      expect(set).toHaveBeenCalledWith('', expect.objectContaining([event1]));
     });
 
-    test('should be ok with no storage provider', () => {
+    test('should save event to the storage provider', async () => {
       const destination = new Destination();
       destination.config = useDefaultConfig();
-      destination.config.storageProvider = undefined;
-      expect(destination.saveEvents()).toBe(undefined);
+      destination.config.storageProvider = {
+        isEnabled: async () => true,
+        get: async () => undefined,
+        set: async () => undefined,
+        remove: async () => undefined,
+        reset: async () => undefined,
+        getRaw: async () => undefined,
+      };
+      const event = { event_type: 'event', insert_id: '1' };
+      const get = jest.spyOn(destination.config.storageProvider, 'get').mockResolvedValueOnce([]);
+      const set = jest.spyOn(destination.config.storageProvider, 'set').mockResolvedValueOnce(undefined);
+      const context = {
+        event: event,
+        attempts: 0,
+        callback: () => undefined,
+        timeout: 0,
+      };
+      await destination.updateEventStorage([], [context]);
+      expect(get).toHaveBeenCalledTimes(1);
+      expect(set).toHaveBeenCalledTimes(1);
+      expect(set).toHaveBeenCalledWith('', expect.objectContaining([event]));
     });
   });