init message hub and chathub observers

Directory.Packages.props

@@ -26,6 +26,7 @@
     <PackageVersion Include="System.IdentityModel.Tokens.Jwt" Version="8.0.0" />
     <PackageVersion Include="System.Memory.Data" Version="8.0.0" />
     <PackageVersion Include="System.Text.Json" Version="8.0.5" />
+    <PackageVersion Include="System.Reactive" Version="6.0.1" />
     <PackageVersion Include="Serilog.Sinks.Console" Version="6.0.0" />
     <PackageVersion Include="Serilog.Extensions.Logging" Version="9.0.0" />
     <PackageVersion Include="Serilog.Sinks.File" Version="6.0.0" />

src/Infrastructure/BotSharp.Abstraction/BotSharp.Abstraction.csproj

@@ -36,6 +36,7 @@
     <PackageReference Include="System.IdentityModel.Tokens.Jwt" />
     <PackageReference Include="System.Memory.Data" />
     <PackageReference Include="System.Text.Json" />
+    <PackageReference Include="System.Reactive" />
     <PackageReference Include="Serilog.Sinks.Console" />
     <PackageReference Include="Serilog.Sinks.File" />
     <PackageReference Include="Rougamo.Fody" />

src/Infrastructure/BotSharp.Abstraction/Observables/Models/HubObserveData.cs

@@ -0,0 +1,6 @@
+namespace BotSharp.Abstraction.Observables.Models;
+
+public class HubObserveData : ObserveDataBase
+{
+    public RoleDialogModel Data { get; set; } = null!;
+}

src/Infrastructure/BotSharp.Abstraction/Observables/Models/ObserveDataBase.cs

@@ -0,0 +1,6 @@
+namespace BotSharp.Abstraction.Observables.Models;
+
+public abstract class ObserveDataBase
+{
+    public IServiceProvider ServiceProvider { get; set; } = null!;
+}

src/Infrastructure/BotSharp.Core/Conversations/ConversationPlugin.cs

@@ -10,6 +10,7 @@ using BotSharp.Core.Messaging;
 using BotSharp.Core.Routing.Reasoning;
 using BotSharp.Core.Templating;
 using BotSharp.Core.Translation;
+using BotSharp.Core.Observables.Queues;
 using Microsoft.Extensions.Configuration;
 
 namespace BotSharp.Core.Conversations;
@@ -41,6 +42,8 @@ public class ConversationPlugin : IBotSharpPlugin
             return settingService.Bind<GoogleApiSettings>("GoogleApi");
         });
 
+        services.AddSingleton<MessageHub>();
+
         services.AddScoped<IConversationStorage, ConversationStorage>();
         services.AddScoped<IConversationService, ConversationService>();
         services.AddScoped<IConversationProgressService, ConversationProgressService>();

src/Infrastructure/BotSharp.Core/Observables/Queues/MessageHub.cs

@@ -0,0 +1,45 @@
+using BotSharp.Abstraction.Observables.Models;
+using System.Reactive.Subjects;
+
+namespace BotSharp.Core.Observables.Queues;
+
+public class MessageHub
+{
+    private readonly ILogger<MessageHub> _logger;
+    private readonly ISubject<HubObserveData> _observable = new Subject<HubObserveData>();
+    public IObservable<HubObserveData> Events => _observable;
+
+    public MessageHub(ILogger<MessageHub> logger)
+    {
+        _logger = logger;
+    }
+
+    /// <summary>
+    /// Push an item to the observers.
+    /// </summary>
+    /// <param name="item"></param>
+    public void Push(HubObserveData item)
+    {
+        _logger.LogInformation($"Pushing item to observers: {item.Data.Content}");
+        _observable.OnNext(item);
+    }
+
+    /// <summary>
+    /// Send a complete notification to the observers.
+    /// This will stop the observers from receiving data.
+    /// </summary>
+    public void Complete()
+    {
+        _observable.OnCompleted();
+    }
+
+    /// <summary>
+    /// Send an error notification to the observers.
+    /// This will stop the observers from receiving data.
+    /// </summary>
+    /// <param name="error"></param>
+    public void Error(Exception error)
+    {
+        _observable.OnError(error);
+    }
+}

src/Plugins/BotSharp.Plugin.ChatHub/ChatHubPlugin.cs

@@ -1,5 +1,8 @@
 using BotSharp.Abstraction.Crontab;
+using BotSharp.Core.Observables.Queues;
 using BotSharp.Plugin.ChatHub.Hooks;
+using BotSharp.Plugin.ChatHub.Observers;
+using Microsoft.AspNetCore.Builder;
 using Microsoft.Extensions.Configuration;
 
 namespace BotSharp.Plugin.ChatHub;
@@ -7,7 +10,7 @@ namespace BotSharp.Plugin.ChatHub;
 /// <summary>
 /// The dialogue channel connects users, AI assistants and customer service representatives.
 /// </summary>
