Merge pull request 'feat: expose terminal consumer failure hook' (#3) from IT-1033 into main

Reviewed-on: #3

AGENTS.md

@@ -0,0 +1,26 @@
+# HrynCo.RabbitMq Agent Rules
+
+## Git workflow
+
+- Treat `main` as the only default and integration branch for this repository.
+- Start every task branch from the latest `origin/main`.
+- Before creating a task branch, fetch the remote, switch to `main`, and fast-forward it from `origin/main`.
+- Create the task branch only after confirming that local `main` matches `origin/main`.
+- Open pull requests from the task branch into `main`.
+- Do not use `development` as the base or pull-request target for new work in this repository.
+- Do not commit, push, publish a NuGet package, or merge unless the repository owner explicitly requests that step.
+
+Recommended branch preparation:
+
+```text
+git fetch origin
+git switch main
+git pull --ff-only origin main
+git switch -c <task-branch>
+```
+
+## Delivery
+
+- Keep changes focused and backward compatible where practical because this repository produces a shared NuGet package.
+- Run the solution tests before handoff.
+- Use Conventional Commit messages with a lowercase subject and finish the commit body with `Ref: <issue-id>` when an issue exists.

Hrynco.RabbitMq.Tests/RabbitMqConsumerBaseExtensionTests.cs

@@ -0,0 +1,206 @@
+namespace Hrynco.RabbitMq.Tests;
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using FluentAssertions;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using Xunit;
+
+public sealed class RabbitMqConsumerBaseExtensionTests
+{
+    private static readonly RabbitMqMessageContext Context = new()
+    {
+        QueueName = "messages.test",
+        Exchange = string.Empty,
+        RoutingKey = "messages.test",
+        MessageId = "message-123",
+        MessageType = "Example.Message.v1",
+        CorrelationId = "correlation-123",
+        ReplyTo = "messages.reply",
+        ContentType = "application/json",
+        Redelivered = true
+    };
+
+    [Fact]
+    public async Task ContextAwareOverload_ReceivesTransportMetadata()
+    {
+        var consumer = new ContextAwareConsumer(CreateOptions());
+        var message = CreateMessage();
+
+        await consumer.InvokeAsync(message, Context);
+
+        consumer.HandledMessage.Should().BeSameAs(message);
+        consumer.HandledContext.Should().BeSameAs(Context);
+    }
+
+    [Fact]
+    public async Task ContextAwareOverload_DelegatesToLegacyOverride()
+    {
+        var consumer = new LegacyConsumer(CreateOptions());
+        var message = CreateMessage();
+
+        await consumer.InvokeAsync(message, Context);
+
+        consumer.HandledMessage.Should().BeSameAs(message);
+    }
+
+    [Fact]
+    public void ValidationHook_IsValidByDefault()
+    {
+        var consumer = new LegacyConsumer(CreateOptions());
+
+        bool isValid = consumer.Validate(CreateMessage(), Context, out string? error);
+
+        isValid.Should().BeTrue();
+        error.Should().BeNull();
+    }
+
+    [Fact]
+    public async Task RetriesExhaustedHook_ReceivesMessageContextAndTerminalException()
+    {
+        var consumer = new TerminalFailureConsumer(CreateOptions());
+        var message = CreateMessage();
+        var exception = new InvalidOperationException("terminal failure");
+
+        await consumer.InvokeRetriesExhaustedAsync(message, Context, exception);
+
+        consumer.FailedMessage.Should().BeSameAs(message);
+        consumer.FailedContext.Should().BeSameAs(Context);
+        consumer.TerminalException.Should().BeSameAs(exception);
+    }
+
+    [Fact]
+    public async Task RetriesExhaustedHook_IsNoOpByDefault()
+    {
+        var consumer = new LegacyConsumer(CreateOptions());
+
+        Func<Task> act = () => consumer.InvokeRetriesExhaustedAsync(
+            CreateMessage(),
+            Context,
+            new InvalidOperationException("terminal failure"));
+
+        await act.Should().NotThrowAsync();
+    }
+
+    private static TestMessage CreateMessage() => new()
+    {
+        CorrelationContext = new CorrelationContext { CorrelationId = "payload-correlation" },
+        Data = "payload"
+    };
+
+    private static IOptionsMonitor<RabbitMqSettings> CreateOptions()
+    {
+        return new TestOptionsMonitor(new RabbitMqSettings
+        {
+            Host = "localhost",
+            User = "guest",
+            Password = "guest"
+        });
+    }
+
+    private sealed class ContextAwareConsumer(IOptionsMonitor<RabbitMqSettings> options)
+        : RabbitMqConsumerBase<TestMessage, string>(options, NullLogger.Instance)
+    {
+        protected override string QueueName => "messages.test";
+        public TestMessage? HandledMessage { get; private set; }
+        public RabbitMqMessageContext? HandledContext { get; private set; }
+
+        protected override Task HandleMessageAsync(
+            TestMessage message,
+            RabbitMqMessageContext context,
+            CancellationToken cancellationToken)
+        {
+            HandledMessage = message;
+            HandledContext = context;
+            return Task.CompletedTask;
+        }
+
+        public Task InvokeAsync(TestMessage message, RabbitMqMessageContext context)
+        {
+            return HandleMessageAsync(message, context, CancellationToken.None);
+        }
+    }
+
+    private sealed class LegacyConsumer(IOptionsMonitor<RabbitMqSettings> options)
+        : RabbitMqConsumerBase<TestMessage, string>(options, NullLogger.Instance)
+    {
+        protected override string QueueName => "messages.test";
+        public TestMessage? HandledMessage { get; private set; }
+
+        protected override Task HandleMessageAsync(TestMessage message, CancellationToken cancellationToken)
+        {
+            HandledMessage = message;
+            return Task.CompletedTask;
+        }
+
+        public Task InvokeAsync(TestMessage message, RabbitMqMessageContext context)
+        {
+            return HandleMessageAsync(message, context, CancellationToken.None);
+        }
+
+        public bool Validate(TestMessage message, RabbitMqMessageContext context, out string? error)
+        {
+            return TryValidateMessage(message, context, out error);
+        }
+
+        public Task InvokeRetriesExhaustedAsync(
+            TestMessage message,
+            RabbitMqMessageContext context,
+            Exception exception)
+        {
+            return HandleMessageRetriesExhaustedAsync(
+                message,
+                context,
+                exception,
+                CancellationToken.None);
+        }
+    }
+
+    private sealed class TerminalFailureConsumer(IOptionsMonitor<RabbitMqSettings> options)
+        : RabbitMqConsumerBase<TestMessage, string>(options, NullLogger.Instance)
+    {
+        protected override string QueueName => "messages.test";
+        public TestMessage? FailedMessage { get; private set; }
+        public RabbitMqMessageContext? FailedContext { get; private set; }
+        public Exception? TerminalException { get; private set; }
+
+        protected override Task HandleMessageRetriesExhaustedAsync(
+            TestMessage message,
+            RabbitMqMessageContext context,
+            Exception exception,
+            CancellationToken cancellationToken)
+        {
+            FailedMessage = message;
+            FailedContext = context;
+            TerminalException = exception;
+            return Task.CompletedTask;
+        }
+
+        public Task InvokeRetriesExhaustedAsync(
+            TestMessage message,
+            RabbitMqMessageContext context,
+            Exception exception)
+        {
+            return HandleMessageRetriesExhaustedAsync(
+                message,
+                context,
+                exception,
+                CancellationToken.None);
+        }
+    }
+
+    private sealed record TestMessage : IRabbitMqMessage<string>
+    {
+        public CorrelationContext CorrelationContext { get; set; } = null!;
+        public string Data { get; set; } = string.Empty;
+    }
+
+    private sealed class TestOptionsMonitor(RabbitMqSettings settings) : IOptionsMonitor<RabbitMqSettings>
+    {
+        public RabbitMqSettings CurrentValue => settings;
+        public RabbitMqSettings Get(string? name) => settings;
+        public IDisposable? OnChange(Action<RabbitMqSettings, string?> listener) => null;
+    }
+}

Hrynco.RabbitMq/README.md

@@ -6,10 +6,29 @@ RabbitMQ publisher and consumer base for HrynCo applications.
 
 - `RabbitMqSettings` — connection settings record (host, port, user, password, virtual host)
 - `IRabbitMqPublisher` / `RabbitMqPublisher` — publishes JSON-serialized messages to a named queue
-- `RabbitMqConsumerBase<TMessage, TMessageData>` — abstract background service base for consumers, with retry + dead-letter support
+- `RabbitMqConsumerBase<TMessage, TMessageData>` — background service base with connection management, manual ACK/NACK, retry, permanent-validation rejection, and backward-compatible context-aware handling
+- `RabbitMqMessageContext` — neutral AMQP delivery metadata (`MessageId`, type, correlation, routing, headers, and redelivery state)
 - `IRabbitMqMessage<TMessageData>` — message contract interface
 - `CorrelationContext` — correlation ID carrier
 
 ## Packaging
 
 This package is intended for reuse through NuGet. The test project is excluded from packing.
+
+## Consumer extension points
+
+Existing consumers can keep overriding `HandleMessageAsync(message, cancellationToken)`.
+Consumers that need AMQP metadata can instead override
+`HandleMessageAsync(message, context, cancellationToken)`. The base class retains
+ownership of acknowledgements and retries.
+
+Override `TryValidateMessage(...)` for application-specific permanent validation.
+Returning `false` nacks the delivery without requeue before retry processing begins.
+Keep validation errors free of credentials and sensitive payload values.
+
+Override `HandleMessageRetriesExhaustedAsync(...)` when a consumer must react once to a
+terminal processing failure, for example by publishing a neutral failure result to the
+requesting client. The hook runs after the final handler exception and before the original
+delivery is nacked without requeue. Hook failures are logged and do not replace the
+original failure or acknowledgement policy. Application shutdown cancellation does not
+invoke the hook.

Hrynco.RabbitMq/RabbitMqConsumerBase.cs

@@ -1,6 +1,7 @@
 namespace Hrynco.RabbitMq;
 
 using System;
+using System.Collections.Generic;
 using System.Text;
 using System.Text.Json;
 using System.Threading;
@@ -45,7 +46,55 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
 
     private RabbitMqSettings Settings => _options.Get(SettingsName);
 
-    protected abstract Task HandleMessageAsync(TMessage message, CancellationToken cancellationToken);
+    /// <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)
     {
@@ -88,19 +137,47 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
             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, cancellationToken);
+                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, message.CorrelationContext?.CorrelationId, RetryDelay.TotalSeconds);
+                    attempt, MaxRetries, QueueName, payloadCorrelationId, RetryDelay.TotalSeconds);
 
                 await Task.Delay(RetryDelay, cancellationToken);
             }
