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;

/// <summary>
/// Handles errors that occur during RabbitMQ message consumption and publishes them using the configured publisher service.
/// </summary>
/// <!-- aidoc:v1 sig=5b1d5b6 -->
public class RabbitConsumerErrorHandler(IPublisherService publisherService)
{
    private const int MaxRetries = 2;
    private static readonly ILogger Logger = Log.ForContext<RabbitConsumerErrorHandler>();

    /// <summary>
    /// Processes a received message by invoking the next handler in the pipeline, and on failure logs the error,
    /// serializes the message body to a string, and delegates to the retry handler using the message properties,
    /// received info, and the captured exception.
    /// </summary>
    /// <param name="message">The deserialized message whose body is inspected or re-serialized for retry handling.</param>
    /// <param name="receivedInfo">Information about the message receipt, passed to the retry handler.</param>
    /// <param name="next">The asynchronous delegate representing the next step in the processing pipeline.</param>
    /// <!-- aidoc:v1 sig=f916bf0 body=5a18c25 -->
    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);
            }
        }

    /// <summary>
    /// Handles retry processing for a received message, either forwarding it to an error queue when the maximum retry count is exceeded or republishing it to the original queue with an incremented retry counter.
    /// </summary>
    /// <param name="receivedInfo">Metadata about the message origin, used to determine the source queue and to build the error message.</param>
    /// <param name="properties">The current <see cref="MessageProperties"/> of the message; its headers are copied so the retry count can be updated without mutating the original instance.</param>
    /// <param name="body">The raw message payload that will be resent to the queue or forwarded to the error queue.</param>
    /// <param name="exception">The <see cref="Exception"/> that triggered the retry, included when building the error message.</param>
    /// <exception cref="Exception">Thrown when the republish to the original queue fails (i.e. <see cref="publisherService"/>.SendMessage returns <c>false</c>). The exception is caught and logged internally.</exception>
    /// <!-- aidoc:v1 sig=d9b4e83 body=c5b7128 -->
    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");
        }
    }

    /// <summary>
    /// Retrieves the retry count from the message properties headers. Returns the deserialized integer value when a "retries" header containing a byte array is present, or 0 when the headers are missing, the key is not found, or the value is not a byte array.
    /// </summary>
    /// <param name="properties">The message properties whose headers are inspected for the "retries" entry.</param>
    /// <returns>The number of retries parsed from the "retries" header, or 0 if the header is absent or not a byte array.</returns>
    /// <!-- aidoc:v1 sig=d7573fc body=b1a585f -->
    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;
        }

    /// <summary>
    /// Constructs a new error message that preserves the original message properties and attaches exception details for downstream processing or dead-letter handling.
    /// </summary>
    /// <param name="receivedInfo">The metadata of the original received message, including exchange, routing key, and queue information used to identify the error source.</param>
    /// <param name="originalProperties">The properties of the original message from which delivery mode, content type, correlation id, and message id are copied.</param>
    /// <param name="body">The raw body of the original message that caused the error.</param>
    /// <param name="exception">The exception that was raised, whose message is included in the error payload.</param>
    /// <param name="headers">The headers to associate with the resulting error message.</param>
    /// <returns>A <see cref="Message{Error}"/> containing the constructed error and the propagated message properties.</returns>
    /// <!-- aidoc:v1 sig=250e4be body=a4fe29e -->
    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);
        }
}