From 7b14c8375d83d03a9541a7581998b26ca1e5d99f Mon Sep 17 00:00:00 2001
From: Haiping Chen <haiping008@gmail.com>
Date: Fri, 27 Dec 2024 18:16:23 +0000
Subject: [PATCH] Add Error channel

---
 .../Infrastructures/Events/RedisPublisher.cs   | 12 +++++++++++-
 .../Infrastructures/Events/RedisSubscriber.cs  | 18 ++++++++++++++----
 2 files changed, 25 insertions(+), 5 deletions(-)

diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs
index b6517535..514f4aa1 100644
--- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs
+++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs
@@ -79,7 +79,7 @@ public class RedisPublisher : IEventPublisher
         return exists;
     }
 
-    private NameValueEntry[] AssembleMessage(RedisValue message)
+    public static NameValueEntry[] AssembleMessage(RedisValue message)
     {
         return
         [
@@ -88,6 +88,16 @@ public class RedisPublisher : IEventPublisher
         ];
     }
 
+    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();
diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs
index d53c9400..b3d910d9 100644
--- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs
+++ b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs
@@ -42,6 +42,8 @@ public class RedisSubscriber : IEventSubscriber
             await CreateConsumerGroup(db, channel, group);
         }
 
+        await CreateConsumerGroup(db, $"{channel}-Error", group);
+
         var consumer = Environment.MachineName;
         if (port.HasValue)
         {
@@ -98,16 +100,24 @@ public class RedisSubscriber : IEventSubscriber
             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;