@@ -108,13 +185,60 @@ public abstract class RabbitMqConsumerBase<TMessage, TMessageData> : BackgroundS
             {
                 _logger.LogError(ex,
                     "All {Max} attempts exhausted for message on queue {Queue} [CorrelationId={CorrelationId}] — nacking without requeue",
-                    MaxRetries, QueueName, message.CorrelationContext?.CorrelationId);
+                    MaxRetries, QueueName, payloadCorrelationId);
 
-                await _channel!.BasicNackAsync(args.DeliveryTag, multiple: false, requeue: false, cancellationToken: cancellationToken);
+                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;

Hrynco.RabbitMq/RabbitMqMessageContext.cs

@@ -0,0 +1,23 @@
+namespace Hrynco.RabbitMq;
+
+using System.Collections.Generic;
+
+/// <summary>
+/// Transport metadata supplied by RabbitMQ for a delivered message.
+/// Application-specific consumers can use this context for validation,
+/// correlation, and diagnostics without taking ownership of acknowledgements.
+/// </summary>
+public sealed record RabbitMqMessageContext
+{
+    public required string QueueName { get; init; }
+    public required string Exchange { get; init; }
+    public required string RoutingKey { get; init; }
+    public string? MessageId { get; init; }
+    public string? MessageType { get; init; }
+    public string? CorrelationId { get; init; }
+    public string? ReplyTo { get; init; }
+    public string? ContentType { get; init; }
+    public bool Redelivered { get; init; }
+    public IReadOnlyDictionary<string, object?> Headers { get; init; }
+        = new Dictionary<string, object?>();
+}