tests

Author: raju-opti

lib/event_processor/event_store.spec.ts

@@ -13,15 +13,16 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-import { vi, describe, it, expect, beforeEach } from 'vitest';
-import { getMockAsyncCache, getMockSyncCache } from '../tests/mock/mock_cache';
-import { SyncStore } from '../utils/cache/store';
+import { vi, describe, it, expect } from 'vitest';
+import { getMockAsyncCache } from '../tests/mock/mock_cache';
 import { EventWithId } from './batch_event_processor';
 import { EventStore, StoredEvent } from './event_store';
 import { createImpressionEvent } from '../tests/mock/create_event';
 
 import { DEFAULT_MAX_EVENTS_IN_STORE } from './event_store';
 import { exhaustMicrotasks } from '../tests/testUtils';
+import { EVENT_STORE_FULL } from '../message/log_message';
+import { OptimizelyError } from '../error/optimizly_error';
 
 type TestStoreConfig = {
   maxSize?: number;
@@ -68,7 +69,7 @@ describe('EventStore', () => {
     expect(saved2).toEqual(expect.objectContaining(event));
   });
 
-  it('should return all keys from getKeys()', async () => {
+  it('should return all keys from getKeys', async () => {
     const { store } = getEventStore();
     const event: EventWithId = {
       id: '1',
@@ -88,7 +89,7 @@ describe('EventStore', () => {
     expect(savedKeys).toEqual(keys);
   });
 
-  it('should limit the number of saved keys', async () => {
+  it('should limit the number of saved keys when set is called concurrently', async () => {
     const { store } = getEventStore();
 
     const event: EventWithId = {
@@ -120,6 +121,24 @@ describe('EventStore', () => {
     expect(savedKeys).toEqual(keys.slice(0, DEFAULT_MAX_EVENTS_IN_STORE));
   });
 
+  it('should limit the number of saved keys when set is called serially', async () => {
+    const { store } = getEventStore({ maxSize: 2 });
+
+    const event: EventWithId = {
+      id: '1',
+      event: createImpressionEvent('test'),
+    }
+
+    await expect(store.set('event-1', event)).resolves.not.toThrow();
+    await expect(store.set('event-2', event)).resolves.not.toThrow();
+    await expect(store.set('event-3', event)).rejects.toThrow(new OptimizelyError(EVENT_STORE_FULL, event.id));
+
+    const savedKeys = await store.getKeys();
+    savedKeys.sort();
+
+    expect(savedKeys).toEqual(['event-1', 'event-2']);
+  });
+
   it('should save keys again when the number of stored events drops below maxSize', async () => {
     const { store } = getEventStore();
 
@@ -182,7 +201,7 @@ describe('EventStore', () => {
     expect(await store.getKeys()).toEqual([]);
   });
 
-  it('should resave events without expireAt on get', async () => {
+  it('should resave events without expiresAt on get', async () => {
     const ttl = 120_000;
     const { mockStore, store } = getEventStore({ ttl });
 
@@ -191,7 +210,6 @@ describe('EventStore', () => {
       event: createImpressionEvent('test'),
     }
 
-
     const originalSet = mockStore.set.bind(mockStore);
 
     let call = 0;
@@ -200,12 +218,12 @@ describe('EventStore', () => {
         return originalSet(key, value);
       }
 
-      // Simulate old stored event without expireAt
-      const eventWithoutExpireAt: StoredEvent = {
+      // Simulate old stored event without expiresAt
+      const eventWithoutExpiresAt: StoredEvent = {
         id: value.id,
         event: value.event,
       };
-      return originalSet(key, eventWithoutExpireAt);
+      return originalSet(key, eventWithoutExpiresAt);
     });
 
     await store.set('test', event);
@@ -222,4 +240,179 @@ describe('EventStore', () => {
     expect(secondCall[1].expiresAt).toBeDefined();
     expect(secondCall[1].expiresAt!).toBeGreaterThanOrEqual(Date.now() + ttl - 10);
   });
+
+  it('should store event when key expires after store being full', async () => {
+    const ttl = 100;
+    const { store } = getEventStore({ ttl, maxSize: 2 });
+
+    const event: EventWithId = {
+      id: '1',
+      event: createImpressionEvent('test'),
+    }
+
+    await store.set('event-1', event);
+    await store.set('event-2', event);
+    await expect(store.set('event-3', event)).rejects.toThrow();
+
+    expect(await store.getKeys().then(keys => keys.sort())).toEqual(['event-1', 'event-2']);
+
+    // wait for ttl to expire
+    await new Promise(resolve => setTimeout(resolve, ttl + 50));
+
+    // both events should be expired now
+    expect(await store.get('event-1')).toBeUndefined();
+    expect(await store.get('event-2')).toBeUndefined();
+
+    // should be able to add new events now
+    await expect(store.set('event-3', event)).resolves.not.toThrow();
+    await expect(store.set('event-4', event)).resolves.not.toThrow();
+
+    const savedEvent3 = await store.get('event-3');
+    expect(savedEvent3).toEqual(expect.objectContaining(event));
+    const savedEvent4 = await store.get('event-4');
+    expect(savedEvent4).toEqual(expect.objectContaining(event));
+  });
+
+  it('should return all requested events correctly from getBatched', async () => {
+    const { store } = getEventStore();
+    const event1: EventWithId = { id: '1', event: createImpressionEvent('test-1') };
+    const event2: EventWithId = { id: '2', event: createImpressionEvent('test-2') };
+    const event3: EventWithId = { id: '3', event: createImpressionEvent('test-3') };
+
+    await store.set('key-1', event1);
+    await store.set('key-2', event2);
+    await store.set('key-3', event3);
+
+    const results = await store.getBatched(['key-3', 'key-1', 'key-2', 'key-4']);
+    
+    expect(results).toHaveLength(4);
+    expect(results[0]).toEqual(expect.objectContaining(event3));
+    expect(results[1]).toEqual(expect.objectContaining(event1));
+    expect(results[2]).toEqual(expect.objectContaining(event2));
+    expect(results[3]).toBeUndefined();
+  });
+
+
+  it('should handle expired events in getBatched', async () => {
+    const ttl = 100;
+    const { store } = getEventStore({ ttl });
+    const event1: EventWithId = { id: '1', event: createImpressionEvent('test-1') };
+    const event2: EventWithId = { id: '2', event: createImpressionEvent('test-2') };
+
+    await store.set('key-1', event1);
+    await new Promise(resolve => setTimeout(resolve, 70));
+    await store.set('key-2', event2);
+
+    // wait for first key to expire but not the second
+    await new Promise(resolve => setTimeout(resolve, 50));
+
+    const results = await store.getBatched(['key-1', 'key-2']);
+    
+    expect(results).toHaveLength(2);
+    expect(results[0]).toBeUndefined();
+    expect(results[1]).toEqual(expect.objectContaining(event2));
+
+    await expect(store.getKeys()).resolves.toEqual(['key-2']);
+  });
+
+  it('should resave events without expiresAt during getBatched', async () => {
+    const ttl = 120_000;
+    const { mockStore, store } = getEventStore({ ttl });
+    const event: EventWithId = { id: '1', event: createImpressionEvent('test') };
+
+    const originalSet = mockStore.set.bind(mockStore);
+
+    let call = 0;
+    const setSpy = vi.spyOn(mockStore, 'set').mockImplementation(async (key: string, value: StoredEvent) => {
+      if (call++ > 0) {
+        return originalSet(key, value);
+      }
+
+      // Simulate old stored event without expiresAt
+      const eventWithoutExpiresAt: StoredEvent = {
+        id: value.id,
+        event: value.event,
+      };
+      return originalSet(key, eventWithoutExpiresAt);
+    });
+
+    await store.set('key-1', event);
+    await store.set('key-2', event);
+
+    const results = await store.getBatched(['key-1', 'key-2']);
+    
+    expect(results).toHaveLength(2);
+    expect(results[0]).toEqual(expect.objectContaining(event));
+    expect(results[1]).toEqual(expect.objectContaining(event));
+
+    await exhaustMicrotasks();
+    expect(setSpy).toHaveBeenCalledTimes(3);
+
+    const secondCall = setSpy.mock.calls[1];
+
+    expect(secondCall[1].expiresAt).toBeDefined();
+    expect(secondCall[1].expiresAt!).toBeGreaterThanOrEqual(Date.now() + ttl - 10);
+  });
+
+  it('should store event when keys expire during getBatched after store being full', async () => {
+    const ttl = 100;
+    const { store } = getEventStore({ ttl, maxSize: 2 });
+
+    const event: EventWithId = {
+      id: '1',
+      event: createImpressionEvent('test'),
+    }
+
+    await store.set('event-1', event);
+    await new Promise(resolve => setTimeout(resolve, 70));
+    await store.set('event-2', event);
+    await expect(store.set('event-3', event)).rejects.toThrow();
+
+    expect(await store.getKeys().then(keys => keys.sort())).toEqual(['event-1', 'event-2']);
+
+    // wait for the first event to expire
+    await new Promise(resolve => setTimeout(resolve, 50));
+
+    const results = await store.getBatched(['event-1', 'event-2']);
+    
+    expect(results).toHaveLength(2);
+    expect(results[0]).toBeUndefined();
+    expect(results[1]).toEqual(expect.objectContaining(event));
+    
+    await new Promise(resolve => setTimeout(resolve));
+
+    // should be able to add new event now
+    await expect(store.set('event-3', event)).resolves.not.toThrow();
+    const savedEvent = await store.get('event-3');
+    expect(savedEvent).toEqual(expect.objectContaining(event));
+    expect(await store.getKeys().then(keys => keys.sort())).toEqual(['event-2', 'event-3']);
+  });
+
+  it('should restore in-memory key consistency after getKeys is called', async () => {
+    const { mockStore, store } = getEventStore({ maxSize: 2 });
+    const event: EventWithId = { id: '1', event: createImpressionEvent('test') };
+
+    const originalSet = mockStore.set.bind(mockStore);
+    
+    let call = 0;
+    vi.spyOn(mockStore, 'set').mockImplementation(async (key: string, value: StoredEvent) => {
+      // only the seconde set call should fail
+      if (call++ != 1) return originalSet(key, value);
+      return Promise.reject(new Error('Simulated set failure'));
+    });
+
+    await expect(store.set('key-1', event)).resolves.not.toThrow();
+    // this should fail, but in memory key list will become full
+    await expect(store.set('key-2', event)).rejects.toThrow('Simulated set failure');
+    await expect(store.set('key-3', event)).rejects.toThrow(new OptimizelyError(EVENT_STORE_FULL, event.id));
+    
+
+    let keys = await store.getKeys();
+    expect(keys.sort()).toEqual(['key-1']);
+    
+    await expect(store.set('key-3', event)).resolves.not.toThrow();
+    
+    keys = await store.getKeys();
+    expect(keys.sort()).toEqual(['key-1', 'key-3']);
+  });
 });

lib/event_processor/event_store.ts

@@ -1,6 +1,6 @@
 import { EventWithId } from "./batch_event_processor";
-import { AsyncPrefixStore, AsyncStore, AsyncStoreWithBatchedGet, OpStore, Store, SyncPrefixStore } from "../utils/cache/store";
-import { Maybe, OpType, OpValue } from "../utils/type";
+import { AsyncPrefixStore, AsyncStore, AsyncStoreWithBatchedGet, Store, StoreWithBatchedGet, SyncPrefixStore } from "../utils/cache/store";
+import { Maybe } from "../utils/type";
 import { SerialRunner } from "../utils/executor/serial_runner";
 import { LoggerFacade } from "../logging/logger";
 import { EVENT_STORE_FULL } from "../message/log_message";
@@ -26,7 +26,7 @@ export type EventStoreConfig = {
 export class EventStore extends AsyncStoreWithBatchedGet<EventWithId> implements AsyncStore<EventWithId> {
   readonly operation = 'async';
 
-  private store: Store<StoredEvent>;
+  private store: StoreWithBatchedGet<StoredEvent>;
   private serializer: SerialRunner = new SerialRunner();
   private logger?: LoggerFacade;
   private maxSize: number;
@@ -75,6 +75,8 @@ export class EventStore extends AsyncStoreWithBatchedGet<EventWithId> implements
   async set(key: string, event: EventWithId): Promise<unknown> {
     await this.readKeys();
 
+    // readKeys might have failed, in that case we cannot enforce max size
+    // that means, the store might grow beyond max size in failure scenarios
     if (this.keys !== undefined && this.keys.size >= this.maxSize) {
       this.logger?.info(EVENT_STORE_FULL, event.event.uuid);
       return Promise.reject(new OptimizelyError(EVENT_STORE_FULL, event.event.uuid));
@@ -90,8 +92,7 @@ export class EventStore extends AsyncStoreWithBatchedGet<EventWithId> implements
     return this.store.set(key, { ...event, expiresAt: Date.now() + this.ttl });
   }
 
-  async get(key: string): Promise<EventWithId | undefined> {
-    const value = await this.store.get(key);
+  private processStoredEvent(key: string, value: StoredEvent | undefined): Maybe<EventWithId> {
     if (!value) return undefined;
 
     // if there is events in the stored saved by old version of the sdk, 
@@ -112,6 +113,12 @@ export class EventStore extends AsyncStoreWithBatchedGet<EventWithId> implements
     return value;
   }
 
+  async get(key: string): Promise<EventWithId | undefined> {
+    const value = await this.store.get(key);
+    
+    return this.processStoredEvent(key, value);
+  }
+
   async remove(key: string): Promise<unknown> {
     await this.store.remove(key);
     this.keys?.delete(key);
@@ -125,6 +132,7 @@ export class EventStore extends AsyncStoreWithBatchedGet<EventWithId> implements
   }
 
   async getBatched(keys: string[]): Promise<Maybe<EventWithId>[]> {
-    return [];
+    const values = await this.store.getBatched(keys);
+    return values.map((value, index) => this.processStoredEvent(keys[index], value));
   }
 }

lib/utils/cache/store.ts

@@ -48,6 +48,8 @@ export abstract class AsyncStoreWithBatchedGet<V> implements AsyncStore<V> {
   abstract getBatched(keys: string[]): Promise<Maybe<V>[]>;
 }
 
+export type StoreWithBatchedGet<V> = SyncStoreWithBatchedGet<V> | AsyncStoreWithBatchedGet<V>;
+
 export const getBatchedSync = <V>(store: SyncStore<V>, keys: string[]): Maybe<V>[] => {
   if (store instanceof SyncStoreWithBatchedGet) {
     return store.getBatched(keys);