chore: adds initMetadata handling to FDv2 datasources (#866)

**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

**Related issues**

@abarker-launchdarkly added initMetadata support while I was working on
FDv2 data sources. This catches the FDv2 datasources up to FDv1
implementations on main.

**Describe the solution you've provided**

Similar to existing FDv1 approach.

---------

Co-authored-by: Todd Anderson <[email protected]>

Author: tanderson-ld

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

@@ -8,6 +8,9 @@ describe('given a polling processor', () => {
     requestAllData: jest.fn(),
   };
   const longInterval = 100000;
+  const headers = {
+    'x-ld-envid': 'envKey',
+  };
   const allFDv2Events = {
     events: [
       {
@@ -66,7 +69,7 @@ describe('given a polling processor', () => {
   });
 
   it('calls callback on success', async () => {
-    requestor.requestAllData = jest.fn((cb) => cb(undefined, fdv2JsonData));
+    requestor.requestAllData = jest.fn((cb) => cb(undefined, fdv2JsonData, headers));
     let dataCallback;
     await new Promise<void>((resolve) => {
       dataCallback = jest.fn(() => {
@@ -77,18 +80,23 @@ describe('given a polling processor', () => {
     });
 
     expect(dataCallback).toHaveBeenNthCalledWith(1, true, {
-      basis: true,
-      id: `mockId`,
-      state: `mockState`,
-      updates: [
-        {
-          kind: `flag`,
-          key: `flagA`,
-          version: 123,
-          object: { objectFieldA: 'objectValueA' },
-        },
-      ],
-      version: 1,
+      initMetadata: {
+        environmentId: 'envKey',
+      },
+      payload: {
+        basis: true,
+        id: `mockId`,
+        state: `mockState`,
+        updates: [
+          {
+            kind: `flag`,
+            key: `flagA`,
+            version: 123,
+            object: { objectFieldA: 'objectValueA' },
+          },
+        ],
+        version: 1,
+      },
     });
   });
 
@@ -99,7 +107,7 @@ describe('given a polling processor', () => {
       new TestLogger(),
       true,
     );
-    requestor.requestAllData = jest.fn((cb) => cb(undefined, fdv1JsonData));
+    requestor.requestAllData = jest.fn((cb) => cb(undefined, fdv1JsonData, headers));
     let dataCallback;
     await new Promise<void>((resolve) => {
       dataCallback = jest.fn(() => {
@@ -110,24 +118,29 @@ describe('given a polling processor', () => {
     });
 
     expect(dataCallback).toHaveBeenNthCalledWith(1, true, {
-      basis: true,
-      id: `FDv1Fallback`,
-      state: `FDv1Fallback`,
-      updates: [
-        {
-          kind: `flag`,
-          key: `flagA`,
-          version: 456,
-          object: { version: 456 },
-        },
-        {
-          kind: `segment`,
-          key: `segmentA`,
-          version: 789,
-          object: { version: 789 },
-        },
-      ],
-      version: 1,
+      initMetadata: {
+        environmentId: 'envKey',
+      },
+      payload: {
+        basis: true,
+        id: `FDv1Fallback`,
+        state: `FDv1Fallback`,
+        updates: [
+          {
+            kind: `flag`,
+            key: `flagA`,
+            version: 456,
+            object: { version: 456 },
+          },
+          {
+            kind: `segment`,
+            key: `segmentA`,
+            version: 789,
+            object: { version: 789 },
+          },
+        ],
+        version: 1,
+      },
     });
   });
 
@@ -138,7 +151,7 @@ describe('given a polling processor', () => {
       new TestLogger(),
       true,
     );
-    requestor.requestAllData = jest.fn((cb) => cb(undefined, fdv2JsonData));
+    requestor.requestAllData = jest.fn((cb) => cb(undefined, fdv2JsonData, headers));
     let dataCallback;
     await new Promise<void>((resolve) => {
       dataCallback = jest.fn(() => {
@@ -159,6 +172,9 @@ describe('given a polling processor with a short poll duration', () => {
     requestAllData: jest.fn(),
   };
   const shortInterval = 0.1;
+  const headers = {
+    'x-ld-envid': 'envKey',
+  };
   const allEvents = {
     events: [
       {
@@ -206,7 +222,7 @@ describe('given a polling processor with a short poll duration', () => {
     const expectedCalls = new Promise<void>((resolve) => {
       let callCount = 0;
       requestor.requestAllData = jest.fn((cb) => {
-        cb(undefined, jsonData);
+        cb(undefined, jsonData, headers);
         callCount += 1;
         if (callCount >= 5) {
           resolve();
@@ -256,7 +272,7 @@ describe('given a polling processor with a short poll duration', () => {
     const expectedCalls = new Promise<void>((resolve) => {
       let callCount = 0;
       requestor.requestAllData = jest.fn((cb) => {
-        cb(undefined, '{sad');
+        cb(undefined, '{sad', headers);
         callCount += 1;
         if (callCount >= 5) {
           resolve();

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

@@ -32,6 +32,9 @@ function getBasicConfiguration(inLogger: LDLogger) {
 
 const dateNowString = '2023-08-10';
 const sdkKey = 'my-sdk-key';
+const openHeaders = {
+  'x-ld-envid': 'envKey',
+};
 const events = {
   'server-intent': {
     data: '{"payloads": [{"intentCode": "xfer-full", "id": "mockId"}]}',
@@ -59,6 +62,7 @@ beforeEach(() => {
 const createMockEventSource = (streamUri: string = '', options: any = {}) => ({
   streamUri,
   options,
+  onopen: jest.fn(),
   onclose: jest.fn(),
   addEventListener: jest.fn(),
   close: jest.fn(),
@@ -71,6 +75,7 @@ describe('given a stream processor with mock event source', () => {
   let mockEventSource: any;
   let mockDataCallback: jest.Mock;
   let mockStatusCallback: jest.Mock;
+  let simulateOpen: () => void;
   let simulateEvents: (e?: any) => void;
   let simulateError: (e: { status: number; message: string }) => boolean;
 
@@ -92,6 +97,11 @@ describe('given a stream processor with mock event source', () => {
         return mockEventSource;
       }),
     } as any;
+    simulateOpen = () => {
+      mockEventSource.onopen({
+        headers: openHeaders,
+      });
+    };
     simulateEvents = (e: any = events) => {
       // positions in these call arrays match order of handler attachment in payloadStreamReader
       mockEventSource.addEventListener.mock.calls[0][1](e['server-intent']); // server intent listener
@@ -230,20 +240,26 @@ describe('given a stream processor with mock event source', () => {
   });
 
   it('executes payload listener', () => {
+    simulateOpen();
     simulateEvents();
     expect(mockDataCallback).toHaveBeenNthCalledWith(1, true, {
-      basis: true,
-      id: `mockId`,
-      state: `mockState`,
-      updates: [
-        {
-          kind: `flag`,
-          key: `flagA`,
-          version: 123,
-          object: { objectFieldA: 'objectValueA' },
-        },
-      ],
-      version: 1,
+      initMetadata: {
+        environmentId: 'envKey',
+      },
+      payload: {
+        basis: true,
+        id: `mockId`,
+        state: `mockState`,
+        updates: [
+          {
+            kind: `flag`,
+            key: `flagA`,
+            version: 123,
+            object: { objectFieldA: 'objectValueA' },
+          },
+        ],
+        version: 1,
+      },
     });
   });
 

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

@@ -17,74 +17,89 @@ beforeEach(() => {
 });
 
 const fullTransferPayload = {
-  id: 'payloadID',
-  version: 99,
-  state: 'initial',
-  basis: true,
-  updates: [
-    {
-      kind: 'flag',
-      key: 'flagkey',
-      version: 1,
-      object: {
+  initMetadata: {
+    environmentId: 'envId',
+  },
+  payload: {
+    id: 'payloadID',
+    version: 99,
+    state: 'initial',
+    basis: true,
+    updates: [
+      {
+        kind: 'flag',
         key: 'flagkey',
         version: 1,
+        object: {
+          key: 'flagkey',
+          version: 1,
+        },
       },
-    },
-    {
-      kind: 'segment',
-      key: 'segkey',
-      version: 1,
-      object: {
+      {
+        kind: 'segment',
         key: 'segkey',
-        version: 2,
+        version: 1,
+        object: {
+          key: 'segkey',
+          version: 2,
+        },
       },
-    },
-  ],
+    ],
+  },
 };
 
 const changesTransferPayload = {
-  id: 'payloadID',
-  version: 99,
-  state: 'changes',
-  basis: false,
-  updates: [
-    {
-      kind: 'flag',
-      key: 'flagkey',
-      version: 1,
-      object: {
+  initMetadata: {
+    environmentId: 'envId',
+  },
+  payload: {
+    id: 'payloadID',
+    version: 99,
+    state: 'changes',
+    basis: false,
+    updates: [
+      {
+        kind: 'flag',
         key: 'flagkey',
         version: 1,
+        object: {
+          key: 'flagkey',
+          version: 1,
+        },
       },
-    },
-    {
-      kind: 'segment',
-      key: 'segkey',
-      version: 2,
-      object: {
+      {
+        kind: 'segment',
         key: 'segkey',
         version: 2,
+        object: {
+          key: 'segkey',
+          version: 2,
+        },
       },
-    },
-    {
-      kind: 'flag',
-      key: 'deletedFlag',
-      version: 3,
-      object: {
+      {
+        kind: 'flag',
         key: 'deletedFlag',
         version: 3,
+        object: {
+          key: 'deletedFlag',
+          version: 3,
+        },
+        deleted: true,
       },
-      deleted: true,
-    },
-  ],
+    ],
+  },
 };
 
 const changesTransferNone = {
-  id: 'payloadID',
-  version: 99,
-  basis: false,
-  updates: [],
+  initMetadata: {
+    environmentId: 'envId',
+  },
+  payload: {
+    id: 'payloadID',
+    version: 99,
+    basis: false,
+    updates: [],
+  },
 };
 
 describe('createPayloadListenerFDv2', () => {
@@ -120,7 +135,7 @@ describe('createPayloadListenerFDv2', () => {
         },
       },
       basisReceived,
-      undefined,
+      { environmentId: 'envId' },
       'initial',
     );
   });
@@ -154,7 +169,7 @@ describe('createPayloadListenerFDv2', () => {
         },
       },
       basisReceived,
-      undefined,
+      { environmentId: 'envId' },
       'changes',
     );
   });