Add Error channel

Author: Oceania2018

src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs

@@ -79,7 +79,7 @@ private bool CheckMessageExists(IDatabase db, string channel, string fieldName,
         return exists;
     }
 
-    private NameValueEntry[] AssembleMessage(RedisValue message)
+    public static NameValueEntry[] AssembleMessage(RedisValue message)
     {
         return
         [
@@ -88,6 +88,16 @@ private NameValueEntry[] AssembleMessage(RedisValue message)
         ];
     }
 
+    public static NameValueEntry[] AssembleErrorMessage(RedisValue message, string error)
+    {
+        return
+        [
+            new NameValueEntry("message", message),
+            new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")),
+            new NameValueEntry("error", error)
+        ];
+    }
+
     public async Task ReDispatchAsync(string channel, int count = 10, string order = "asc")
     {
         var db = _redis.GetDatabase();

src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs

@@ -42,6 +42,8 @@ public async Task SubscribeAsync(string channel, string group, int? port, bool p
             await CreateConsumerGroup(db, channel, group);
         }
 
+        await CreateConsumerGroup(db, $"{channel}-Error", group);
+
         var consumer = Environment.MachineName;
         if (port.HasValue)
         {
@@ -98,16 +100,24 @@ private async Task<int> HandleGroupMessage(IDatabase db, string channel, string
             try
             {
                 await received(channel, entry.Values[0].Value);
-
-                // Optionally delete the message to save space
-                await db.StreamDeleteAsync(channel, [entry.Id]);
             }
             catch (Exception ex)
             {
-                _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}");
+                _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id} {entry.Values[0].Value}");
+
+                // Add a message to the Error stream, keeping only the latest 1 million messages
+                await db.StreamAddAsync($"{channel}-Error",
+                    RedisPublisher.AssembleErrorMessage(entry.Values[0].Value, ex.Message),
+                    messageId: entry.Id,
+                    maxLength: 1000 * 10000);
+
                 // Slow down the consumer if there are errors
                 await Task.Delay(1000 * 10);
             }
+            finally
+            {
+                await db.StreamDeleteAsync(channel, [entry.Id]);
+            }
         }
 
         return entries.Length;