[FME-12059] SDK_UPDATE with metadata

Author: ZamoraEmmanuel

src/readiness/__tests__/readinessManager.spec.ts

@@ -3,6 +3,7 @@ import { EventEmitter } from '../../utils/MinEvents';
 import { IReadinessManager } from '../types';
 import { SDK_READY, SDK_UPDATE, SDK_SPLITS_ARRIVED, SDK_SEGMENTS_ARRIVED, SDK_READY_FROM_CACHE, SDK_SPLITS_CACHE_LOADED, SDK_READY_TIMED_OUT } from '../constants';
 import { ISettings } from '../../types';
+import { EventMetadata, SdkUpdateMetadataKeys } from '../../sync/polling/types';
 
 const settings = {
   startup: {
@@ -300,3 +301,62 @@ test('READINESS MANAGER / Destroy before it was ready and timedout', (done) => {
   }, settingsWithTimeout.startup.readyTimeout * 1.5);
 
 });
+
+test('READINESS MANAGER / SDK_UPDATE should emit with metadata', () => {
+  const readinessManager = readinessManagerFactory(EventEmitter, settings);
+
+  // SDK_READY
+  readinessManager.splits.emit(SDK_SPLITS_ARRIVED);
+  readinessManager.segments.emit(SDK_SEGMENTS_ARRIVED);
+
+  const metadata: EventMetadata = {
+    [SdkUpdateMetadataKeys.UPDATED_FLAGS]: ['flag1', 'flag2']
+  };
+
+  let receivedMetadata: EventMetadata | undefined;
+  readinessManager.gate.on(SDK_UPDATE, (meta: EventMetadata) => {
+    receivedMetadata = meta;
+  });
+
+  readinessManager.splits.emit(SDK_SPLITS_ARRIVED, metadata);
+
+  expect(receivedMetadata).toEqual(metadata);
+});
+
+test('READINESS MANAGER / SDK_UPDATE should handle undefined metadata', () => {
+  const readinessManager = readinessManagerFactory(EventEmitter, settings);
+
+  // SDK_READY
+  readinessManager.splits.emit(SDK_SPLITS_ARRIVED);
+  readinessManager.segments.emit(SDK_SEGMENTS_ARRIVED);
+
+  let receivedMetadata: any;
+  readinessManager.gate.on(SDK_UPDATE, (meta: EventMetadata) => {
+    receivedMetadata = meta;
+  });
+
+  readinessManager.splits.emit(SDK_SPLITS_ARRIVED);
+
+  expect(receivedMetadata).toBeUndefined();
+});
+
+test('READINESS MANAGER / SDK_UPDATE should forward metadata from segments', () => {
+  const readinessManager = readinessManagerFactory(EventEmitter, settings);
+
+  // SDK_READY
+  readinessManager.splits.emit(SDK_SPLITS_ARRIVED);
+  readinessManager.segments.emit(SDK_SEGMENTS_ARRIVED);
+
+  const metadata: EventMetadata = {
+    [SdkUpdateMetadataKeys.UPDATED_SEGMENTS]: ['segment1', 'segment2']
+  };
+
+  let receivedMetadata: EventMetadata | undefined;
+  readinessManager.gate.on(SDK_UPDATE, (meta: EventMetadata) => {
+    receivedMetadata = meta;
+  });
+
+  readinessManager.segments.emit(SDK_SEGMENTS_ARRIVED, metadata);
+
+  expect(receivedMetadata).toEqual(metadata);
+});

src/readiness/readinessManager.ts

@@ -3,6 +3,7 @@ import { ISettings } from '../types';
 import SplitIO from '../../types/splitio';
 import { SDK_SPLITS_ARRIVED, SDK_SPLITS_CACHE_LOADED, SDK_SEGMENTS_ARRIVED, SDK_READY_TIMED_OUT, SDK_READY_FROM_CACHE, SDK_UPDATE, SDK_READY } from './constants';
 import { IReadinessEventEmitter, IReadinessManager, ISegmentsEventEmitter, ISplitsEventEmitter } from './types';
