Merge branch 'master' into lida_Dev

src/Infrastructure/BotSharp.Abstraction/Users/IUserService.cs

@@ -17,6 +17,7 @@ public interface IUserService
     Task<Token?> GetAffiliateToken(string authorization);
     Task<Token?> GetAdminToken(string authorization);
     Task<Token?> GetToken(string authorization);
+    Task<Token> CreateTokenByUser(User user);
     Task<User> GetMyProfile();
     Task<bool> VerifyUserNameExisting(string userName);
     Task<bool> VerifyEmailExisting(string email);

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

@@ -39,10 +39,7 @@ public class RedisPublisher : IEventPublisher
 
         // Add a message to the stream, keeping only the latest 1 million messages
         var messageId = await db.StreamAddAsync(channel, 
-            [
-                new NameValueEntry("message", message),
-                new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o"))
-            ],
+            AssembleMessage(message),
             maxLength: 1000 * 10000);
 
         _logger.LogInformation($"Published message {channel} {message} ({messageId})");
@@ -82,6 +79,17 @@ public class RedisPublisher : IEventPublisher
         return exists;
     }
 
+    private NameValueEntry[] AssembleMessage(RedisValue message, int retry = 0)
+    {
+        return
+        [
+            new NameValueEntry("message", message),
+            new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o")),
+            new NameValueEntry("machine", Environment.MachineName),
+            new NameValueEntry("retry", retry),
+        ];
+    }
+
     public async Task ReDispatchAsync(string channel, int count = 10, string order = "asc")
     {
         var db = _redis.GetDatabase();
@@ -93,10 +101,12 @@ public class RedisPublisher : IEventPublisher
 
             try
             {
-                var messageId = await db.StreamAddAsync(channel, [
-                    new NameValueEntry("message", entry.Values[0].Value),
-                    new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o"))
-                ]);
+                var message = entry.Values.First(x => x.Name == "message").Value;
+                var retryKv = entry.Values.FirstOrDefault(x => x.Name == "retry");
+                int.TryParse(retryKv.Value, out int retry);
+                var messageId = await db.StreamAddAsync(channel, 
+                    AssembleMessage(message, retry: retry + 1),
+                    maxLength: 1000 * 10000);
 
                 _logger.LogWarning($"ReDispatched message: {channel} {entry.Values[0].Value} ({messageId})");
 

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

@@ -58,23 +58,31 @@ public class RedisSubscriber : IEventSubscriber
                 break;
             }
 
-            if (priorityEnabled)
+            try
             {
-                if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, consumer, received) > 0)
+                if (priorityEnabled)
                 {
-                    continue;
-                }
+                    if (await HandleGroupMessage(db, $"{channel}-{EventPriority.High}", group, consumer, received) > 0)
+                    {
+                        continue;
+                    }
 
-                if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received) > 0)
+                    if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received) > 0)
+                    {
+                        continue;
+                    }
+
+                    await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, consumer, received);
+                }
+                else
                 {
-                    continue;
+                    await HandleGroupMessage(db, channel, group, consumer, received);
                 }
-
-                await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, consumer, received);
             }
-            else
+            catch (Exception ex)
             {
-                await HandleGroupMessage(db, channel, group, consumer, received);
+                _logger.LogError($"Error processing message: {ex.Message}\r\n{ex}");
+                await Task.Delay(1000 * 60);
             }
         }
     }

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

@@ -506,6 +506,20 @@ public class UserService : IUserService
         return token;
     }
 
+    public async Task<Token> CreateTokenByUser(User user)
+    {
+        var accessToken = GenerateJwtToken(user);
+        var jwt = new JwtSecurityTokenHandler().ReadJwtToken(accessToken);
+        var token = new Token
+        {
+            AccessToken = accessToken,
+            ExpireTime = jwt.Payload.Exp.Value,
+            TokenType = "Bearer",
+            Scope = "api"
+        };
+        return token;
+    }
+
     public async Task<bool> VerifyUserNameExisting(string userName)
     {
         if (string.IsNullOrEmpty(userName))