feat: Add support for migrations.

contract-tests/index.js

@@ -28,6 +28,11 @@ app.get('/', (req, res) => {
       'tags',
       'big-segments',
       'user-type',
+      'migrations',
+      'event-sampling',
+      'config-override-kind',
+      'metric-kind',
+      'strongly-typed',
     ],
   });
 });

contract-tests/sdkClientEntity.js

@@ -1,4 +1,12 @@
-import ld from 'node-server-sdk';
+import got from 'got';
+import ld, {
+  LDConcurrentExecution,
+  LDExecutionOrdering,
+  LDMigrationError,
+  LDMigrationSuccess,
+  LDSerialExecution,
+  createMigration,
+} from 'node-server-sdk';
 
 import BigSegmentTestStore from './BigSegmentTestStore.js';
 import { Log, sdkLogger } from './log.js';
@@ -54,6 +62,30 @@ export function makeSdkConfig(options, tag) {
   return cf;
 }
 
+function getExecution(order) {
+  switch (order) {
+    case 'serial': {
+      return new LDSerialExecution(LDExecutionOrdering.Fixed);
+    }
+    case 'random': {
+      return new LDSerialExecution(LDExecutionOrdering.Random);
+    }
+    case 'concurrent': {
+      return new LDConcurrentExecution();
+    }
+    default: {
+      throw new Error('Unsupported execution order.');
+    }
+  }
+}
+
+function makeMigrationPostOptions(payload) {
+  if (payload) {
+    return { body: payload };
+  }
+  return {};
+}
+
 export async function newSdkClientEntity(options) {
   const c = {};
   const log = Log(options.tag);
@@ -92,10 +124,30 @@ export async function newSdkClientEntity(options) {
       case 'evaluate': {
         const pe = params.evaluate;
         if (pe.detail) {
-          return await client.variationDetail(pe.flagKey, pe.context || pe.user, pe.defaultValue);
+          switch(pe.valueType) {
+            case "bool":
+              return await client.boolVariationDetail(pe.flagKey, pe.context || pe.user, pe.defaultValue);
+            case "int": // Intentional fallthrough.
+            case "double":
+              return await client.numberVariationDetail(pe.flagKey, pe.context || pe.user, pe.defaultValue);
+            case "string":
+              return await client.stringVariationDetail(pe.flagKey, pe.context || pe.user, pe.defaultValue);
+            default:
+              return await client.variationDetail(pe.flagKey, pe.context || pe.user, pe.defaultValue);
+          }
+
         } else {
-          const value = await client.variation(pe.flagKey, pe.context || pe.user, pe.defaultValue);
-          return { value };
+          switch(pe.valueType) {
+            case "bool":
+              return {value: await client.boolVariation(pe.flagKey, pe.context || pe.user, pe.defaultValue)};
+            case "int": // Intentional fallthrough.
+            case "double":
+              return {value: await client.numberVariation(pe.flagKey, pe.context || pe.user, pe.defaultValue)};
+            case "string":
+              return {value: await client.stringVariation(pe.flagKey, pe.context || pe.user, pe.defaultValue)};
+            default:
+              return {value: await client.variation(pe.flagKey, pe.context || pe.user, pe.defaultValue)};
+          }
         }
       }
 
@@ -126,6 +178,101 @@ export async function newSdkClientEntity(options) {
       case 'getBigSegmentStoreStatus':
         return await client.bigSegmentStoreStatusProvider.requireStatus();
 
+      case 'migrationVariation':
+        const migrationVariation = params.migrationVariation;
+        const res = await client.migrationVariation(
+          migrationVariation.key,
+          migrationVariation.context,
+          migrationVariation.defaultStage,
+        );
+        return { result: res.value };
+
+      case 'migrationOperation':
+        const migrationOperation = params.migrationOperation;
+        const readExecutionOrder = migrationOperation.readExecutionOrder;
+
+        const migration = createMigration(client, {
+          execution: getExecution(readExecutionOrder),
+          latencyTracking: migrationOperation.trackLatency,
+          errorTracking: migrationOperation.trackErrors,
+          check: migrationOperation.trackConsistency ? (a, b) => a === b : undefined,
+          readNew: async (payload) => {
+            try {
+              const res = await got.post(
+                migrationOperation.newEndpoint,
+                makeMigrationPostOptions(payload),
+              );
+              return LDMigrationSuccess(res.body);
+            } catch (err) {
+              return LDMigrationError(err.message);
+            }
+          },
+          writeNew: async (payload) => {
+            try {
+              const res = await got.post(
+                migrationOperation.newEndpoint,
+                makeMigrationPostOptions(payload),
+              );
+              return LDMigrationSuccess(res.body);
+            } catch (err) {
+              return LDMigrationError(err.message);
+            }
+          },
+          readOld: async (payload) => {
+            try {
+              const res = await got.post(
+                migrationOperation.oldEndpoint,
+                makeMigrationPostOptions(payload),
+              );
+              return LDMigrationSuccess(res.body);
+            } catch (err) {
+              return LDMigrationError(err.message);
+            }
+          },
+          writeOld: async (payload) => {
+            try {
+              const res = await got.post(
+                migrationOperation.oldEndpoint,
+                makeMigrationPostOptions(payload),
+              );
+              return LDMigrationSuccess(res.body);
+            } catch (err) {
+              return LDMigrationError(err.message);
+            }
+          },
+        });
+
+        switch (migrationOperation.operation) {
+          case 'read': {
+            const res = await migration.read(
+              migrationOperation.key,
+              migrationOperation.context,
+              migrationOperation.defaultStage,
+              migrationOperation.payload,
+            );
+            if (res.success) {
+              return { result: res.result };
+            } else {
+              return { result: res.error };
+            }
+          }
+          case 'write': {
+            const res = await migration.write(
+              migrationOperation.key,
+              migrationOperation.context,
+              migrationOperation.defaultStage,
+              migrationOperation.payload,
+            );
+
+            if (res.authoritative.success) {
+              return { result: res.authoritative.result };
+            } else {
+              return { result: res.authoritative.error };
+            }
+          }
+        }
+        return undefined;
+
       default:
         throw badCommandError;
     }

packages/shared/akamai-edgeworker-sdk/src/featureStore/index.ts

@@ -60,7 +60,7 @@ export class EdgeFeatureStore implements LDFeatureStore {
           callback(item.segments[dataKey]);
           break;
         default:
-          throw new Error(`Unsupported DataKind: ${namespace}`);
+          callback(null);
       }
     } catch (err) {
       this.logger.error(err);

packages/shared/common/__tests__/internal/events/EventProcessor.test.ts

@@ -28,6 +28,12 @@ import {
 } from '../../../src/api/subsystem';
 import { EventProcessor, InputIdentifyEvent } from '../../../src/internal';
 import { EventProcessorOptions } from '../../../src/internal/events/EventProcessor';
+import shouldSample from '../../../src/internal/events/sampling';
+
+jest.mock('../../../src/internal/events/sampling', () => ({
+  __esModule: true,
+  default: jest.fn(() => true),
+}));
 
 const user = { key: 'userKey', name: 'Red' };
 const userWithFilteredName = {
@@ -195,6 +201,8 @@ describe('given an event processor', () => {
   beforeEach(() => {
     eventSender = new MockEventSender();
     contextDeduplicator = new MockContextDeduplicator();
+    // @ts-ignore
+    shouldSample.mockImplementation(() => true);
 
     eventProcessor = new EventProcessor(
       eventProcessorConfig,
@@ -286,6 +294,8 @@ describe('given an event processor', () => {
       value: 'value',
       trackEvents: true,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -302,6 +312,70 @@ describe('given an event processor', () => {
     ]);
   });
 
+  it('uses sampling ratio for feature events', async () => {
+    Date.now = jest.fn(() => 1000);
+    eventProcessor.sendEvent({
+      kind: 'feature',
+      creationDate: 1000,
+      context: Context.fromLDContext(user),
+      key: 'flagkey',
+      version: 11,
+      variation: 1,
+      value: 'value',
+      trackEvents: true,
+      default: 'default',
+      samplingRatio: 2,
+      withReasons: true,
+    });
+
+    await eventProcessor.flush();
+    const request = await eventSender.queue.take();
+    expect(shouldSample).toHaveBeenCalledWith(2);
+
+    expect(request.data).toEqual([
+      {
+        kind: 'index',
+        creationDate: 1000,
+        context: { ...user, kind: 'user' },
+      },
+      { ...makeFeatureEvent(1000, 11), samplingRatio: 2 },
+      makeSummary(1000, 1000, 1, 11),
+    ]);
+  });
+
+  it('excludes feature events that are not sampled', async () => {
+    // @ts-ignore
+    shouldSample.mockImplementation((ratio) => ratio !== 2);
+    Date.now = jest.fn(() => 1000);
+    eventProcessor.sendEvent({
+      kind: 'feature',
+      creationDate: 1000,
+      context: Context.fromLDContext(user),
+      key: 'flagkey',
+      version: 11,
+      variation: 1,
+      value: 'value',
+      trackEvents: true,
+      default: 'default',
+      samplingRatio: 2,
+      withReasons: true,
+    });
+
+    await eventProcessor.flush();
+    const request = await eventSender.queue.take();
+    expect(shouldSample).toHaveBeenCalledWith(2);
+    expect(shouldSample).toHaveBeenCalledWith(1);
+
+    expect(request.data).toEqual([
+      {
+        kind: 'index',
+        creationDate: 1000,
+        context: { ...user, kind: 'user' },
+      },
+      makeSummary(1000, 1000, 1, 11),
+    ]);
+  });
+
   it('handles the version being 0', async () => {
     Date.now = jest.fn(() => 1000);
     eventProcessor.sendEvent({
@@ -314,6 +388,8 @@ describe('given an event processor', () => {
       value: 'value',
       trackEvents: true,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -344,6 +420,8 @@ describe('given an event processor', () => {
       trackEvents: false,
       debugEventsUntilDate: 2000,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -373,6 +451,8 @@ describe('given an event processor', () => {
       trackEvents: true,
       debugEventsUntilDate: 2000,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -409,6 +489,8 @@ describe('given an event processor', () => {
       trackEvents: false,
       debugEventsUntilDate: 1500,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -438,6 +520,8 @@ describe('given an event processor', () => {
       value: 'value',
       trackEvents: true,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
     eventProcessor.sendEvent({
       kind: 'feature',
@@ -449,6 +533,8 @@ describe('given an event processor', () => {
       value: 'carrot',
       trackEvents: true,
       default: 'potato',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -511,6 +597,8 @@ describe('given an event processor', () => {
       value: 'value',
       trackEvents: false,
       default: 'default',
+      samplingRatio: 1,
+      withReasons: true,
     });
     eventProcessor.sendEvent({
       kind: 'feature',
@@ -522,6 +610,8 @@ describe('given an event processor', () => {
       value: 'carrot',
       trackEvents: false,
       default: 'potato',
+      samplingRatio: 1,
+      withReasons: true,
     });
 
     await eventProcessor.flush();
@@ -577,6 +667,7 @@ describe('given an event processor', () => {
       context: Context.fromLDContext(user),
       key: 'eventkey',
       data: { thing: 'stuff' },
+      samplingRatio: 1,
     });
 
     await eventProcessor.flush();
@@ -607,6 +698,7 @@ describe('given an event processor', () => {
       context: Context.fromLDContext(anonUser),
       key: 'eventkey',
       data: { thing: 'stuff' },
+      samplingRatio: 1,
     });
 
     await eventProcessor.flush();
@@ -639,6 +731,7 @@ describe('given an event processor', () => {
       key: 'eventkey',
       data: { thing: 'stuff' },
       metricValue: 1.5,
+      samplingRatio: 1,
     });
 
     await eventProcessor.flush();