From 45bf766ef87514f9973ad253972c10b0b8572a11 Mon Sep 17 00:00:00 2001 From: Aden <128471249+adenchen123@users.noreply.github.com> Date: Thu, 28 Nov 2024 18:31:20 +0800 Subject: [PATCH 1/4] Update ConversationService.cs --- .../Conversations/Services/ConversationService.cs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.cs b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.cs index f6326a7b..55d80c18 100644 --- a/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.cs +++ b/src/Infrastructure/BotSharp.Core/Conversations/Services/ConversationService.cs @@ -173,12 +173,14 @@ public partial class ConversationService : IConversationService var state = _services.GetRequiredService(); 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); } From da90856638f0e791247355669a271b1f25155494 Mon Sep 17 00:00:00 2001 From: AnonymousDotNet <18776095145@163.com> Date: Fri, 29 Nov 2024 17:08:27 +0800 Subject: [PATCH 2/4] fix: Add regionCode verification --- .../BotSharp.Core/Users/Services/UserService.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs b/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs index 0cddc5b7..9bc436a7 100644 --- a/src/Infrastructure/BotSharp.Core/Users/Services/UserService.cs +++ b/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 RemoveDashboardConversation(string userId, string conversationId) { var db = _services.GetRequiredService(); From 039054ad9ea5b18bafd7c49a67ae91e23f748803 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Fri, 29 Nov 2024 17:12:44 -0600 Subject: [PATCH 3/4] allow CancellationToken --- .../Infrastructures/Events/IEventSubscriber.cs | 6 +++++- .../Infrastructures/Events/RedisSubscriber.cs | 10 +++++++++- 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs index fa96bb35..e04c2488 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventSubscriber.cs +++ b/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 received); - Task SubscribeAsync(string channel, string group, bool priorityEnabled, Func received); + Task SubscribeAsync(string channel, string group, bool priorityEnabled, + Func received, + CancellationToken? stoppingToken = null); } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs index 89b38a8d..34ff0079 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisSubscriber.cs +++ b/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 received) + public async Task SubscribeAsync(string channel, string group, bool priorityEnabled, + Func 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) From 5f31ffe6587e3321238ee346146b938feff5dff2 Mon Sep 17 00:00:00 2001 From: Haiping Chen Date: Sat, 30 Nov 2024 15:16:04 +0000 Subject: [PATCH 4/4] Allow Event to be removed. --- .../Infrastructures/Events/IEventPublisher.cs | 2 ++ .../Infrastructures/Events/RedisPublisher.cs | 22 +++++++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs b/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs index 31a708bc..58b18cba 100644 --- a/src/Infrastructure/BotSharp.Abstraction/Infrastructures/Events/IEventPublisher.cs +++ b/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); } diff --git a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs b/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs index 32889f4f..89de2453 100644 --- a/src/Infrastructure/BotSharp.Core/Infrastructures/Events/RedisPublisher.cs +++ b/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}"); + } + } + } }