fix(events): Fix two event ingestion bugs (#175)

Author: felipecsl

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

@@ -28,6 +28,7 @@ const createDispatcher = (
     retryIntervalMs: 300,
     maxRetryDelayMs: 5000,
     maxRetries: 3,
+    sdkKey: 'test-sdk-key',
   };
   const config = { ...defaultConfig, ...configOverrides };
   const batchProcessor = new BatchEventProcessor(eventQueue, batchSize);
@@ -107,30 +108,34 @@ describe('DefaultEventDispatcher', () => {
         'http://example.com',
         expect.objectContaining({
           method: 'POST',
-          headers: { 'Content-Type': 'application/json' },
+          headers: { 'Content-Type': 'application/json', 'x-eppo-token': 'test-sdk-key' },
         }),
       );
 
       let fetchOptions = fetch.mock.calls[0][1];
       let payload = JSON.parse(fetchOptions.body);
-      expect(payload).toEqual([
-        expect.objectContaining({ payload: { foo: 'event1' } }),
-        expect.objectContaining({ payload: { foo: 'event2' } }),
-      ]);
+      expect(payload).toEqual({
+        eppo_events: [
+          expect.objectContaining({ payload: { foo: 'event1' } }),
+          expect.objectContaining({ payload: { foo: 'event2' } }),
+        ],
+      });
 
       await new Promise((resolve) => setTimeout(resolve, 100));
 
       expect(global.fetch).toHaveBeenCalledWith(
         'http://example.com',
         expect.objectContaining({
           method: 'POST',
-          headers: { 'Content-Type': 'application/json' },
+          headers: { 'Content-Type': 'application/json', 'x-eppo-token': 'test-sdk-key' },
         }),
       );
 
       fetchOptions = fetch.mock.calls[1][1];
       payload = JSON.parse(fetchOptions.body);
-      expect(payload).toEqual([expect.objectContaining({ payload: { foo: 'event3' } })]);
+      expect(payload).toEqual({
+        eppo_events: [expect.objectContaining({ payload: { foo: 'event3' } })],
+      });
     });
 
     it('does not schedule delivery if the queue is empty', async () => {

src/events/default-event-dispatcher.ts

@@ -10,6 +10,8 @@ import NoOpEventDispatcher from './no-op-event-dispatcher';
 import SdkKeyDecoder from './sdk-key-decoder';
 
 export type EventDispatcherConfig = {
+  // The Eppo SDK key
+  sdkKey: string;
   // target url to deliver events to
   ingestionUrl: string;
   // number of milliseconds to wait between each batch delivery
@@ -25,7 +27,10 @@ export type EventDispatcherConfig = {
 // TODO: Have more realistic default batch size based on average event payload size once we have
 //  more concrete data.
 export const DEFAULT_EVENT_DISPATCHER_BATCH_SIZE = 100;
-export const DEFAULT_EVENT_DISPATCHER_CONFIG: Omit<EventDispatcherConfig, 'ingestionUrl'> = {
+export const DEFAULT_EVENT_DISPATCHER_CONFIG: Omit<
+  EventDispatcherConfig,
+  'ingestionUrl' | 'sdkKey'
+> = {
   deliveryIntervalMs: 10_000,
   retryIntervalMs: 5_000,
   maxRetryDelayMs: 30_000,
@@ -51,11 +56,12 @@ export default class DefaultEventDispatcher implements EventDispatcher {
     config: EventDispatcherConfig,
   ) {
     this.ensureConfigFields(config);
-    this.eventDelivery = new EventDelivery(config.ingestionUrl);
+    const { sdkKey, ingestionUrl, retryIntervalMs, maxRetryDelayMs, maxRetries = 3 } = config;
+    this.eventDelivery = new EventDelivery(sdkKey, ingestionUrl);
     this.retryManager = new BatchRetryManager(this.eventDelivery, {
-      retryIntervalMs: config.retryIntervalMs,
-      maxRetryDelayMs: config.maxRetryDelayMs,
-      maxRetries: config.maxRetries || 3,
+      retryIntervalMs,
+      maxRetryDelayMs,
+      maxRetries,
     });
     this.deliveryIntervalMs = config.deliveryIntervalMs;
     this.networkStatusListener.onNetworkStatusChange((isOffline) => {
@@ -134,7 +140,7 @@ export function newDefaultEventDispatcher(
   networkStatusListener: NetworkStatusListener,
   sdkKey: string,
   batchSize: number = DEFAULT_EVENT_DISPATCHER_BATCH_SIZE,
-  config: Omit<EventDispatcherConfig, 'ingestionUrl'> = DEFAULT_EVENT_DISPATCHER_CONFIG,
+  config: Omit<EventDispatcherConfig, 'ingestionUrl' | 'sdkKey'> = DEFAULT_EVENT_DISPATCHER_CONFIG,
 ): EventDispatcher {
   const sdkKeyDecoder = new SdkKeyDecoder();
   const ingestionUrl = sdkKeyDecoder.decodeEventIngestionUrl(sdkKey);
@@ -147,6 +153,6 @@ export function newDefaultEventDispatcher(
   return new DefaultEventDispatcher(
     new BatchEventProcessor(eventQueue, batchSize),
     networkStatusListener,
-    { ...config, ingestionUrl },
+    { ...config, ingestionUrl, sdkKey },
   );
 }

src/events/event-delivery.ts

@@ -3,7 +3,7 @@ import { logger } from '../application-logger';
 import { Event } from './event-dispatcher';
 
 export default class EventDelivery {
-  constructor(private readonly ingestionUrl: string) {}
+  constructor(private readonly sdkKey: string, private readonly ingestionUrl: string) {}
 
   async deliver(batch: Event[]): Promise<boolean> {
     try {
@@ -12,9 +12,8 @@ export default class EventDelivery {
       );
       const response = await fetch(this.ingestionUrl, {
         method: 'POST',
-        headers: { 'Content-Type': 'application/json' },
-        // TODO: Figure out proper request body encoding format for batch, using JSON for now
-        body: JSON.stringify(batch),
+        headers: { 'Content-Type': 'application/json', 'x-eppo-token': this.sdkKey },
+        body: JSON.stringify({ eppo_events: batch }),
       });
       // TODO: Parse response to check `failed_event_uploads` for any failed event ingestions in the batch
       return response.ok;