From f58ab24f85cc2f448e00c5cbf91da3e9519c993a Mon Sep 17 00:00:00 2001
From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com>
Date: Wed, 4 Mar 2026 16:25:01 -0800
Subject: [PATCH 1/2] feat: Add DataSourceUpdateSink for client-side FDv2

Implements SDK-1921 (FDCS-06): the update sink that processes FDv2
ChangeSetResult payloads from the orchestrator and applies them to
FlagManager. Handles full payloads (init), partial payloads (upsert
per flag), and none payloads. Manages the selector (basis state) in
memory only (Req 6.2.1) and tracks environmentId from response headers.
---
 .../fdv2/DataSourceUpdateSink.test.ts         | 350 ++++++++++++++++++
 .../datasource/fdv2/FDv2DataSource.test.ts    |  44 +--
 .../datasource/fdv2/DataSourceUpdateSink.ts   | 114 ++++++
 .../src/datasource/fdv2/FDv2DataSource.ts     |  10 +-
 .../sdk-client/src/datasource/fdv2/index.ts   |   3 +
 5 files changed, 495 insertions(+), 26 deletions(-)
 create mode 100644 packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts
 create mode 100644 packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts

diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts
new file mode 100644
index 0000000000..c09d304e71
--- /dev/null
+++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts
@@ -0,0 +1,350 @@
+import { Context, internal } from '@launchdarkly/js-sdk-common';
+
+import { createDataSourceUpdateSink } from '../../../src/datasource/fdv2/DataSourceUpdateSink';
+import { ChangeSetResult } from '../../../src/datasource/fdv2/FDv2SourceResult';
+import { FlagManager } from '../../../src/flag-manager/FlagManager';
+import { makeLogger } from './orchestrationTestHelpers';
+
+function makeContext(key: string = 'user-key'): Context {
+  return Context.fromLDContext({ kind: 'user', key });
+}
+
+function makeFlagManager(): jest.Mocked<Pick<FlagManager, 'init' | 'upsert' | 'get' | 'getAll'>> {
+  return {
+    init: jest.fn().mockResolvedValue(undefined),
+    upsert: jest.fn().mockResolvedValue(true),
+    get: jest.fn(),
+    getAll: jest.fn(),
+  };
+}
+
+function makeResult(
+  payload: internal.Payload,
+  opts: { fdv1Fallback?: boolean; environmentId?: string } = {},
+): ChangeSetResult {
+  return {
+    type: 'changeSet',
+    payload,
+    fdv1Fallback: opts.fdv1Fallback ?? false,
+    environmentId: opts.environmentId,
+  };
+}
+
+function makePayload(overrides: Partial<internal.Payload> = {}): internal.Payload {
+  return {
+    id: 'test-payload',
+    version: 1,
+    state: 'test-selector',
+    type: 'full',
+    updates: [],
+    ...overrides,
+  };
+}
+
+function makeFlagEvalUpdate(
+  key: string,
+  version: number,
+  value: unknown,
+  opts: { deleted?: boolean } = {},
+): internal.Update {
+  if (opts.deleted) {
+    return { kind: 'flagEval', key, version, deleted: true };
+  }
+  return {
+    kind: 'flagEval',
+    key,
+    version,
+    object: { value, trackEvents: false },
+  };
+}
+
+// -- full payload --
+
+it('calls flagManager.init with converted descriptors on full payload', () => {
+  const flagManager = makeFlagManager();
+  const context = makeContext();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => context,
+  });
+
+  const payload = makePayload({
+    type: 'full',
+    updates: [makeFlagEvalUpdate('flag-1', 10, true), makeFlagEvalUpdate('flag-2', 20, 'hello')],
+  });
+
+  sink.handleChangeSet(makeResult(payload));
+
+  expect(flagManager.init).toHaveBeenCalledTimes(1);
+  expect(flagManager.init).toHaveBeenCalledWith(context, {
+    'flag-1': {
+      version: 10,
+      flag: { value: true, trackEvents: false, version: 10 },
+    },
+    'flag-2': {
+      version: 20,
+      flag: { value: 'hello', trackEvents: false, version: 20 },
+    },
+  });
+  expect(flagManager.upsert).not.toHaveBeenCalled();
+});
+
+// -- partial payload --
+
+it('calls flagManager.upsert for each flag on partial payload', () => {
+  const flagManager = makeFlagManager();
+  const context = makeContext();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => context,
+  });
+
+  const payload = makePayload({
+    type: 'partial',
+    updates: [makeFlagEvalUpdate('flag-a', 5, 42), makeFlagEvalUpdate('flag-b', 6, false)],
+  });
+
+  sink.handleChangeSet(makeResult(payload));
+
+  expect(flagManager.upsert).toHaveBeenCalledTimes(2);
+  expect(flagManager.upsert).toHaveBeenCalledWith(context, 'flag-a', {
+    version: 5,
+    flag: { value: 42, trackEvents: false, version: 5 },
+  });
+  expect(flagManager.upsert).toHaveBeenCalledWith(context, 'flag-b', {
+    version: 6,
+    flag: { value: false, trackEvents: false, version: 6 },
+  });
+  expect(flagManager.init).not.toHaveBeenCalled();
+});
+
+// -- none payload --
+
+it('does not call flagManager on none payload', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  sink.handleChangeSet(makeResult(makePayload({ type: 'none', updates: [] })));
+
+  expect(flagManager.init).not.toHaveBeenCalled();
+  expect(flagManager.upsert).not.toHaveBeenCalled();
+});
+
+// -- selector management --
+
+it('stores selector from payload state', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  expect(sink.getSelector()).toBeUndefined();
+
+  sink.handleChangeSet(makeResult(makePayload({ state: 'selector-1' })));
+
+  expect(sink.getSelector()).toBe('selector-1');
+});
+
+it('updates selector on subsequent payloads', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  sink.handleChangeSet(makeResult(makePayload({ state: 'a' })));
+  sink.handleChangeSet(makeResult(makePayload({ state: 'b' })));
+
+  expect(sink.getSelector()).toBe('b');
+});
+
+it('does not clear selector when payload has no state', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  sink.handleChangeSet(makeResult(makePayload({ state: 'a' })));
+  sink.handleChangeSet(makeResult(makePayload({ state: undefined })));
+
+  expect(sink.getSelector()).toBe('a');
+});
+
+it('does not set selector from empty string state', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  sink.handleChangeSet(makeResult(makePayload({ state: '' })));
+
+  expect(sink.getSelector()).toBeUndefined();
+});
+
+// -- environmentId tracking --
+
+it('tracks environmentId from ChangeSetResult', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  expect(sink.getEnvironmentId()).toBeUndefined();
+
+  sink.handleChangeSet(makeResult(makePayload(), { environmentId: 'env-123' }));
+
+  expect(sink.getEnvironmentId()).toBe('env-123');
+});
+
+it('updates environmentId on subsequent results', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  sink.handleChangeSet(makeResult(makePayload(), { environmentId: 'env-1' }));
+  sink.handleChangeSet(makeResult(makePayload(), { environmentId: 'env-2' }));
+
+  expect(sink.getEnvironmentId()).toBe('env-2');
+});
+
+// -- non-flagEval kinds --
+
+it('ignores non-flagEval update kinds', () => {
+  const flagManager = makeFlagManager();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+  });
+
+  const payload = makePayload({
+    type: 'full',
+    updates: [
+      { kind: 'segment', key: 'seg-1', version: 1, object: {} },
+      { kind: 'unknown', key: 'unk-1', version: 1, object: {} },
+    ],
+  });
+
+  sink.handleChangeSet(makeResult(payload));
+
+  expect(flagManager.init).toHaveBeenCalledWith(expect.anything(), {});
+});
+
+// -- delete handling --
+
+it('handles delete updates in full payload', () => {
+  const flagManager = makeFlagManager();
+  const context = makeContext();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => context,
+  });
+
+  const payload = makePayload({
+    type: 'full',
+    updates: [makeFlagEvalUpdate('deleted-flag', 7, undefined, { deleted: true })],
+  });
+
+  sink.handleChangeSet(makeResult(payload));
+
+  expect(flagManager.init).toHaveBeenCalledWith(context, {
+    'deleted-flag': {
+      version: 7,
+      flag: {
+        version: 7,
+        deleted: true,
+        value: undefined,
+        trackEvents: false,
+      },
+    },
+  });
+});
+
+it('handles delete updates in partial payload', () => {
+  const flagManager = makeFlagManager();
+  const context = makeContext();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => context,
+  });
+
+  const payload = makePayload({
+    type: 'partial',
+    updates: [makeFlagEvalUpdate('deleted-flag', 7, undefined, { deleted: true })],
+  });
+
+  sink.handleChangeSet(makeResult(payload));
+
+  expect(flagManager.upsert).toHaveBeenCalledWith(context, 'deleted-flag', {
+    version: 7,
+    flag: {
+      version: 7,
+      deleted: true,
+      value: undefined,
+      trackEvents: false,
+    },
+  });
+});
+
+// -- context getter --
+
+it('uses current context from getter on each call', () => {
+  const flagManager = makeFlagManager();
+  const contextA = makeContext('user-a');
+  const contextB = makeContext('user-b');
+  let currentContext = contextA;
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => currentContext,
+  });
+
+  sink.handleChangeSet(makeResult(makePayload({ type: 'full', updates: [] })));
+  expect(flagManager.init).toHaveBeenLastCalledWith(contextA, {});
+
+  currentContext = contextB;
+  sink.handleChangeSet(makeResult(makePayload({ type: 'full', updates: [] })));
+  expect(flagManager.init).toHaveBeenLastCalledWith(contextB, {});
+});
+
+// -- logging --
+
+it('logs debug messages for each payload type', () => {
+  const flagManager = makeFlagManager();
+  const logger = makeLogger();
+
+  const sink = createDataSourceUpdateSink({
+    flagManager: flagManager as unknown as FlagManager,
+    contextGetter: () => makeContext(),
+    logger,
+  });
+
+  sink.handleChangeSet(makeResult(makePayload({ type: 'full', updates: [] })));
+  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('full payload'));
+
+  sink.handleChangeSet(makeResult(makePayload({ type: 'partial', updates: [] })));
+  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('partial payload'));
+
+  sink.handleChangeSet(makeResult(makePayload({ type: 'none', updates: [] })));
+  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('none'));
+});
diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts
index 13e6fc0b88..aa86741e80 100644
--- a/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts
+++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/FDv2DataSource.test.ts
@@ -41,7 +41,7 @@ it('resolves start() when initializer returns changeSet with selector', async ()
 
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   expect(statusManager.requestStateUpdate).toHaveBeenCalledWith('INITIALIZING');
   expect(statusManager.requestStateUpdate).toHaveBeenCalledWith('VALID');
   ds.close();
@@ -114,7 +114,7 @@ it('continues past initializer errors', async () => {
 
   expect(logger.warn).toHaveBeenCalled();
   expect(statusManager.reportError).toHaveBeenCalled();
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -135,7 +135,7 @@ it('continues past terminal errors in initializers', async () => {
   });
 
   await ds.start();
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -157,7 +157,7 @@ it('skips to synchronizers when no initializers are configured', async () => {
 
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -181,7 +181,7 @@ it('delivers changeSet from synchronizer to callback', async () => {
 
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   expect(statusManager.requestStateUpdate).toHaveBeenCalledWith('VALID');
   ds.close();
 });
@@ -213,7 +213,7 @@ it('blocks synchronizer on terminal error and moves to next', async () => {
 
   expect(logger.error).toHaveBeenCalled();
   expect(statusManager.reportError).toHaveBeenCalled();
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -239,7 +239,7 @@ it('continues on interrupted results from synchronizer', async () => {
   await ds.start();
 
   expect(statusManager.reportError).toHaveBeenCalled();
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -261,7 +261,7 @@ it('continues on goodbye results from synchronizer', async () => {
 
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -315,8 +315,8 @@ it('triggers fdv1 fallback when synchronizer changeSet has fdv1Fallback flag', a
   await statusManager.waitForState('VALID', 2);
 
   expect(dataCallback).toHaveBeenCalledTimes(2);
-  expect(dataCallback).toHaveBeenCalledWith(fdv2Payload);
-  expect(dataCallback).toHaveBeenCalledWith(fdv1Payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload: fdv2Payload }));
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload: fdv1Payload }));
   ds.close();
 });
 
@@ -348,7 +348,7 @@ it('triggers fdv1 fallback on terminal error with fdv1Fallback flag', async () =
   await ds.start();
 
   expect(logger.error).toHaveBeenCalled();
-  expect(dataCallback).toHaveBeenCalledWith(fdv1Payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload: fdv1Payload }));
   ds.close();
 });
 
@@ -398,7 +398,7 @@ it('falls back to next synchronizer when fallback condition fires', async () =>
   await ds.start();
 
   expect(logger.warn).toHaveBeenCalledWith(expect.stringContaining('Fallback condition fired'));
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -471,7 +471,7 @@ it('recovers to primary synchronizer when recovery condition fires', async () =>
   await ds.start();
 
   expect(logger.info).toHaveBeenCalledWith(expect.stringContaining('Recovery condition fired'));
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -614,7 +614,7 @@ it('resolves with initializer data even when no synchronizers exist', async () =
 
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -678,9 +678,9 @@ it('delivers multiple changeSets from synchronizer in order', async () => {
   await statusManager.waitForState('VALID', 3);
 
   expect(dataCallback).toHaveBeenCalledTimes(3);
-  expect(dataCallback).toHaveBeenNthCalledWith(1, payload1);
-  expect(dataCallback).toHaveBeenNthCalledWith(2, payload2);
-  expect(dataCallback).toHaveBeenNthCalledWith(3, payload3);
+  expect(dataCallback).toHaveBeenNthCalledWith(1, expect.objectContaining({ payload: payload1 }));
+  expect(dataCallback).toHaveBeenNthCalledWith(2, expect.objectContaining({ payload: payload2 }));
+  expect(dataCallback).toHaveBeenNthCalledWith(3, expect.objectContaining({ payload: payload3 }));
   ds.close();
 });
 
@@ -709,7 +709,7 @@ it('first initializer with selector prevents second initializer from running', a
   await ds.start();
 
   expect(dataCallback).toHaveBeenCalledTimes(1);
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   expect(secondInitFactory).not.toHaveBeenCalled();
   ds.close();
 });
@@ -804,7 +804,7 @@ it('fdv1 fallback not triggered when fdv1Fallback flag is absent', async () => {
 
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   expect(fdv1Factory).not.toHaveBeenCalled();
   ds.close();
 });
@@ -840,7 +840,7 @@ it('fdv1 fallback blocks other synchronizers', async () => {
 
   // FDv1 fallback should block non-FDv1 synchronizers — second sync should not be called
   expect(secondSyncFactory).not.toHaveBeenCalled();
-  expect(dataCallback).toHaveBeenCalledWith(fdv1Payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload: fdv1Payload }));
   ds.close();
 });
 
@@ -864,7 +864,7 @@ it('fdv1 fallback ignored when no FDv1 synchronizer is configured', async () =>
   await ds.start();
 
   // Should process the changeSet normally without error
-  expect(dataCallback).toHaveBeenCalledWith(payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload }));
   ds.close();
 });
 
