Merge pull request #811 from Qtoss-AI/master

WebDriver enhancement

Author: Oceania2018

src/Infrastructure/BotSharp.Abstraction/Browsing/IWebBrowser.cs

@@ -28,4 +28,5 @@ public interface IWebBrowser
     Task<BrowserActionResult> CloseCurrentPage(MessageInfo message);
     Task<BrowserActionResult> SendHttpRequest(MessageInfo message, HttpRequestParams actionParams);
     Task<BrowserActionResult> GetAttributeValue(MessageInfo message, ElementLocatingArgs location);
+    Task<BrowserActionResult> SetAttributeValue(MessageInfo message, ElementLocatingArgs location);
 }

src/Infrastructure/BotSharp.Abstraction/Browsing/Models/BrowserActionArgs.cs

@@ -3,4 +3,6 @@ namespace BotSharp.Abstraction.Browsing.Models;
 public class BrowserActionArgs
 {
     public bool Headless { get; set; }
+    public string? UserDataDir { get; set; }
+    public string? RemoteHostUrl { get; set; }
 }

src/Infrastructure/BotSharp.Abstraction/Browsing/Models/BrowserActionResult.cs

@@ -8,6 +8,10 @@ public class BrowserActionResult
     public string? StackTrace { get; set; }
     public string? Selector { get; set; }
     public string? Body { get; set; }
+    /// <summary>
+    /// Page open in new tab after button click
+    /// </summary>
+    public string? UrlAfterAction { get; set; }
     public bool IsHighlighted { get; set; }
 
     public override string ToString()

src/Infrastructure/BotSharp.Abstraction/Repositories/IBotSharpRepository.cs

@@ -33,6 +33,7 @@ public interface IBotSharpRepository : IHaveServiceProvider
     List<User> GetUserByIds(List<string> ids) => throw new NotImplementedException();
     List<User> GetUsersByAffiliateId(string affiliateId) => throw new NotImplementedException();
     User? GetUserByUserName(string userName) => throw new NotImplementedException();
+    void UpdateUserName(string userId, string userName) => throw new NotImplementedException();
     Dashboard? GetDashboard(string id = null) => throw new NotImplementedException();
     void CreateUser(User user) => throw new NotImplementedException();
     void UpdateExistUser(string userId, User user) => throw new NotImplementedException();
@@ -49,6 +50,7 @@ public interface IBotSharpRepository : IHaveServiceProvider
     PagedItems<User> GetUsers(UserFilter filter) => throw new NotImplementedException();
     User? GetUserDetails(string userId, bool includeAgent = false) => throw new NotImplementedException();
     bool UpdateUser(User user, bool updateUserAgents = false) => throw new NotImplementedException();
+
     #endregion
 
     #region Agent

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.Abstraction/Users/Models/User.cs

@@ -23,7 +23,6 @@ public class User
     public bool Verified { get; set; }
     public string RegionCode { get; set; } = "CN";
     public string? AffiliateId { get; set; }
-    public string? ReferralCode { get; set; }
     public string? EmployeeId { get; set; }
     public bool IsDisabled { get; set; }
     public IEnumerable<string> Permissions { get; set; } = [];

src/Infrastructure/BotSharp.Abstraction/Users/Models/UserActivationModel.cs

@@ -4,4 +4,5 @@ public class UserActivationModel
 {
     public string UserName { get; set; }
     public string VerificationCode { get; set; }
+    public string RegionCode { get; set; } = "CN";
 }

src/Infrastructure/BotSharp.Core/BotSharp.Core.csproj

@@ -190,10 +190,10 @@
   <ItemGroup>
     <PackageReference Include="Aspects.Cache" Version="2.0.4" />
     <PackageReference Include="DistributedLock.Redis" Version="1.0.3" />
-    <PackageReference Include="EntityFrameworkCore.BootKit" Version="8.6.0" />
+    <PackageReference Include="EntityFrameworkCore.BootKit" Version="8.8.0" />
     <PackageReference Include="Fluid.Core" Version="2.11.1" />
     <PackageReference Include="Microsoft.Extensions.Caching.Memory" Version="8.0.1" />
-    <PackageReference Include="Microsoft.Extensions.Http" Version="8.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Http" Version="8.0.1" />
     <PackageReference Include="Nanoid" Version="3.1.0" />
   </ItemGroup>
 

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

@@ -39,10 +39,7 @@ public async Task BroadcastAsync(string channel, string message)
 
         // 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,25 @@ private bool CheckMessageExists(IDatabase db, string channel, string fieldName,
         return exists;
     }
 
+    public static NameValueEntry[] AssembleMessage(RedisValue message)
+    {
+        return
+        [
+            new NameValueEntry("message", message),
+            new NameValueEntry("timestamp", DateTime.UtcNow.ToString("o"))
+        ];
+    }
+
+    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();
@@ -93,10 +109,10 @@ public async Task ReDispatchAsync(string channel, int count = 10, string order =
 
             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 messageId = await db.StreamAddAsync(channel, 
+                    AssembleMessage(message),
+                    maxLength: 1000 * 10000);
 
                 _logger.LogWarning($"ReDispatched message: {channel} {entry.Values[0].Value} ({messageId})");
 
@@ -164,20 +180,7 @@ 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}");
-            }
-        }
+        var deletedCount = await db.StreamDeleteAsync(channel, entries.Select(x => x.Id).ToArray());
+        _logger.LogWarning($"Deleted {deletedCount} messages from Redis stream {channel}");
     }
 }

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)
         {
@@ -58,28 +60,36 @@ public async Task SubscribeAsync(string channel, string group, int? port, bool p
                 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, $"{channel}-Error") > 0)
+                    {
+                        continue;
+                    }
 
-                if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received) > 0)
+                    if (await HandleGroupMessage(db, $"{channel}-{EventPriority.Medium}", group, consumer, received, $"{channel}-Error") > 0)
+                    {
+                        continue;
+                    }
+
+                    await HandleGroupMessage(db, $"{channel}-{EventPriority.Low}", group, consumer, received, $"{channel}-Error");
+                }
+                else
                 {
-                    continue;
+                    await HandleGroupMessage(db, channel, group, consumer, received, $"{channel}-Error");
                 }
-
-                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);
             }
         }
     }
 
-    private async Task<int> HandleGroupMessage(IDatabase db, string channel, string group, string consumer, Func<string, string, Task> received)
+    private async Task<int> HandleGroupMessage(IDatabase db, string channel, string group, string consumer, Func<string, string, Task> received, string errorChannel)
     {
         var entries = await db.StreamReadGroupAsync(channel, group, consumer, count: 1);
         foreach (var entry in entries)
@@ -90,13 +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(errorChannel,
+                    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
+            {
+                var deletedCount = await db.StreamDeleteAsync(channel, [entry.Id]);
+                _logger.LogInformation($"Handled message {entry.Id}: {deletedCount == 1}");
             }
         }