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": [

src/client/eppo-client.ts

@@ -1,3 +1,5 @@
+import { randomUUID } from 'crypto';
+
 import ApiEndpoints from '../api-endpoints';
 import { logger } from '../application-logger';
 import { IAssignmentEvent, IAssignmentLogger } from '../assignment-logger';
@@ -21,7 +23,10 @@ import { EppoValue } from '../eppo_value';
 import { Evaluator, FlagEvaluation, noneResult } from '../evaluator';
 import ArrayBackedNamedEventQueue from '../events/array-backed-named-event-queue';
 import { BoundedEventQueue } from '../events/bounded-event-queue';
+import DefaultEventDispatcher from '../events/default-event-dispatcher';
+import EventDispatcher from '../events/event-dispatcher';
 import NamedEventQueue from '../events/named-event-queue';
+import NoOpEventDispatcher from '../events/no-op-event-dispatcher';
 import {
   FlagEvaluationDetailsBuilder,
   IFlagEvaluationDetails,
@@ -77,8 +82,17 @@ export interface IContainerExperiment<T> {
   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 +113,23 @@ export default class EppoClient {
   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;
@@ -909,9 +923,10 @@ export default class EppoClient {
     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 +944,9 @@ export default class EppoClient {
     return {
       configFetchedAt: this.flagConfigurationStore.getConfigFetchedAt() ?? '',
       configPublishedAt: this.flagConfigurationStore.getConfigPublishedAt() ?? '',
-      configEnvironment: this.flagConfigurationStore.getEnvironment() ?? { name: '' },
+      configEnvironment: this.flagConfigurationStore.getEnvironment() ?? {
+        name: '',
+      },
     };
   }
 

src/evaluator.ts

@@ -159,7 +159,10 @@ export class Evaluator {
     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();
-  }
 }