feat(events): Add EventDispatcher and related classes (#138)

* feat(events): Add `EventDispatcher` and related classes

* nits

* moar nits

* add uuid

* ignore explicit any

* code review nits

Author: felipecsl

.eslintrc.js

@@ -24,6 +24,7 @@ module.exports = {
     ],
     'import/named': 'off',
     'import/no-unresolved': 'off',
+    '@typescript-eslint/no-explicit-any': 'off',
     'import/order': [
       'warn',
       {
@@ -33,7 +34,7 @@ module.exports = {
             group: 'parent',
             position: 'before',
           },
-       ],
+        ],
         groups: ['builtin', 'external', 'parent', 'sibling', 'index'],
         'newlines-between': 'always',
         alphabetize: {

package.json

@@ -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';
@@ -20,7 +22,8 @@ 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 NamedEventQueue from '../events/named-event-queue';
+import EventDispatcher from '../events/event-dispatcher';
+import NoOpEventDispatcher from '../events/no-op-event-dispatcher';
 import {
   FlagEvaluationDetailsBuilder,
   IFlagEvaluationDetails,
@@ -76,8 +79,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,
+  flushIntervalMs: 10_000,
+  retryIntervalMs: 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> =
@@ -98,23 +110,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;
@@ -908,9 +920,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 {
@@ -928,7 +941,9 @@ export default class EppoClient {
     return {
       configFetchedAt: this.flagConfigurationStore.getConfigFetchedAt() ?? '',
       configPublishedAt: this.flagConfigurationStore.getConfigPublishedAt() ?? '',
-      configEnvironment: this.flagConfigurationStore.getEnvironment() ?? { name: '' },
+      configEnvironment: this.flagConfigurationStore.getEnvironment() ?? {
+        name: '',
+      },
     };
   }
 
@@ -1131,6 +1146,6 @@ export function checkValueTypeMatch(
   }
 }
 
-export function newBoundedArrayEventQueue<T>(name: string): BoundedEventQueue<T> {
+function newBoundedArrayEventQueue<T>(name: string): BoundedEventQueue<T> {
   return new BoundedEventQueue<T>(new ArrayBackedNamedEventQueue<T>(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();
-  }
 }