feat: data change notify error handling (#11)

Author: deleteLater

src/Api/DataSynchronizer/DataSyncMessageHandler.cs

@@ -1,14 +1,14 @@
-using System.Text;
 using System.Text.Json;
-using System.Text.Json.Nodes;
 using Api.Messaging;
 using Api.Store;
-using Domain.Messages;
 using Streaming.Protocol;
 
 namespace Api.DataSynchronizer;
 
-public class DataSyncMessageHandler(IAgentStore agentStore, IDataChangeNotifier dataChangeNotifier)
+public class DataSyncMessageHandler(
+    IAgentStore agentStore,
+    IDataChangeNotifier dataChangeNotifier,
+    ILogger<DataSyncMessageHandler> logger)
     : IDataSyncMessageHandler
 {
     public ValueTask<byte[]> GetRequestPayloadAsync() => agentStore.GetDataSyncPayloadAsync();
@@ -31,66 +31,36 @@ public async Task HandleAsync(JsonElement message)
                     .ToArray();
                 await agentStore.UpdateAsync(items);
 
-                await OnItemsUpdated(items);
+                await SafeNotifyAsync(items);
                 break;
             }
             case DataSyncEventTypes.Patch:
             {
                 var patchDataSet = PatchDataSet.FromJson(data);
                 await agentStore.UpdateAsync(patchDataSet.Items);
 
-                await OnItemsUpdated(patchDataSet.Items);
+                await SafeNotifyAsync(patchDataSet.Items);
                 break;
             }
         }
     }
 
-    private async Task OnItemsUpdated(StoreItem[] items)
+    private async Task SafeNotifyAsync(StoreItem[] items)
     {
-        List<DataChangeMessage> dataChanges = [];
-
         foreach (var item in items)
         {
-            await AddDataChange(item);
-        }
-
-        await dataChangeNotifier.NotifyAsync(dataChanges.ToArray());
-
-        return;
-
-        async Task AddDataChange(StoreItem item)
-        {
-            if (item.Type == StoreItemType.Flag)
+            try
             {
-                var flagChange = new DataChangeMessage(
-                    Topics.FeatureFlagChange,
-                    item.Id,
-                    Encoding.UTF8.GetString(item.JsonBytes)
-                );
-
-                dataChanges.Add(flagChange);
+                await dataChangeNotifier.NotifyAsync(item);
             }
-
-            if (item.Type == StoreItemType.Segment)
+            catch (Exception ex)
             {
-                using var segment = JsonDocument.Parse(item.JsonBytes);
-
-                var envId = segment.RootElement.GetProperty("envId").GetGuid();
-                var affectedIds = await agentStore.GetFlagReferencesAsync(envId, item.Id);
-
-                JsonObject payload = new()
-                {
-                    ["segment"] = JsonSerializer.SerializeToNode(segment),
-                    ["affectedFlagIds"] = JsonSerializer.SerializeToNode(affectedIds)
-                };
-
-                var segmentChange = new DataChangeMessage(
-                    Topics.SegmentChange,
+                logger.LogError(
+                    ex,
+                    "Exception occurred while processing StoreItem Id {Id} of type {Type} for data change notification.",
                     item.Id,
-                    JsonSerializer.Serialize(payload)
+                    item.Type
                 );
-
-                dataChanges.Add(segmentChange);
             }
         }
     }

src/Api/Messaging/DataChangeNotifier.cs

@@ -1,3 +1,7 @@
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Nodes;
+using Api.Store;
 using Domain.Messages;
 using Streaming.Connections;
 using Streaming.Protocol;
@@ -9,17 +13,20 @@ public class DataChangeNotifier : IDataChangeNotifier
 {
     private readonly IConnectionManager _connectionManager;
     private readonly IDataSyncService _dataSyncService;
+    private readonly IAgentStore _agentStore;
     private readonly Dictionary<string, IMessageConsumer> _dataChangeHandlers;
     private readonly ILogger<DataChangeNotifier> _logger;
 
     public DataChangeNotifier(
         IConnectionManager connectionManager,
         IDataSyncService dataSyncService,
+        IAgentStore agentStore,
         IEnumerable<IMessageConsumer> messageHandlers,
         ILogger<DataChangeNotifier> logger)
     {
         _connectionManager = connectionManager;
         _dataSyncService = dataSyncService;
+        _agentStore = agentStore;
         _dataChangeHandlers = messageHandlers.ToDictionary(x => x.Topic, x => x);
         _logger = logger;
     }
@@ -60,27 +67,86 @@ async Task NotifyConnectionAsync(Connection connection)
         }
     }
 
