feat: Environment ID support for hooks

packages/sdk/server-node/package.json

@@ -47,7 +47,7 @@
   "dependencies": {
     "@launchdarkly/js-server-sdk-common": "2.14.0",
     "https-proxy-agent": "^5.0.1",
-    "launchdarkly-eventsource": "2.0.3"
+    "launchdarkly-eventsource": "2.1.0"
   },
   "devDependencies": {
     "@trivago/prettier-plugin-sort-imports": "^4.1.1",

packages/shared/common/__tests__/internal/metadata/InitMetadata.test.ts

@@ -0,0 +1,17 @@
+import { initMetadataFromHeaders } from '../../../src/internal/metadata';
+
+it('handles passing undefined headers', () => {
+  expect(initMetadataFromHeaders()).toBeUndefined();
+});
+
+it('handles missing x-ld-envid header', () => {
+  expect(initMetadataFromHeaders({})).toBeUndefined();
+});
+
+it('retrieves environmentId from headers', () => {
+  expect(initMetadataFromHeaders({ 'x-ld-envid': '12345' })).toEqual({ environmentId: '12345' });
+});
+
+it('retrieves environmentId from mixed case header', () => {
+  expect(initMetadataFromHeaders({ 'X-LD-EnvId': '12345' })).toEqual({ environmentId: '12345' });
+});

packages/shared/common/src/api/platform/EventSource.ts

@@ -4,13 +4,13 @@ export type EventName = string;
 export type EventListener = (event?: { data?: any }) => void;
 export type ProcessStreamResponse = {
   deserializeData: (data: string) => any;
-  processJson: (json: any) => void;
+  processJson: (json: any, initHeaders?: { [key: string]: string }) => void;
 };
 
 export interface EventSource {
   onclose: (() => void) | undefined;
   onerror: ((err?: HttpErrorResponse) => void) | undefined;
-  onopen: (() => void) | undefined;
+  onopen: ((e: { headers?: { [key: string]: string } }) => void) | undefined;
   onretrying: ((e: { delayMillis: number }) => void) | undefined;
 
   addEventListener(type: EventName, listener: EventListener): void;

packages/shared/common/src/internal/index.ts

@@ -3,3 +3,4 @@ export * from './diagnostics';
 export * from './evaluation';
 export * from './events';
 export * from './fdv2';
+export * from './metadata';

packages/shared/common/src/internal/metadata/InitMetadata.ts

@@ -0,0 +1,26 @@
+/**
+ * Metadata used to initialize an LDFeatureStore.
+ */
+export interface InitMetadata {
+  environmentId: string;
+}
+
+/**
+ * Creates an InitMetadata object from initialization headers.
+ *
+ * @param initHeaders Initialization headers received when establishing
+ * a streaming or polling connection to LD.
+ * @returns InitMetadata object, or undefined if initHeaders is undefined
+ * or missing the required header values.
+ */
+export function initMetadataFromHeaders(initHeaders?: {
+  [key: string]: string;
+}): InitMetadata | undefined {
+  if (initHeaders) {
+    const envIdKey = Object.keys(initHeaders).find((key) => key.toLowerCase() === 'x-ld-envid');
+    if (envIdKey) {
+      return { environmentId: initHeaders[envIdKey] };
+    }
+  }
+  return undefined;
+}

packages/shared/common/src/internal/metadata/index.ts

@@ -0,0 +1,3 @@
+import { InitMetadata, initMetadataFromHeaders } from './InitMetadata';
+
+export { InitMetadata, initMetadataFromHeaders };

packages/shared/sdk-server/__tests__/data_sources/DataSourceUpdates.test.ts

@@ -1,11 +1,29 @@
 import { AsyncQueue } from 'launchdarkly-js-test-helpers';
 
+import { internal } from '@launchdarkly/js-sdk-common';
+
 import { LDFeatureStore } from '../../src/api/subsystems';
 import promisify from '../../src/async/promisify';
 import DataSourceUpdates from '../../src/data_sources/DataSourceUpdates';
 import InMemoryFeatureStore from '../../src/store/InMemoryFeatureStore';
 import VersionedDataKinds from '../../src/store/VersionedDataKinds';
 
+type InitMetadata = internal.InitMetadata;
+
+it('passes initialization metadata to underlying feature store', () => {
+  const metadata: InitMetadata = { environmentId: '12345' };
+  const store = new InMemoryFeatureStore();
+  store.init = jest.fn();
+  const updates = new DataSourceUpdates(
+    store,
+    () => false,
+    () => {},
+  );
+  updates.init({}, () => {}, metadata);
+  expect(store.init).toHaveBeenCalledTimes(1);
+  expect(store.init).toHaveBeenNthCalledWith(1, expect.any(Object), expect.any(Function), metadata);
+});
+
 describe.each([true, false])(
   'given a DataSourceUpdates with in memory store and change listeners: %s',
   (listen) => {

packages/shared/sdk-server/__tests__/data_sources/PollingProcessor.test.ts

@@ -76,6 +76,18 @@ describe('given an event processor', () => {
     expect(flags).toEqual(allData.flags);
     expect(segments).toEqual(allData.segments);
   });
+
+  it('initializes the feature store with metadata', () => {
+    const initHeaders = {
+      'x-ld-envid': '12345',
+    };
+    requestor.requestAllData = jest.fn((cb) => cb(undefined, jsonData, initHeaders));
+
+    processor.start();
+    const metadata = storeFacade.getInitMetadata?.();
+
+    expect(metadata).toEqual({ environmentId: '12345' });
+  });
 });
 
 describe('given a polling processor with a short poll duration', () => {

packages/shared/sdk-server/__tests__/data_sources/Requestor.test.ts

@@ -49,7 +49,7 @@ describe('given a requestor', () => {
               throw new Error('Function not implemented.');
             },
             entries(): Iterable<[string, string]> {
-              throw new Error('Function not implemented.');
+              return testHeaders ? Object.entries(testHeaders) : [];
             },
             has(_name: string): boolean {
               throw new Error('Function not implemented.');
@@ -115,7 +115,9 @@ describe('given a requestor', () => {
   });
 
   it('stores and sends etags', async () => {
-    testHeaders.etag = 'abc123';
+    testHeaders = {
+      etag: 'abc123',
+    };
     testResponse = 'a response';
     const res1 = await promisify<{ err: any; body: any }>((cb) => {
       requestor.requestAllData((err, body) => cb({ err, body }));
@@ -134,4 +136,17 @@ describe('given a requestor', () => {
     expect(req1.options.headers?.['if-none-match']).toBe(undefined);
     expect(req2.options.headers?.['if-none-match']).toBe((testHeaders.etag = 'abc123'));
   });
+
+  it('passes response headers to callback', async () => {
+    testHeaders = {
+      header1: 'value1',
+      header2: 'value2',
+      header3: 'value3',
+    };
+    const res = await promisify<{ err: any; body: any; headers: any }>((cb) => {
+      requestor.requestAllData((err, body, headers) => cb({ err, body, headers }));
+    });
+
+    expect(res.headers).toEqual(testHeaders);
+  });
 });

packages/shared/sdk-server/__tests__/data_sources/StreamingProcessor.test.ts

@@ -138,7 +138,7 @@ describe('given a stream processor with mock event source', () => {
   });
 
   it('uses expected uri and eventSource init args', () => {
-    expect(basicPlatform.requests.createEventSource).toBeCalledWith(
+    expect(basicPlatform.requests.createEventSource).toHaveBeenCalledWith(
       `${serviceEndpoints.streaming}/all`,
       {
         errorFilter: expect.any(Function),
@@ -200,32 +200,44 @@ describe('given a stream processor with mock event source', () => {
     const patchHandler = mockEventSource.addEventListener.mock.calls[1][1];
     patchHandler(event);
 
-    expect(mockListener.deserializeData).toBeCalledTimes(2);
-    expect(mockListener.processJson).toBeCalledTimes(2);
+    expect(mockListener.deserializeData).toHaveBeenCalledTimes(2);
+    expect(mockListener.processJson).toHaveBeenCalledTimes(2);
+  });
+
+  it('passes initialization headers to listener', () => {
+    const headers = {
+      header1: 'value1',
+      header2: 'value2',
+      header3: 'value3',
+    };
+    mockEventSource.onopen({ type: 'open', headers });
+    simulatePutEvent();
+    expect(mockListener.processJson).toHaveBeenCalledTimes(1);
+    expect(mockListener.processJson).toHaveBeenNthCalledWith(1, expect.any(Object), headers);
   });
 
   it('passes error to callback if json data is malformed', async () => {
     (mockListener.deserializeData as jest.Mock).mockReturnValue(false);
     simulatePutEvent();
 
-    expect(logger.error).toBeCalledWith(expect.stringMatching(/invalid data in "put"/));
-    expect(logger.debug).toBeCalledWith(expect.stringMatching(/invalid json/i));
+    expect(logger.error).toHaveBeenCalledWith(expect.stringMatching(/invalid data in "put"/));
+    expect(logger.debug).toHaveBeenCalledWith(expect.stringMatching(/invalid json/i));
     expect(mockErrorHandler.mock.lastCall[0].message).toMatch(/malformed json/i);
   });
 
   it('calls error handler if event.data prop is missing', async () => {
     simulatePutEvent({ flags: {} });
 
-    expect(mockListener.deserializeData).not.toBeCalled();
-    expect(mockListener.processJson).not.toBeCalled();
+    expect(mockListener.deserializeData).not.toHaveBeenCalled();
+    expect(mockListener.processJson).not.toHaveBeenCalled();
     expect(mockErrorHandler.mock.lastCall[0].message).toMatch(/unexpected payload/i);
   });
 
   it('closes and stops', async () => {
     streamingProcessor.close();
 
-    expect(streamingProcessor.stop).toBeCalled();
-    expect(mockEventSource.close).toBeCalled();
+    expect(streamingProcessor.stop).toHaveBeenCalled();
+    expect(mockEventSource.close).toHaveBeenCalled();
     // @ts-ignore
     expect(streamingProcessor.eventSource).toBeUndefined();
   });
@@ -249,8 +261,8 @@ describe('given a stream processor with mock event source', () => {
       const willRetry = simulateError(testError);
 
       expect(willRetry).toBeTruthy();
-      expect(mockErrorHandler).not.toBeCalled();
-      expect(logger.warn).toBeCalledWith(
+      expect(mockErrorHandler).not.toHaveBeenCalled();
+      expect(logger.warn).toHaveBeenCalledWith(
         expect.stringMatching(new RegExp(`${status}.*will retry`)),
       );
 
@@ -270,10 +282,10 @@ describe('given a stream processor with mock event source', () => {
       const willRetry = simulateError(testError);
 
       expect(willRetry).toBeFalsy();
-      expect(mockErrorHandler).toBeCalledWith(
+      expect(mockErrorHandler).toHaveBeenCalledWith(
         new LDStreamingError(DataSourceErrorKind.Unknown, testError.message, testError.status),
       );
-      expect(logger.error).toBeCalledWith(
+      expect(logger.error).toHaveBeenCalledWith(
         expect.stringMatching(new RegExp(`${status}.*permanently`)),
       );
 

packages/shared/sdk-server/__tests__/data_sources/createStreamListeners.test.ts

@@ -94,13 +94,36 @@ describe('createStreamListeners', () => {
 
       processJson(allData);
 
-      expect(logger.debug).toBeCalledWith(expect.stringMatching(/initializing/i));
-      expect(dataSourceUpdates.init).toBeCalledWith(
+      expect(logger.debug).toHaveBeenCalledWith(expect.stringMatching(/initializing/i));
+      expect(dataSourceUpdates.init).toHaveBeenCalledWith(
         {
           features: flags,
           segments,
         },
         onPutCompleteHandler,
+        undefined,
+      );
+    });
+
+    test('data source init is called with initialization metadata', async () => {
+      const listeners = createStreamListeners(dataSourceUpdates, logger, onCompleteHandlers);
+      const { processJson } = listeners.get('put')!;
+      const {
+        data: { flags, segments },
+      } = allData;
+      const initHeaders = {
+        'x-ld-envid': '12345',
+      };
+      processJson(allData, initHeaders);
+
+      expect(logger.debug).toHaveBeenCalledWith(expect.stringMatching(/initializing/i));
+      expect(dataSourceUpdates.init).toHaveBeenCalledWith(
+        {
+          features: flags,
+          segments,
+        },
+        onPutCompleteHandler,
+        { environmentId: '12345' },
       );
     });
   });
@@ -121,8 +144,8 @@ describe('createStreamListeners', () => {
 
       processJson(patchData);
 
-      expect(logger.debug).toBeCalledWith(expect.stringMatching(/updating/i));
-      expect(dataSourceUpdates.upsert).toBeCalledWith(kind, data, onPatchCompleteHandler);
+      expect(logger.debug).toHaveBeenCalledWith(expect.stringMatching(/updating/i));
+      expect(dataSourceUpdates.upsert).toHaveBeenCalledWith(kind, data, onPatchCompleteHandler);
     });
 
     test('data source upsert not called missing kind', async () => {
@@ -132,7 +155,7 @@ describe('createStreamListeners', () => {
 
       processJson(missingKind);
 
-      expect(dataSourceUpdates.upsert).not.toBeCalled();
+      expect(dataSourceUpdates.upsert).not.toHaveBeenCalled();
     });
 
     test('data source upsert not called wrong namespace path', async () => {
@@ -142,7 +165,7 @@ describe('createStreamListeners', () => {
 
       processJson(wrongKey);
 
-      expect(dataSourceUpdates.upsert).not.toBeCalled();
+      expect(dataSourceUpdates.upsert).not.toHaveBeenCalled();
     });
   });
 
@@ -162,8 +185,8 @@ describe('createStreamListeners', () => {
 
       processJson(deleteData);
 
-      expect(logger.debug).toBeCalledWith(expect.stringMatching(/deleting/i));
-      expect(dataSourceUpdates.upsert).toBeCalledWith(
+      expect(logger.debug).toHaveBeenCalledWith(expect.stringMatching(/deleting/i));
+      expect(dataSourceUpdates.upsert).toHaveBeenCalledWith(
         kind,
         { key: 'flagkey', version, deleted: true },
         onDeleteCompleteHandler,
@@ -177,7 +200,7 @@ describe('createStreamListeners', () => {
 
       processJson(missingKind);
 
-      expect(dataSourceUpdates.upsert).not.toBeCalled();
+      expect(dataSourceUpdates.upsert).not.toHaveBeenCalled();
     });
 
     test('data source upsert not called wrong namespace path', async () => {
@@ -187,7 +210,7 @@ describe('createStreamListeners', () => {
 
       processJson(wrongKey);
 
-      expect(dataSourceUpdates.upsert).not.toBeCalled();
+      expect(dataSourceUpdates.upsert).not.toHaveBeenCalled();
     });
   });
 });

packages/shared/sdk-server/__tests__/hooks/HookRunner.test.ts

@@ -36,6 +36,7 @@ describe('given a HookRunner', () => {
         reason: { kind: 'ERROR', errorKind: 'FLAG_NOT_FOUND' },
         variationIndex: null,
       }),
+      '12345',
     );
 
     testHook.verifyAfter(
@@ -44,6 +45,7 @@ describe('given a HookRunner', () => {
         context: { ...defaultUser },
         defaultValue: false,
         method: 'LDClient.variation',
+        environmentId: '12345',
       },
       { added: 'added data' },
       {
@@ -187,13 +189,15 @@ it('can add a hook after initialization', async () => {
       reason: { kind: 'FALLTHROUGH' },
       variationIndex: 0,
     }),
+    '12345',
   );
   testHook.verifyBefore(
     {
       flagKey: 'flagKey',
       context: { ...defaultUser },
       defaultValue: false,
       method: 'LDClient.variation',
+      environmentId: '12345',
     },
     {},
   );
@@ -203,6 +207,7 @@ it('can add a hook after initialization', async () => {
       context: { ...defaultUser },
       defaultValue: false,
       method: 'LDClient.variation',
+      environmentId: '12345',
     },
     {},
     {