hrynco-rabbitmq/Hrynco.RabbitMq/RabbitMqConsumerBase.cs

namespace Hrynco.RabbitMq;

using System;
using System.Collections.Generic;
using System.Text;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

/// <summary>
/// Base class for RabbitMQ consumers. Handles connection management, manual ack,
/// and retry with backoff before dead-lettering.
/// Override <see cref="SettingsName"/> to use a named <see cref="RabbitMqSettings"/> instance.
/// </summary>
public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundService
    where TMessage : class, IRabbitMqMessage<TMessageData>
{
    private readonly IOptionsMonitor<RabbitMqSettings> _options;
    private readonly ILogger _logger;

    private IConnection? _connection;
    private IChannel? _channel;

    protected RabbitMqConsumerBase(IOptionsMonitor<RabbitMqSettings> options, ILogger logger)
    {
        _options = options;
        _logger = logger;
    }

    protected abstract string QueueName { get; }

    /// <summary>
    /// Name of the <see cref="RabbitMqSettings"/> instance to use.
    /// Override to use a named instance when multiple RabbitMQ connections are configured.
    /// Defaults to <see cref="Options.DefaultName"/> (the unnamed instance).
    /// </summary>
    protected virtual string SettingsName => Options.DefaultName;

    protected virtual int MaxRetries => 3;
    protected virtual TimeSpan RetryDelay => TimeSpan.FromSeconds(5);

    private RabbitMqSettings Settings => _options.Get(SettingsName);

    /// <summary>
    /// Handles a deserialized message. Existing consumers can continue overriding this overload.
    /// New consumers that need transport metadata can override the context-aware overload instead.
    /// </summary>
    protected virtual Task HandleMessageAsync(TMessage message, CancellationToken cancellationToken)
    {
        throw new NotSupportedException(
            $"{GetType().Name} must override a HandleMessageAsync overload.");
    }

    /// <summary>
    /// Handles a deserialized message together with neutral RabbitMQ delivery metadata.
    /// The default implementation preserves compatibility by delegating to the original overload.
    /// </summary>
    protected virtual Task HandleMessageAsync(
        TMessage message,
        RabbitMqMessageContext context,
        CancellationToken cancellationToken)
    {
        return HandleMessageAsync(message, cancellationToken);
    }

    /// <summary>
    /// Performs application-specific validation before retries begin.
    /// Return false for a permanently invalid message; it will be nacked without requeue.
    /// Validation errors should describe the rule without including sensitive payload values.
    /// </summary>
    protected virtual bool TryValidateMessage(
        TMessage message,
        RabbitMqMessageContext context,
        out string? validationError)
    {
        validationError = null;
        return true;
    }

    /// <summary>
    /// Handles a terminal processing failure after all message retries are exhausted.
    /// The default implementation is a no-op. Implementations should avoid throwing;
    /// failures from this hook are logged and the original message is still nacked.
    /// </summary>
    protected virtual Task HandleMessageRetriesExhaustedAsync(
        TMessage message,
        RabbitMqMessageContext context,
        Exception exception,
        CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await EnsureConnectionAsync(stoppingToken);

        var consumer = new AsyncEventingBasicConsumer(_channel!);

        consumer.ReceivedAsync += async (_, args) =>
        {
            await ProcessMessageAsync(args, stoppingToken);
        };

        await _channel!.BasicConsumeAsync(queue: QueueName, autoAck: false, consumer: consumer,
            cancellationToken: stoppingToken);

        // Hold until cancellation — consumer events fire on the channel thread
        await Task.Delay(Timeout.Infinite, stoppingToken).ConfigureAwait(ConfigureAwaitOptions.SuppressThrowing);
    }

    private async Task ProcessMessageAsync(BasicDeliverEventArgs args, CancellationToken cancellationToken)
    {
        TMessage? message = null;

        try
        {
            var json = Encoding.UTF8.GetString(args.Body.ToArray());
            message = JsonSerializer.Deserialize<TMessage>(json);

            if (message is null)
            {
                _logger.LogWarning("Received null message on queue {Queue} — nacking without requeue", QueueName);
                await _channel!.BasicNackAsync(args.DeliveryTag, multiple: false, requeue: false, cancellationToken: cancellationToken);
                return;
            }
        }
        catch (JsonException ex)
        {
            _logger.LogError(ex, "Failed to deserialize message on queue {Queue} — nacking without requeue", QueueName);
            await _channel!.BasicNackAsync(args.DeliveryTag, multiple: false, requeue: false, cancellationToken: cancellationToken);
            return;
        }

        RabbitMqMessageContext context = CreateMessageContext(args);
        string? payloadCorrelationId = message.CorrelationContext?.CorrelationId;

        using IDisposable? scope = _logger.BeginScope(new Dictionary<string, object?>
        {
            ["MessageId"] = context.MessageId,
            ["MessageType"] = context.MessageType,
            ["CorrelationId"] = payloadCorrelationId ?? context.CorrelationId,
            ["BrokerCorrelationId"] = context.CorrelationId,
            ["Queue"] = context.QueueName,
            ["RoutingKey"] = context.RoutingKey,
            ["Redelivered"] = context.Redelivered
        });

        if (!TryValidateMessage(message, context, out string? validationError))
        {
            _logger.LogWarning(
                "Rejected invalid message on queue {Queue}: {ValidationError} — nacking without requeue",
                QueueName,
                validationError ?? "No validation error was provided");
            await NackWithoutRequeueAsync(args.DeliveryTag, cancellationToken);
            return;
        }

        for (int attempt = 1; attempt <= MaxRetries; attempt++)
        {
            try
            {
                await HandleMessageAsync(message, context, cancellationToken);
                await _channel!.BasicAckAsync(args.DeliveryTag, multiple: false, cancellationToken: cancellationToken);
                return;
            }
            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
            {
                throw;
            }
            catch (Exception ex) when (attempt < MaxRetries)
            {
                _logger.LogWarning(ex,
                    "Attempt {Attempt}/{Max} failed for message on queue {Queue} [CorrelationId={CorrelationId}] — retrying in {Delay}s",
                    attempt, MaxRetries, QueueName, payloadCorrelationId, RetryDelay.TotalSeconds);

                await Task.Delay(RetryDelay, cancellationToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex,
                    "All {Max} attempts exhausted for message on queue {Queue} [CorrelationId={CorrelationId}] — nacking without requeue",
                    MaxRetries, QueueName, payloadCorrelationId);

                try
                {
                    await HandleMessageRetriesExhaustedAsync(
                        message,
                        context,
                        ex,
                        cancellationToken);
                }
                catch (Exception hookException)
                {
                    _logger.LogError(
                        hookException,
                        "Terminal failure hook failed for message on queue {Queue} [CorrelationId={CorrelationId}]",
                        QueueName,
                        payloadCorrelationId);
                }

                await NackWithoutRequeueAsync(args.DeliveryTag, cancellationToken);
            }
        }
    }

    private RabbitMqMessageContext CreateMessageContext(BasicDeliverEventArgs args)
    {
        IReadOnlyDictionary<string, object?> headers = args.BasicProperties.Headers is null
            ? new Dictionary<string, object?>()
            : new Dictionary<string, object?>(args.BasicProperties.Headers);

        return new RabbitMqMessageContext
        {
            QueueName = QueueName,
            Exchange = args.Exchange,
            RoutingKey = args.RoutingKey,
            MessageId = args.BasicProperties.MessageId,
            MessageType = args.BasicProperties.Type,
            CorrelationId = args.BasicProperties.CorrelationId,
            ReplyTo = args.BasicProperties.ReplyTo,
            ContentType = args.BasicProperties.ContentType,
            Redelivered = args.Redelivered,
            Headers = headers
        };
    }

    private Task NackWithoutRequeueAsync(ulong deliveryTag, CancellationToken cancellationToken)
    {
        return _channel!.BasicNackAsync(
            deliveryTag,
            multiple: false,
            requeue: false,
            cancellationToken: cancellationToken).AsTask();
    }

    private async Task EnsureConnectionAsync(CancellationToken cancellationToken)
    {
        var s = Settings;
        var factory = new ConnectionFactory
        {
            HostName = s.Host,
            Port = s.Port,
            UserName = s.User,
            Password = s.Password,
            VirtualHost = s.VirtualHost
        };

        _connection = await factory.CreateConnectionAsync(cancellationToken);
        _channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);

        await _channel.QueueDeclareAsync(QueueName, durable: true, exclusive: false, autoDelete: false,
            cancellationToken: cancellationToken);
        await _channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false,
            cancellationToken: cancellationToken);

        _logger.LogInformation("RabbitMQ consumer connected to queue {Queue} (settings: {SettingsName})",
            QueueName, SettingsName);
    }

    public override void Dispose()
    {
        _channel?.DisposeAsync().AsTask().GetAwaiter().GetResult();
        _connection?.DisposeAsync().AsTask().GetAwaiter().GetResult();
        base.Dispose();
    }
}