@@ -892,6 +892,6 @@ it('fdv1 fallback triggered on interrupted result with fdv1Fallback flag', async
   // Start resolves when the fdv1 synchronizer delivers its changeSet.
   await ds.start();
 
-  expect(dataCallback).toHaveBeenCalledWith(fdv1Payload);
+  expect(dataCallback).toHaveBeenCalledWith(expect.objectContaining({ payload: fdv1Payload }));
   ds.close();
 });
diff --git a/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts b/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts
new file mode 100644
index 0000000000..40b1959aae
--- /dev/null
+++ b/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts
@@ -0,0 +1,114 @@
+import { Context, LDLogger } from '@launchdarkly/js-sdk-common';
+
+import { FlagManager } from '../../flag-manager/FlagManager';
+import { flagEvalPayloadToItemDescriptors } from '../flagEvalMapper';
+import { ChangeSetResult } from './FDv2SourceResult';
+
+/**
+ * Configuration for creating a {@link DataSourceUpdateSink}.
+ */
+export interface DataSourceUpdateSinkConfig {
+  /** FlagManager to apply flag updates to. */
+  flagManager: FlagManager;
+
+  /**
+   * Getter for the current evaluation context. A getter is used instead of a
+   * stored context because the context changes on `identify()`.
+   */
+  contextGetter: () => Context;
+
+  /** Optional logger. */
+  logger?: LDLogger;
+}
+
+/**
+ * Processes FDv2 {@link ChangeSetResult} payloads and applies them to
+ * {@link FlagManager}. Manages the selector (basis state) for delta sync
+ * and tracks the environmentId from response headers.
+ *
+ * This is the client-side equivalent of the server SDK's payload listener,
+ * adapted for the Initializer/Synchronizer/FDv2DataSource architecture.
+ */
+export interface DataSourceUpdateSink {
+  /**
+   * Processes a {@link ChangeSetResult} and applies flag updates to
+   * FlagManager. Pass this as the `dataCallback` to {@link FDv2DataSource}.
+   */
+  handleChangeSet(result: ChangeSetResult): void;
+
+  /**
+   * Returns the current selector string for use as the `basis` query
+   * parameter. Returns undefined if no selector has been received yet.
+   * Pass this as the `selectorGetter` to {@link FDv2DataSource}.
+   *
+   * The selector is stored in memory only and is NOT persisted with cache
+   * (Requirement 6.2.1).
+   */
+  getSelector(): string | undefined;
+
+  /**
+   * Returns the environment ID from the most recent response headers.
+   */
+  getEnvironmentId(): string | undefined;
+}
+
+/**
+ * Creates a {@link DataSourceUpdateSink}.
+ */
+export function createDataSourceUpdateSink(
+  config: DataSourceUpdateSinkConfig,
+): DataSourceUpdateSink {
+  const { flagManager, contextGetter, logger } = config;
+
+  let selector: string | undefined;
+  let environmentId: string | undefined;
+
+  return {
+    handleChangeSet(result: ChangeSetResult): void {
+      const { payload } = result;
+
+      // Update selector if present in payload (Requirement 6.2.1: in-memory only).
+      if (payload.state) {
+        selector = payload.state;
+      }
+
+      // Track environmentId from response headers (Requirement 4.2.1).
+      if (result.environmentId) {
+        environmentId = result.environmentId;
+      }
+
+      const context = contextGetter();
+
+      switch (payload.type) {
+        case 'full': {
+          const descriptors = flagEvalPayloadToItemDescriptors(payload.updates);
+          logger?.debug(`FDv2 full payload: initializing ${Object.keys(descriptors).length} flags`);
+          flagManager.init(context, descriptors);
+          break;
+        }
+        case 'partial': {
+          const descriptors = flagEvalPayloadToItemDescriptors(payload.updates);
+          logger?.debug(`FDv2 partial payload: upserting ${Object.keys(descriptors).length} flags`);
+          Object.entries(descriptors).forEach(([key, descriptor]) => {
+            flagManager.upsert(context, key, descriptor);
+          });
+          break;
+        }
+        case 'none':
+          logger?.debug('FDv2 payload type "none": no flag updates needed');
+          break;
+        default:
+          logger?.warn(`Unknown FDv2 payload type: ${payload.type}`);
+          break;
+      }
+    },
+
+    getSelector(): string | undefined {
+      return selector;
+    },
+
+    getEnvironmentId(): string | undefined {
+      return environmentId;
+    },
+  };
+}
diff --git a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts
index d8c122c908..2682f78766 100644
--- a/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts
+++ b/packages/shared/sdk-client/src/datasource/fdv2/FDv2DataSource.ts
@@ -1,4 +1,4 @@
-import { internal, LDLogger } from '@launchdarkly/js-sdk-common';
+import { LDLogger } from '@launchdarkly/js-sdk-common';
 
 import { DataSourceStatusManager } from '../DataSourceStatusManager';
 import {
@@ -12,9 +12,11 @@ import { ChangeSetResult, FDv2SourceResult, StatusResult } from './FDv2SourceRes
 import { createSourceManager, InitializerFactory, SynchronizerSlot } from './SourceManager';
 
 /**
- * Callback invoked when the orchestrator produces a changeSet payload.
+ * Callback invoked when the orchestrator produces a changeSet result.
+ * Receives the full {@link ChangeSetResult} so consumers have access to both
+ * the payload and metadata (environmentId, fdv1Fallback).
  */
-export type DataCallback = (payload: internal.Payload) => void;
+export type DataCallback = (result: ChangeSetResult) => void;
 
 /**
  * Configuration for the {@link FDv2DataSource} orchestrator.
@@ -107,7 +109,7 @@ export function createFDv2DataSource(config: FDv2DataSourceConfig): FDv2DataSour
   }
 
   function applyChangeSet(result: ChangeSetResult) {
-    dataCallback(result.payload);
+    dataCallback(result);
     statusManager.requestStateUpdate('VALID');
   }
 
diff --git a/packages/shared/sdk-client/src/datasource/fdv2/index.ts b/packages/shared/sdk-client/src/datasource/fdv2/index.ts
index 78879c5ff9..136026bfe6 100644
--- a/packages/shared/sdk-client/src/datasource/fdv2/index.ts
+++ b/packages/shared/sdk-client/src/datasource/fdv2/index.ts
@@ -1,6 +1,9 @@
 export type { AsyncQueue } from './AsyncQueue';
 export { createAsyncQueue } from './AsyncQueue';
 
+export type { DataSourceUpdateSink, DataSourceUpdateSinkConfig } from './DataSourceUpdateSink';
+export { createDataSourceUpdateSink } from './DataSourceUpdateSink';
+
 export type { FDv2PollResponse, FDv2Requestor } from './FDv2Requestor';
 export { makeFDv2Requestor } from './FDv2Requestor';
 

From 841dce76cafb855264eb8cb43bb3b6aa4e2be1de Mon Sep 17 00:00:00 2001
From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com>
Date: Thu, 5 Mar 2026 14:38:11 -0800
Subject: [PATCH 2/2] feat: Add applyChanges to FlagManager for changeset
 processing

Moves selector and environmentId management into FlagManager, following
the server SDK's LDTransactionalFeatureStore.applyChanges pattern.
FlagManager.applyChanges accepts pre-converted ItemDescriptors with
basis flag, selector, and environmentId. DataSourceUpdateSink simplified
to a thin adapter that converts FDv2 protocol types and delegates.
---
 .../fdv2/DataSourceUpdateSink.test.ts         | 235 +++++++-----------
 .../flag-manager/FlagManager.test.ts          | 122 +++++++++
 .../datasource/fdv2/DataSourceUpdateSink.ts   |  78 ++----
 .../src/flag-manager/FlagManager.ts           |  59 +++++
 .../src/flag-manager/FlagPersistence.ts       |  20 ++
 5 files changed, 306 insertions(+), 208 deletions(-)

diff --git a/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts b/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts
index c09d304e71..95e6628a44 100644
--- a/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts
+++ b/packages/shared/sdk-client/__tests__/datasource/fdv2/DataSourceUpdateSink.test.ts
@@ -9,12 +9,9 @@ function makeContext(key: string = 'user-key'): Context {
   return Context.fromLDContext({ kind: 'user', key });
 }
 
-function makeFlagManager(): jest.Mocked<Pick<FlagManager, 'init' | 'upsert' | 'get' | 'getAll'>> {
+function makeFlagManager(): jest.Mocked<Pick<FlagManager, 'applyChanges'>> {
   return {
-    init: jest.fn().mockResolvedValue(undefined),
-    upsert: jest.fn().mockResolvedValue(true),
-    get: jest.fn(),
-    getAll: jest.fn(),
+    applyChanges: jest.fn().mockResolvedValue(undefined),
   };
 }
 
@@ -58,9 +55,9 @@ function makeFlagEvalUpdate(
   };
 }
 
-// -- full payload --
+// -- full payload delegation --
 
-it('calls flagManager.init with converted descriptors on full payload', () => {
+it('delegates full payload to flagManager.applyChanges with basis=true', () => {
   const flagManager = makeFlagManager();
   const context = makeContext();
 
@@ -71,28 +68,27 @@ it('calls flagManager.init with converted descriptors on full payload', () => {
 
   const payload = makePayload({
     type: 'full',
+    state: 'sel-1',
     updates: [makeFlagEvalUpdate('flag-1', 10, true), makeFlagEvalUpdate('flag-2', 20, 'hello')],
   });
 
-  sink.handleChangeSet(makeResult(payload));
+  sink.handleChangeSet(makeResult(payload, { environmentId: 'env-1' }));
 
-  expect(flagManager.init).toHaveBeenCalledTimes(1);
-  expect(flagManager.init).toHaveBeenCalledWith(context, {
-    'flag-1': {
-      version: 10,
-      flag: { value: true, trackEvents: false, version: 10 },
-    },
-    'flag-2': {
-      version: 20,
-      flag: { value: 'hello', trackEvents: false, version: 20 },
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(
+    context,
+    {
+      'flag-1': { version: 10, flag: { value: true, trackEvents: false, version: 10 } },
+      'flag-2': { version: 20, flag: { value: 'hello', trackEvents: false, version: 20 } },
     },
-  });
-  expect(flagManager.upsert).not.toHaveBeenCalled();
+    true,
+    'sel-1',
+    'env-1',
+  );
 });
 
-// -- partial payload --
+// -- partial payload delegation --
 
-it('calls flagManager.upsert for each flag on partial payload', () => {
+it('delegates partial payload to flagManager.applyChanges with basis=false', () => {
   const flagManager = makeFlagManager();
   const context = makeContext();
 
@@ -103,85 +99,44 @@ it('calls flagManager.upsert for each flag on partial payload', () => {
 
   const payload = makePayload({
     type: 'partial',
-    updates: [makeFlagEvalUpdate('flag-a', 5, 42), makeFlagEvalUpdate('flag-b', 6, false)],
+    state: 'sel-2',
+    updates: [makeFlagEvalUpdate('flag-a', 5, 42)],
   });
 
   sink.handleChangeSet(makeResult(payload));
 
-  expect(flagManager.upsert).toHaveBeenCalledTimes(2);
-  expect(flagManager.upsert).toHaveBeenCalledWith(context, 'flag-a', {
-    version: 5,
-    flag: { value: 42, trackEvents: false, version: 5 },
-  });
-  expect(flagManager.upsert).toHaveBeenCalledWith(context, 'flag-b', {
-    version: 6,
-    flag: { value: false, trackEvents: false, version: 6 },
-  });
-  expect(flagManager.init).not.toHaveBeenCalled();
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(
+    context,
+    { 'flag-a': { version: 5, flag: { value: 42, trackEvents: false, version: 5 } } },
+    false,
+    'sel-2',
+    undefined,
+  );
 });
 
-// -- none payload --
+// -- none payload delegation --
 
-it('does not call flagManager on none payload', () => {
-  const flagManager = makeFlagManager();
-
-  const sink = createDataSourceUpdateSink({
-    flagManager: flagManager as unknown as FlagManager,
-    contextGetter: () => makeContext(),
-  });
-
-  sink.handleChangeSet(makeResult(makePayload({ type: 'none', updates: [] })));
-
-  expect(flagManager.init).not.toHaveBeenCalled();
-  expect(flagManager.upsert).not.toHaveBeenCalled();
-});
-
-// -- selector management --
-
-it('stores selector from payload state', () => {
-  const flagManager = makeFlagManager();
-
-  const sink = createDataSourceUpdateSink({
-    flagManager: flagManager as unknown as FlagManager,
-    contextGetter: () => makeContext(),
-  });
-
-  expect(sink.getSelector()).toBeUndefined();
-
-  sink.handleChangeSet(makeResult(makePayload({ state: 'selector-1' })));
-
-  expect(sink.getSelector()).toBe('selector-1');
-});
-
-it('updates selector on subsequent payloads', () => {
+it('delegates none payload to flagManager.applyChanges with empty updates', () => {
   const flagManager = makeFlagManager();
+  const context = makeContext();
 
   const sink = createDataSourceUpdateSink({
     flagManager: flagManager as unknown as FlagManager,
-    contextGetter: () => makeContext(),
+    contextGetter: () => context,
   });
 
-  sink.handleChangeSet(makeResult(makePayload({ state: 'a' })));
-  sink.handleChangeSet(makeResult(makePayload({ state: 'b' })));
+  sink.handleChangeSet(
+    makeResult(makePayload({ type: 'none', state: 'sel-3', updates: [] }), {
+      environmentId: 'env-2',
+    }),
+  );
 
-  expect(sink.getSelector()).toBe('b');
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(context, {}, false, 'sel-3', 'env-2');
 });
 
-it('does not clear selector when payload has no state', () => {
-  const flagManager = makeFlagManager();
-
-  const sink = createDataSourceUpdateSink({
-    flagManager: flagManager as unknown as FlagManager,
-    contextGetter: () => makeContext(),
-  });
-
-  sink.handleChangeSet(makeResult(makePayload({ state: 'a' })));
-  sink.handleChangeSet(makeResult(makePayload({ state: undefined })));
+// -- selector passthrough --
 
-  expect(sink.getSelector()).toBe('a');
-});
-
-it('does not set selector from empty string state', () => {
+it('passes undefined selector when payload state is empty string', () => {
   const flagManager = makeFlagManager();
 
   const sink = createDataSourceUpdateSink({
@@ -191,12 +146,16 @@ it('does not set selector from empty string state', () => {
 
   sink.handleChangeSet(makeResult(makePayload({ state: '' })));
 
-  expect(sink.getSelector()).toBeUndefined();
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(
+    expect.anything(),
+    expect.anything(),
+    true,
+    undefined,
+    undefined,
+  );
 });
 
-// -- environmentId tracking --
-
-it('tracks environmentId from ChangeSetResult', () => {
+it('passes undefined selector when payload state is undefined', () => {
   const flagManager = makeFlagManager();
 
   const sink = createDataSourceUpdateSink({
@@ -204,30 +163,20 @@ it('tracks environmentId from ChangeSetResult', () => {
     contextGetter: () => makeContext(),
   });
 
-  expect(sink.getEnvironmentId()).toBeUndefined();
-
-  sink.handleChangeSet(makeResult(makePayload(), { environmentId: 'env-123' }));
-
-  expect(sink.getEnvironmentId()).toBe('env-123');
-});
-
-it('updates environmentId on subsequent results', () => {
-  const flagManager = makeFlagManager();
-
-  const sink = createDataSourceUpdateSink({
-    flagManager: flagManager as unknown as FlagManager,
-    contextGetter: () => makeContext(),
-  });
-
-  sink.handleChangeSet(makeResult(makePayload(), { environmentId: 'env-1' }));
-  sink.handleChangeSet(makeResult(makePayload(), { environmentId: 'env-2' }));
+  sink.handleChangeSet(makeResult(makePayload({ state: undefined })));
 
-  expect(sink.getEnvironmentId()).toBe('env-2');
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(
+    expect.anything(),
+    expect.anything(),
+    true,
+    undefined,
+    undefined,
+  );
 });
 
 // -- non-flagEval kinds --
 
-it('ignores non-flagEval update kinds', () => {
+it('ignores non-flagEval update kinds in conversion', () => {
   const flagManager = makeFlagManager();
 
   const sink = createDataSourceUpdateSink({
@@ -245,12 +194,18 @@ it('ignores non-flagEval update kinds', () => {
 
   sink.handleChangeSet(makeResult(payload));
 
-  expect(flagManager.init).toHaveBeenCalledWith(expect.anything(), {});
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(
+    expect.anything(),
+    {},
+    true,
+    expect.anything(),
+    undefined,
+  );
 });
 
 // -- delete handling --
 
-it('handles delete updates in full payload', () => {
+it('converts delete updates to tombstone descriptors', () => {
   const flagManager = makeFlagManager();
   const context = makeContext();
 
@@ -266,44 +221,18 @@ it('handles delete updates in full payload', () => {
 
   sink.handleChangeSet(makeResult(payload));
 
-  expect(flagManager.init).toHaveBeenCalledWith(context, {
-    'deleted-flag': {
-      version: 7,
-      flag: {
+  expect(flagManager.applyChanges).toHaveBeenCalledWith(
+    context,
+    {
+      'deleted-flag': {
         version: 7,
-        deleted: true,
-        value: undefined,
-        trackEvents: false,
+        flag: { version: 7, deleted: true, value: undefined, trackEvents: false },
       },
     },
-  });
-});
-
-it('handles delete updates in partial payload', () => {
-  const flagManager = makeFlagManager();
-  const context = makeContext();
-
-  const sink = createDataSourceUpdateSink({
-    flagManager: flagManager as unknown as FlagManager,
-    contextGetter: () => context,
-  });
-
-  const payload = makePayload({
-    type: 'partial',
-    updates: [makeFlagEvalUpdate('deleted-flag', 7, undefined, { deleted: true })],
-  });
-
-  sink.handleChangeSet(makeResult(payload));
-
-  expect(flagManager.upsert).toHaveBeenCalledWith(context, 'deleted-flag', {
-    version: 7,
-    flag: {
-      version: 7,
-      deleted: true,
-      value: undefined,
-      trackEvents: false,
-    },
-  });
+    true,
+    expect.anything(),
+    undefined,
+  );
 });
 
 // -- context getter --
@@ -320,11 +249,23 @@ it('uses current context from getter on each call', () => {
   });
 
   sink.handleChangeSet(makeResult(makePayload({ type: 'full', updates: [] })));
-  expect(flagManager.init).toHaveBeenLastCalledWith(contextA, {});
+  expect(flagManager.applyChanges).toHaveBeenLastCalledWith(
+    contextA,
+    {},
+    true,
+    expect.anything(),
+    undefined,
+  );
 
   currentContext = contextB;
   sink.handleChangeSet(makeResult(makePayload({ type: 'full', updates: [] })));
-  expect(flagManager.init).toHaveBeenLastCalledWith(contextB, {});
+  expect(flagManager.applyChanges).toHaveBeenLastCalledWith(
+    contextB,
+    {},
+    true,
+    expect.anything(),
+    undefined,
+  );
 });
 
 // -- logging --
@@ -340,10 +281,10 @@ it('logs debug messages for each payload type', () => {
   });
 
   sink.handleChangeSet(makeResult(makePayload({ type: 'full', updates: [] })));
-  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('full payload'));
+  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('full'));
 
   sink.handleChangeSet(makeResult(makePayload({ type: 'partial', updates: [] })));
-  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('partial payload'));
+  expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('partial'));
 
   sink.handleChangeSet(makeResult(makePayload({ type: 'none', updates: [] })));
   expect(logger.debug).toHaveBeenCalledWith(expect.stringContaining('none'));
diff --git a/packages/shared/sdk-client/__tests__/flag-manager/FlagManager.test.ts b/packages/shared/sdk-client/__tests__/flag-manager/FlagManager.test.ts
index d8fe440004..6134bb9950 100644
--- a/packages/shared/sdk-client/__tests__/flag-manager/FlagManager.test.ts
+++ b/packages/shared/sdk-client/__tests__/flag-manager/FlagManager.test.ts
@@ -87,6 +87,128 @@ function makeMockItemDescriptor(version: number = 1, value: any = 'test-value'):
   };
 }
 
+describe('given a flag manager', () => {
+  let flagManager: DefaultFlagManager;
+  let mockPlatform: Platform;
+  let mockLogger: LDLogger;
+
+  beforeEach(() => {
+    mockLogger = makeMockLogger();
+    mockPlatform = makeMockPlatform(makeMemoryStorage(), makeMockCrypto());
+    flagManager = new DefaultFlagManager(
+      mockPlatform,
+      TEST_SDK_KEY,
+      TEST_MAX_CACHED_CONTEXTS,
+      mockLogger,
+    );
+  });
+
+  it('applies a full changeset replacing all flags', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+
+    await flagManager.applyChanges(
+      context,
+      {
+        'flag-a': makeMockItemDescriptor(1, 'a'),
+        'flag-b': makeMockItemDescriptor(2, 'b'),
+      },
+      true,
+    );
+
+    expect(flagManager.get('flag-a')?.flag.value).toBe('a');
+    expect(flagManager.get('flag-b')?.flag.value).toBe('b');
+  });
+
+  it('applies a full changeset that replaces previous flags', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+
+    await flagManager.applyChanges(context, { old: makeMockItemDescriptor(1, 'old') }, true);
+    await flagManager.applyChanges(context, { new: makeMockItemDescriptor(2, 'new') }, true);
+
+    expect(flagManager.get('old')).toBeUndefined();
+    expect(flagManager.get('new')?.flag.value).toBe('new');
+  });
+
+  it('applies a partial changeset upserting individual flags', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+
+    await flagManager.applyChanges(
+      context,
+      {
+        'flag-a': makeMockItemDescriptor(1, 'a'),
+        'flag-b': makeMockItemDescriptor(1, 'b'),
+      },
+      true,
+    );
+
+    await flagManager.applyChanges(context, { 'flag-b': makeMockItemDescriptor(2, 'b2') }, false);
+
+    expect(flagManager.get('flag-a')?.flag.value).toBe('a');
+    expect(flagManager.get('flag-b')?.flag.value).toBe('b2');
+  });
+
+  it('applies a partial changeset that adds new flags', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+
+    await flagManager.applyChanges(context, { existing: makeMockItemDescriptor(1, 'e') }, true);
+    await flagManager.applyChanges(context, { added: makeMockItemDescriptor(1, 'new') }, false);
+
+    expect(flagManager.get('existing')?.flag.value).toBe('e');
+    expect(flagManager.get('added')?.flag.value).toBe('new');
+  });
+
+  it('applies empty changeset without error', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, false);
+    expect(Object.keys(flagManager.getAll())).toHaveLength(0);
+  });
+
+  it('returns undefined for selector and environmentId initially', () => {
+    expect(flagManager.getSelector()).toBeUndefined();
+    expect(flagManager.getEnvironmentId()).toBeUndefined();
+  });
+
+  it('stores selector from applyChanges', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, true, 'sel-1');
+    expect(flagManager.getSelector()).toBe('sel-1');
+  });
+
+  it('updates selector on subsequent applyChanges', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, true, 'sel-1');
+    await flagManager.applyChanges(context, {}, false, 'sel-2');
+    expect(flagManager.getSelector()).toBe('sel-2');
+  });
+
+  it('does not clear selector when not provided', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, true, 'sel-1');
+    await flagManager.applyChanges(context, {}, false);
+    expect(flagManager.getSelector()).toBe('sel-1');
+  });
+
+  it('stores environmentId from applyChanges', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, true, undefined, 'env-123');
+    expect(flagManager.getEnvironmentId()).toBe('env-123');
+  });
+
+  it('updates environmentId on subsequent applyChanges', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, true, undefined, 'env-1');
+    await flagManager.applyChanges(context, {}, false, undefined, 'env-2');
+    expect(flagManager.getEnvironmentId()).toBe('env-2');
+  });
+
+  it('does not clear environmentId when not provided', async () => {
+    const context = Context.fromLDContext({ kind: 'user', key: 'user-key' });
+    await flagManager.applyChanges(context, {}, true, undefined, 'env-1');
+    await flagManager.applyChanges(context, {}, false);
+    expect(flagManager.getEnvironmentId()).toBe('env-1');
+  });
+});
+
 describe('FlagManager override tests', () => {
   let flagManager: DefaultFlagManager;
   let mockPlatform: Platform;
diff --git a/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts b/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts
index 40b1959aae..572d5e104a 100644
--- a/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts
+++ b/packages/shared/sdk-client/src/datasource/fdv2/DataSourceUpdateSink.ts
@@ -22,12 +22,11 @@ export interface DataSourceUpdateSinkConfig {
 }
 
 /**
- * Processes FDv2 {@link ChangeSetResult} payloads and applies them to
- * {@link FlagManager}. Manages the selector (basis state) for delta sync
- * and tracks the environmentId from response headers.
+ * Thin adapter between the FDv2 protocol layer and {@link FlagManager}.
+ * Converts FDv2 protocol types (payload updates) to {@link ItemDescriptor}
+ * format and delegates to {@link FlagManager.applyChanges}.
  *
- * This is the client-side equivalent of the server SDK's payload listener,
- * adapted for the Initializer/Synchronizer/FDv2DataSource architecture.
+ * Selector and environmentId are managed by FlagManager, not the sink.
  */
 export interface DataSourceUpdateSink {
   /**
@@ -35,21 +34,6 @@ export interface DataSourceUpdateSink {
    * FlagManager. Pass this as the `dataCallback` to {@link FDv2DataSource}.
    */
   handleChangeSet(result: ChangeSetResult): void;
-
-  /**
-   * Returns the current selector string for use as the `basis` query
-   * parameter. Returns undefined if no selector has been received yet.
-   * Pass this as the `selectorGetter` to {@link FDv2DataSource}.
-   *
-   * The selector is stored in memory only and is NOT persisted with cache
-   * (Requirement 6.2.1).
-   */
-  getSelector(): string | undefined;
-
-  /**
-   * Returns the environment ID from the most recent response headers.
-   */
-  getEnvironmentId(): string | undefined;
 }
 
 /**
@@ -60,55 +44,27 @@ export function createDataSourceUpdateSink(
 ): DataSourceUpdateSink {
   const { flagManager, contextGetter, logger } = config;
 
-  let selector: string | undefined;
-  let environmentId: string | undefined;
-
   return {
     handleChangeSet(result: ChangeSetResult): void {
       const { payload } = result;
-
-      // Update selector if present in payload (Requirement 6.2.1: in-memory only).
-      if (payload.state) {
-        selector = payload.state;
-      }
-
-      // Track environmentId from response headers (Requirement 4.2.1).
-      if (result.environmentId) {
-        environmentId = result.environmentId;
-      }
-
       const context = contextGetter();
+      const selector = payload.state || undefined;
+      const { environmentId } = result;
 
-      switch (payload.type) {
-        case 'full': {
-          const descriptors = flagEvalPayloadToItemDescriptors(payload.updates);
-          logger?.debug(`FDv2 full payload: initializing ${Object.keys(descriptors).length} flags`);
-          flagManager.init(context, descriptors);
-          break;
-        }
-        case 'partial': {
-          const descriptors = flagEvalPayloadToItemDescriptors(payload.updates);
-          logger?.debug(`FDv2 partial payload: upserting ${Object.keys(descriptors).length} flags`);
-          Object.entries(descriptors).forEach(([key, descriptor]) => {
-            flagManager.upsert(context, key, descriptor);
-          });
-          break;
-        }
-        case 'none':
-          logger?.debug('FDv2 payload type "none": no flag updates needed');
-          break;
-        default:
-          logger?.warn(`Unknown FDv2 payload type: ${payload.type}`);
-          break;
+      if (payload.type === 'none') {
+        logger?.debug('FDv2 payload type "none": no flag updates needed');
+        flagManager.applyChanges(context, {}, false, selector, environmentId);
+        return;
       }
-    },
 
-    getSelector(): string | undefined {
-      return selector;
-    },
+      const descriptors = flagEvalPayloadToItemDescriptors(payload.updates);
+      const basis = payload.type === 'full';
+
+      logger?.debug(
+        `FDv2 ${payload.type} payload: ${basis ? 'initializing' : 'upserting'} ${Object.keys(descriptors).length} flags`,
+      );
 
-    getEnvironmentId(): string | undefined {
-      return environmentId;
+      flagManager.applyChanges(context, descriptors, basis, selector, environmentId);
     },
   };
 }
diff --git a/packages/shared/sdk-client/src/flag-manager/FlagManager.ts b/packages/shared/sdk-client/src/flag-manager/FlagManager.ts
index 094819bf56..b9702873ba 100644
--- a/packages/shared/sdk-client/src/flag-manager/FlagManager.ts
+++ b/packages/shared/sdk-client/src/flag-manager/FlagManager.ts
@@ -55,6 +55,39 @@ export interface FlagManager {
    */
   setBootstrap(context: Context, newFlags: { [key: string]: ItemDescriptor }): void;
 
+  /**
+   * Applies a changeset to the flag store. If {@link basis} is true, replaces
+   * all flags (like {@link init}). If false, upserts individual flags (like
+   * calling {@link upsert} for each entry).
+   *
+   * Stores the {@link selector} in memory for use as the `basis` query
+   * parameter on subsequent requests. The selector is NOT persisted with
+   * cache (Requirement 6.2.1). Tracks {@link environmentId} from response
+   * headers (Requirement 4.2.1).
+   *
+   * This follows the same pattern as the server SDK's
+   * `LDTransactionalFeatureStore.applyChanges`.
+   */
+  applyChanges(
+    context: Context,
+    updates: { [key: string]: ItemDescriptor },
+    basis: boolean,
+    selector?: string,
+    environmentId?: string,
+  ): Promise<void>;
+
+  /**
+   * Returns the current selector string for use as the `basis` query
+   * parameter. Returns undefined if no selector has been received yet.
+   * The selector is stored in memory only (Requirement 6.2.1).
+   */
+  getSelector(): string | undefined;
+
+  /**
+   * Returns the environment ID from the most recent changeset.
+   */
+  getEnvironmentId(): string | undefined;
+
   /**
    * Register a flag change callback.
    */
@@ -121,6 +154,8 @@ export default class DefaultFlagManager implements FlagManager {
   private _flagUpdater: FlagUpdater;
   private _flagPersistencePromise: Promise<FlagPersistence>;
   private _overrides?: { [key: string]: LDFlagValue };
+  private _selector?: string;
+  private _environmentId?: string;
 
   /**
    * @param platform implementation of various platform provided functionality
@@ -212,6 +247,30 @@ export default class DefaultFlagManager implements FlagManager {
     return (await this._flagPersistencePromise).loadCached(context);
   }
 
+  async applyChanges(
+    context: Context,
+    updates: { [key: string]: ItemDescriptor },
+    basis: boolean,
+    selector?: string,
+    environmentId?: string,
+  ): Promise<void> {
+    if (selector) {
+      this._selector = selector;
+    }
+    if (environmentId) {
+      this._environmentId = environmentId;
+    }
+    return (await this._flagPersistencePromise).applyChanges(context, updates, basis);
+  }
+
+  getSelector(): string | undefined {
+    return this._selector;
+  }
+
+  getEnvironmentId(): string | undefined {
+    return this._environmentId;
+  }
+
   on(callback: FlagsChangeCallback): void {
     this._flagUpdater.on(callback);
   }
diff --git a/packages/shared/sdk-client/src/flag-manager/FlagPersistence.ts b/packages/shared/sdk-client/src/flag-manager/FlagPersistence.ts
index 118b35df3b..6e2c95ebb5 100644
--- a/packages/shared/sdk-client/src/flag-manager/FlagPersistence.ts
+++ b/packages/shared/sdk-client/src/flag-manager/FlagPersistence.ts
@@ -51,6 +51,26 @@ export default class FlagPersistence {
     return false;
   }
 
+  /**
+   * Applies a changeset to the flag store. If {@link basis} is true, replaces all
+   * flags via {@link FlagUpdater.init}. If false, upserts individual flags via
+   * {@link FlagUpdater.upsert}. Persists changes after applying.
+   */
+  async applyChanges(
+    context: Context,
+    updates: { [key: string]: ItemDescriptor },
+    basis: boolean,
+  ): Promise<void> {
+    if (basis) {
+      this._flagUpdater.init(context, updates);
+    } else {
+      Object.entries(updates).forEach(([key, descriptor]) => {
+        this._flagUpdater.upsert(context, key, descriptor);
+      });
+    }
+    await this._storeCache(context);
+  }
+
   /**
    * Loads the flags from persistence for the provided context and gives those to the
    * {@link FlagUpdater} this {@link FlagPersistence} was constructed with.