feat: Environment ID support for hooks (#823)

**Requirements**

- [x] I have added test coverage for new or changed functionality
- [x] I have followed the repository's [pull request submission
guidelines](../blob/main/CONTRIBUTING.md#submitting-pull-requests)
- [x] I have validated my changes against all supported platform
versions

**Describe the solution you've provided**

If present, `environmentId` is now passed to the HookRunner in the
evaluation series.

For streaming connections:
- Response headers are now attached to the `open` event
(launchdarkly/js-eventsource#33)
- `StreamingProcessor` passes these headers to the stream listeners via
`processJson`. The listener for the PUT event extracts the
`environmentId` from the headers and passes this as initialization
metadata to the underlying feature store.

For polling connections:
- `PollingProcessor` retrieves the response headers via the underlying
`Requestor` and extracts the `environmentId` from the headers and passes
this as initialization metadata to the underlying feature store.

LDClient will then call `getInitMetaData()` on the underlying feature
store (if the feature store supports it) when executing a hook and pass
`environmentId` in the execution series data.

Currently only `InMemoryFeatureStore` has been modified to support
initialization metadata. This functionality can be added to other
feature stores by modifying `init()` to accept the optional
`initMetadata` parameter and implementing the optional
`getInitMetadata()` method.

Author: abarker-launchdarkly

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`)),
       );