Playa/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs

using System.Net;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Telegram.Bot;
using Telegram.Bot.Args;
using Telegram.Bot.Exceptions;
using Telegram.Bot.Polling;
using Telegram.Bot.Requests;
using Telegram.Bot.Types;
using Telegram.Bot.Types.Enums;

namespace Volcanic.AgenticCore.Telegram.Polling;

/// <summary>Receives one update. Must return quickly: the polling loop waits for it before the next update.</summary>
public interface IUpdateRouter
{
    Task RouteAsync(Update update, CancellationToken cancellationToken);
}

/// <summary>
/// Runs Telegram.Bot's built-in long-polling loop and makes it survive connection loss.
/// <para>
/// The built-in loop polls again right after a failed <c>getUpdates</c> (a busy loop while offline) and ends for good
/// if the error handler throws. So the error handler here never throws and waits out a backoff delay, which is what
/// holds the loop back; a successful <c>getUpdates</c> response resets the backoff. <c>401</c> is the only fatal error.
/// If the loop still ends while the host runs, it is restarted from the last offset, so no update is lost or repeated.
/// </para>
/// </summary>
public sealed class PollingService(
    ITelegramBotClient bot,
    IUpdateRouter router,
    ReconnectPolicy policy,
    IHostApplicationLifetime lifetime,
    ILogger<PollingService> logger) : BackgroundService, IUpdateHandler
{
    internal static readonly UpdateType[] AllowedUpdates = [UpdateType.Message, UpdateType.CallbackQuery, UpdateType.MyChatMember];

    private int _offset;

    /// <summary>The next update id to request; everything below it has been handed to the router.</summary>
    public int Offset => Volatile.Read(ref _offset);

    /// <summary>Set when polling stopped the application because retrying cannot help (a rejected token).</summary>
    public Exception? FatalError { get; private set; }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        bot.OnApiResponseReceived += OnApiResponseReceived;
        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    logger.LogInformation("Polling Telegram for updates");
                    await bot.ReceiveAsync(this, new ReceiverOptions { Offset = Offset, AllowedUpdates = AllowedUpdates }, stoppingToken)
                        .ConfigureAwait(false);
                }
                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
                {
                    return;
                }
                catch (FatalPollingException ex)
                {
                    logger.LogCritical(ex.InnerException, "{Reason}. Stopping.", ex.Message);
                    FatalError = ex;
                    lifetime.StopApplication();
                    return;
                }
                catch (Exception ex)
                {
                    logger.LogError(ex, "The Telegram polling loop failed; restarting it");
                }

                if (stoppingToken.IsCancellationRequested) return;
                if (!await DelayAsync(policy.NextDelay(), stoppingToken).ConfigureAwait(false)) return;
            }
        }
        finally
        {
            bot.OnApiResponseReceived -= OnApiResponseReceived;
        }
    }

    public async Task HandleUpdateAsync(ITelegramBotClient botClient, Update update, CancellationToken cancellationToken)
    {
        Volatile.Write(ref _offset, update.Id + 1);
        MarkConnected();
        try
        {
            await router.RouteAsync(update, cancellationToken).ConfigureAwait(false);
        }
        catch (Exception ex) when (ex is not OperationCanceledException)
        {
            logger.LogError(ex, "Handling update {UpdateId} failed", update.Id);
        }
    }

    public async Task HandleErrorAsync(ITelegramBotClient botClient, Exception exception, HandleErrorSource source, CancellationToken cancellationToken)
    {
        if (source != HandleErrorSource.PollingError)
        {
            logger.LogError(exception, "Telegram update handling failed ({Source})", source);
            return;
        }

        if (exception is ApiRequestException { ErrorCode: (int)HttpStatusCode.Unauthorized })
            throw new FatalPollingException("Telegram rejected the bot token (401 Unauthorized)", exception);

        var delay = policy.NextDelay();
        var failures = policy.Failures;
        if (exception is ApiRequestException { ErrorCode: (int)HttpStatusCode.Conflict })
            logger.LogWarning("Telegram reports another poller or a webhook for this bot (409): {Message}. Retrying in {Delay:0.#} s",
                exception.Message, delay.TotalSeconds);
        else if (failures == 1)
            // A single failure is usually the long-poll timeout racing the HTTP timeout; only repeated ones matter.
            logger.LogDebug(exception, "Polling Telegram failed; retrying in {Delay:0.#} s", delay.TotalSeconds);
        else
            logger.LogWarning("Telegram unreachable (attempt {Attempt}): {Message}. Retrying in {Delay:0.#} s",
                failures, exception.Message, delay.TotalSeconds);

        await DelayAsync(delay, cancellationToken).ConfigureAwait(false);
    }

    private ValueTask OnApiResponseReceived(ITelegramBotClient botClient, ApiResponseEventArgs args, CancellationToken cancellationToken)
    {
        if (args.ApiRequestEventArgs.Request is GetUpdatesRequest && args.ResponseMessage.IsSuccessStatusCode) MarkConnected();
        return ValueTask.CompletedTask;
    }

    private void MarkConnected()
    {
        var failures = policy.Reset();
        if (failures > 1) logger.LogInformation("Telegram connection restored after {Failures} failed attempts", failures);
    }

    /// <returns>False when the host is stopping.</returns>
    private static async Task<bool> DelayAsync(TimeSpan delay, CancellationToken cancellationToken)
    {
        try
        {
            await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
            return true;
        }
        catch (OperationCanceledException)
        {
            return false;
        }
    }
}

/// <summary>A polling error retrying cannot fix.</summary>
public sealed class FatalPollingException(string message, Exception inner) : Exception(message, inner);