-public class ChatHubPlugin : IBotSharpPlugin
+public class ChatHubPlugin : IBotSharpPlugin, IBotSharpAppPlugin
 {
     public string Id => "6e52d42d-1e23-406b-8599-36af36c83209";
     public string Name => "Chat Hub";
@@ -28,4 +31,12 @@ public class ChatHubPlugin : IBotSharpPlugin
         services.AddScoped<IContentGeneratingHook, StreamingLogHook>();
         services.AddScoped<ICrontabHook, ChatHubCrontabHook>();
     }
+
+    public void Configure(IApplicationBuilder app)
+    {
+        var services = app.ApplicationServices;
+        var queue = services.GetRequiredService<MessageHub>();
+        var logger = services.GetRequiredService<ILogger<MessageHub>>();
+        queue.Events.Subscribe(new ChatHubObserver(logger));
+    }
 }

src/Plugins/BotSharp.Plugin.ChatHub/Observers/ChatHubObserver.cs

@@ -0,0 +1,119 @@
+using BotSharp.Abstraction.Conversations.Dtos;
+using BotSharp.Abstraction.Observables.Models;
+using BotSharp.Abstraction.SideCar;
+using BotSharp.Abstraction.Users.Dtos;
+using BotSharp.Plugin.ChatHub.Hooks;
+using Microsoft.AspNetCore.SignalR;
+
+namespace BotSharp.Plugin.ChatHub.Observers;
+
+public class ChatHubObserver : IObserver<HubObserveData>
+{
+    private readonly ILogger _logger;
+    private IServiceProvider _services;
+    private IUserIdentity _user;
+
+    private const string RECEIVE_CLIENT_MESSAGE = "OnMessageReceivedFromClient";
+    private const string GENERATE_SENDER_ACTION = "OnSenderActionGenerated";
+
+    public ChatHubObserver(ILogger logger)
+    {
+        _logger = logger;
+    }
+
+    public void OnCompleted()
+    {
+        _logger.LogInformation($"{nameof(ChatHubObserver)} receives complete notification.");
+    }
+
+    public void OnError(Exception error)
+    {
+        _logger.LogError(error, $"{nameof(ChatHubObserver)} receives error notification: {error.Message}");
+    }
+
+    public void OnNext(HubObserveData value)
+    {
+        _services = value.ServiceProvider;
+        _user = _services.GetRequiredService<IUserIdentity>();
+        
+        ReceiveMessage(value.Data).ConfigureAwait(false).GetAwaiter().GetResult();
+    }
+
+    private async Task ReceiveMessage(RoleDialogModel message)
+    {
+        if (!AllowSendingMessage()) return;
+
+        var conv = _services.GetRequiredService<IConversationService>();
+        var userService = _services.GetRequiredService<IUserService>();
+        var sender = await userService.GetMyProfile();
+
+        // Update console conversation UI for CSR
+        var model = new ChatResponseDto()
+        {
+            ConversationId = conv.ConversationId,
+            MessageId = message.MessageId,
+            Payload = message.Payload,
+            Text = !string.IsNullOrEmpty(message.SecondaryContent) ? message.SecondaryContent : message.Content,
+            Sender = UserDto.FromUser(sender)
+        };
+        await ReceiveClientMessage(conv.ConversationId, model);
+
+        // Send typing-on to client
+        var action = new ConversationSenderActionModel
+        {
+            ConversationId = conv.ConversationId,
+            SenderAction = SenderActionEnum.TypingOn
+        };
+
+        await GenerateSenderAction(conv.ConversationId, action);
+    }
+
+    private async Task ReceiveClientMessage(string conversationId, ChatResponseDto model)
+    {
+        try
+        {
+            var settings = _services.GetRequiredService<ChatHubSettings>();
+            var chatHub = _services.GetRequiredService<IHubContext<SignalRHub>>();
+
+            if (settings.EventDispatchBy == EventDispatchType.Group)
+            {
+                await chatHub.Clients.Group(conversationId).SendAsync(RECEIVE_CLIENT_MESSAGE, model);
+            }
+            else
+            {
+                await chatHub.Clients.User(_user.Id).SendAsync(RECEIVE_CLIENT_MESSAGE, model);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, $"Failed to receive assistant message in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})");
+        }
+    }
+
+    private bool AllowSendingMessage()
+    {
+        var sidecar = _services.GetService<IConversationSideCar>();
+        return sidecar == null || !sidecar.IsEnabled();
+    }
+
+    private async Task GenerateSenderAction(string conversationId, ConversationSenderActionModel action)
+    {
+        try
+        {
+            var settings = _services.GetRequiredService<ChatHubSettings>();
+            var chatHub = _services.GetRequiredService<IHubContext<SignalRHub>>();
+            if (settings.EventDispatchBy == EventDispatchType.Group)
+            {
+                await chatHub.Clients.Group(conversationId).SendAsync(GENERATE_SENDER_ACTION, action);
+            }
+            else
+            {
+                await chatHub.Clients.User(_user.Id).SendAsync(GENERATE_SENDER_ACTION, action);
+            }
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, $"Failed to generate sender action in {nameof(ChatHubConversationHook)} (conversation id: {conversationId})");
+        }
+    }
+}