migrate conv latest states

Author: Jicheng Lu

src/Infrastructure/BotSharp.Abstraction/Conversations/IConversationService.cs

@@ -70,4 +70,6 @@ Task<bool> SendMessage(string agentId,
     /// <param name="preLoad">if pre-loading, then keys are not filter by the search query</param>
     /// <returns></returns>
     Task<List<string>> GetConversationStateSearhKeys(string query, int convLimit = 100, int keyLimit = 10, bool preload = false);
+
+    Task<bool> MigrateLatestStates(int batchSize = 100, int errorLimit = 10);
 }

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

@@ -151,6 +151,10 @@ List<string> TruncateConversation(string conversationId, string messageId, bool
          => throw new NotImplementedException();
     List<string> GetConversationStateSearchKeys(int messageLowerLimit = 2, int convUpperLimit = 100)
          => throw new NotImplementedException();
+    List<string> GetConversationsToMigrate(int batchSize = 100)
+        => throw new NotImplementedException();
+    bool MigrateConvsersationLatestStates(string conversationId)
+         => throw new NotImplementedException();
     #endregion
 
     #region LLM Completion Log

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

@@ -0,0 +1,98 @@
+using NetTopologySuite.Algorithm;
+using System.Diagnostics;
+
+namespace BotSharp.Core.Conversations.Services;
+
+public partial class ConversationService
+{
+    public async Task<bool> MigrateLatestStates(int batchSize = 100, int errorLimit = 10)
+    {
+        var db = _services.GetRequiredService<IBotSharpRepository>();
+        var isSuccess = true;
+        var errorCount = 0;
+        var batchNum = 0;
+        var info = string.Empty;
+        var error = string.Empty;
+
+#if DEBUG
+        Console.WriteLine($"\r\n#Start migrating Conversation Latest States...\r\n");
+#else
+        _logger.LogInformation($"#Start migrating Conversation Latest States...");
+#endif
+        var sw = Stopwatch.StartNew();
+
+        var convIds = db.GetConversationsToMigrate(batchSize);
+        
+        while (!convIds.IsNullOrEmpty())
+        {
+            batchNum++;
+            var innerSw = Stopwatch.StartNew();
+#if DEBUG
+            Console.WriteLine($"\r\n#Start migrating Conversation Latest States (batch number: {batchNum})\r\n");
+#else
+            _logger.LogInformation($"#Start migrating Conversation Latest States (batch number: {batchNum})");
+#endif
+
+            for (int i = 0; i < convIds.Count; i++)
+            {
+                var convId = convIds.ElementAt(i);
+                try
+                {
+                    var done = db.MigrateConvsersationLatestStates(convId);
+                    info = $"Conversation {convId} latest states have been migrated ({i + 1}/{convIds.Count})!";
+#if DEBUG
+                    Console.WriteLine($"\r\n{info}\r\n");
+#else
+                    _logger.LogInformation($"{info}");
+#endif
+                }
+                catch (Exception ex)
+                {
+                    errorCount++;
+                    error = $"Conversation {convId} latest states fail to be migrated! ({i + 1}/{convIds.Count})\r\n{ex.Message}\r\n{ex.InnerException}";
+#if DEBUG
+                    Console.WriteLine($"\r\n{error}\r\n");
+#else
+                    _logger.LogError($"{error}");
+#endif
+                }
+            }
+
+            if (errorCount >= errorLimit)
+            {
+                error = $"\r\nErrors exceed limit => stop the migration!\r\n";
+#if DEBUG
+                Console.WriteLine($"{error}");
+#else
+                _logger.LogError($"{error}");
+#endif
+                innerSw.Stop();
+                isSuccess = false;
+                break;
+            }
+
+            innerSw.Stop();
+            info = $"#Done migrating Conversation Latest States (batch number: {batchNum}) " +
+                $"(Total time: {innerSw.Elapsed.Hours} hrs, {innerSw.Elapsed.Minutes} mins, {innerSw.Elapsed.Seconds} seconds)";
+#if DEBUG
+            Console.WriteLine($"\r\n{info}\r\n");
+#else
+            _logger.LogInformation($"{info}");
+#endif
+
+            await Task.Delay(100);
+            convIds = db.GetConversationsToMigrate(batchSize);
+        }
+
+        sw.Stop();
+        info = $"#Done with migrating Conversation Latest States! " +
+            $"(Total time: {sw.Elapsed.Days} days, {sw.Elapsed.Hours} hrs, {sw.Elapsed.Minutes} mins, {sw.Elapsed.Seconds} seconds)";
+#if DEBUG
+        Console.WriteLine($"\r\n{info}\r\n");
+#else
+        _logger.LogInformation($"{info}");
+#endif
+
+        return isSuccess;
+    }
+}

src/Infrastructure/BotSharp.Core/Loggers/Services/LoggerService.Instruction.cs

@@ -41,7 +41,7 @@ public async Task<PagedItems<InstructionLogModel>> GetInstructionLogs(InstructLo
 
         var items = logs.Items.Select(x =>
         {
-            x.AgentId = !string.IsNullOrEmpty(x.AgentId) ? agents.FirstOrDefault(a => a.Id == x.AgentId)?.Name : null;
+            x.AgentName = !string.IsNullOrEmpty(x.AgentId) ? agents.FirstOrDefault(a => a.Id == x.AgentId)?.Name : null;
 
             if (!isAdmin)
             {

src/Infrastructure/BotSharp.Core/Repository/FileRepository/FileRepository.Conversation.cs

@@ -687,6 +687,56 @@ public List<string> GetConversationStateSearchKeys(int messageLowerLimit = 2, in
     }
 
 
+
+    public List<string> GetConversationsToMigrate(int batchSize = 100)
+    {
+        var baseDir = Path.Combine(_dbSettings.FileRepository, _conversationSettings.DataDir);
+        if (!Directory.Exists(baseDir)) return [];
+
+        var convIds = new List<string>();
+        var dirs = Directory.GetDirectories(baseDir);
+
+        foreach (var dir in dirs)
+        {
+            var latestStateFile = Path.Combine(dir, CONV_LATEST_STATE_FILE);
+            if (File.Exists(latestStateFile)) continue;
+
+            var convId = dir.Split(Path.DirectorySeparatorChar).Last();
+            if (string.IsNullOrEmpty(convId)) continue;
+
+            convIds.Add(convId);
+            if (convIds.Count >= batchSize)
+            {
+                break;
+            }
+        }
+
+        return convIds;
+    }
+
+
+    public bool MigrateConvsersationLatestStates(string conversationId)
+    {
+        if (string.IsNullOrEmpty(conversationId)) return false;
+
+        var convDir = FindConversationDirectory(conversationId);
+        if (string.IsNullOrEmpty(convDir))
+        {
+            return false;
+        }
+
+        var stateFile = Path.Combine(convDir, STATE_FILE);
+        var states = CollectConversationStates(stateFile);
+        var latestStates = BuildLatestStates(states);
+
+        var latestStateFile = Path.Combine(convDir, CONV_LATEST_STATE_FILE);
+        var stateStr = JsonSerializer.Serialize(latestStates, _options);
+        File.WriteAllText(latestStateFile, stateStr);
+
+        return true;
+    }
+
+
     #region Private methods
     private string? FindConversationDirectory(string conversationId)
     {
@@ -883,6 +933,11 @@ private Dictionary<string, JsonDocument> CollectConversationLatestStates(string
     private Dictionary<string, JsonDocument> BuildLatestStates(List<StateKeyValue> states)
     {
         var endNodes = new Dictionary<string, JsonDocument>();
+        if (states.IsNullOrEmpty())
+        {
+            return endNodes;
+        }
+
         foreach (var pair in states)
         {
             var value = pair.Values?.LastOrDefault();

src/Infrastructure/BotSharp.OpenAPI/Controllers/ConversationController.cs

@@ -563,6 +563,16 @@ public async Task<List<string>> GetConversationStateKeys([FromQuery] string quer
     }
     #endregion
 
+    #region Migrate Latest States
+    [HttpPost("/conversation/latest-state/migrate")]
+    public async Task<bool> MigrateConversationLatestStates([FromBody] MigrateLatestStateRequest request)
+    {
+        var convService = _services.GetRequiredService<IConversationService>();
+        var res = await convService.MigrateLatestStates(request.BatchSize, request.ErrorLimit);
+        return res;
+    }
+    #endregion
+
     #region Private methods
     private void SetStates(IConversationService conv, NewMessageModel input)
     {

src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/MigrateLatestStateRequest.cs

@@ -0,0 +1,7 @@
+namespace BotSharp.OpenAPI.ViewModels.Conversations;
+
+public class MigrateLatestStateRequest
+{
+    public int BatchSize { get; set; } = 1000;
+    public int ErrorLimit { get; set; } = 10;
+}

src/Plugins/BotSharp.Plugin.MongoStorage/Repository/MongoRepository.Conversation.cs

@@ -1,5 +1,6 @@
 using BotSharp.Abstraction.Conversations.Models;
 using BotSharp.Abstraction.Repositories.Filters;
+using MongoDB.Driver;
 using System.Text.Json;
 
 namespace BotSharp.Plugin.MongoStorage.Repository;
@@ -661,6 +662,37 @@ public List<string> GetConversationStateSearchKeys(int messageLowerLimit = 2, in
         return keys;
     }
 
+
+
+    public List<string> GetConversationsToMigrate(int batchSize = 100)
+    {
+        var convFilter = Builders<ConversationDocument>.Filter.Exists(x => x.LatestStates, false);
+        var sortDef = Builders<ConversationDocument>.Sort.Ascending(x => x.CreatedTime);
+        var convIds = _dc.Conversations.Find(convFilter).Sort(sortDef)
+                                       .Limit(batchSize).ToEnumerable()
+                                       .Select(x => x.Id).ToList();
+        return convIds ?? [];
+    }
+
+    public bool MigrateConvsersationLatestStates(string conversationId)
+    {
+        if (string.IsNullOrEmpty(conversationId)) return false;
+
+        var stateFilter = Builders<ConversationStateDocument>.Filter.Eq(x => x.ConversationId, conversationId);
+        var foundStates = _dc.ConversationStates.Find(stateFilter).FirstOrDefault();
+        if (foundStates?.States == null) return false;
+
+        var states = foundStates.States.ToList();
+        var latestStates = BuildLatestStates(states);
+
+        var convFilter = Builders<ConversationDocument>.Filter.Eq(x => x.Id, conversationId);
+        var convUpdate = Builders<ConversationDocument>.Update.Set(x => x.LatestStates, latestStates);
+        _dc.Conversations.UpdateOne(convFilter, convUpdate);
+
+        return true;
+    }
+
+    #region Private methods
     private string ConvertSnakeCaseToPascalCase(string snakeCase)
     {
         string[] words = snakeCase.Split('_');
@@ -682,6 +714,11 @@ private string ConvertSnakeCaseToPascalCase(string snakeCase)
     private Dictionary<string, BsonDocument> BuildLatestStates(List<StateMongoElement> states)
     {
         var endNodes = new Dictionary<string, BsonDocument>();
+        if (states.IsNullOrEmpty())
+        {
+            return endNodes;
+        }
+
         foreach (var pair in states)
         {
             var value = pair.Values?.LastOrDefault();
@@ -703,4 +740,5 @@ private Dictionary<string, BsonDocument> BuildLatestStates(List<StateMongoElemen
 
         return endNodes;
     }
+    #endregion
 }