adas-core/adas-core.Infrastructure/Utils/RabbitConsumerErrorStrategy.cs

using adas_core.Application.Services.Interfaces;
using EasyNetQ;
using EasyNetQ.SystemMessages;
using Newtonsoft.Json;
using Serilog;
using System.Text;
using ILogger = Serilog.ILogger;

namespace adas_core.Infrastructure.Utils;

public class RabbitConsumerErrorHandler(IPublisherService publisherService)
{
    private const int MaxRetries = 2;
    private static readonly ILogger Logger = Log.ForContext<RabbitConsumerErrorHandler>();

    public async Task HandleAsync<T>(
        Message<T> message,
        MessageReceivedInfo receivedInfo,
        Func<Task> next)
    {
        try
        {
            await next();
        }
        catch (Exception exception)
        {
            Logger.Error(exception, "Consumer error {Message}", exception.Message);

            var properties = message.Properties;

            var body = Encoding.UTF8.GetString(
                message.Body is byte[] bytes
                    ? bytes
                    : Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message.Body)));

            HandleRetries(receivedInfo, properties, body, exception);
        }
    }

    private void HandleRetries(
        MessageReceivedInfo receivedInfo,
        MessageProperties properties,
        string body,
        Exception exception)
    {
        try
        {
            var headers = properties.Headers != null
                ? new Dictionary<string, object>(properties.Headers)
                : [];

            var retries = GetRetries(properties);

            // MAX RETRIES → ERROR QUEUE
            if (retries > MaxRetries)
            {
                if (!receivedInfo.Queue.StartsWith("Error"))
                {
                    headers["retries"] = BitConverter.GetBytes(MaxRetries + 1);

                    var errorMsg = CreateErrorMessage(
                        receivedInfo,
                        properties,
                        body,
                        exception,
                        headers
                    );

                    _ = publisherService.SendMessageError(
                        errorMsg,
                        $"Error{receivedInfo.Queue}");
                }

                return;
            }

            // RETRY NORMAL
            headers["retries"] = BitConverter.GetBytes(retries + 1);

            var newProps = new MessageProperties
            {
                DeliveryMode = 2,
                Headers = headers,
                ContentType = properties.ContentType,
                CorrelationId = properties.CorrelationId,
                MessageId = properties.MessageId
            };

            var message = new Message<string>(body, newProps);

            var result = publisherService
                .SendMessage(message, receivedInfo.Queue)
                .GetAwaiter()
                .GetResult();

            if (!result)
                throw new Exception("Requeue failed");
        }
        catch (Exception e)
        {
            Logger.Error(e, "Exception processing rabbit message");
        }
    }

    private int GetRetries(MessageProperties properties)
    {
        if (properties.Headers != null &&
            properties.Headers.TryGetValue("retries", out var retriesObj) &&
            retriesObj is byte[] bytes)
        {
            return BitConverter.ToInt32(bytes, 0);
        }

        return 0;
    }

    private static Message<Error> CreateErrorMessage(
        MessageReceivedInfo receivedInfo,
        MessageProperties originalProperties,
        string body,
        Exception exception,
        Dictionary<string, object> headers)
    {
        var props = new MessageProperties
        {
            Headers = headers,
            DeliveryMode = originalProperties.DeliveryMode,
            ContentType = originalProperties.ContentType,
            CorrelationId = originalProperties.CorrelationId,
            MessageId = originalProperties.MessageId
        };

        var error = new Error(
            body,
            exception.Message,
            receivedInfo.Exchange,
            receivedInfo.RoutingKey,
            receivedInfo.Queue,
            DateTime.UtcNow,
            props
        );

        return new Message<Error>(error, props);
    }
}