chore: Add initial FDv2 streaming data source. (#185)

Adds an FDv2 Streaming Data Source which handles an FDv2 stream, but
produces FDv1 store updates.

Later the event source will be updated to produce output in FDv2 format
for the transactional store.


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> Adds an FDv2 SSE data source that translates FDv2 changes into FDv1
store updates, introduces JsonElement-based deserialization paths, and
includes supporting utilities and tests.
> 
> - **Data sources**:
> - **FDv2 streaming**: New
`Internal/FDv2DataSources/FDv2StreamingDataSource` handling SSE events,
selector-based query params, header-aware init, error/status handling,
and upserts via FDv1 APIs.
> - **FDv2→FDv1 translation**: New `FDv2ChangeSetTranslator` converts
FDv2 change sets into `PutData`/`PatchData` for existing store
interfaces.
> - **Endpoints/Logging**: Add
`StandardEndpoints.FDv2StreamingRequestPath` and
`FDv2PollingRequestPath`; add `LogNames.FDv2DataSourceSubLog`.
> - **Model/serialization**:
> - Extend `DataStoreTypes.DataKind` with `DeserializeFromJsonElement`;
update `DataModel.Features`/`Segments` to wire new deserializers.
> - In FDv2 payloads, change `PutObject.Object` to `JsonElement`; add
`FDv2Event.TryDeserializeFromJsonString`.
> - **Utilities**:
> - New `Internal/FDv2DataSources/QueryStringHelper` for
parsing/building query strings.
> - **Tests**:
> - Add comprehensive tests for `FDv2StreamingDataSource`,
`QueryStringHelper`, and FDv2 payload/protocol behavior; update protocol
tests for `JsonElement` objects.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
59b4d95. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: kinyoklion

pkgs/sdk/server/src/DataModel.cs

@@ -1,7 +1,6 @@
 using System.Collections.Generic;
 using System.Text.Json;
 using LaunchDarkly.Sdk.Server.Internal.Model;
-
 using static LaunchDarkly.Sdk.Server.Subsystems.DataStoreTypes;
 
 namespace LaunchDarkly.Sdk.Server
@@ -25,7 +24,8 @@ public static class DataModel
         /// Applications should not need to reference this object directly. It is public so that custom integrations
         /// and test code can serialize or deserialize data or inject it into a data store.
         /// </remarks>
-        public static DataKind Features = new DataKind("features", SerializeFlag, DeserializeFlag);
+        public static DataKind Features =
+            new DataKind("features", SerializeFlag, DeserializeFlag, DeserializeFlagFromJsonElement);
 
         /// <summary>
         /// The <see cref="DataKind"/> instance that describes user segment data.
@@ -34,7 +34,8 @@ public static class DataModel
         /// Applications should not need to reference this object directly. It is public so that custom integrations
         /// and test code can serialize or deserialize data or inject it into a data store.
         /// </remarks>
-        public static DataKind Segments = new DataKind("segments", SerializeSegment, DeserializeSegment);
+        public static DataKind Segments = new DataKind("segments", SerializeSegment, DeserializeSegment,
+            DeserializeSegmentFromJsonElement);
 
         /// <summary>
         /// An enumeration of all supported <see cref="DataKind"/>s.
@@ -59,8 +60,13 @@ private static void SerializeFlag(object o, Utf8JsonWriter w) =>
         private static ItemDescriptor DeserializeFlag(ref Utf8JsonReader r)
         {
             var flag = FeatureFlagSerialization.Instance.Read(ref r, null, null) as FeatureFlag;
-            return flag.Deleted ? ItemDescriptor.Deleted(flag.Version) :
-                new ItemDescriptor(flag.Version, flag);
+            return flag.Deleted ? ItemDescriptor.Deleted(flag.Version) : new ItemDescriptor(flag.Version, flag);
+        }
+
+        private static ItemDescriptor DeserializeFlagFromJsonElement(JsonElement element)
+        {
+            var flag = element.Deserialize<FeatureFlag>();
+            return flag.Deleted ? ItemDescriptor.Deleted(flag.Version) : new ItemDescriptor(flag.Version, flag);
         }
 
         private static void SerializeSegment(object o, Utf8JsonWriter w) =>
@@ -69,8 +75,15 @@ private static void SerializeSegment(object o, Utf8JsonWriter w) =>
         private static ItemDescriptor DeserializeSegment(ref Utf8JsonReader r)
         {
             var segment = SegmentSerialization.Instance.Read(ref r, null, null) as Segment;
-            return segment.Deleted ? ItemDescriptor.Deleted(segment.Version) :
-                new ItemDescriptor(segment.Version, segment);
+            return segment.Deleted
+                ? ItemDescriptor.Deleted(segment.Version)
+                : new ItemDescriptor(segment.Version, segment);
+        }
+
+        private static ItemDescriptor DeserializeSegmentFromJsonElement(JsonElement element)
+        {
+            var flag = element.Deserialize<Segment>();
+            return flag.Deleted ? ItemDescriptor.Deleted(flag.Version) : new ItemDescriptor(flag.Version, flag);
         }
     }
 }

