feat(events): Add event context (#213)

* feat(events): Add event context

* valida context value type

* ensure context limits

* fix lint

* fix lint v2

* Update src/events/default-event-dispatcher.ts

Co-authored-by: Eric Petzel <[email protected]>

* Update src/events/default-event-dispatcher.ts

Co-authored-by: Eric Petzel <[email protected]>

---------

Co-authored-by: Eric Petzel <[email protected]>

Author: felipecsl

src/client/eppo-client.ts

@@ -165,6 +165,21 @@ export default class EppoClient {
     this.eventDispatcher = eventDispatcher;
   }
 
+  /**
+   * Attaches a context to be included with all events dispatched by the EventDispatcher.
+   * The context is delivered as a top-level object in the ingestion request payload.
+   * An existing key can be removed by providing a `null` value.
+   * Calling this method with same key multiple times causes only the last value to be used for the
+   * given key.
+   *
+   * @param key - The context entry key.
+   * @param value - The context entry value, must be a string, number, boolean, or null. If value is
+   * an object or an array, will throw an ArgumentError.
+   */
+  setContext(key: string, value: string | number | boolean | null) {
+    this.eventDispatcher?.attachContext(key, value);
+  }
+
   // noinspection JSUnusedGlobalSymbols
   setBanditModelConfigurationStore(
     banditModelConfigurationStore: IConfigurationStore<BanditParameters>,

src/events/batch-retry-manager.spec.ts

@@ -19,15 +19,15 @@ describe('BatchRetryManager', () => {
 
   it('should successfully retry and deliver a batch with no failures', async () => {
     mockDelivery.deliver.mockResolvedValueOnce({ failedEvents: [] });
-    const result = await batchRetryManager.retry(mockBatch);
+    const result = await batchRetryManager.retry(mockBatch, {});
     expect(result).toEqual([]);
     expect(mockDelivery.deliver).toHaveBeenCalledTimes(1);
-    expect(mockDelivery.deliver).toHaveBeenCalledWith(mockBatch);
+    expect(mockDelivery.deliver).toHaveBeenCalledWith(mockBatch, {});
   });
 
   it('should retry failed deliveries up to maxRetries times and return last failed batch', async () => {
     mockDelivery.deliver.mockResolvedValue({ failedEvents: [{ id: 'event1' }] });
-    const result = await batchRetryManager.retry(mockBatch);
+    const result = await batchRetryManager.retry(mockBatch, {});
     expect(result).toEqual([{ id: 'event1' }]);
     expect(mockDelivery.deliver).toHaveBeenCalledTimes(maxRetries);
   });
@@ -40,7 +40,7 @@ describe('BatchRetryManager', () => {
 
     jest.useFakeTimers();
 
-    const retryPromise = batchRetryManager.retry(mockBatch);
+    const retryPromise = batchRetryManager.retry(mockBatch, {});
 
     // 1st retry: 100ms
     // 2nd retry: 200ms
@@ -67,7 +67,7 @@ describe('BatchRetryManager', () => {
 
     jest.useFakeTimers();
 
-    const retryPromise = batchRetryManager.retry(mockBatch);
+    const retryPromise = batchRetryManager.retry(mockBatch, {});
     // 100ms + 200ms + 300ms (maxRetryDelayMs) = 600ms
     await jest.advanceTimersByTimeAsync(600);
     const result = await retryPromise;

src/events/batch-retry-manager.ts

@@ -1,11 +1,12 @@
 import { logger } from '../application-logger';
 
+import { EventContext } from './default-event-dispatcher';
 import Event from './event';
 import { EventDeliveryResult } from './event-delivery';
 
-export type IEventDelivery = {
-  deliver(batch: Event[]): Promise<EventDeliveryResult>;
-};
+export interface IEventDelivery {
+  deliver(batch: Event[], context: EventContext): Promise<EventDeliveryResult>;
+}
 
 /**
  * Attempts to retry delivering a batch of events to the ingestionUrl up to `maxRetries` times
@@ -27,22 +28,22 @@ export default class BatchRetryManager {
   ) {}
 
   /** Re-attempts delivery of the provided batch, returns the UUIDs of events that failed retry. */
-  async retry(batch: Event[], attempt = 0): Promise<Event[]> {
+  async retry(batch: Event[], context: EventContext, attempt = 0): Promise<Event[]> {
     const { retryIntervalMs, maxRetryDelayMs, maxRetries } = this.config;
     const delay = Math.min(retryIntervalMs * Math.pow(2, attempt), maxRetryDelayMs);
     logger.info(
       `[BatchRetryManager] Retrying batch delivery of ${batch.length} events in ${delay}ms...`,
     );
     await new Promise((resolve) => setTimeout(resolve, delay));
 
-    const { failedEvents } = await this.delivery.deliver(batch);
+    const { failedEvents } = await this.delivery.deliver(batch, context);
     if (failedEvents.length === 0) {
       logger.info(`[BatchRetryManager] Batch delivery successfully after ${attempt + 1} tries.`);
       return [];
     }
     // attempts are zero-indexed while maxRetries is not
     if (attempt < maxRetries - 1) {
-      return this.retry(failedEvents, attempt + 1);
+      return this.retry(failedEvents, context, attempt + 1);
     } else {
       logger.warn(`[BatchRetryManager] Failed to deliver batch after ${maxRetries} tries, bailing`);
       return batch;

src/events/default-event-dispatcher.spec.ts

@@ -1,3 +1,5 @@
+import { v4 as randomUUID } from 'uuid';
+
 import ArrayBackedNamedEventQueue from './array-backed-named-event-queue';
 import BatchEventProcessor from './batch-event-processor';
 import DefaultEventDispatcher, {
@@ -120,6 +122,7 @@ describe('DefaultEventDispatcher', () => {
       let fetchOptions = fetch.mock.calls[0][1];
       let payload = JSON.parse(fetchOptions.body);
       expect(payload).toEqual({
+        context: {},
         eppo_events: [
           expect.objectContaining({ payload: { foo: 'event1' } }),
           expect.objectContaining({ payload: { foo: 'event2' } }),
@@ -139,6 +142,7 @@ describe('DefaultEventDispatcher', () => {
       fetchOptions = fetch.mock.calls[1][1];
       payload = JSON.parse(fetchOptions.body);
       expect(payload).toEqual({
+        context: {},
         eppo_events: [expect.objectContaining({ payload: { foo: 'event3' } })],
       });
     });
@@ -319,6 +323,57 @@ describe('DefaultEventDispatcher', () => {
     });
   });
 
+  describe('attachContext', () => {
+    it('should throw an error if the value is an object', () => {
+      const eventQueue = new ArrayBackedNamedEventQueue<Event>('test-queue');
+      const { dispatcher } = createDispatcher({ maxRetries: 1 }, eventQueue);
+      expect(() => dispatcher.attachContext('foo', {} as any)).toThrow();
+      expect(() => dispatcher.attachContext('foo', [] as any)).toThrow();
+    });
+
+    it('should not throw an error if the value is a string, number, boolean, or null', () => {
+      const eventQueue = new ArrayBackedNamedEventQueue<Event>('test-queue');
+      const { dispatcher } = createDispatcher({ maxRetries: 1 }, eventQueue);
+      expect(() => dispatcher.attachContext('foo', 'bar')).not.toThrow();
+      expect(() => dispatcher.attachContext('foo', 1)).not.toThrow();
+      expect(() => dispatcher.attachContext('foo', true)).not.toThrow();
+      expect(() => dispatcher.attachContext('foo', null)).not.toThrow();
+    });
+
+    it('should throw an error if the context value is too long', () => {
+      const eventQueue = new ArrayBackedNamedEventQueue<Event>('test-queue');
+      const { dispatcher } = createDispatcher({ maxRetries: 1 }, eventQueue);
+      expect(() => dispatcher.attachContext('foo', 'a'.repeat(2049))).toThrow();
+    });
+
+    it('attaches a context to be included with all events dispatched by this dispatcher', async () => {
+      const eventQueue = new ArrayBackedNamedEventQueue<Event>('test-queue');
+      const { dispatcher } = createDispatcher({ maxRetries: 1 }, eventQueue);
+      dispatcher.attachContext('foo', 'bar');
+      dispatcher.attachContext('baz', 'qux');
+      const event = {
+        uuid: randomUUID(),
+        payload: { foo: 'event1' },
+        timestamp: new Date().getTime(),
+        type: 'foo',
+      };
+      dispatcher.dispatch(event);
+      const fetch = global.fetch as jest.Mock;
+      fetch.mockResolvedValue({ ok: true, json: () => Promise.resolve([]) });
+
+      await new Promise((resolve) => setTimeout(resolve, 100));
+
+      expect(global.fetch).toHaveBeenCalledWith('http://example.com', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json', 'x-eppo-token': 'test-sdk-key' },
+        body: JSON.stringify({
+          eppo_events: [event],
+          context: { foo: 'bar', baz: 'qux' },
+        }),
+      });
+    });
+  });
+
   describe('validation', () => {
     it('should throw an error if the serialized event is too long', () => {
       const { dispatcher } = createDispatcher();

src/events/default-event-dispatcher.ts

@@ -25,7 +25,11 @@ export type EventDispatcherConfig = {
   maxRetries?: number;
 };
 
-export const MAX_EVENT_SERIALIZED_LENGTH = 4096;
+export type EventContext = Record<string, string | number | boolean | null>;
+
+const MAX_CONTEXT_SERIALIZED_LENGTH = 2048;
+const MAX_EVENT_SERIALIZED_LENGTH = 4096;
+
 export const DEFAULT_EVENT_DISPATCHER_BATCH_SIZE = 1_000;
 export const DEFAULT_EVENT_DISPATCHER_CONFIG: Omit<
   EventDispatcherConfig,
@@ -47,6 +51,7 @@ export default class DefaultEventDispatcher implements EventDispatcher {
   private readonly eventDelivery: EventDelivery;
   private readonly retryManager: BatchRetryManager;
   private readonly deliveryIntervalMs: number;
+  private readonly context: EventContext = {};
   private dispatchTimer: NodeJS.Timeout | null = null;
   private isOffline = false;
 
@@ -75,15 +80,36 @@ export default class DefaultEventDispatcher implements EventDispatcher {
     });
   }
 
+  attachContext(key: string, value: string | number | boolean | null): void {
+    this.ensureValidContext(key, value);
+    this.context[key] = value;
+  }
+
   dispatch(event: Event) {
     this.ensureValidEvent(event);
     this.batchProcessor.push(event);
     this.maybeScheduleNextDelivery();
   }
 
+  private ensureValidContext(key: string, value: string | number | boolean | null) {
+    if (value && (typeof value === 'object' || Array.isArray(value))) {
+      throw new Error('Context value must be a string, number, boolean, or null');
+    }
+    if (
+      value &&
+      JSON.stringify({ ...this.context, [key]: value }).length > MAX_CONTEXT_SERIALIZED_LENGTH
+    ) {
+      throw new Error(
+        `The total context size must be less than ${MAX_CONTEXT_SERIALIZED_LENGTH} characters`,
+      );
+    }
+  }
+
   private ensureValidEvent(event: Event) {
     if (JSON.stringify(event).length > MAX_EVENT_SERIALIZED_LENGTH) {
-      throw new Error('Event serialized length exceeds maximum allowed length of 4096');
+      throw new Error(
+        `Event serialized length exceeds maximum allowed length of ${MAX_EVENT_SERIALIZED_LENGTH}`,
+      );
     }
   }
 
@@ -100,10 +126,12 @@ export default class DefaultEventDispatcher implements EventDispatcher {
       return;
     }
 
-    const { failedEvents } = await this.eventDelivery.deliver(batch);
+    // make a defensive copy of the context to avoid mutating the original
+    const context = { ...this.context };
+    const { failedEvents } = await this.eventDelivery.deliver(batch, context);
     if (failedEvents.length > 0) {
       logger.warn('[EventDispatcher] Failed to deliver some events from batch, retrying...');
-      const failedRetry = await this.retryManager.retry(failedEvents);
+      const failedRetry = await this.retryManager.retry(failedEvents, context);
       if (failedRetry.length > 0) {
         // re-enqueue events that failed to retry
         this.batchProcessor.push(...failedRetry);

src/events/event-delivery.spec.ts

@@ -20,40 +20,40 @@ describe('EventDelivery', () => {
   it('should deliver events successfully when response is OK', async () => {
     const mockResponse = { ok: true, json: async () => ({}) };
     (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
-    const result = await eventDelivery.deliver(testBatch);
+    const result = await eventDelivery.deliver(testBatch, {});
     expect(result).toEqual({ failedEvents: [] });
     expect(global.fetch).toHaveBeenCalledWith(ingestionUrl, {
       method: 'POST',
       headers: { 'Content-Type': 'application/json', 'x-eppo-token': sdkKey },
-      body: JSON.stringify({ eppo_events: testBatch }),
+      body: JSON.stringify({ eppo_events: testBatch, context: {} }),
     });
   });
 
   it('should return failed result if response is not OK', async () => {
     const mockResponse = { ok: false };
     (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
-    const result = await eventDelivery.deliver(testBatch);
+    const result = await eventDelivery.deliver(testBatch, {});
     expect(result).toEqual({ failedEvents: testBatch });
   });
 
   it('should return failed events when response includes failed events', async () => {
     const failedEvents = ['1', '2'];
     const mockResponse = { ok: true, json: async () => ({ failed_events: failedEvents }) };
     (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
-    const result = await eventDelivery.deliver(testBatch);
+    const result = await eventDelivery.deliver(testBatch, {});
     expect(result).toEqual({ failedEvents: [testBatch[0], testBatch[1]] });
   });
 
   it('should return success=true if no failed events in the response', async () => {
     const mockResponse = { ok: true, json: async () => ({}) };
     (global.fetch as jest.Mock).mockResolvedValue(mockResponse);
-    const result = await eventDelivery.deliver(testBatch);
+    const result = await eventDelivery.deliver(testBatch, {});
     expect(result).toEqual({ failedEvents: [] });
   });
 
   it('should handle fetch errors gracefully', async () => {
     (global.fetch as jest.Mock).mockRejectedValue(new Error('Network error'));
-    const result = await eventDelivery.deliver(testBatch);
+    const result = await eventDelivery.deliver(testBatch, {});
     expect(result).toEqual({ failedEvents: testBatch });
   });
 });

src/events/event-delivery.ts

@@ -1,12 +1,14 @@
 import { logger } from '../application-logger';
 
+import { IEventDelivery } from './batch-retry-manager';
+import { EventContext } from './default-event-dispatcher';
 import Event from './event';
 
 export type EventDeliveryResult = {
   failedEvents: Event[];
 };
 
-export default class EventDelivery {
+export default class EventDelivery implements IEventDelivery {
   constructor(
     private readonly sdkKey: string,
     private readonly ingestionUrl: string,
@@ -16,15 +18,15 @@ export default class EventDelivery {
    * Delivers a batch of events to the ingestion URL endpoint. Returns the UUIDs of any events from
    * the batch that failed ingestion.
    */
-  async deliver(batch: Event[]): Promise<EventDeliveryResult> {
+  async deliver(batch: Event[], context: EventContext): Promise<EventDeliveryResult> {
     try {
       logger.info(
         `[EventDispatcher] Delivering batch of ${batch.length} events to ${this.ingestionUrl}...`,
       );
       const response = await fetch(this.ingestionUrl, {
         method: 'POST',
         headers: { 'Content-Type': 'application/json', 'x-eppo-token': this.sdkKey },
-        body: JSON.stringify({ eppo_events: batch }),
+        body: JSON.stringify({ eppo_events: batch, context }),
       });
       if (response.ok) {
         return await this.parseFailedEvents(response, batch);

src/events/event-dispatcher.ts

@@ -3,4 +3,15 @@ import Event from './event';
 export default interface EventDispatcher {
   /** Dispatches (enqueues) an event for eventual delivery. */
   dispatch(event: Event): void;
+  /**
+   * Attaches a context to be included with all events dispatched by this dispatcher.
+   * The context is delivered as a top-level object in the ingestion request payload.
+   * An existing key can be removed by providing a `null` value.
+   * Calling this method with same key multiple times causes only the last value to be used for the
+   * given key.
+   *
+   * @param key - The context entry key.
+   * @param value - The context entry value.
+   */
+  attachContext(key: string, value: string | number | boolean | null): void;
 }

src/events/no-op-event-dispatcher.ts

@@ -2,6 +2,11 @@ import Event from './event';
 import EventDispatcher from './event-dispatcher';
 
 export default class NoOpEventDispatcher implements EventDispatcher {
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  attachContext(key: string, value: string | number | boolean | null): void {
+    // Do nothing
+  }
+
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
   dispatch(_: Event): void {
     // Do nothing