update

Author: raju-opti

lib/event_processor/batch_event_processor.react_native.spec.ts

@@ -53,6 +53,8 @@ import { EventWithId } from './batch_event_processor';
 import { buildLogEvent } from './event_builder/log_event';
 import { createImpressionEvent } from '../tests/mock/create_event';
 import { ProcessableEvent } from './event_processor';
+import { wait } from '../tests/testUtils';
+import { mock } from 'node:test';
 
 const getMockDispatcher = () => {
   return {
@@ -77,6 +79,10 @@ describe('ReactNativeNetInfoEventProcessor', () => {
     const dispatchRepeater = getMockRepeater();
     const failedEventRepeater = getMockRepeater();
 
+        // fail the first dispatch attempt that is made on start
+    eventDispatcher.dispatchEvent = vi.fn().mockResolvedValueOnce({ statusCode: 500 })
+      .mockResolvedValue({ statusCode: 200 });
+
     const cache = getMockAsyncCache<EventWithId>();
     const events: ProcessableEvent[] = [];
 
@@ -98,18 +104,27 @@ describe('ReactNativeNetInfoEventProcessor', () => {
     processor.start();
     await processor.onRunning();
 
+    await exhaustMicrotasks();
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
+
     mockNetInfo.pushState(true);
-    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+    await exhaustMicrotasks();
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
 
     mockNetInfo.pushState(true);
-    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+    await exhaustMicrotasks();
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
   });
 
-  it('should retry failed events when network becomes reachable', async () => {
+  it('should retry failed events immediately when network becomes reachable', async () => {
     const eventDispatcher = getMockDispatcher();
     const dispatchRepeater = getMockRepeater();
     const failedEventRepeater = getMockRepeater();
 
+    // fail the first dispatch attempt that is made on start
+    eventDispatcher.dispatchEvent = vi.fn().mockResolvedValueOnce({ statusCode: 500 })
+      .mockResolvedValue({ statusCode: 200 });
+
     const cache = getMockAsyncCache<EventWithId>();
     const events: ProcessableEvent[] = [];
 
@@ -131,14 +146,17 @@ describe('ReactNativeNetInfoEventProcessor', () => {
     processor.start();
     await processor.onRunning();
 
-    mockNetInfo.pushState(false);
-    expect(eventDispatcher.dispatchEvent).not.toHaveBeenCalled();
+    await exhaustMicrotasks();
+    
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(1);
 
-    mockNetInfo.pushState(true);
+    mockNetInfo.pushState(false);
+    mockNetInfo.pushState(true); 
 
     await exhaustMicrotasks();
 
-    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledWith(buildLogEvent(events));
+    expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(2);
+    expect(eventDispatcher.dispatchEvent).toHaveBeenNthCalledWith(2, buildLogEvent(events));
   });
 
   it('should unsubscribe from netinfo listener when stopped', async () => {

lib/event_processor/batch_event_processor.spec.ts

@@ -368,124 +368,6 @@ describe('BatchEventProcessor', async () => {
       expect(events).toEqual(eventsInStore);
     });
 
-    it('should not store the event in the eventStore but still dispatch if the \
-        number of pending events is greater than the limit', async () => {
-      const eventDispatcher = getMockDispatcher();
-      const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;
-      mockDispatch.mockResolvedValue(resolvablePromise().promise);
-
-      const eventStore = getMockSyncCache<EventWithId>();
-
-      const idGenerator = new IdGenerator();
-
-      for (let i = 0; i < 505; i++) {
-        const event = createImpressionEvent(`id-${i}`);
-        const cacheId = idGenerator.getId();
-        await eventStore.set(cacheId, { id: cacheId, event });
-      }
-
-      expect(eventStore.size()).toEqual(505);
-
-      const processor = new BatchEventProcessor({
-        eventDispatcher,
-        dispatchRepeater: getMockRepeater(),
-        batchSize: 1,
-        eventStore,
-      });
-
-      processor.start();
-      await processor.onRunning();
-
-      const events: ProcessableEvent[] = [];
-      for(let i = 0; i < 2; i++) {
-        const event = createImpressionEvent(`id-${i}`);
-        events.push(event);
-        await processor.process(event)
-      }
-
-      expect(eventStore.size()).toEqual(505);
-      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(507);
-      expect(eventDispatcher.dispatchEvent.mock.calls[505][0]).toEqual(buildLogEvent([events[0]]));
-      expect(eventDispatcher.dispatchEvent.mock.calls[506][0]).toEqual(buildLogEvent([events[1]]));
-    });
-
-    it('should store events in the eventStore when the number of events in the store\
-        becomes lower than the limit', async () => {
-      const eventDispatcher = getMockDispatcher();
-
-      const dispatchResponses: ResolvablePromise<any>[] = [];
-
-      const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;
-      mockDispatch.mockImplementation((arg) => {
-        const dispatchResponse = resolvablePromise();
-        dispatchResponses.push(dispatchResponse);
-        return dispatchResponse.promise;
-      });
-
-      const eventStore = getMockSyncCache<EventWithId>();
-
-      const idGenerator = new IdGenerator();
-
-      for (let i = 0; i < 502; i++) {
-        const event = createImpressionEvent(`id-${i}`);
-        const cacheId = String(i);
-        await eventStore.set(cacheId, { id: cacheId, event });
-      }
-
-      expect(eventStore.size()).toEqual(502);
-
-      const processor = new BatchEventProcessor({
-        eventDispatcher,
-        dispatchRepeater: getMockRepeater(),
-        batchSize: 1,
-        eventStore,
-      });
-
-      processor.start();
-      await processor.onRunning();
-
-      let events: ProcessableEvent[] = [];
-      for(let i = 0; i < 2; i++) {
-        const event = createImpressionEvent(`id-${i + 502}`);
-        events.push(event);
-        await processor.process(event)
-      }
-
-      expect(eventStore.size()).toEqual(502);
-      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(504);
-
-      expect(eventDispatcher.dispatchEvent.mock.calls[502][0]).toEqual(buildLogEvent([events[0]]));
-      expect(eventDispatcher.dispatchEvent.mock.calls[503][0]).toEqual(buildLogEvent([events[1]]));
-
-      // resolve the dispatch for events not saved in the store
-      dispatchResponses[502].resolve({ statusCode: 200 });
-      dispatchResponses[503].resolve({ statusCode: 200 });
-
-      await exhaustMicrotasks();
-      expect(eventStore.size()).toEqual(502);
-
-      // resolve the dispatch for 3 events in store, making the store size 499 which is lower than the limit
-      dispatchResponses[0].resolve({ statusCode: 200 });
-      dispatchResponses[1].resolve({ statusCode: 200 });
-      dispatchResponses[2].resolve({ statusCode: 200 });
-
-      await exhaustMicrotasks();
-      expect(eventStore.size()).toEqual(499);
-
-      // process 2 more events
-      events = [];
-      for(let i = 0; i < 2; i++) {
-        const event = createImpressionEvent(`id-${i + 504}`);
-        events.push(event);
-        await processor.process(event)
-      }
-
-      expect(eventStore.size()).toEqual(500);
-      expect(eventDispatcher.dispatchEvent).toHaveBeenCalledTimes(506);
-      expect(eventDispatcher.dispatchEvent.mock.calls[504][0]).toEqual(buildLogEvent([events[0]]));
-      expect(eventDispatcher.dispatchEvent.mock.calls[505][0]).toEqual(buildLogEvent([events[1]]));      
-    });
-
     it('should still dispatch events even if the store save fails', async () => {
       const eventDispatcher = getMockDispatcher();
       const mockDispatch: MockInstance<typeof eventDispatcher.dispatchEvent> = eventDispatcher.dispatchEvent;

lib/event_processor/batch_event_processor.ts

@@ -31,7 +31,6 @@ import { FAILED_TO_DISPATCH_EVENTS, SERVICE_NOT_RUNNING } from "error_message";
 import { OptimizelyError } from "../error/optimizly_error";
 import { sprintf } from "../utils/fns";
 import { SERVICE_STOPPED_BEFORE_RUNNING } from "../service";
-import { EVENT_STORE_FULL } from "../message/log_message";
 
 export const DEFAULT_MIN_BACKOFF = 1000;
 export const DEFAULT_MAX_BACKOFF = 32000;
@@ -40,7 +39,7 @@ export const MAX_EVENTS_IN_STORE = 500;
 export type EventWithId = {
   id: string;
   event: ProcessableEvent;
-  notStored?: boolean;
+  // notStored?: boolean;
 };
 
 export type RetryConfig = {
@@ -73,9 +72,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
   private eventQueue: EventWithId[] = [];
   private batchSize: number;
   private eventStore?: Store<EventWithId>;
-  private eventCountInStore: Maybe<number> = undefined;
-  private eventCountWaitPromise: Promise<unknown> = Promise.resolve();
-  private maxEventsInStore: number = MAX_EVENTS_IN_STORE;
   private dispatchRepeater: Repeater;
   private failedEventRepeater?: Repeater;
   private idGenerator: IdGenerator = new IdGenerator();
@@ -96,7 +92,7 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     this.dispatchRepeater = config.dispatchRepeater;
     this.dispatchRepeater.setTask(() => this.flush());
 
-    this.maxEventsInStore = Math.max(2 * config.batchSize, MAX_EVENTS_IN_STORE);
+    // this.maxEventsInStore = Math.max(2 * config.batchSize, MAX_EVENTS_IN_STORE);
     this.failedEventRepeater = config.failedEventRepeater;
     this.failedEventRepeater?.setTask(() => this.retryFailedEvents());
     if (config.logger) {
@@ -215,9 +211,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     runResult.result.then((res) => {
       events.forEach((event) => {
         this.eventStore?.remove(event.id);
-        if (!event.notStored && this.eventCountInStore) {
-          this.eventCountInStore--;
-        }
       });
       return Promise.resolve();
     }).catch((err) => {
@@ -250,7 +243,10 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
       event: event,
     };
 
-    await this.storeEvent(eventWithId);
+    // store in eventStore first to avoid data loss
+    await Promise.resolve(this.eventStore?.set(eventWithId.id, eventWithId)).catch((e) => {
+      this.logger?.error(e);
+    });
 
     if (this.eventQueue.length > 0 && !areEventContextsEqual(this.eventQueue[0].event, event)) {
       this.flush();
@@ -265,46 +261,6 @@ export class BatchEventProcessor extends BaseService implements EventProcessor {
     }
   }
 
-  private async readEventCountInStore(store: Store<EventWithId>): Promise<void> {
-    if (this.eventCountInStore !== undefined) {
-      return;
-    }
-    
-    try {
-      const keys = await store.getKeys();
-      this.eventCountInStore = keys.length;
-    } catch (e) {
-      this.logger?.error(e);
-    }
-  }
-
-  private async findEventCountInStore(): Promise<unknown> {
-    if (this.eventStore && this.eventCountInStore === undefined) {
-      const store = this.eventStore;
-      this.eventCountWaitPromise = this.eventCountWaitPromise.then(() => this.readEventCountInStore(store));
-      return this.eventCountWaitPromise;
-    }
-    return Promise.resolve();
-  }
-
-  private async storeEvent(eventWithId: EventWithId): Promise<void> {
-    await this.findEventCountInStore();
-    if (this.eventCountInStore !== undefined && this.eventCountInStore >= this.maxEventsInStore) {
-      this.logger?.info(EVENT_STORE_FULL, eventWithId.event.uuid);
-      eventWithId.notStored = true;
-      return;
-    }
-
-    await Promise.resolve(this.eventStore?.set(eventWithId.id, eventWithId)).then(() => {
-      if (this.eventCountInStore !== undefined) {
-        this.eventCountInStore++;
-      }
-    }).catch((e) => {
-      eventWithId.notStored = true;
-      this.logger?.error(e);
-    });
-  }
-
   start(): void {
     if (!this.isNew()) {
       return;

lib/event_processor/event_processor_factory.browser.ts

@@ -22,12 +22,14 @@ import {
   OpaqueEventProcessor,
   wrapEventProcessor,
   getForwardingEventProcessor,
+  transformEventStore,
+  getPrefixEventStore,
 } from './event_processor_factory';
 import defaultEventDispatcher from './event_dispatcher/default_dispatcher.browser';
 import sendBeaconEventDispatcher from './event_dispatcher/send_beacon_dispatcher.browser';
 import { LocalStorageCache } from '../utils/cache/local_storage_cache.browser';
-import { SyncPrefixStore } from '../utils/cache/store';
-import { EVENT_STORE_PREFIX, FAILED_EVENT_RETRY_INTERVAL } from './event_processor_factory';
+import { FAILED_EVENT_RETRY_INTERVAL } from './event_processor_factory';
+import { DEFAULT_MAX_EVENTS_IN_STORE, EventStore } from './event_store';
 
 export const DEFAULT_EVENT_BATCH_SIZE = 10;
 export const DEFAULT_EVENT_FLUSH_INTERVAL = 1_000;
@@ -38,17 +40,15 @@ export const createForwardingEventProcessor = (
   return wrapEventProcessor(getForwardingEventProcessor(eventDispatcher));
 };
 
-const identity = <T>(v: T): T => v;
-
 export const createBatchEventProcessor = (
   options: BatchEventProcessorOptions = {}
 ): OpaqueEventProcessor => {
-  const localStorageCache = new LocalStorageCache<EventWithId>();
-  const eventStore = new SyncPrefixStore<EventWithId, EventWithId>(
-    localStorageCache, EVENT_STORE_PREFIX,
-    identity,
-    identity,
-  );
+  const eventStore = options.eventStore ? getPrefixEventStore(options.eventStore) : new EventStore({
+    store: new LocalStorageCache<EventWithId>(),
+    maxSize: options.batchSize ? Math.max(options.batchSize * 2, DEFAULT_MAX_EVENTS_IN_STORE)
+      : DEFAULT_MAX_EVENTS_IN_STORE,
+    ttl: options.storeTtl,
+  });
 
   return getOpaqueBatchEventProcessor({
     eventDispatcher: options.eventDispatcher || defaultEventDispatcher,
@@ -63,5 +63,6 @@ export const createBatchEventProcessor = (
     },
     failedEventRetryInterval: FAILED_EVENT_RETRY_INTERVAL,
     eventStore,
+    storeTtl: options.storeTtl,
   });
 };

lib/event_processor/event_processor_factory.react_native.spec.ts

@@ -111,8 +111,10 @@ describe('createBatchEventProcessor', () => {
     MockAsyncPrefixStore.mockClear();
   });
 
-  it('uses AsyncStorageCache and AsyncPrefixStore to create eventStore if no eventStore is provided', () => {
-    const processor = createBatchEventProcessor({});
+  it('uses an EventStore instance with correct options if no eventStore is provided', () => {
+    const processor = createBatchEventProcessor({
+      storeTtl: 60_000,
+    });
 
     expect(Object.is(processor, mockGetOpaqueBatchEventProcessor.mock.results[0].value)).toBe(true);
     const eventStore = mockGetOpaqueBatchEventProcessor.mock.calls[0][0].eventStore;