pkgs/sdk/server/src/Internal/DataSources/StreamingDataSource.cs

@@ -41,6 +41,12 @@ internal class StreamingDataSource : IDataSource
         private readonly Logger _log;
 
         private readonly IEventSource _es;
+        /// <summary>
+        /// When the store enters a failed state, and we don't have "data source monitoring", we want to log
+        /// a message that we are restarting the event source. We don't want to log this message on multiple
+        /// sequential failures. This boolean is used to determine if the previous attempt to write also
+        /// failed, and in which case we will not log.
+        /// </summary>
         private volatile bool _lastStoreUpdateFailed = false;
         internal DateTime _esStarted; // exposed for testing
 

pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ChangeSet.cs

@@ -72,7 +72,7 @@ internal sealed class FDv2Change
         /// <summary>
         /// The raw JSON string representing the object data (only present for Put operations).
         /// </summary>
-        public string Object { get; }
+        public JsonElement? Object { get; }
 
         /// <summary>
         /// Constructs a new Change.
@@ -83,7 +83,7 @@ internal sealed class FDv2Change
         /// <param name="version">The version of the change.</param>
         /// <param name="obj">The raw JSON string representing the object data (required for Put operations).</param>
         /// <exception cref="ArgumentNullException">Thrown when <paramref name="kind"/> or <paramref name="key"/> is null.</exception>
