diff --git a/Luna.Channels.Abstractions/ChannelConnectionEventArgs.cs b/Luna.Channels.Abstractions/ChannelConnectionEventArgs.cs new file mode 100644 index 0000000..d5b1cbf --- /dev/null +++ b/Luna.Channels.Abstractions/ChannelConnectionEventArgs.cs @@ -0,0 +1,27 @@ +namespace Luna.Channels.Abstractions; + +/// +/// Event arguments for channel connection state changes +/// +public class ChannelConnectionEventArgs : EventArgs +{ + /// + /// The channel whose connection state changed + /// + public required IChannel Channel { get; init; } + + /// + /// The previous connection state + /// + public bool WasConnected { get; init; } + + /// + /// The new connection state + /// + public bool IsConnected { get; init; } + + /// + /// Optional error message if connection failed + /// + public string? ErrorMessage { get; init; } +} \ No newline at end of file diff --git a/Luna.Channels.Abstractions/ChannelMessageReceivedEventArgs.cs b/Luna.Channels.Abstractions/ChannelMessageReceivedEventArgs.cs new file mode 100644 index 0000000..7061719 --- /dev/null +++ b/Luna.Channels.Abstractions/ChannelMessageReceivedEventArgs.cs @@ -0,0 +1,25 @@ +using Luna.Shared; + +namespace Luna.Channels.Abstractions; + +/// +/// Event arguments for when a message is received from a channel +/// +public class ChannelMessageReceivedEventArgs : EventArgs +{ + /// + /// The received message + /// + public required ChannelMessage Message { get; init; } + + /// + /// The channel that received the message + /// + public required IChannel Channel { get; init; } + + /// + /// Set to true to indicate the message has been handled and should not be processed further + /// + public bool IsHandled { get; set; } + +} diff --git a/Luna.Channels.Abstractions/ChannelType.cs b/Luna.Channels.Abstractions/ChannelType.cs new file mode 100644 index 0000000..d135f14 --- /dev/null +++ b/Luna.Channels.Abstractions/ChannelType.cs @@ -0,0 +1,8 @@ +namespace Luna.Channels.Abstractions; + +public static class ChannelType +{ + public const string Cli = "cli"; + public const string Web = "web"; + public const string Telegram = "telegram"; +} \ No newline at end of file diff --git a/Luna.Channels.Abstractions/IChannel.cs b/Luna.Channels.Abstractions/IChannel.cs new file mode 100644 index 0000000..00f133a --- /dev/null +++ b/Luna.Channels.Abstractions/IChannel.cs @@ -0,0 +1,60 @@ +using Luna.Shared; + +namespace Luna.Channels.Abstractions; + +public delegate Task ChannelMessageReceivedEventHandler(object? sender, ChannelMessageReceivedEventArgs args); +public delegate Task ChannelConnectionChangedEventHandler(object? sender, ChannelConnectionEventArgs args); + +/// +/// Represents a communication channel for the Luna AI Assistant. +/// Channels handle sending and receiving messages via different platforms +/// (e.g., CLI, Telegram, Discord, Email). +/// +public interface IChannel : IDisposable +{ + /// + /// Unique identifier for this channel instance + /// + string ChannelId { get; } + + /// + /// Type of channel (e.g., "cli", "telegram", "discord") + /// + string ChannelType { get; } + + /// + /// Display name for the channel + /// + string DisplayName { get; } + + /// + /// Indicates whether the channel is currently connected and ready + /// + bool IsConnected { get; } + + /// + /// Event raised when a message is received from the channel + /// + event ChannelMessageReceivedEventHandler? MessageReceived; + + /// + /// Event raised when the channel connection state changes + /// + event ChannelConnectionChangedEventHandler? ConnectionStateChanged; + + /// + /// Send a message to the channel + /// + /// The message to send + /// Cancellation token + /// Task representing the send operation + Task SendMessageAsync(ChannelMessage message, CancellationToken cancellationToken = default); + + /// + /// Send a streaming response to the channel + /// + /// Async enumerable of response updates + /// Cancellation token + /// Task representing the send operation + Task SendStreamingMessageAsync(IAsyncEnumerable messageStream, CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/Luna.Channels.Abstractions/IChannelManager.cs b/Luna.Channels.Abstractions/IChannelManager.cs new file mode 100644 index 0000000..e586faf --- /dev/null +++ b/Luna.Channels.Abstractions/IChannelManager.cs @@ -0,0 +1,37 @@ +namespace Luna.Channels.Abstractions; + +/// +/// Manages the lifecycle and routing of all registered communication channels. +/// +public interface IChannelManager +{ + /// + /// Registers a channel with this manager. + /// + /// + /// Thrown if a channel with the same ID is already registered. + void RegisterChannel(IChannel channel); + + /// + /// Unregisters a channel by its ID. No-op if the channel is not found. + /// + /// The ID of the channel to remove. + void UnregisterChannel(string channelId); + + /// + /// Retrieves a registered channel by its ID. + /// + /// The ID of the channel to retrieve. + /// The channel, or null if not found. + IChannel? GetChannel(string channelId); + + /// + /// Returns all currently registered channels. + /// + IReadOnlyList GetAllChannels(); + + /// + /// Raised after a message has been received and routed. + /// + event EventHandler? MessageRouted; +} diff --git a/Luna.Channels.Abstractions/Luna.Channels.Abstractions.csproj b/Luna.Channels.Abstractions/Luna.Channels.Abstractions.csproj new file mode 100644 index 0000000..86df37a --- /dev/null +++ b/Luna.Channels.Abstractions/Luna.Channels.Abstractions.csproj @@ -0,0 +1,13 @@ + + + + net10.0 + enable + enable + + + + + + + diff --git a/Luna.Channels/ChannelManager.cs b/Luna.Channels/ChannelManager.cs new file mode 100644 index 0000000..c728370 --- /dev/null +++ b/Luna.Channels/ChannelManager.cs @@ -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 logger) : IChannelManager +{ + private readonly Dictionary channels = new(); + private readonly Dictionary sessionMap = new(); + + public event EventHandler? 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 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); + } + } +} \ No newline at end of file diff --git a/Luna.Channels/Cli/CliChannel.cs b/Luna.Channels/Cli/CliChannel.cs new file mode 100644 index 0000000..b15da0b --- /dev/null +++ b/Luna.Channels/Cli/CliChannel.cs @@ -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 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() { } +} \ No newline at end of file diff --git a/Luna.Channels/Extensions/ServiceCollectionExtensions.cs b/Luna.Channels/Extensions/ServiceCollectionExtensions.cs new file mode 100644 index 0000000..696616f --- /dev/null +++ b/Luna.Channels/Extensions/ServiceCollectionExtensions.cs @@ -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(); + + // Telegram + var telegramOptions = new TelegramOptions { BotToken = TelegramBotToken }; + services.AddSingleton(telegramOptions); + services.AddSingleton(_ => new TelegramBotClient(TelegramBotToken)); + services.AddHostedService(); + + return services; + } + } +} diff --git a/Luna.Channels/Luna.Channels.csproj b/Luna.Channels/Luna.Channels.csproj new file mode 100644 index 0000000..48e9cc5 --- /dev/null +++ b/Luna.Channels/Luna.Channels.csproj @@ -0,0 +1,22 @@ + + + + net10.0 + enable + enable + + + + + + + + + + + + + + + + diff --git a/Luna.Channels/Telegram/TelegramAdapter.cs b/Luna.Channels/Telegram/TelegramAdapter.cs new file mode 100644 index 0000000..7d98e67 --- /dev/null +++ b/Luna.Channels/Telegram/TelegramAdapter.cs @@ -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 adapterLogger, + ILogger channelLogger) : IHostedService, IUpdateHandler +{ + private readonly Dictionary 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); + } +} diff --git a/Luna.Channels/Telegram/TelegramChannel.cs b/Luna.Channels/Telegram/TelegramChannel.cs new file mode 100644 index 0000000..40b18d1 --- /dev/null +++ b/Luna.Channels/Telegram/TelegramChannel.cs @@ -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 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 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(); + } + + /// + /// Called by TelegramAdapter when an incoming message is received for this chat. + /// Applies AllowedUserIds filtering and null/empty text filtering before raising the event. + /// + 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); + } + } + } +} diff --git a/Luna.Channels/Web/WebChannel.cs b/Luna.Channels/Web/WebChannel.cs new file mode 100644 index 0000000..09a01c5 --- /dev/null +++ b/Luna.Channels/Web/WebChannel.cs @@ -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 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() { } +} \ No newline at end of file