telegram-bot/modules/RichSiteSummary/Service/MessageSender.cs

using System;
using System.Collections.Concurrent;
using System.Text;
using System.Threading.Tasks;
using BotFramework;
using BotFramework.Abstractions;
using Kruzya.TelegramBot.Core.Service;
using Microsoft.Extensions.Logging;
using Telegram.Bot;
using Telegram.Bot.Exceptions;
using Telegram.Bot.Types;

namespace Kruzya.TelegramBot.RichSiteSummary.Service
{
    public class MessageSender : AbstractTimedHostedService
    {
        protected override TimeSpan TimerPeriod => TimeSpan.FromSeconds(1);
        
        protected readonly ConcurrentQueue<UserMessage> _queue;
        protected readonly ConcurrentQueue<UserUnsubscribe> _unsubscribeQueue;
        
        public MessageSender(ILogger<MessageSender> logger, IBotInstance bot, ConcurrentQueue<UserMessage> userMessageQueue, ConcurrentQueue<UserUnsubscribe> userUnsubscribeQueue) : base(logger, bot)
        {
            _unsubscribeQueue = userUnsubscribeQueue;
            _queue = userMessageQueue;
        }

        protected override async Task OnRun()
        {
            if (_queue.Count == 0)
            {
                return;
            }

            UserMessage message;
            if (!_queue.TryDequeue(out message))
            {
                return;
            }

            _logger.LogDebug($"Message for {message.ChatId.Identifier} dequeued.");
            try
            {
                if (string.IsNullOrWhiteSpace(message.ImageUrl))
                    await _bot.BotClient.SendTextMessageAsync(message.ChatId, message.Text, message.ParseMode, null,
                        message.DisableWebPagePreview);
                else
                    await _bot.BotClient.SendPhotoAsync(message.ChatId, message.ImageUrl, message.Text,
                        message.ParseMode, null, message.DisableWebPagePreview);
            }
            catch (ApiRequestException e)
            {
                _logger.LogDebug($"Message for {message.ChatId.Identifier} is failed: {e.Message}.");
                if (!e.Message.Contains("bot was blocked by user"))
                {
                    ReEnqueue(message, e, message.ChatId); // looks like a network issue
                    return;
                }

                // User added bot to blacklist.
                // Unsubscribe him.
                _unsubscribeQueue.Enqueue(new UserUnsubscribe()
                {
                    ChatId = message.ChatId
                });
            }
            catch (Exception e)
            {
                _logger.LogDebug($"Message for {message.ChatId.Identifier} is failed: {e.Message}.");
                ReEnqueue(message, e);
            }
        }
        
        private void ReEnqueue(UserMessage message, Exception e = null, ChatId chatId = null)
        {
            if (e != null)
            {
                var eMessage = new StringBuilder();
                eMessage.Append(e.Message);
                if (chatId != null)
                {
                    eMessage.Append($" ({chatId.Identifier})");
                }

                _logger.LogError(eMessage.ToString());
            }

            _queue.Enqueue(message);
        }
    }
}