feat: Translate FDv2 payloads at the data source layer (#309)

BEGIN_COMMIT_OVERRIDE
feat: Translate FDv2 payloads at the data source layer
feat: Add FDv2 streaming source factories and query-parameter
authentication
END_COMMIT_OVERRIDE

## What this changes

Moves flag-eval translation out of the apply path and into the data
source layer, and introduces a typed `ChangeSet` distinct from the
wire-level `Payload`.

Previously the streaming and polling sources emitted a `ChangeSetResult`
carrying the raw wire `Payload` (a list of `Update`s with unparsed
object maps), and the flag-eval objects were converted to typed
descriptors later, at apply time in
`DataSourceEventHandler.handlePayload`. A payload that parsed at the
protocol level but whose flag objects were malformed therefore failed
*after* acquisition — surfacing as an invalid-message that drove a
connection restart, on a fixed interval, without ever arming the
orchestrator's fallback timer.

Now:

- `ChangeSet` carries typed `Map<String, ItemDescriptor>` updates plus
the type and selector. `translatePayload` converts a wire `Payload` into
a `ChangeSet`, throwing if any flag-eval object cannot be parsed.
- The streaming and polling sources translate at acquisition. A
translation failure becomes an **interrupted** source result, exactly
like a malformed protocol body — so the orchestrator's fallback timer
governs recovery (a source stuck on invalid data falls back after the
timeout instead of retrying forever). Streaming discards the partial
handler state and keeps the SSE connection; the server's next valid
payload is processed normally.
- `ChangeSetResult` carries the `ChangeSet`, and `handlePayload` applies
it directly with no conversion and no failure path of its own.
- The cache initializer builds its `ChangeSet` straight from the
already-typed cached evaluation results, dropping a JSON round-trip that
previously serialized typed results only to re-parse them.

This is foundational for the FDv2 data system: the streaming-source and
orchestrator PRs build on `ChangeSet`. Nothing constructs the FDv2
sources in production yet, so behavior is unchanged for shipping code.

## Testing

Source-layer tests cover a protocol-valid payload whose flag data cannot
be parsed: polling returns interrupted, and streaming returns
interrupted then recovers on the next valid payload. The flag manager /
event handler tests apply typed change sets directly (full replace,
partial without per-item version comparison, none). Cache initializer
tests assert the evaluation result is carried through without
re-parsing. Full `common_client` suite passes.

SDK-2186

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Touches FDv2 acquisition, URL/auth composition, and
error-classification paths that govern fallback and flag updates;
changes are largely behind not-yet-production FDv2 wiring but alter
recovery semantics when invalid flag data arrives.
> 
> **Overview**
> FDv2 now separates wire **`Payload`** from typed **`ChangeSet`**
(`Map<String, ItemDescriptor>`). **`translatePayload`** runs in
polling/streaming (and cache builds descriptors directly), so
**`ChangeSetResult`** and **`handlePayload`** only apply already-typed
updates.
> 
> **Malformed flag-eval data** is surfaced as **`interrupted`** at the
source layer (like other transient parse errors) instead of failing
later in the event handler—so the orchestrator’s fallback timer can kick
in rather than endless connection retries.
> 
> Adds shared **`buildFDv2Uri`** (relay query params, `basis`,
`withReasons`) for polling and streaming; **`SourceFactoryContext`**
carries credential-driven **`auth`** query params. **Streaming
synchronizer factories** are implemented (SSE client, per-reconnect
URI/`basis`, legacy **`ping`** → one-shot poll, env ID fallback).
Streaming also maps **`UnrecoverableStatusError`** to terminal errors
with optional FDv1 fallback.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
38646ce. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: kinyoklion

packages/common_client/lib/src/data_sources/data_source_event_handler.dart

@@ -6,7 +6,6 @@ import '../flag_manager/flag_manager.dart';
 import '../item_descriptor.dart';
 import 'data_source_status.dart';
 import 'data_source_status_manager.dart';
-import 'fdv2/flag_eval_mapper.dart';
 import 'fdv2/payload.dart';
 
 enum MessageStatus { messageHandled, invalidMessage, unhandledVerb }
@@ -98,24 +97,23 @@ final class DataSourceEventHandler {
     }
   }
 
-  /// Applies an FDv2 payload to the flag store.
+  /// Applies an FDv2 change set to the flag store.
   ///