+import { SdkUpdateMetadata } from '../sync/polling/types';
 
 function splitsEventEmitterFactory(EventEmitter: new () => SplitIO.IEventEmitter): ISplitsEventEmitter {
   const splitsEventEmitter = objectAssign(new EventEmitter(), {
@@ -15,7 +16,7 @@ function splitsEventEmitterFactory(EventEmitter: new () => SplitIO.IEventEmitter
   // `isSplitKill` condition avoids an edge-case of wrongly emitting SDK_READY if:
   // - `/memberships` fetch and SPLIT_KILL occurs before `/splitChanges` fetch, and
   // - storage has cached splits (for which case `splitsStorage.killLocally` can return true)
-  splitsEventEmitter.on(SDK_SPLITS_ARRIVED, (isSplitKill: boolean) => { if (!isSplitKill) splitsEventEmitter.splitsArrived = true; });
+  splitsEventEmitter.on(SDK_SPLITS_ARRIVED, (metadata: SdkUpdateMetadata, isSplitKill: boolean) => { if (!isSplitKill) splitsEventEmitter.splitsArrived = true; });
   splitsEventEmitter.once(SDK_SPLITS_CACHE_LOADED, () => { splitsEventEmitter.splitsCacheLoaded = true; });
 
   return splitsEventEmitter;
@@ -98,12 +99,12 @@ export function readinessManagerFactory(
     }
   }
 
-  function checkIsReadyOrUpdate(diff: any) {
+  function checkIsReadyOrUpdate(metadata: SdkUpdateMetadata) {
     if (isDestroyed) return;
     if (isReady) {
       try {
         syncLastUpdate();
-        gate.emit(SDK_UPDATE, diff);
+        gate.emit(SDK_UPDATE, metadata);
       } catch (e) {
         // throws user callback exceptions in next tick
         setTimeout(() => { throw e; }, 0);

src/storages/inRedis/__tests__/TelemetryCacheInRedis.spec.ts

@@ -10,78 +10,88 @@ const latencyKey = `${prefix}.telemetry.latencies`;
 const initKey = `${prefix}.telemetry.init`;
 const fieldVersionablePrefix = `${metadata.s}/${metadata.n}/${metadata.i}`;
 
-test('TELEMETRY CACHE IN REDIS', async () => {
+describe('TELEMETRY CACHE IN REDIS', () => {
+  let connection: RedisAdapter;
+  let cache: TelemetryCacheInRedis;
+  let keysBuilder: KeyBuilderSS;
 
-  const keysBuilder = new KeyBuilderSS(prefix, metadata);
-  const connection = new RedisAdapter(loggerMock);
-  const cache = new TelemetryCacheInRedis(loggerMock, keysBuilder, connection);
+  beforeEach(async () => {
+    keysBuilder = new KeyBuilderSS(prefix, metadata);
+    connection = new RedisAdapter(loggerMock);
+    cache = new TelemetryCacheInRedis(loggerMock, keysBuilder, connection);
 
-  // recordException
-  expect(await cache.recordException('tr')).toBe(1);
-  expect(await cache.recordException('tr')).toBe(2);
-  expect(await cache.recordException('tcfs')).toBe(1);
+    await connection.del(exceptionKey);
+    await connection.del(latencyKey);
+    await connection.del(initKey);
+  });
 
-  expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/track')).toBe('2');
-  expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatment')).toBe(null);
-  expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatmentsWithConfigByFlagSets')).toBe('1');
+  test('TELEMETRY CACHE IN REDIS', async () => {
 
-  // recordLatency
-  expect(await cache.recordLatency('tr', 1.6)).toBe(1);
-  expect(await cache.recordLatency('tr', 1.6)).toBe(2);
-  expect(await cache.recordLatency('tfs', 1.6)).toBe(1);
+    // recordException
+    expect(await cache.recordException('tr')).toBe(1);
+    expect(await cache.recordException('tr')).toBe(2);
+    expect(await cache.recordException('tcfs')).toBe(1);
 
-  expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/track/2')).toBe('2');
-  expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatment/2')).toBe(null);
-  expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatmentsByFlagSets/2')).toBe('1');
+    expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/track')).toBe('2');
+    expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatment')).toBe(null);
+    expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/treatmentsWithConfigByFlagSets')).toBe('1');
 
-  // recordConfig
-  expect(await cache.recordConfig()).toBe(1);
-  expect(JSON.parse(await connection.hget(initKey, fieldVersionablePrefix) as string)).toEqual({
-    oM: 1,
-    st: 'redis',
-    aF: 0,
-    rF: 0
-  });
+    // recordLatency
+    expect(await cache.recordLatency('tr', 1.6)).toBe(1);
+    expect(await cache.recordLatency('tr', 1.6)).toBe(2);
+    expect(await cache.recordLatency('tfs', 1.6)).toBe(1);
 
-  // popLatencies
-  const latencies = await cache.popLatencies();
-  latencies.forEach((latency, m) => {
-    expect(JSON.parse(m)).toEqual(metadata);
-    expect(latency).toEqual({
-      tfs: [0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
-      tr: [0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
-    });
-  });
-  expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/track/2')).toBe(null);
-
-  // popExceptions
-  const exceptions = await cache.popExceptions();
-  exceptions.forEach((exception, m) => {
-    expect(JSON.parse(m)).toEqual(metadata);
-    expect(exception).toEqual({
-      tcfs: 1,
-      tr: 2,
-    });
-  });
-  expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/track')).toBe(null);
+    expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/track/2')).toBe('2');
+    expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatment/2')).toBe(null);
+    expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/treatmentsByFlagSets/2')).toBe('1');
 
-  // popConfig
-  const configs = await cache.popConfigs();
-  configs.forEach((config, m) => {
-    expect(JSON.parse(m)).toEqual(metadata);
-    expect(config).toEqual({
+    // recordConfig
+    expect(await cache.recordConfig()).toBe(1);
+    expect(JSON.parse(await connection.hget(initKey, fieldVersionablePrefix) as string)).toEqual({
       oM: 1,
       st: 'redis',
       aF: 0,
       rF: 0
     });
-  });
-  expect(await connection.hget(initKey, fieldVersionablePrefix)).toBe(null);
 
-  // pops when there is no data
-  expect((await cache.popLatencies()).size).toBe(0);
-  expect((await cache.popExceptions()).size).toBe(0);
-  expect((await cache.popConfigs()).size).toBe(0);
+    // popLatencies
+    const latencies = await cache.popLatencies();
+    latencies.forEach((latency, m) => {
+      expect(JSON.parse(m)).toEqual(metadata);
+      expect(latency).toEqual({
+        tfs: [0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+        tr: [0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
+      });
+    });
+    expect(await connection.hget(latencyKey, fieldVersionablePrefix + '/track/2')).toBe(null);
+
+    // popExceptions
+    const exceptions = await cache.popExceptions();
+    exceptions.forEach((exception, m) => {
+      expect(JSON.parse(m)).toEqual(metadata);
+      expect(exception).toEqual({
+        tcfs: 1,
+        tr: 2,
+      });
+    });
+    expect(await connection.hget(exceptionKey, fieldVersionablePrefix + '/track')).toBe(null);
 
-  await connection.disconnect();
+    // popConfig
+    const configs = await cache.popConfigs();
+    configs.forEach((config, m) => {
+      expect(JSON.parse(m)).toEqual(metadata);
+      expect(config).toEqual({
+        oM: 1,
+        st: 'redis',
+        aF: 0,
+        rF: 0
+      });
+    });
+    expect(await connection.hget(initKey, fieldVersionablePrefix)).toBe(null);
+
+    // pops when there is no data
+    expect((await cache.popLatencies()).size).toBe(0);
+    expect((await cache.popExceptions()).size).toBe(0);
+    expect((await cache.popConfigs()).size).toBe(0);
+  });
 });

src/sync/polling/types.ts

@@ -31,3 +31,13 @@ export interface IPollingManagerCS extends IPollingManager {
   remove(matchingKey: string): void;
   get(matchingKey: string): IMySegmentsSyncTask | undefined
 }
+
+export enum SdkUpdateMetadataKeys {
+  UPDATED_FLAGS = 'updatedFlags',
+  UPDATED_SEGMENTS = 'updatedSegments'
+}
+
+export type SdkUpdateMetadata = {
+  [SdkUpdateMetadataKeys.UPDATED_FLAGS]?: string[]
+  [SdkUpdateMetadataKeys.UPDATED_SEGMENTS]?: string[]
+}

src/sync/polling/updaters/__tests__/segmentChangesUpdater.spec.ts

@@ -0,0 +1,52 @@
+import { readinessManagerFactory } from '../../../../readiness/readinessManager';
+import { SegmentsCacheInMemory } from '../../../../storages/inMemory/SegmentsCacheInMemory';
+import { segmentChangesUpdaterFactory } from '../segmentChangesUpdater';
+import { fullSettings } from '../../../../utils/settingsValidation/__tests__/settings.mocks';
+import { EventEmitter } from '../../../../utils/MinEvents';
+import { loggerMock } from '../../../../logger/__tests__/sdkLogger.mock';
+import { ISegmentChangesFetcher } from '../../fetchers/types';
+import { ISegmentChangesResponse } from '../../../../dtos/types';
+import { SDK_SEGMENTS_ARRIVED } from '../../../../readiness/constants';
+
+describe('segmentChangesUpdater', () => {
+  const segments = new SegmentsCacheInMemory();
+  const updateSegments = jest.spyOn(segments, 'update');
+
+  const readinessManager = readinessManagerFactory(EventEmitter, fullSettings);
+  const segmentsEmitSpy = jest.spyOn(readinessManager.segments, 'emit');
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    segments.clear();
+    readinessManager.segments.segmentsArrived = false;
+  });
+
+  test('test with segments update - should emit updatedSegments and NOT updatedFlags', async () => {
+    const segmentName = 'test-segment';
+    const segmentChange: ISegmentChangesResponse = {
+      name: segmentName,
+      added: ['key1', 'key2'],
+      removed: [],
+      since: -1,
+      till: 123
+    };
+
+    const mockSegmentChangesFetcher: ISegmentChangesFetcher = jest.fn().mockResolvedValue([segmentChange]);
+
+    const segmentChangesUpdater = segmentChangesUpdaterFactory(
+      loggerMock,
+      mockSegmentChangesFetcher,
+      segments,
+      readinessManager,
+      1000,
+      1
+    );
+
+    segments.registerSegments([segmentName]);
+
+    await segmentChangesUpdater(undefined, segmentName);
+
+    expect(updateSegments).toHaveBeenCalledWith(segmentName, segmentChange.added, segmentChange.removed, segmentChange.till);
+    expect(segmentsEmitSpy).toBeCalledWith(SDK_SEGMENTS_ARRIVED, { updatedSegments: [segmentName] });
+  });
+});