-        public FDv2Change(FDv2ChangeType type, string kind, string key, int version, string obj = null)
+        public FDv2Change(FDv2ChangeType type, string kind, string key, int version, JsonElement? obj = null)
         {
             Type = type;
             Kind = kind ?? throw new ArgumentNullException(nameof(kind));

pkgs/sdk/server/src/Internal/FDv2DataSources/FDv2ChangeSetTranslator.cs

@@ -0,0 +1,126 @@
+using System.Collections.Generic;
+using System.Linq;
+using LaunchDarkly.Logging;
+using LaunchDarkly.Sdk.Server.Internal.DataSources;
+using LaunchDarkly.Sdk.Server.Subsystems;
+
+namespace LaunchDarkly.Sdk.Server.Internal.FDv2DataSources
+{
+    /// <summary>
+    /// Translates FDv2 changesets into FDv1 format.
+    /// </summary>
+    internal static class FDv2ChangeSetTranslator
+    {
+        /// <summary>
+        /// Translates an FDv2 changeset with Full or None type into PutData.
+        /// </summary>
+        /// <param name="changeset">The changeset to translate.</param>
+        /// <param name="log">Logger for diagnostic messages.</param>
+        /// <returns>PutData containing the full dataset organized by data kind.</returns>
+        public static StreamProcessorEvents.PutData TranslatePutData(FDv2ChangeSet changeset, Logger log)
+        {
+            var dataBuilder = System.Collections.Immutable.ImmutableList
+                .CreateBuilder<KeyValuePair<DataStoreTypes.DataKind,
+                    DataStoreTypes.KeyedItems<DataStoreTypes.ItemDescriptor>>>();
+
+            var changesByKind = changeset.Changes.GroupBy(c => c.Kind);
+
+            foreach (var kindGroup in changesByKind)
+            {
+                var kind = kindGroup.Key;
+                var dataKind = GetDataKind(kind);
+
+                if (dataKind == null)
+                {
+                    log.Warn($"Unknown data kind '{kind}' in changeset, skipping");
+                    continue;
+                }
+
+                var itemsBuilder = System.Collections.Immutable.ImmutableList
+                    .CreateBuilder<KeyValuePair<string, DataStoreTypes.ItemDescriptor>>();
+
+                foreach (var change in kindGroup)
+                {
+                    if (change.Type != FDv2ChangeType.Put || !change.Object.HasValue) continue;
+                    var item = dataKind.DeserializeFromJsonElement(change.Object.Value);
+                    itemsBuilder.Add(new KeyValuePair<string, DataStoreTypes.ItemDescriptor>(change.Key, item));
+                    // Note: Delete operations in a Full changeset would be unusual, but we skip them
+                    // since a full transfer should only contain items that exist
+                }
+
+                dataBuilder.Add(
+                    new KeyValuePair<DataStoreTypes.DataKind, DataStoreTypes.KeyedItems<DataStoreTypes.ItemDescriptor>>(
+                        dataKind,
+                        new DataStoreTypes.KeyedItems<DataStoreTypes.ItemDescriptor>(itemsBuilder.ToImmutable())
+                    ));
+            }
+
+            return new StreamProcessorEvents.PutData("/",
+                new DataStoreTypes.FullDataSet<DataStoreTypes.ItemDescriptor>(dataBuilder.ToImmutable()));
+        }
+
+        /// <summary>
+        /// Translates an FDv2 changeset with Partial type into a list of PatchData.
+        /// </summary>
+        /// <param name="changeset">The changeset to translate.</param>
+        /// <param name="log">Logger for diagnostic messages.</param>
+        /// <returns>List of PatchData representing individual upserts or deletes.</returns>
+        public static List<StreamProcessorEvents.PatchData> TranslatePatchData(FDv2ChangeSet changeset, Logger log)
+        {
+            var patches = new List<StreamProcessorEvents.PatchData>();
+
+            foreach (var change in changeset.Changes)
+            {
+                var dataKind = GetDataKind(change.Kind);
+
+                if (dataKind == null)
+                {
+                    log.Warn($"Unknown data kind '{change.Kind}' in change, skipping");
+                    continue;
+                }
+
+                DataStoreTypes.ItemDescriptor item;
+
+                switch (change.Type)
+                {
+                    case FDv2ChangeType.Put when !change.Object.HasValue:
+                        log.Warn($"Put operation for {change.Kind}/{change.Key} missing object data, skipping");
+                        continue;
+                    // Deserialize the object using the DataKind's deserializer
+                    case FDv2ChangeType.Put:
+                        item = dataKind.DeserializeFromJsonElement(change.Object.Value);
+                        break;
+                    case FDv2ChangeType.Delete:
+                        // For deletes, create a deleted ItemDescriptor with the version
+                        item = DataStoreTypes.ItemDescriptor.Deleted(change.Version);
+                        break;
+                    default:
+                        log.Warn($"Unknown change type for {change.Kind}/{change.Key}, skipping");
+                        continue;
+                }
+
+                patches.Add(new StreamProcessorEvents.PatchData(dataKind, change.Key, item));
+            }
+
+            return patches;
+        }
+
+        /// <summary>
+        /// Maps an FDv2 object kind to the corresponding DataKind.
+        /// </summary>
+        /// <param name="kind">The kind string from the FDv2 change.</param>
+        /// <returns>The corresponding DataKind, or null if the kind is not recognized.</returns>
+        private static DataStoreTypes.DataKind GetDataKind(string kind)
+        {
+            switch (kind)
+            {
+                case "flag":
+                    return DataModel.Features;
+                case "segment":
+                    return DataModel.Segments;
+                default:
+                    return null;
+            }
+        }
+    }
+}