Merge remote-tracking branch 'origin/master' into jason_dev

src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs

@@ -15,4 +15,6 @@ public interface IEventPublisher
     Task ReDispatchAsync(string channel, int count = 10, string order = "asc");
 
     Task ReDispatchPendingAsync(string channel, string group, int count = 10);
+
+    Task RemoveAsync(string channel, int count = 10);
 }

src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs

@@ -1,8 +1,12 @@
+using System.Threading;
+
 namespace BotSharp.Abstraction.Infrastructures.Events;
 
 public interface IEventSubscriber
 {
     Task SubscribeAsync(string channel, Func<string, string, Task> received);
 
-    Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func<string, string, Task> received);
+    Task SubscribeAsync(string channel, string group, bool priorityEnabled, 
+        Func<string, string, Task> received,
+        CancellationToken? stoppingToken = null);
 }

src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.cs

@@ -173,12 +173,14 @@ public partial class ConversationService : IConversationService
             var state = _services.GetRequiredService<IConversationStateService>();
             var channel = state.GetState("channel");
             var channelId = state.GetState("channel_id");
+            var userId = state.GetState("current_user_id");
             var sess = new Conversation
             {
                 Id = _conversationId,
                 Channel = channel,
                 ChannelId = channelId,
-                AgentId = agentId
+                AgentId = agentId,
+                UserId = userId,
             };
             converation = await NewConversation(sess);
         }

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

@@ -158,4 +158,26 @@ public class RedisPublisher : IEventPublisher
             Console.WriteLine($"Redis error: {ex.Message}");
         }
     }
+
+    public async Task RemoveAsync(string channel, int count = 10)
+    {
+        var db = _redis.GetDatabase();
+
+        var entries = await db.StreamRangeAsync(channel, "-", "+", count: count, messageOrder: Order.Ascending);
+        foreach (var entry in entries)
+        {
+            _logger.LogInformation($"Fetched message: {channel} {entry.Values[0].Value} ({entry.Id})");
+
+            try
+            {
+                await db.StreamDeleteAsync(channel, [entry.Id]);
+
+                _logger.LogWarning($"Deleted message: {channel} {entry.Values[0].Value} ({entry.Id})");
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError($"Error processing message: {ex.Message}, event id: {channel} {entry.Id}\r\n{ex}");
+            }
+        }
+    }
 }

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

@@ -25,7 +25,9 @@ public class RedisSubscriber : IEventSubscriber
         });
     }
 
-    public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func<string, string, Task> received)
+    public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, 
+        Func<string, string, Task> received, 
+        CancellationToken? stoppingToken = null)
     {
         var db = _redis.GetDatabase();
 
@@ -44,6 +46,12 @@ public class RedisSubscriber : IEventSubscriber
         {
             await Task.Delay(100);
 
+            if (stoppingToken.HasValue && stoppingToken.Value.IsCancellationRequested)
+            {
+                _logger.LogInformation($"Stopping consumer channel & group: [{channel}, {group}]");
+                break;
+            }
+
             if (priorityEnabled)
             {
                 if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, received) > 0)

src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs

@@ -570,7 +570,7 @@ public class UserService : IUserService
 
         if (!string.IsNullOrEmpty(user.Phone))
         {
-            record = db.GetUserByPhone(user.Phone);
+            record = db.GetUserByPhone(user.Phone, regionCode: user.RegionCode);
         }
 
         if (!string.IsNullOrEmpty(user.Email))
@@ -745,7 +745,7 @@ public class UserService : IUserService
         await Task.CompletedTask;
         return true;
     }
-    
+
     public async Task<bool> RemoveDashboardConversation(string userId, string conversationId)
     {
         var db = _services.GetRequiredService<IBotSharpRepository>();