feat(events): Add EventDispatcher and related classes

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@eppo/js-client-sdk-common",
-  "version": "4.3.0",
+  "version": "4.4.0",
   "description": "Eppo SDK for client-side JavaScript applications (base for both web and react native)",
   "main": "dist/index.js",
   "files": [
@@ -46,6 +46,7 @@
     "@types/lodash": "^4.17.5",
     "@types/md5": "^2.3.2",
     "@types/semver": "^7.5.6",
+    "@types/uuid": "^10.0.0",
     "@typescript-eslint/eslint-plugin": "^5.13.0",
     "@typescript-eslint/parser": "^5.13.0",
     "eslint": "^8.17.0",
@@ -71,7 +72,8 @@
     "js-base64": "^3.7.7",
     "md5": "^2.3.0",
     "pino": "^8.19.0",
-    "semver": "^7.5.4"
+    "semver": "^7.5.4",
+    "uuid": "^8.3.2"
   },
   "packageManager": "[email protected]+sha512.a6b2f7906b721bba3d67d4aff083df04dad64c399707841b7acf00f6b133b7ac24255f2652fa22ae3534329dc6180534e98d17432037ff6fd140556e2bb3137e"
 }

src/client/eppo-client.ts

@@ -1,3 +1,5 @@
+import { v4 as randomUUID } from 'uuid';
+
 import ApiEndpoints from '../api-endpoints';
 import { logger } from '../application-logger';
 import { IAssignmentEvent, IAssignmentLogger } from '../assignment-logger';
@@ -21,7 +23,8 @@
 import { Evaluator, FlagEvaluation, noneResult } from '../evaluator';
 import ArrayBackedNamedEventQueue from '../events/array-backed-named-event-queue';
 import { BoundedEventQueue } from '../events/bounded-event-queue';
-import NamedEventQueue from '../events/named-event-queue';
+import EventDispatcher from '../events/event-dispatcher';
+import NoOpEventDispatcher from '../events/no-op-event-dispatcher';
 import {
   FlagEvaluationDetailsBuilder,
   IFlagEvaluationDetails,
@@ -77,8 +80,17 @@
   treatmentVariationEntries: Array<T>;
 }
 
+const DEFAULT_EVENT_DISPATCHER_CONFIG = {
+  // TODO: Replace with actual ingestion URL
+  ingestionUrl: 'https://example.com/events',
+  batchSize: 10,
+  flushInterval: 10_000,
+  retryInterval: 5_000,
+  maxRetries: 3,
+};
+
 export default class EppoClient {
-  private readonly eventQueue: NamedEventQueue<unknown>;
+  private readonly eventDispatcher: EventDispatcher;
   private readonly assignmentEventsQueue: BoundedEventQueue<IAssignmentEvent> =
     newBoundedArrayEventQueue<IAssignmentEvent>('assignments');
   private readonly banditEventsQueue: BoundedEventQueue<IBanditEvent> =
@@ -99,23 +111,23 @@
   private readonly evaluator = new Evaluator();
 
   constructor({
-    eventQueue = new ArrayBackedNamedEventQueue('events'),
+    eventDispatcher = new NoOpEventDispatcher(),
     isObfuscated = false,
     flagConfigurationStore,
     banditVariationConfigurationStore,
     banditModelConfigurationStore,
     configurationRequestParameters,
   }: {
-    // Queue for arbitrary, application-level events (not to be confused with Eppo specific assignment
+    // Dispatcher for arbitrary, application-level events (not to be confused with Eppo specific assignment
     // or bandit events). These events are application-specific and captures by EppoClient#track API.
-    eventQueue?: NamedEventQueue<unknown>;
+    eventDispatcher?: EventDispatcher;
     flagConfigurationStore: IConfigurationStore<Flag | ObfuscatedFlag>;
     banditVariationConfigurationStore?: IConfigurationStore<BanditVariation[]>;
     banditModelConfigurationStore?: IConfigurationStore<BanditParameters>;
     configurationRequestParameters?: FlagConfigurationRequestParameters;
     isObfuscated?: boolean;
   }) {
-    this.eventQueue = eventQueue;
+    this.eventDispatcher = eventDispatcher;
     this.flagConfigurationStore = flagConfigurationStore;
     this.banditVariationConfigurationStore = banditVariationConfigurationStore;
     this.banditModelConfigurationStore = banditModelConfigurationStore;
@@ -546,7 +558,7 @@
         );
         evaluationDetails.banditAction = action;
       }
     } catch (err: any) {
       logger.error('Error determining bandit action', err);
       if (!this.isGracefulFailureMode) {
         throw err;
@@ -781,7 +793,7 @@
         expectedVariationType,
       );
       return this.parseVariationWithDetails(result, defaultValue, expectedVariationType);
     } catch (error: any) {
       const eppoValue = this.rethrowIfNotGraceful(error, defaultValue);
       if (error instanceof FlagEvaluationError && error.flagEvaluationDetails) {
         return {
@@ -814,7 +826,7 @@
       }
       const eppoValue = EppoValue.valueOf(variation.value, expectedVariationType);
       return { eppoValue, flagEvaluationDetails };
     } catch (error: any) {
       const eppoValue = this.rethrowIfNotGraceful(error, defaultValue);
       return { eppoValue, flagEvaluationDetails };
     }
@@ -909,9 +921,10 @@
     return result;
   }
 
