Merge pull request #205 from seplz/feature/CloseIdleConversations

add ConversationTimeoutService

Author: Oceania2018

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

@@ -8,6 +8,7 @@ public interface IConversationService
     void SetConversationId(string conversationId, List<string> states);
     Task<Conversation> GetConversation(string id);
     Task<List<Conversation>> GetConversations();
+    Task<List<Conversation>> GetLastConversations();
     Task DeleteConversation(string id);
 
     /// <summary>

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

@@ -34,6 +34,7 @@ List<Agent> GetAgents(string? name = null, bool? disabled = null, bool? allowRou
     void UpdateConversationStates(string conversationId, List<StateKeyValue> states);
     Conversation GetConversation(string conversationId);
     List<Conversation> GetConversations(string userId);
+    List<Conversation> GetLastConversations();
     void AddExectionLogs(string conversationId, List<string> logs);
     List<string> GetExectionLogs(string conversationId);
     #endregion

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

@@ -52,6 +52,12 @@ public async Task<List<Conversation>> GetConversations()
         return conversations.OrderByDescending(x => x.CreatedTime).ToList();
     }
 
+    public async Task<List<Conversation>> GetLastConversations()
+    {
+        var db = _services.GetRequiredService<IBotSharpRepository>();
+        return db.GetLastConversations();
+    }
+
     public async Task<Conversation> NewConversation(Conversation sess)
     {
         var db = _services.GetRequiredService<IBotSharpRepository>();

src/Infrastructure/BotSharp.Core/Repository/BotSharpDbContext.cs

@@ -131,6 +131,11 @@ public List<Conversation> GetConversations(string userId)
         throw new NotImplementedException();
     }
 
+    public List<Conversation> GetLastConversations()
+    {
+        throw new NotImplementedException();
+    }
+
     public string GetConversationDialog(string conversationId)
     {
         throw new NotImplementedException();

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

@@ -743,6 +743,28 @@ public List<Conversation> GetConversations(string userId)
         return records;
     }
 
