Update splitChangesUpdater, splitChangesFetcher and related types

EmilianoSanchez · EmilianoSanchez · commit 67a164b2df24 · 2025-02-27T12:53:41.000-03:00
diff --git a/src/dtos/types.ts b/src/dtos/types.ts
@@ -228,8 +228,16 @@ export type ISplitPartial = Pick<ISplit, 'conditions' | 'configurations' | 'traf
 
 /** Interface of the parsed JSON response of `/splitChanges` */
 export interface ISplitChangesResponse {
-  till: number,
-  splits: ISplit[]
+  ff?: {
+    t: number,
+    s?: number,
+    d: ISplit[]
+  },
+  rbs?: {
+    t: number,
+    s?: number,
+    d: IRBSegment[]
+  }
 }
 
 /** Interface of the parsed JSON response of `/segmentChanges/{segmentName}` */
diff --git a/src/logger/constants.ts b/src/logger/constants.ts
@@ -21,6 +21,7 @@ export const RETRIEVE_MANAGER = 29;
 export const SYNC_OFFLINE_DATA = 30;
 export const SYNC_SPLITS_FETCH = 31;
 export const SYNC_SPLITS_UPDATE = 32;
+export const SYNC_RBS_UPDATE = 33;
 export const STREAMING_NEW_MESSAGE = 35;
 export const SYNC_TASK_START = 36;
 export const SYNC_TASK_EXECUTE = 37;