-  /// Full payloads replace the stored flags, partial payloads apply each
-  /// update, and a payload of type none confirms the SDK is up to date
-  /// without changing data. All three mark the data source valid.
-  Future<MessageStatus> handlePayload(LDContext context, Payload payload,
+  /// Full change sets replace the stored flags, partial change sets apply
+  /// each update, and a change set of type none confirms the SDK is up to
+  /// date without changing data. All three mark the data source valid.
+  Future<MessageStatus> handlePayload(LDContext context, ChangeSet changeSet,
       {String? environmentId}) async {
     try {
-      final updates = mapUpdatesToItemDescriptors(payload.updates);
-      await _flagManager.applyChanges(context, updates, payload.type,
+      await _flagManager.applyChanges(
+          context, changeSet.updates, changeSet.type,
           environmentId: environmentId);
       _statusManager.setValid();
       return MessageStatus.messageHandled;
     } catch (err) {
-      _logger.error('FDv2 payload contained invalid flag data: '
-          '${err.runtimeType}');
+      _logger.error('Failed to apply an FDv2 change set: ${err.runtimeType}');
       _statusManager.setErrorByKind(
-          ErrorKind.invalidData, 'FDv2 payload contained invalid data');
+          ErrorKind.invalidData, 'Failed to apply an FDv2 change set');
       return MessageStatus.invalidMessage;
     }
   }

packages/common_client/lib/src/data_sources/fdv2/cache_initializer.dart

@@ -1,8 +1,7 @@
 import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart';
 
-import 'flag_eval_mapper.dart';
+import '../../item_descriptor.dart';
 import 'payload.dart';
-import 'selector.dart';
 import 'source.dart';
 import 'source_result.dart';
 
@@ -67,20 +66,15 @@ final class CacheInitializer implements Initializer {
       return _miss();
     }
 
-    final updates = <Update>[];
+    final updates = <String, ItemDescriptor>{};
     cached.flags.forEach((key, evalResult) {
-      updates.add(Update(
-        kind: flagEvalKind,
-        key: key,
-        version: evalResult.version,
-        object: LDEvaluationResultSerialization.toJson(evalResult),
-      ));
+      updates[key] =
+          ItemDescriptor(version: evalResult.version, flag: evalResult);
     });
 
     return ChangeSetResult(
-      payload: Payload(
+      changeSet: ChangeSet(
         type: PayloadType.full,
-        selector: Selector.empty,
         updates: updates,
       ),
       environmentId: cached.environmentId,
@@ -95,9 +89,9 @@ final class CacheInitializer implements Initializer {
   }
 
   ChangeSetResult _miss() => ChangeSetResult(
-        payload: const Payload(
+        changeSet: const ChangeSet(
           type: PayloadType.none,
-          updates: [],
+          updates: {},
         ),
         freshness: _now(),
         persist: false,

packages/common_client/lib/src/data_sources/fdv2/endpoints.dart

@@ -1,3 +1,7 @@
+import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart';
+
+import 'selector.dart';
+
 /// FDv2 endpoint paths.
 ///
 /// These paths are uniform across mobile and browser SDKs; FDv2 does
@@ -20,3 +24,39 @@ abstract final class FDv2Endpoints {
   static String streamingGet(String encodedContext) =>
       '$streaming/$encodedContext';
 }
+
+/// Builds an FDv2 request URI: appends [addedPath] to [baseUri]'s path and
+/// merges the [withReasons], [basis], and [additionalQueryParameters]
+/// query parameters onto the base URL's own.
+///
+/// Composes against the parsed [baseUri] so a custom URL carrying its own
+/// query parameters (e.g. a relay proxy with a token) is preserved --
+/// including repeated keys, via `queryParametersAll`, which a plain
+/// `queryParameters` map would collapse to the last value. String
+/// concatenation against the base would instead land the appended path
+/// inside the query component.
+///
+/// Shared by the polling requestor and the streaming source so the two
+/// transports build URLs identically.
+Uri buildFDv2Uri({
+  required Uri baseUri,
+  required String addedPath,
+  required bool withReasons,
+  required Selector basis,
+  Map<String, String> additionalQueryParameters = const {},
+}) {
+  final mergedPath = appendPath(baseUri.path, addedPath);
+  final mergedQuery = <String, dynamic>{}
+    ..addAll(baseUri.queryParametersAll)
+    ..addAll(additionalQueryParameters);
+  if (withReasons) {
+    mergedQuery['withReasons'] = 'true';
+  }
+  if (basis.state case final state? when state.isNotEmpty) {
+    mergedQuery['basis'] = state;
+  }
+  return baseUri.replace(
+    path: mergedPath,
+    queryParameters: mergedQuery.isEmpty ? null : mergedQuery,
+  );
+}

packages/common_client/lib/src/data_sources/fdv2/entry_factories.dart

@@ -2,9 +2,14 @@ import 'dart:convert';
 
 import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart'
     hide ServiceEndpoints;
+import 'package:launchdarkly_event_source_client/launchdarkly_event_source_client.dart';
 
+import '../../config/defaults/default_config.dart';
 import '../../config/service_endpoints.dart';
+import '../streaming_data_source.dart' show LDLoggerToEventSourceAdapter;
 import 'cache_initializer.dart' as cache_src;
+import 'endpoints.dart';
+import 'protocol_types.dart';
 import 'source_factory_context.dart';
 import 'mode_definition.dart' as mode;
 import 'polling_base.dart';
@@ -13,6 +18,8 @@ import 'polling_synchronizer.dart';
 import 'requestor.dart';
 import 'selector.dart';
 import 'source.dart';
+import 'streaming_base.dart';
+import 'streaming_synchronizer.dart';
 
 /// Merges per-entry [mode.EndpointConfig] overrides into [base].
 ServiceEndpoints mergeServiceEndpoints(
@@ -50,6 +57,7 @@ FDv2PollingBase _buildPollingBase({
     contextJson: ctx.contextJson,
     usePost: usePost,
     withReasons: ctx.withReasons,
+    additionalQueryParameters: ctx.additionalQueryParameters,
     httpProperties: ctx.httpProperties,
     httpClientFactory: ctx.httpClientFactory ?? _defaultHttpClientFactory,
   );
@@ -63,6 +71,69 @@ HttpClient _defaultHttpClientFactory(HttpProperties httpProperties) {
   return HttpClient(httpProperties: httpProperties);
 }
 
+/// Constructs the [SSEClient] used by a streaming source. [uriProvider]
+/// is re-invoked on every connection attempt so the `basis` query
+/// parameter reflects the current selector. Tests inject a fake.
+typedef FDv2SseClientFactory = SSEClient Function({
+  required Uri Function() uriProvider,
+  required HttpProperties httpProperties,
+  required String? body,
+  required SseHttpMethod method,
+  required EventSourceLogger logger,
+});
+
+/// FDv2 event names subscribed on the streaming connection. Includes the
+/// legacy `ping` bridge event.
+const Set<String> _fdv2StreamEventNames = {
+  FDv2EventTypes.serverIntent,
+  FDv2EventTypes.putObject,
+  FDv2EventTypes.deleteObject,
+  FDv2EventTypes.payloadTransferred,
+  FDv2EventTypes.goodbye,
+  FDv2EventTypes.error,
+  FDv2EventTypes.heartbeat,
+  'ping',
+};
+
+/// Constructs the production [SSEClient]. The default for the factory
+/// builders; tests inject a fake through the same parameter.
+SSEClient defaultSseClientFactory({
+  required Uri Function() uriProvider,
+  required HttpProperties httpProperties,
+  required String? body,
+  required SseHttpMethod method,
+  required EventSourceLogger logger,
+}) {
+  return SSEClient(uriProvider(), _fdv2StreamEventNames,
+      headers: httpProperties.baseHeaders,
+      body: body,
+      httpMethod: method,
+      logger: logger,
+      uriProvider: uriProvider);
+}
+
+/// Builds the streaming URI for the current state. Invoked per
+/// connection attempt so the `basis` parameter tracks the selector.
+Uri _buildStreamingUri({
+  required ServiceEndpoints endpoints,
+  required String contextEncoded,
+  required bool usePost,
+  required bool withReasons,
+  required Selector basis,
+  Map<String, String> additionalQueryParameters = const {},
+}) {
+  final addedPath = usePost
+      ? FDv2Endpoints.streaming
+      : FDv2Endpoints.streamingGet(contextEncoded);
+  return buildFDv2Uri(
+    baseUri: Uri.parse(endpoints.streaming),
+    addedPath: addedPath,
+    withReasons: withReasons,
+    basis: basis,
+    additionalQueryParameters: additionalQueryParameters,
+  );
+}
+
 /// A factory for creating [Initializer] instances.
 final class InitializerFactory {
   /// True for cache initializers.
@@ -130,12 +201,11 @@ InitializerFactory createInitializerFactoryFromEntry(
 }
 
 /// Builds a [SynchronizerFactory] for a single [mode.SynchronizerEntry].
-///
-/// Throws [UnsupportedError] for unsupported entry types.
 SynchronizerFactory createSynchronizerFactoryFromEntry(
   mode.SynchronizerEntry entry,
-  SourceFactoryContext ctx,
-) {
+  SourceFactoryContext ctx, {
+  FDv2SseClientFactory sseClientFactory = defaultSseClientFactory,
+}) {
   switch (entry) {
     case final mode.PollingSynchronizer e:
       final interval = e.pollInterval ?? ctx.defaultPollingInterval;
@@ -155,9 +225,47 @@ SynchronizerFactory createSynchronizerFactoryFromEntry(
           );
         },
       );
-    case mode.StreamingSynchronizer():
-      throw UnsupportedError(
-        'FDv2 StreamingSynchronizer factories are not implemented yet',
+    case final mode.StreamingSynchronizer e:
+      return SynchronizerFactory(
+        create: (SelectorGetter selectorGetter) {
+          final endpointsResolved =
+              mergeServiceEndpoints(ctx.serviceEndpoints, e.endpoints);
+          Uri uriProvider() => _buildStreamingUri(
+                endpoints: endpointsResolved,
+                contextEncoded: base64UrlEncode(utf8.encode(ctx.contextJson)),
+                usePost: e.usePost,
+                withReasons: ctx.withReasons,
+                basis: selectorGetter(),
+                additionalQueryParameters: ctx.additionalQueryParameters,
+              );
+          final sseClient = sseClientFactory(
+            uriProvider: uriProvider,
+            httpProperties: ctx.httpProperties,
+            body: e.usePost ? ctx.contextJson : null,
+            method: e.usePost ? SseHttpMethod.post : SseHttpMethod.get,
+            logger: LDLoggerToEventSourceAdapter(ctx.logger),
+          );
+
+          // Legacy ping events trigger a one-shot poll.
+          final pingPollingBase = _buildPollingBase(
+            endpoints: e.endpoints,
+            usePost: e.usePost,
+            ctx: ctx,
+          );
+
+          return FDv2StreamingSynchronizer(
+            base: FDv2StreamingBase(
+              sseClient: sseClient,
+              pingHandler: () =>
+                  pingPollingBase.pollOnce(basis: selectorGetter()),
+              logger: ctx.logger,
+              // Used when the transport exposes no response headers (the
+              // browser EventSource) to read x-ld-envid from.
+              defaultEnvironmentId: DefaultConfig.credentialConfig
+                  .environmentIdFallback(ctx.credential),
+            ),
+          );
+        },
       );
   }
 }
@@ -173,9 +281,11 @@ List<InitializerFactory> buildInitializerFactories(
 /// One factory per entry, in list order.
 List<SynchronizerFactory> buildSynchronizerFactories(
   List<mode.SynchronizerEntry> entries,
-  SourceFactoryContext ctx,
-) {
+  SourceFactoryContext ctx, {
+  FDv2SseClientFactory sseClientFactory = defaultSseClientFactory,
+}) {
   return entries
-      .map((e) => createSynchronizerFactoryFromEntry(e, ctx))
+      .map((e) => createSynchronizerFactoryFromEntry(e, ctx,
+          sseClientFactory: sseClientFactory))
       .toList();
 }

packages/common_client/lib/src/data_sources/fdv2/flag_eval_mapper.dart

@@ -6,6 +6,19 @@ import 'payload.dart';
 /// The object kind for client-side flag evaluation results.
 const String flagEvalKind = 'flag-eval';
 
+/// Translates a wire-level [Payload] into a typed [ChangeSet] ready for
+/// the flag store.
+///
+/// Throws if any flag-eval object cannot be parsed, so the data source
+/// layer can report it as a data source error and recover the connection.
+ChangeSet translatePayload(Payload payload) {
+  return ChangeSet(
+    selector: payload.selector,
+    type: payload.type,
+    updates: mapUpdatesToItemDescriptors(payload.updates),
+  );
+}
+
 /// Converts FDv2 [Update] objects to a map of [ItemDescriptor]s suitable
 /// for the flag store.
 ///

packages/common_client/lib/src/data_sources/fdv2/payload.dart

@@ -1,5 +1,6 @@
 import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart';
 
+import '../../item_descriptor.dart';
 import 'selector.dart';
 
 /// The type of payload transfer.
@@ -95,3 +96,35 @@ final class Payload {
   String toString() => 'Payload(selector: $selector, type: $type, '
       'updates: ${updates.length})';
 }
+
+/// A [Payload] translated into typed flag descriptors, ready to apply to
+/// the flag store.
+///
+/// The wire-level [Payload] carries raw [Update] objects; producing a
+/// [ChangeSet] converts each flag-eval object into an [ItemDescriptor].
+/// The data source layer performs this conversion (see `translatePayload`),
+/// so a malformed object is reported as a data source error where the
+/// connection can recover.
+final class ChangeSet {
+  /// The selector for this change set, carried over from the originating
+  /// [Payload]. [Selector.empty] when the source provided none (cached
+  /// data or an intent of none).
+  final Selector selector;
+
+  /// Whether this is a full, partial, or no-op change set.
+  final PayloadType type;
+
+  /// The typed updates, keyed by flag key. A tombstone (deletion) is an
+  /// [ItemDescriptor] with a null flag.
+  final Map<String, ItemDescriptor> updates;
+
+  const ChangeSet({
+    this.selector = Selector.empty,
+    required this.type,
+    required this.updates,
+  });
+
+  @override
+  String toString() => 'ChangeSet(selector: $selector, type: $type, '
+      'updates: ${updates.length})';
+}