+    public List<Conversation> GetLastConversations()
+    {
+        var records = new List<Conversation>();
+        var dir = Path.Combine(_dbSettings.FileRepository, _conversationSettings.DataDir);
+
+        foreach (var d in Directory.GetDirectories(dir))
+        {
+            var path = Path.Combine(d, "conversation.json");
+            if (!File.Exists(path)) continue;
+
+            var json = File.ReadAllText(path);
+            var record = JsonSerializer.Deserialize<Conversation>(json, _options);
+            if (record != null)
+            {
+                records.Add(record);
+            }
+        }
+        return records.GroupBy(r => r.UserId)
+            .Select(g => g.OrderByDescending(x => x.CreatedTime).First())
+            .ToList();
+    }
+
     public void AddExectionLogs(string conversationId, List<string> logs)
     {
         if (string.IsNullOrEmpty(conversationId) || logs.IsNullOrEmpty()) return;

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

@@ -6,7 +6,7 @@ namespace BotSharp.Core.Users.Services;
 public class UserIdentity : IUserIdentity
 {
     private readonly IHttpContextAccessor _contextAccessor;
-    private IEnumerable<Claim> _claims => _contextAccessor.HttpContext.User.Claims;
+    private IEnumerable<Claim> _claims => _contextAccessor.HttpContext?.User.Claims!;
 
     public UserIdentity(IHttpContextAccessor contextAccessor)
     {
@@ -15,14 +15,14 @@ public UserIdentity(IHttpContextAccessor contextAccessor)
 
 
     public string Id 
-        => _claims.FirstOrDefault(x => x.Type == ClaimTypes.NameIdentifier)?.Value;
+        => _claims?.FirstOrDefault(x => x.Type == ClaimTypes.NameIdentifier)?.Value!;
 
     public string Email 
-        => _claims.FirstOrDefault(x => x.Type == ClaimTypes.Email)?.Value;
+        => _claims?.FirstOrDefault(x => x.Type == ClaimTypes.Email)?.Value!;
 
     public string FirstName 
-        => _claims.FirstOrDefault(x => x.Type == ClaimTypes.GivenName)?.Value;
+        => _claims?.FirstOrDefault(x => x.Type == ClaimTypes.GivenName)?.Value!;
 
     public string LastName 
-        => _claims.FirstOrDefault(x => x.Type == ClaimTypes.Surname)?.Value;
+        => _claims?.FirstOrDefault(x => x.Type == ClaimTypes.Surname)?.Value!;
 }

src/Infrastructure/BotSharp.OpenAPI/BackgroundServices/ConversationTimeoutService.cs

@@ -0,0 +1,80 @@
+using BotSharp.Abstraction.Agents.Enums;
+using BotSharp.Abstraction.Conversations.Models;
+using Microsoft.Extensions.Hosting;
+
+namespace BotSharp.OpenAPI.BackgroundServices
+{
+    public class ConversationTimeoutService : BackgroundService
+    {
+        private readonly IServiceProvider _services;
+        private readonly ILogger<ConversationTimeoutService> _logger;
+
+        public ConversationTimeoutService(IServiceProvider services, ILogger<ConversationTimeoutService> logger)
+        {
+            _services = services;
+            _logger = logger;
+        }
+
+        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        {
+            _logger.LogInformation("Conversation Timeout Service is running.");
+            try
+            {
+                while (true)
+                {
+                    stoppingToken.ThrowIfCancellationRequested();
+                    var delay = Task.Delay(TimeSpan.FromMinutes(1));
+                    try
+                    {
+                        await CloseIdleConversationsAsync(TimeSpan.FromMinutes(10));
+                    }
+                    catch (Exception ex)
+                    {
+                        _logger.LogError(ex, $"Error occurred closing conversations.");
+                    }
+                    await delay;
+                }
+            }
+            catch (OperationCanceledException) { }
+        }
+
+        public override async Task StopAsync(CancellationToken stoppingToken)
+        {
+            _logger.LogInformation("Conversation Timeout Service is stopping.");
+            await base.StopAsync(stoppingToken);
+        }
+
+        private async Task CloseIdleConversationsAsync(TimeSpan conversationIdleTimeout)
+        {
+            using var scope = _services.CreateScope();
+            var conversationService = scope.ServiceProvider.GetRequiredService<IConversationService>();
+            var hooks = scope.ServiceProvider.GetServices<IConversationHook>()
+                .OrderBy(x => x.Priority)
+                .ToList();
+            var moment = DateTime.UtcNow.Add(-conversationIdleTimeout);
+            var conversations =
+                (await conversationService.GetLastConversations())
+                .Where(c => c.CreatedTime <= moment);
+            foreach (var conversation in conversations)
+            {
+                try
+                {
+                    var response = new RoleDialogModel(AgentRole.Assistant, "End the conversation due to timeout.")
+                    {
+                        StopCompletion = true,
+                        FunctionName = "conversation_end"
+                    };
+
+                    foreach (var hook in hooks)
+                    {
+                        await hook.OnConversationEnding(response);
+                    }
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, $"Error occurred closing conversation #{conversation.Id}.");
+                }
+            }
+        }
+    }
+}

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

@@ -732,6 +732,24 @@ public List<Conversation> GetConversations(string userId)
         return records;
     }
 
+    public List<Conversation> GetLastConversations()
+    {
+        var records = new List<Conversation>();
+        var conversations = _dc.Conversations.Aggregate()
+            .Group(c => c.UserId,
+                g => g.OrderByDescending(x => x.CreatedTime).First())
+            .ToList();
+        return conversations.Select(c => new Conversation()
+        {
+            Id = c.Id.ToString(),
+            AgentId = c.AgentId.ToString(),
+            UserId = c.UserId.ToString(),
+            Title = c.Title,
+            CreatedTime = c.CreatedTime,
+            UpdatedTime = c.UpdatedTime
+        }).ToList();
+    }
+
     public void AddExectionLogs(string conversationId, List<string> logs)
     {
         if (string.IsNullOrEmpty(conversationId) || logs.IsNullOrEmpty()) return;