+  /** TODO */
   // noinspection JSUnusedGlobalSymbols
   track(event: unknown, params: Record<string, unknown>) {
-    this.eventQueue.push({ event, params });
+    this.eventDispatcher.dispatch({ id: randomUUID(), data: event, params });
   }
 
   private newFlagEvaluationDetailsBuilder(flagKey: string): FlagEvaluationDetailsBuilder {
@@ -929,7 +942,9 @@
     return {
       configFetchedAt: this.flagConfigurationStore.getConfigFetchedAt() ?? '',
       configPublishedAt: this.flagConfigurationStore.getConfigPublishedAt() ?? '',
-      configEnvironment: this.flagConfigurationStore.getEnvironment() ?? { name: '' },
+      configEnvironment: this.flagConfigurationStore.getEnvironment() ?? {
+        name: '',
+      },
     };
   }
 
@@ -1037,7 +1052,7 @@
     eventsToFlush.forEach((event) => {
       try {
         logFunction(event);
       } catch (error: any) {
         logger.error(`[Eppo SDK] Error flushing event to logger: ${error.message}`);
       }
     });
@@ -1085,7 +1100,7 @@
         allocationKey: allocationKey ?? '__eppo_no_allocation',
         variationKey: variation?.key ?? '__eppo_no_variation',
       });
     } catch (error: any) {
       logger.error(`[Eppo SDK] Error logging assignment event: ${error.message}`);
     }
   }

src/evaluator.ts