diff --git a/src/logger/messages/debug.ts b/src/logger/messages/debug.ts
@@ -20,8 +20,9 @@ export const codesDebug: [number, string][] = codesInfo.concat([
   [c.RETRIEVE_MANAGER, 'Retrieving manager instance.'],
   // synchronizer
   [c.SYNC_OFFLINE_DATA, c.LOG_PREFIX_SYNC_OFFLINE + 'Feature flags data: \n%s'],
-  [c.SYNC_SPLITS_FETCH, c.LOG_PREFIX_SYNC_SPLITS + 'Spin up feature flags update using since = %s'],
-  [c.SYNC_SPLITS_UPDATE, c.LOG_PREFIX_SYNC_SPLITS + 'New feature flags %s. Removed feature flags %s. Segment names collected %s'],
+  [c.SYNC_SPLITS_FETCH, c.LOG_PREFIX_SYNC_SPLITS + 'Spin up feature flags update using since = %s and rbSince = %s.'],
+  [c.SYNC_SPLITS_UPDATE, c.LOG_PREFIX_SYNC_SPLITS + 'New feature flags %s. Removed feature flags %s.'],
+  [c.SYNC_RBS_UPDATE, c.LOG_PREFIX_SYNC_SPLITS + 'New rule-based segments %s. Removed rule-based segments %s.'],
   [c.STREAMING_NEW_MESSAGE, c.LOG_PREFIX_SYNC_STREAMING + 'New SSE message received, with data: %s.'],
   [c.SYNC_TASK_START, c.LOG_PREFIX_SYNC + ': Starting %s. Running each %s millis'],
   [c.SYNC_TASK_EXECUTE, c.LOG_PREFIX_SYNC + ': Running %s'],
diff --git a/src/services/types.ts b/src/services/types.ts
@@ -35,7 +35,7 @@ export type ISplitHttpClient = (url: string, options?: IRequestOptions, latencyT
 
 export type IFetchAuth = (userKeys?: string[]) => Promise<IResponse>
 
-export type IFetchSplitChanges = (since: number, noCache?: boolean, till?: number) => Promise<IResponse>
+export type IFetchSplitChanges = (since: number, noCache?: boolean, till?: number, rbSince?: number) => Promise<IResponse>
 
 export type IFetchSegmentChanges = (since: number, segmentName: string, noCache?: boolean, till?: number) => Promise<IResponse>
 
diff --git a/src/sync/polling/fetchers/splitChangesFetcher.ts b/src/sync/polling/fetchers/splitChangesFetcher.ts
@@ -11,11 +11,12 @@ export function splitChangesFetcherFactory(fetchSplitChanges: IFetchSplitChanges
     since: number,
     noCache?: boolean,
     till?: number,
+    rbSince?: number,
     // Optional decorator for `fetchSplitChanges` promise, such as timeout or time tracker
     decorator?: (promise: Promise<IResponse>) => Promise<IResponse>
   ) {
 
-    let splitsPromise = fetchSplitChanges(since, noCache, till);
+    let splitsPromise = fetchSplitChanges(since, noCache, till, rbSince);
     if (decorator) splitsPromise = decorator(splitsPromise);
 
     return splitsPromise.then(resp => resp.json());
diff --git a/src/sync/polling/fetchers/types.ts b/src/sync/polling/fetchers/types.ts
@@ -5,6 +5,7 @@ export type ISplitChangesFetcher = (
   since: number,
   noCache?: boolean,
   till?: number,
+  rbSince?: number,
   decorator?: (promise: Promise<IResponse>) => Promise<IResponse>
 ) => Promise<ISplitChangesResponse>
 
diff --git a/src/sync/polling/updaters/__tests__/splitChangesUpdater.spec.ts b/src/sync/polling/updaters/__tests__/splitChangesUpdater.spec.ts
@@ -4,14 +4,15 @@ import { splitApiFactory } from '../../../../services/splitApi';
 import { SegmentsCacheInMemory } from '../../../../storages/inMemory/SegmentsCacheInMemory';
 import { SplitsCacheInMemory } from '../../../../storages/inMemory/SplitsCacheInMemory';
 import { splitChangesFetcherFactory } from '../../fetchers/splitChangesFetcher';
-import { splitChangesUpdaterFactory, parseSegments, computeSplitsMutation } from '../splitChangesUpdater';
+import { splitChangesUpdaterFactory, parseSegments, computeMutation } from '../splitChangesUpdater';
 import splitChangesMock1 from '../../../../__tests__/mocks/splitchanges.since.-1.json';
 import fetchMock from '../../../../__tests__/testUtils/fetchMock';
 import { fullSettings, settingsSplitApi } from '../../../../utils/settingsValidation/__tests__/settings.mocks';
 import { EventEmitter } from '../../../../utils/MinEvents';
 import { loggerMock } from '../../../../logger/__tests__/sdkLogger.mock';
 import { telemetryTrackerFactory } from '../../../../trackers/telemetryTracker';
 import { splitNotifications } from '../../../streaming/__tests__/dataMocks';
+import { RBSegmentsCacheInMemory } from '../../../../storages/inMemory/RBSegmentsCacheInMemory';
 
 const ARCHIVED_FF = 'ARCHIVED';
 
@@ -94,58 +95,60 @@ test('splitChangesUpdater / segments parser', () => {
 test('splitChangesUpdater / compute splits mutation', () => {
   const splitFiltersValidation = { queryString: null, groupedFilters: { bySet: [], byName: [], byPrefix: [] }, validFilters: [] };
 
-  let splitsMutation = computeSplitsMutation([activeSplitWithSegments, archivedSplit] as ISplit[], splitFiltersValidation);
+  let segments = new Set<string>();
+  let splitsMutation = computeMutation([activeSplitWithSegments, archivedSplit] as ISplit[], segments, splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([activeSplitWithSegments]);
   expect(splitsMutation.removed).toEqual([archivedSplit]);
-  expect(splitsMutation.segments).toEqual(['A', 'B']);
+  expect(Array.from(segments)).toEqual(['A', 'B']);
 
   // SDK initialization without sets
   // should process all the notifications
-  splitsMutation = computeSplitsMutation([testFFSetsAB, test2FFSetsX] as ISplit[], splitFiltersValidation);
+  segments = new Set<string>();
+  splitsMutation = computeMutation([testFFSetsAB, test2FFSetsX] as ISplit[], segments, splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([testFFSetsAB, test2FFSetsX]);
   expect(splitsMutation.removed).toEqual([]);
-  expect(splitsMutation.segments).toEqual([]);
+  expect(Array.from(segments)).toEqual([]);
 });
 
 test('splitChangesUpdater / compute splits mutation with filters', () => {
   // SDK initialization with sets: [set_a, set_b]
   let splitFiltersValidation = { queryString: '&sets=set_a,set_b', groupedFilters: { bySet: ['set_a', 'set_b'], byName: ['name_1'], byPrefix: [] }, validFilters: [] };
 
   // fetching new feature flag in sets A & B
-  let splitsMutation = computeSplitsMutation([testFFSetsAB], splitFiltersValidation);
+  let splitsMutation = computeMutation([testFFSetsAB], new Set(), splitFiltersValidation);
 
   // should add it to mutations
   expect(splitsMutation.added).toEqual([testFFSetsAB]);
   expect(splitsMutation.removed).toEqual([]);
 
   // fetching existing test feature flag removed from set B
-  splitsMutation = computeSplitsMutation([testFFRemoveSetB], splitFiltersValidation);
+  splitsMutation = computeMutation([testFFRemoveSetB], new Set(), splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([testFFRemoveSetB]);
   expect(splitsMutation.removed).toEqual([]);
 
   // fetching existing test feature flag removed from set B
-  splitsMutation = computeSplitsMutation([testFFRemoveSetA], splitFiltersValidation);
+  splitsMutation = computeMutation([testFFRemoveSetA], new Set(), splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([]);
   expect(splitsMutation.removed).toEqual([testFFRemoveSetA]);
 
   // fetching existing test feature flag removed from set B
-  splitsMutation = computeSplitsMutation([testFFEmptySet], splitFiltersValidation);
+  splitsMutation = computeMutation([testFFEmptySet], new Set(), splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([]);
   expect(splitsMutation.removed).toEqual([testFFEmptySet]);
 
   // SDK initialization with names: ['test2']
   splitFiltersValidation = { queryString: '&names=test2', groupedFilters: { bySet: [], byName: ['test2'], byPrefix: [] }, validFilters: [] };
-  splitsMutation = computeSplitsMutation([testFFSetsAB], splitFiltersValidation);
+  splitsMutation = computeMutation([testFFSetsAB], new Set(), splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([]);
   expect(splitsMutation.removed).toEqual([testFFSetsAB]);
 
-  splitsMutation = computeSplitsMutation([test2FFSetsX, testFFEmptySet], splitFiltersValidation);
+  splitsMutation = computeMutation([test2FFSetsX, testFFEmptySet], new Set(), splitFiltersValidation);
 
   expect(splitsMutation.added).toEqual([test2FFSetsX]);
   expect(splitsMutation.removed).toEqual([testFFEmptySet]);
@@ -161,10 +164,13 @@ describe('splitChangesUpdater', () => {
   const splits = new SplitsCacheInMemory();
   const updateSplits = jest.spyOn(splits, 'update');
 
+  const rbSegments = new RBSegmentsCacheInMemory();
+  // @TODO spy on rbSegments
+
   const segments = new SegmentsCacheInMemory();
   const registerSegments = jest.spyOn(segments, 'registerSegments');
 
-  const storage = { splits, segments };
+  const storage = { splits, rbSegments, segments };
 
   const readinessManager = readinessManagerFactory(EventEmitter, fullSettings);
   const splitsEmitSpy = jest.spyOn(readinessManager.splits, 'emit');
diff --git a/src/sync/polling/updaters/splitChangesUpdater.ts b/src/sync/polling/updaters/splitChangesUpdater.ts
@@ -1,13 +1,13 @@
 import { ISegmentsCacheBase, IStorageBase } from '../../../storages/types';
 import { ISplitChangesFetcher } from '../fetchers/types';
-import { ISplit, ISplitChangesResponse, ISplitFiltersValidation } from '../../../dtos/types';
+import { IRBSegment, ISplit, ISplitChangesResponse, ISplitFiltersValidation, MaybeThenable } from '../../../dtos/types';
 import { ISplitsEventEmitter } from '../../../readiness/types';
 import { timeout } from '../../../utils/promise/timeout';
 import { SDK_SPLITS_ARRIVED, SDK_SPLITS_CACHE_LOADED } from '../../../readiness/constants';
 import { ILogger } from '../../../logger/types';
-import { SYNC_SPLITS_FETCH, SYNC_SPLITS_UPDATE, SYNC_SPLITS_FETCH_FAILS, SYNC_SPLITS_FETCH_RETRY } from '../../../logger/constants';
+import { SYNC_SPLITS_FETCH, SYNC_SPLITS_UPDATE, SYNC_RBS_UPDATE, SYNC_SPLITS_FETCH_FAILS, SYNC_SPLITS_FETCH_RETRY } from '../../../logger/constants';
 import { startsWith } from '../../../utils/lang';
-import { IN_SEGMENT } from '../../../utils/constants';
+import { IN_RULE_BASED_SEGMENT, IN_SEGMENT } from '../../../utils/constants';
 import { setToArray } from '../../../utils/lang/sets';
 
 type ISplitChangesUpdater = (noCache?: boolean, till?: number, splitUpdateNotification?: { payload: ISplit, changeNumber: number }) => Promise<boolean>
@@ -27,24 +27,23 @@ function checkAllSegmentsExist(segments: ISegmentsCacheBase): Promise<boolean> {
  * Collect segments from a raw split definition.
  * Exported for testing purposes.
  */
-export function parseSegments({ conditions }: ISplit): Set<string> {
+export function parseSegments({ conditions }: ISplit | IRBSegment, matcherType: typeof IN_SEGMENT | typeof IN_RULE_BASED_SEGMENT = IN_SEGMENT): Set<string> {
   let segments = new Set<string>();
 
   for (let i = 0; i < conditions.length; i++) {
     const matchers = conditions[i].matcherGroup.matchers;
 
     matchers.forEach(matcher => {
-      if (matcher.matcherType === IN_SEGMENT) segments.add(matcher.userDefinedSegmentMatcherData.segmentName);
+      if (matcher.matcherType === matcherType) segments.add(matcher.userDefinedSegmentMatcherData.segmentName);
     });
   }
 
   return segments;
 }
 
-interface ISplitMutations {
-  added: ISplit[],
-  removed: ISplit[],
-  segments: string[]
+interface ISplitMutations<T extends ISplit | IRBSegment> {
+  added: T[],
+  removed: T[]
 }
 
 /**
@@ -68,30 +67,30 @@ function matchFilters(featureFlag: ISplit, filters: ISplitFiltersValidation) {
   return matchNames || matchPrefix;
 }
 
+function isFF(ruleBasedEntity: IRBSegment | ISplit): ruleBasedEntity is ISplit {
+  return (ruleBasedEntity as ISplit).defaultTreatment !== undefined;
+}
+
 /**
  * Given the list of splits from /splitChanges endpoint, it returns the mutations,
  * i.e., an object with added splits, removed splits and used segments.
  * Exported for testing purposes.
  */
-export function computeSplitsMutation(entries: ISplit[], filters: ISplitFiltersValidation): ISplitMutations {
-  const segments = new Set<string>();
-  const computed = entries.reduce((accum, split) => {
-    if (split.status === 'ACTIVE' && matchFilters(split, filters)) {
-      accum.added.push(split);
+export function computeMutation<T extends ISplit | IRBSegment>(rules: Array<T>, segments: Set<string>, filters?: ISplitFiltersValidation): ISplitMutations<T> {
 
-      parseSegments(split).forEach((segmentName: string) => {
+  return rules.reduce((accum, ruleBasedEntity) => {
+    if (ruleBasedEntity.status === 'ACTIVE' && (!filters || matchFilters(ruleBasedEntity as ISplit, filters))) {
+      accum.added.push(ruleBasedEntity);
+
+      parseSegments(ruleBasedEntity).forEach((segmentName: string) => {
         segments.add(segmentName);
       });
     } else {
-      accum.removed.push(split);
+      accum.removed.push(ruleBasedEntity);
     }
 
     return accum;
-  }, { added: [], removed: [], segments: [] } as ISplitMutations);
-
-  computed.segments = setToArray(segments);
-
-  return computed;
+  }, { added: [], removed: [] } as ISplitMutations<T>);
 }
 
 /**
@@ -111,14 +110,14 @@ export function computeSplitsMutation(entries: ISplit[], filters: ISplitFiltersV
 export function splitChangesUpdaterFactory(
   log: ILogger,
   splitChangesFetcher: ISplitChangesFetcher,
-  storage: Pick<IStorageBase, 'splits' | 'segments'>,
+  storage: Pick<IStorageBase, 'splits' | 'rbSegments' | 'segments'>,
   splitFiltersValidation: ISplitFiltersValidation,
   splitsEventEmitter?: ISplitsEventEmitter,
   requestTimeoutBeforeReady: number = 0,
   retriesOnFailureBeforeReady: number = 0,
   isClientSide?: boolean
 ): ISplitChangesUpdater {
-  const { splits, segments } = storage;
+  const { splits, rbSegments, segments } = storage;
 
   let startingUp = true;
 
@@ -135,35 +134,60 @@ export function splitChangesUpdaterFactory(
    * @param noCache - true to revalidate data to fetch
    * @param till - query param to bypass CDN requests
    */
-  return function splitChangesUpdater(noCache?: boolean, till?: number, splitUpdateNotification?: { payload: ISplit, changeNumber: number }) {
+  return function splitChangesUpdater(noCache?: boolean, till?: number, splitUpdateNotification?: { payload: ISplit | IRBSegment, changeNumber: number }) {
 
     /**
      * @param since - current changeNumber at splitsCache
      * @param retry - current number of retry attempts
      */
-    function _splitChangesUpdater(since: number, retry = 0): Promise<boolean> {
-      log.debug(SYNC_SPLITS_FETCH, [since]);
-      const fetcherPromise = Promise.resolve(splitUpdateNotification ?
-        { splits: [splitUpdateNotification.payload], till: splitUpdateNotification.changeNumber } :
-        splitChangesFetcher(since, noCache, till, _promiseDecorator)
+    function _splitChangesUpdater(sinces: [number, number], retry = 0): Promise<boolean> {
+      const [since, rbSince] = sinces;
+      log.debug(SYNC_SPLITS_FETCH, sinces);
+      const fetcherPromise = Promise.resolve(
+        splitUpdateNotification ?
+          isFF(splitUpdateNotification.payload) ?
+            // IFFU edge case: a change to a flag that adds an IN_RULE_BASED_SEGMENT matcher that is not present yet
+            Promise.resolve(rbSegments.contains(parseSegments(splitUpdateNotification.payload, IN_RULE_BASED_SEGMENT))).then((contains) => {
+              return contains ?
+                { ff: { d: [splitUpdateNotification.payload as ISplit], t: splitUpdateNotification.changeNumber } } :
+                splitChangesFetcher(since, noCache, till, rbSince, _promiseDecorator);
+            }) :
+            { rbs: { d: [splitUpdateNotification.payload as IRBSegment], t: splitUpdateNotification.changeNumber } } :
+          splitChangesFetcher(since, noCache, till, rbSince, _promiseDecorator)
       )
         .then((splitChanges: ISplitChangesResponse) => {
           startingUp = false;
 
-          const mutation = computeSplitsMutation(splitChanges.splits, splitFiltersValidation);
+          const usedSegments = new Set<string>();
+
+          let ffUpdate: MaybeThenable<boolean> = false;
+          if (splitChanges.ff) {
+            const { added, removed } = computeMutation(splitChanges.ff.d, usedSegments, splitFiltersValidation);
+            log.debug(SYNC_SPLITS_UPDATE, [added.length, removed.length]);
+            ffUpdate = splits.update(added, removed, splitChanges.ff.t);
+          }
 
-          log.debug(SYNC_SPLITS_UPDATE, [mutation.added.length, mutation.removed.length, mutation.segments.length]);
+          let rbsUpdate: MaybeThenable<boolean> = false;
+          if (splitChanges.rbs) {
+            const { added, removed } = computeMutation(splitChanges.rbs.d, usedSegments);
+            log.debug(SYNC_RBS_UPDATE, [added.length, removed.length]);
+            rbsUpdate = rbSegments.update(added, removed, splitChanges.rbs.t);
+          }
 
-          // Write into storage
-          // @TODO call `setChangeNumber` only if the other storage operations have succeeded, in order to keep storage consistency
-          return Promise.all([
-            splits.update(mutation.added, mutation.removed, splitChanges.till),
-            segments.registerSegments(mutation.segments)
-          ]).then(([isThereUpdate]) => {
+          return Promise.all([ffUpdate, rbsUpdate,
+            // @TODO if at least 1 segment fetch fails due to 404 and other segments are updated in the storage, SDK_UPDATE is not emitted
+            segments.registerSegments(setToArray(usedSegments))
+          ]).then(([ffChanged, rbsChanged]) => {
             if (splitsEventEmitter) {
               // To emit SDK_SPLITS_ARRIVED for server-side SDK, we must check that all registered segments have been fetched
-              return Promise.resolve(!splitsEventEmitter.splitsArrived || (since !== splitChanges.till && isThereUpdate && (isClientSide || checkAllSegmentsExist(segments))))
-                .catch(() => false /** noop. just to handle a possible `checkAllSegmentsExist` rejection, before emitting SDK event */)
+              return Promise.resolve(!splitsEventEmitter.splitsArrived ||
+                (
+                  (!splitChanges.ff || since !== splitChanges.ff.t) &&
+                  (!splitChanges.rbs || rbSince !== splitChanges.rbs.t) &&
+                  (ffChanged || rbsChanged) &&
+                  (isClientSide || checkAllSegmentsExist(segments))
+                )
+              )
                 .then(emitSplitsArrivedEvent => {
                   // emit SDK events
                   if (emitSplitsArrivedEvent) splitsEventEmitter.emit(SDK_SPLITS_ARRIVED);
@@ -179,7 +203,7 @@ export function splitChangesUpdaterFactory(
           if (startingUp && retriesOnFailureBeforeReady > retry) {
             retry += 1;
             log.info(SYNC_SPLITS_FETCH_RETRY, [retry, error]);
-            return _splitChangesUpdater(since, retry);
+            return _splitChangesUpdater(sinces, retry);
           } else {
             startingUp = false;
           }
@@ -196,7 +220,7 @@ export function splitChangesUpdaterFactory(
       return fetcherPromise;
     }
 
-    let sincePromise = Promise.resolve(splits.getChangeNumber()); // `getChangeNumber` never rejects or throws error
-    return sincePromise.then(_splitChangesUpdater);
+    // `getChangeNumber` never rejects or throws error
+    return Promise.all([splits.getChangeNumber(), rbSegments.getChangeNumber()]).then(_splitChangesUpdater);
   };
 }
