NATS-Powered Telegram Notification System

public interface IEventSubjectProvider
{
    string EnvironmentPrefix { get; }
    
    // Document lifecycle subjects
    string OcrRequested { get; }
    string OcrCompleted { get; }
    string EmbeddingsCompleted { get; }
    
    // Notification subjects
    string NotificationTelegram { get; }
    string NotificationEmail { get; }
    string NotificationWebhook { get; }
    
    // Telegram link subjects (NEW)
    string TelegramLinkTokenValidate { get; }
    string TelegramPushLinkRequest { get; }
    
    string ApplyPrefix(string subject);
}

public class EventSubjectProvider : IEventSubjectProvider
{
    private readonly EnvironmentOptions _options;

    public EventSubjectProvider(IOptions<EnvironmentOptions> options)
    {
        _options = options.Value;

        // Pre-compute all static subjects for performance
        OcrRequested = _options.ApplySubjectPrefix("archives.documents.ocr.requested");
        OcrCompleted = _options.ApplySubjectPrefix("archives.documents.ocr.completed");
        EmbeddingsCompleted = _options.ApplySubjectPrefix("archives.documents.embeddings.completed");
        
        // Notification subjects
        NotificationTelegram = _options.ApplySubjectPrefix("notifications.telegram");
        NotificationEmail = _options.ApplySubjectPrefix("notifications.email");
        NotificationWebhook = _options.ApplySubjectPrefix("notifications.webhook");
        
        // Telegram link subjects
        TelegramLinkTokenValidate = _options.ApplySubjectPrefix("telegram.link-token.validate");
        TelegramPushLinkRequest = _options.ApplySubjectPrefix("telegram.push-link.request");
    }

    public string EnvironmentPrefix => _options.Prefix;
    
    public string ApplyPrefix(string subject) => _options.ApplySubjectPrefix(subject);
    
    // Cached properties...
    public string OcrRequested { get; }
    public string NotificationTelegram { get; }
    public string TelegramLinkTokenValidate { get; }
    public string TelegramPushLinkRequest { get; }
    // etc.
}

public class TelegramNotificationService : INotificationService
{
    private readonly IMessagingConnection _messaging;
    private readonly IUserRepository _userRepository;
    private readonly IEventSubjectProvider _subjects;
    private readonly ILogger<TelegramNotificationService> _logger;

    public TelegramNotificationService(
        IUserRepository userRepository,
        IMessagingConnection messaging,
        IEventSubjectProvider subjects,
        ILogger<TelegramNotificationService> logger)
    {
        _userRepository = userRepository;
        _messaging = messaging;
        _subjects = subjects;
        _logger = logger;
    }

    public async Task NotifyDocumentProcessedAsync(
        BlueRobinId userId,
        BlueRobinId documentId,
        string documentName,
        bool success,
        string? errorMessage = null,
        CancellationToken cancellationToken = default)
    {
        var user = await _userRepository.GetByIdAsync(userId, cancellationToken);
        if (user == null) return;

        // Check notification preferences
        if (success && user.Preferences?.NotifyOnDocumentProcessed != true) return;
        if (!success && user.Preferences?.NotifyOnDocumentFailed != true) return;

        var chatId = user.GetTelegramChatId();
        if (!chatId.HasValue) return;

        var notification = new DocumentProcessedNotification
        {
            MessageId = Guid.NewGuid().ToString("N"),
            ChatId = chatId.Value,
            Title = success ? "✅ Document Processed" : "❌ Processing Failed",
            Body = success
                ? $"Your document *{documentName}* has been processed successfully."
                : $"Failed to process *{documentName}*.\n\nError: {errorMessage ?? "Unknown"}",
            UserId = userId.Value,
            DocumentId = documentId.Value,
            DocumentName = documentName,
            Success = success,
            Priority = success ? NotificationPriority.Normal : NotificationPriority.High,
            ParseMarkdown = true
        };

        await PublishTelegramNotificationAsync(notification, cancellationToken);
    }

    private async Task PublishTelegramNotificationAsync(
        TelegramNotificationMessage notification,
        CancellationToken cancellationToken)
    {
        try
        {
            // Use the environment-prefixed subject
            await _messaging.PublishAsync(
                _subjects.NotificationTelegram, 
                notification, 
                cancellationToken);
                
            _logger.LogDebug(
                "Published notification {MessageId} to {Subject} for chat {ChatId}",
                notification.MessageId, 
                _subjects.NotificationTelegram, 
                notification.ChatId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to publish notification {MessageId}", notification.MessageId);
            throw;
        }
    }
}

/// <summary>
/// Background service that listens to NATS Core for outbound Telegram notifications
/// and sends them via the Telegram Bot API.
/// 
/// Uses Core NATS subscription to match the publisher (TelegramNotificationService)
/// which publishes via IMessagingConnection.PublishAsync().
/// Uses IEventSubjectProvider for environment-prefixed subjects.
/// </summary>
public class TelegramNotificationWorker : BackgroundService
{
    private readonly INatsConnection _natsConnection;
    private readonly TelegramBotOptions _options;
    private readonly IEventSubjectProvider _subjects;
    private readonly ILogger<TelegramNotificationWorker> _logger;
    private TelegramBotClient? _botClient;

    public TelegramNotificationWorker(
        INatsConnection natsConnection,
        IOptions<TelegramBotOptions> options,
        IEventSubjectProvider subjects,
        ILogger<TelegramNotificationWorker> logger)
    {
        _natsConnection = natsConnection;
        _options = options.Value;
        _subjects = subjects;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        if (string.IsNullOrWhiteSpace(_options.BotToken))
        {
            _logger.LogError("Telegram bot token not configured");
            return;
        }

        _botClient = new TelegramBotClient(_options.BotToken);

        // Use environment-prefixed notification subject
        var subject = _subjects.NotificationTelegram;
        
        _logger.LogInformation(
            "Telegram notification worker starting, subscribing to {Subject}", 
            subject);

        try
        {
            // Subscribe using Core NATS (matching publisher pattern)
            await foreach (var msg in _natsConnection.SubscribeAsync<TelegramNotificationMessage>(
                subject, cancellationToken: stoppingToken))
            {
                try
                {
                    if (msg.Data != null)
                    {
                        await ProcessNotificationAsync(msg.Data, stoppingToken);
                    }
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error processing notification");
                    // Continue processing other messages
                }
            }
        }
        catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
        {
            _logger.LogInformation("Notification worker stopping...");
        }
    }

    private async Task ProcessNotificationAsync(
        TelegramNotificationMessage notification, 
        CancellationToken cancellationToken)
    {
        if (_botClient == null) return;

        _logger.LogDebug(
            "Sending notification to chat {ChatId}: {Title}",
            notification.ChatId, notification.Title);

        var message = string.IsNullOrEmpty(notification.Title)
            ? notification.Body
            : $"*{notification.Title}*\n\n{notification.Body}";

        await _botClient.SendMessage(
            chatId: notification.ChatId,
            text: message,
            parseMode: notification.ParseMarkdown ? ParseMode.Markdown : ParseMode.None,
            cancellationToken: cancellationToken);

        _logger.LogInformation(
            "Sent notification {MessageId} to chat {ChatId}",
            notification.MessageId, notification.ChatId);
    }
}

apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization

namespace: archives-staging

resources:
  - ../../base

patches:
  - target:
      kind: ConfigMap
      name: archives-telegram-bot-config
    patch: |-
      - op: replace
        path: /data/Environment__Prefix
        value: "staging"
      - op: replace
        path: /data/Telegram__MiniAppUrl
        value: "https://web-staging.bluerobin.local/telegram/setup-passkey"