Add channel system with CLI, Telegram, and Web channel implementations
Define channel abstractions (IChannel, IChannelManager) with event-driven message and connection handling. Implement ChannelManager for multi-channel orchestration, CliChannel for terminal I/O, TelegramChannel with adapter pattern for bot API integration, and WebChannel stub. Includes DI registration and channel type enumeration.
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
using System.Collections.Immutable;
|
||||
using Luna.Channels.Abstractions;
|
||||
using Luna.Core.Abstractions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Luna.Channels;
|
||||
|
||||
public class ChannelManager(ISessionManager sessionManager, ILogger<ChannelManager> logger) : IChannelManager
|
||||
{
|
||||
private readonly Dictionary<string, IChannel> channels = new();
|
||||
private readonly Dictionary<string, string> sessionMap = new();
|
||||
|
||||
public event EventHandler<ChannelMessageReceivedEventArgs>? MessageRouted;
|
||||
|
||||
public void RegisterChannel(IChannel channel)
|
||||
{
|
||||
if (!channels.TryAdd(channel.ChannelId, channel))
|
||||
throw new InvalidOperationException($"Channel with id {channel.ChannelId} is already registered.");
|
||||
|
||||
channel.MessageReceived += OnMessageReceivedAsync;
|
||||
}
|
||||
|
||||
public void UnregisterChannel(string channelId)
|
||||
{
|
||||
if (!channels.TryGetValue(channelId, out var channel))
|
||||
return;
|
||||
|
||||
channel.MessageReceived -= OnMessageReceivedAsync;
|
||||
channels.Remove(channelId);
|
||||
}
|
||||
|
||||
public IChannel? GetChannel(string channelId)
|
||||
{
|
||||
channels.TryGetValue(channelId, out var channel);
|
||||
return channel;
|
||||
}
|
||||
|
||||
public IReadOnlyList<IChannel> GetAllChannels()
|
||||
=> channels.Values.ToImmutableList();
|
||||
|
||||
private async Task OnMessageReceivedAsync(object? sender, ChannelMessageReceivedEventArgs args)
|
||||
{
|
||||
if (args.IsHandled)
|
||||
return;
|
||||
|
||||
var channel = args.Channel;
|
||||
var message = args.Message;
|
||||
|
||||
try
|
||||
{
|
||||
var response = sessionManager
|
||||
.RouteMessagesAsync(message.Content, message.ConversationId, channel.ChannelId);
|
||||
|
||||
await channel.SendStreamingMessageAsync(response);
|
||||
MessageRouted?.Invoke(this, args);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Message routing for channel of type {Type} with id {Id} failed", channel.ChannelType, channel.ChannelId);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
using Luna.Channels.Abstractions;
|
||||
using Luna.Core.Abstractions;
|
||||
using Luna.Shared;
|
||||
|
||||
namespace Luna.Channels.Cli;
|
||||
|
||||
public class CliChannel(string channelId, IChatHubContext hubContext) : IChannel
|
||||
{
|
||||
public string ChannelId { get; } = channelId;
|
||||
public string ChannelType => Abstractions.ChannelType.Cli;
|
||||
public string DisplayName { get; } = "Cli";
|
||||
public bool IsConnected { get; }
|
||||
|
||||
public event ChannelMessageReceivedEventHandler? MessageReceived;
|
||||
public event ChannelConnectionChangedEventHandler? ConnectionStateChanged;
|
||||
|
||||
public async Task SendStreamingMessageAsync(IAsyncEnumerable<ChatStreamUpdate> messageStream, CancellationToken cancellationToken = default)
|
||||
{
|
||||
await hubContext.SendStreamingResponseAsync(ChannelId, messageStream, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task SendMessageAsync(ChannelMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
await hubContext.SendResponseAsync(ChannelId, message, cancellationToken);
|
||||
}
|
||||
|
||||
public void RaiseMessageReceived(object? sender, ChannelMessage message)
|
||||
{
|
||||
MessageReceived?.Invoke(sender, new ChannelMessageReceivedEventArgs
|
||||
{
|
||||
Channel = this,
|
||||
Message = message,
|
||||
IsHandled = false
|
||||
});
|
||||
}
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
using Luna.Channels.Abstractions;
|
||||
using Luna.Channels.Telegram;
|
||||
using Luna.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Telegram.Bot;
|
||||
|
||||
namespace Luna.Channels.Extensions;
|
||||
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
private const string TelegramBotToken = "REDACTED-TELEGRAM-BOT-TOKEN";
|
||||
|
||||
extension(IServiceCollection services)
|
||||
{
|
||||
public IServiceCollection AddChannels()
|
||||
{
|
||||
services.AddSingleton<IChannelManager, ChannelManager>();
|
||||
|
||||
// Telegram
|
||||
var telegramOptions = new TelegramOptions { BotToken = TelegramBotToken };
|
||||
services.AddSingleton(telegramOptions);
|
||||
services.AddSingleton<ITelegramBotClient>(_ => new TelegramBotClient(TelegramBotToken));
|
||||
services.AddHostedService<TelegramAdapter>();
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Luna.Channels.Abstractions\Luna.Channels.Abstractions.csproj" />
|
||||
<ProjectReference Include="..\Luna.Configuration\Luna.Configuration.csproj" />
|
||||
<ProjectReference Include="..\Luna.Core.Abstractions\Luna.Core.Abstractions.csproj" />
|
||||
<ProjectReference Include="..\Luna.Shared\Luna.Shared.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.5" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.5" />
|
||||
<PackageReference Include="Telegram.Bot" Version="22.9.5.3" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
@@ -0,0 +1,123 @@
|
||||
using Luna.Channels.Abstractions;
|
||||
using Luna.Configuration;
|
||||
using Luna.Shared;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Telegram.Bot;
|
||||
using Telegram.Bot.Polling;
|
||||
using Telegram.Bot.Types;
|
||||
|
||||
namespace Luna.Channels.Telegram;
|
||||
|
||||
public class TelegramAdapter(
|
||||
IChannelManager channelManager,
|
||||
ITelegramBotClient botClient,
|
||||
TelegramOptions options,
|
||||
ILogger<TelegramAdapter> adapterLogger,
|
||||
ILogger<TelegramChannel> channelLogger) : IHostedService, IUpdateHandler
|
||||
{
|
||||
private readonly Dictionary<long, TelegramChannel> channels = new();
|
||||
private CancellationTokenSource? cts;
|
||||
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
cts = new CancellationTokenSource();
|
||||
botClient.StartReceiving(this, new ReceiverOptions(), cts.Token);
|
||||
adapterLogger.LogInformation("Telegram adapter started polling");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
cts?.Cancel();
|
||||
cts?.Dispose();
|
||||
|
||||
foreach (var (_, channel) in channels)
|
||||
{
|
||||
channelManager.UnregisterChannel(channel.ChannelId);
|
||||
channel.Dispose();
|
||||
}
|
||||
|
||||
channels.Clear();
|
||||
adapterLogger.LogInformation("Telegram adapter stopped");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
public async Task HandleUpdateAsync(ITelegramBotClient client, Update update, CancellationToken cancellationToken)
|
||||
{
|
||||
if (update.Message is not { } message)
|
||||
return;
|
||||
|
||||
if (message.Text is null or "")
|
||||
return;
|
||||
|
||||
var chat = message.Chat;
|
||||
var channel = GetOrCreateChannel(chat.Id);
|
||||
|
||||
if (message.Text.StartsWith('/'))
|
||||
{
|
||||
await HandleCommandAsync(channel, message, cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
var channelMessage = new ChannelMessage(
|
||||
MessageId: message.Id.ToString(),
|
||||
ConversationId: chat.Id.ToString(),
|
||||
SenderId: message.From?.Id.ToString() ?? "unknown",
|
||||
SenderName: message.From?.FirstName ?? "Unknown",
|
||||
Content: message.Text,
|
||||
Timestamp: new DateTimeOffset(message.Date, TimeSpan.Zero));
|
||||
|
||||
channel.RaiseMessageReceived(channelMessage);
|
||||
}
|
||||
|
||||
public Task HandleErrorAsync(ITelegramBotClient client, Exception exception, HandleErrorSource source,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
adapterLogger.LogError(exception, "Telegram polling error from {Source}", source);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private TelegramChannel GetOrCreateChannel(long chatId)
|
||||
{
|
||||
if (channels.TryGetValue(chatId, out var existing))
|
||||
return existing;
|
||||
|
||||
var channel = new TelegramChannel(
|
||||
chatId.ToString(),
|
||||
botClient,
|
||||
channelLogger,
|
||||
options);
|
||||
|
||||
channels[chatId] = channel;
|
||||
channelManager.RegisterChannel(channel);
|
||||
adapterLogger.LogInformation("Created TelegramChannel for chat {ChatId}", chatId);
|
||||
|
||||
return channel;
|
||||
}
|
||||
|
||||
private async Task HandleCommandAsync(TelegramChannel channel, Message message, CancellationToken cancellationToken)
|
||||
{
|
||||
var command = message.Text!.Split(' ')[0].ToLowerInvariant();
|
||||
|
||||
var responseText = command switch
|
||||
{
|
||||
"/start" => "Welcome to Luna AI Assistant! Send me a message and I'll do my best to help.",
|
||||
"/help" => "Luna AI Assistant\n\nCommands:\n/start - Start the bot\n/help - Show this help message\n\nJust send a message to chat with Luna!",
|
||||
_ => null
|
||||
};
|
||||
|
||||
if (responseText is null)
|
||||
return;
|
||||
|
||||
var responseMessage = new ChannelMessage(
|
||||
MessageId: Guid.NewGuid().ToString(),
|
||||
ConversationId: message.Chat.Id.ToString(),
|
||||
SenderId: "luna",
|
||||
SenderName: "Luna",
|
||||
Content: responseText,
|
||||
Timestamp: DateTimeOffset.UtcNow);
|
||||
|
||||
await channel.SendMessageAsync(responseMessage, cancellationToken);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
using System.Text;
|
||||
using Luna.Channels.Abstractions;
|
||||
using Luna.Configuration;
|
||||
using Luna.Shared;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Telegram.Bot;
|
||||
using Telegram.Bot.Types.Enums;
|
||||
|
||||
namespace Luna.Channels.Telegram;
|
||||
|
||||
public class TelegramChannel(
|
||||
string chatId,
|
||||
ITelegramBotClient botClient,
|
||||
ILogger<TelegramChannel> logger,
|
||||
TelegramOptions options) : IChannel
|
||||
{
|
||||
private const int TelegramMessageCharLimit = 4096;
|
||||
|
||||
public string ChannelId { get; } = chatId;
|
||||
public string ChannelType => Abstractions.ChannelType.Telegram;
|
||||
public string DisplayName { get; } = $"Telegram Chat {chatId}";
|
||||
public bool IsConnected { get; private set; } = true;
|
||||
|
||||
public event ChannelMessageReceivedEventHandler? MessageReceived;
|
||||
public event ChannelConnectionChangedEventHandler? ConnectionStateChanged;
|
||||
|
||||
public async Task SendMessageAsync(ChannelMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
try
|
||||
{
|
||||
var targetChatId = long.Parse(ChannelId);
|
||||
await botClient.SendMessage(targetChatId, message.Content, cancellationToken: cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Failed to send message to Telegram chat {ChatId}", ChannelId);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task SendStreamingMessageAsync(
|
||||
IAsyncEnumerable<ChatStreamUpdate> messageStream,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
var targetChatId = long.Parse(ChannelId);
|
||||
int? currentMessageId = null;
|
||||
|
||||
var typingCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
var typingTask = KeepTypingAsync(targetChatId, typingCts.Token);
|
||||
|
||||
var messageBuilder = new StringBuilder();
|
||||
await foreach (var update in messageStream.WithCancellation(cancellationToken))
|
||||
{
|
||||
if (update.Content is not null)
|
||||
messageBuilder.Append(update.Content);
|
||||
|
||||
if (messageBuilder.Length <= TelegramMessageCharLimit)
|
||||
continue;
|
||||
|
||||
try
|
||||
{
|
||||
var currentMessageContent = messageBuilder.ToString();
|
||||
var splitByParagraph = currentMessageContent.Split('\n');
|
||||
|
||||
var i = 0;
|
||||
var builder = new StringBuilder();
|
||||
foreach (var paragraph in splitByParagraph)
|
||||
{
|
||||
if (builder.Length + paragraph.Length < TelegramMessageCharLimit)
|
||||
{
|
||||
builder.AppendLine(paragraph);
|
||||
i++;
|
||||
continue;
|
||||
}
|
||||
|
||||
await botClient.SendMessage(
|
||||
targetChatId,
|
||||
builder.ToString(),
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
messageBuilder.Clear();
|
||||
messageBuilder.Append(string.Join('\n', splitByParagraph[i..]));
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await typingCts.CancelAsync();
|
||||
logger.LogError(ex, "Failed to send streaming update to Telegram chat {ChatId}", ChannelId);
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
if (messageBuilder.Length > 0)
|
||||
{
|
||||
await botClient.SendMessage(
|
||||
targetChatId,
|
||||
messageBuilder.ToString(),
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await typingCts.CancelAsync();
|
||||
logger.LogError(ex, "Failed to send streaming update to Telegram chat {ChatId}", ChannelId);
|
||||
}
|
||||
|
||||
await typingCts.CancelAsync();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Called by TelegramAdapter when an incoming message is received for this chat.
|
||||
/// Applies AllowedUserIds filtering and null/empty text filtering before raising the event.
|
||||
/// </summary>
|
||||
public void RaiseMessageReceived(ChannelMessage message)
|
||||
{
|
||||
if (string.IsNullOrEmpty(message.Content))
|
||||
return;
|
||||
|
||||
if (options.AllowedUserIds.Length > 0 &&
|
||||
!options.AllowedUserIds.Contains(message.SenderId))
|
||||
return;
|
||||
|
||||
MessageReceived?.Invoke(this, new ChannelMessageReceivedEventArgs
|
||||
{
|
||||
Channel = this,
|
||||
Message = message,
|
||||
IsHandled = false
|
||||
});
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
IsConnected = false;
|
||||
ConnectionStateChanged?.Invoke(this, new ChannelConnectionEventArgs
|
||||
{
|
||||
Channel = this,
|
||||
IsConnected = false
|
||||
});
|
||||
}
|
||||
|
||||
private async Task KeepTypingAsync(long chatId, CancellationToken ct)
|
||||
{
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await botClient.SendChatAction(chatId, ChatAction.Typing, null, null, ct);
|
||||
await Task.Delay(4000, ct);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
break;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
logger.LogError(ex, "Failed to send typing action to Telegram chat {ChatId}", chatId);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
using Luna.Channels.Abstractions;
|
||||
using Luna.Core.Abstractions;
|
||||
using Luna.Shared;
|
||||
|
||||
namespace Luna.Channels.Web;
|
||||
|
||||
public class WebChannel(string channelId, IChatHubContext hubContext) : IChannel
|
||||
{
|
||||
public string ChannelId { get; } = channelId;
|
||||
public string ChannelType => Abstractions.ChannelType.Web;
|
||||
public string DisplayName { get; } = "Web Interface";
|
||||
public bool IsConnected { get; }
|
||||
|
||||
public event ChannelMessageReceivedEventHandler? MessageReceived;
|
||||
public event ChannelConnectionChangedEventHandler? ConnectionStateChanged;
|
||||
|
||||
public async Task SendStreamingMessageAsync(IAsyncEnumerable<ChatStreamUpdate> messageStream, CancellationToken cancellationToken = default)
|
||||
{
|
||||
await hubContext.SendStreamingResponseAsync(ChannelId, messageStream, cancellationToken);
|
||||
}
|
||||
|
||||
public async Task SendMessageAsync(ChannelMessage message, CancellationToken cancellationToken = default)
|
||||
{
|
||||
await hubContext.SendResponseAsync(ChannelId, message, cancellationToken);
|
||||
}
|
||||
|
||||
public void RaiseMessageReceived(object? sender, ChannelMessage message)
|
||||
{
|
||||
MessageReceived?.Invoke(sender, new ChannelMessageReceivedEventArgs
|
||||
{
|
||||
Channel = this,
|
||||
Message = message,
|
||||
IsHandled = false
|
||||
});
|
||||
}
|
||||
|
||||
public void Dispose() { }
|
||||
}
|
||||
Reference in New Issue
Block a user