-    public async Task NotifyAsync(DataChangeMessage[] dataChanges)
+    public async Task NotifyAsync(DataChangeMessage dataChange)
     {
-        if (dataChanges.Length == 0)
+        if (!_dataChangeHandlers.TryGetValue(dataChange.Topic, out var handler))
         {
+            _logger.LogWarning("No data change handler found for topic {Topic}.", dataChange.Topic);
             return;
         }
 
-        foreach (var dataChange in dataChanges)
+        try
         {
-            if (!_dataChangeHandlers.TryGetValue(dataChange.Topic, out var handler))
-            {
-                _logger.LogWarning("No data change handler found for topic {Topic}.", dataChange.Topic);
-                continue;
-            }
-
             await handler.HandleAsync(dataChange.Message, CancellationToken.None);
 
             _logger.LogInformation(
-                "Handled data change message for topic {Topic} (Item Id: {Id})",
+                "Notified data change for topic {Topic} (Item Id: {Id})",
                 dataChange.Topic, dataChange.Id
             );
         }
+        catch (Exception ex)
+        {
+            _logger.LogError(
+                ex,
+                "Exception occurred while notifying data change for topic {Topic} (Item Id: {Id})",
+                dataChange.Topic,
+                dataChange.Id
+            );
+        }
+    }
+
+    public async Task NotifyAsync(StoreItem item)
+    {
+        var dataChange = await ItemToDataChangeAsync();
+        if (dataChange != null)
+        {
+            await NotifyAsync(dataChange);
+        }
+
+        return;
+
+        async ValueTask<DataChangeMessage?> ItemToDataChangeAsync()
+        {
+            if (item.Type == StoreItemType.Flag)
+            {
+                var flagChange = new DataChangeMessage(
+                    Topics.FeatureFlagChange,
+                    item.Id,
+                    Encoding.UTF8.GetString(item.JsonBytes)
+                );
+
+                return flagChange;
+            }
+
+            if (item.Type == StoreItemType.Segment)
+            {
+                using var segment = JsonDocument.Parse(item.JsonBytes);
+
+                var envId = segment.RootElement.GetProperty("envId").GetGuid();
+                var affectedIds = await _agentStore.GetFlagReferencesAsync(envId, item.Id);
+
+                JsonObject payload = new()
+                {
+                    ["segment"] = JsonSerializer.SerializeToNode(segment),
+                    ["affectedFlagIds"] = JsonSerializer.SerializeToNode(affectedIds)
+                };
+
+                var segmentChange = new DataChangeMessage(
+                    Topics.SegmentChange,
+                    item.Id,
+                    JsonSerializer.Serialize(payload)
+                );
+
+                return segmentChange;
+            }
+
+            _logger.LogWarning(
+                "Unsupported StoreItem type {Type} for item Id {Id}, skipping data change notification for it.",
+                item.Type,
+                item.Id
+            );
+
+            return null;
+        }
     }
 }

src/Api/Messaging/IDataChangeNotifier.cs

@@ -1,8 +1,12 @@
+using Api.Store;
+
 namespace Api.Messaging;
 
 public interface IDataChangeNotifier
 {
     Task NotifyAsync(Guid envId);
 
-    Task NotifyAsync(DataChangeMessage[] dataChanges);
+    Task NotifyAsync(DataChangeMessage dataChange);
+
+    Task NotifyAsync(StoreItem item);
 }