@@ -134,7 +134,7 @@
           'No allocations matched. Falling back to "Default Allocation", serving NULL',
         ),
       );
     } catch (err: any) {
       const flagEvaluationDetails = flagEvaluationDetailsBuilder.gracefulBuild(
         'ASSIGNMENT_ERROR',
         `Assignment Error: ${err.message}`,
@@ -159,7 +159,10 @@
     split: Split,
     subjectKey: string,
     expectedVariationType: VariationType | undefined,
-  ): { flagEvaluationCode: FlagEvaluationCode; flagEvaluationDescription: string } => {
+  ): {
+    flagEvaluationCode: FlagEvaluationCode;
+    flagEvaluationDescription: string;
+  } => {
     if (!checkValueTypeMatch(expectedVariationType, variation.value)) {
       const { key: vKey, value: vValue } = variation;
       return {

src/events/array-backed-named-event-queue.ts

@@ -1,6 +1,11 @@
 import NamedEventQueue from './named-event-queue';
 
-/** A named event queue backed by an array. */
+/**
+ * @internal
+ * A named event queue backed by an **unbounded** array.
+ * This class probably should NOT be used directly, but only as a backing store for
+ * {@link BoundedEventQueue}.
+ */
 export default class ArrayBackedNamedEventQueue<T> implements NamedEventQueue<T> {
   private readonly events: T[] = [];
 
@@ -22,7 +27,11 @@ export default class ArrayBackedNamedEventQueue<T> implements NamedEventQueue<T>
     return this.events[Symbol.iterator]();
   }
 
-  shift(): T | undefined {
-    return this.events.shift();
+  splice(count: number): T[] {
+    return this.events.splice(0, count);
+  }
+
+  isEmpty(): boolean {
+    return this.events.length === 0;
   }
 }

src/events/batch-event-processor.spec.ts

@@ -0,0 +1,24 @@
+import ArrayBackedNamedEventQueue from './array-backed-named-event-queue';
+import BatchEventProcessor from './batch-event-processor';
+
+describe('BatchEventProcessor', () => {
+  describe('nextBatch', () => {
+    it('should return a batch and remove items from the queue', () => {
+      const eventQueue = new ArrayBackedNamedEventQueue('test-queue');
+      const processor = new BatchEventProcessor(eventQueue, 2);
+      expect(processor.isEmpty()).toBeTruthy();
+      expect(processor.nextBatch()).toHaveLength(0);
+      processor.push({ id: 'foo-1', data: 'event1', params: {} });
+      processor.push({ id: 'foo-2', data: 'event2', params: {} });
+      processor.push({ id: 'foo-3', data: 'event3', params: {} });
+      expect(processor.isEmpty()).toBeFalsy();
+      const batch = processor.nextBatch();
+      expect(batch).toEqual([
+        { id: 'foo-1', data: 'event1', params: {} },
+        { id: 'foo-2', data: 'event2', params: {} },
+      ]);
+      expect(processor.nextBatch()).toEqual([{ id: 'foo-3', data: 'event3', params: {} }]);
+      expect(processor.isEmpty()).toBeTruthy();
+    });
+  });
+});

src/events/batch-event-processor.ts

@@ -0,0 +1,20 @@
+import NamedEventQueue from './named-event-queue';
+
+export default class BatchEventProcessor {
+  constructor(
+    private readonly eventQueue: NamedEventQueue<unknown>,
+    private readonly batchSize: number,
+  ) {}
+
+  nextBatch(): unknown[] {
+    return this.eventQueue.splice(this.batchSize);
+  }
+
+  push(event: unknown): void {
+    this.eventQueue.push(event);
+  }
+
+  isEmpty(): boolean {
+    return this.eventQueue.isEmpty();
+  }
+}

src/events/batch-retry-manager.ts

@@ -0,0 +1,44 @@
+import { logger } from '../application-logger';
+
+import EventDelivery from './event-delivery';
+
+/**
+ * Attempts to retry delivering a batch of events to the ingestionUrl up to `maxRetries` times
+ * using exponential backoff.
+ */
+export default class BatchRetryManager {
+  /**
+   * @param config.retryInterval - The minimum retry interval in milliseconds
+   * @param config.maxRetryDelayMs - The maximum retry delay in milliseconds
+   * @param config.maxRetries - The maximum number of retries
+   */
+  constructor(
+    private readonly delivery: EventDelivery,
+    private readonly config: {
+      retryIntervalMs: number;
+      maxRetryDelayMs: number;
+      maxRetries: number;
+    },
+  ) {}
+
+  async retry(batch: unknown[], attempt = 0): Promise<void> {
+    const { retryIntervalMs, maxRetryDelayMs, maxRetries } = this.config;
+    const delay = Math.min(retryIntervalMs * Math.pow(2, attempt), maxRetryDelayMs);
+    logger.info(`[BatchRetryManager] Retrying batch delivery in ${delay}ms...`);
+    await new Promise((resolve) => setTimeout(resolve, delay));
+
+    const success = await this.delivery.deliver(batch);
+    if (success) {
+      logger.info(`[BatchRetryManager] Batch delivery successfully after ${attempt} retries.`);
+      return;
+    }
+    if (attempt < maxRetries) {
+      return this.retry(batch, attempt + 1);
+    } else {
+      // TODO: Persist batch to avoid data loss
+      logger.warn(
+        `[BatchRetryManager] Failed to deliver batch after ${maxRetries} retries, bailing`,
+      );
+    }
+  }
+}

src/events/bounded-event-queue.ts

@@ -24,9 +24,4 @@ export class BoundedEventQueue<T> {
     this.queue.length = 0;
     return events;
   }
-
-  /** Returns the first event in the queue and removes it. */
-  shift(): T | undefined {
-    return this.queue.shift();
-  }
 }