Message Brokers

using System.Text;
using System.Text.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

var builder = Host.CreateApplicationBuilder(args);
var rabbit = builder.Configuration.GetSection("RabbitMq");
var factory = new ConnectionFactory
{
    // Required settings, no guest/guest fallback; the password comes from a secret store.
    HostName = rabbit["Host"] ?? throw new InvalidOperationException("RabbitMq:Host is not configured"),
    UserName = rabbit["User"] ?? throw new InvalidOperationException("RabbitMq:User is not configured"),
    Password = rabbit["Password"] ?? throw new InvalidOperationException("RabbitMq:Password is not configured"),
    ClientProvidedName = "inventory-worker",
};
// One connection for the process, opened asynchronously before the host starts. A broker that is
// down at boot fails start-up so the orchestrator restarts the pod; automatic recovery handles later drops.
await using var connection = await factory.CreateConnectionAsync();
builder.Services.AddSingleton<IConnection>(connection);
builder.Services.AddSingleton<OrderEventPublisher>();
builder.Services.AddHostedService<OrderPlacedConsumer>();
await builder.Build().RunAsync();

public sealed record OrderPlaced(Guid OrderId, int CustomerId, decimal Total, string[] Skus);

public sealed class OrderPlacedConsumer(IConnection connection, ILogger<OrderPlacedConsumer> logger) : BackgroundService
{
    private const string Queue = "inventory.order-placed";

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        await using var channel = await connection.CreateChannelAsync(cancellationToken: stoppingToken);
        await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 20, global: false, stoppingToken);

        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            try
            {
                var message = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, JsonSerializerOptions.Web)
                    ?? throw new JsonException("Empty message");
                await ReserveStockAsync(message, stoppingToken);
                await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
            }
            catch (JsonException ex)
            {
                logger.LogError(ex, "Unreadable message {MessageId}; dead-lettering", ea.BasicProperties.MessageId);
                await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: false);
            }
            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
            {
                logger.LogWarning(ex, "Message {MessageId} failed (redelivered: {Redelivered}); requeueing in 5 s",
                    ea.BasicProperties.MessageId, ea.Redelivered);
                // An immediate requeue comes straight back and burns the delivery limit in milliseconds.
                // Waiting also pauses this consumer, since handlers run one at a time by default.
                await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
                await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: true);
            }
        };

        await channel.BasicConsumeAsync(Queue, autoAck: false, consumer, stoppingToken);
        await Task.Delay(Timeout.Infinite, stoppingToken).ContinueWith(_ => { }, TaskScheduler.Default);
    }

    private Task ReserveStockAsync(OrderPlaced message, CancellationToken ct)
    {
        logger.LogInformation("Reserving {Count} SKUs for order {OrderId}", message.Skus.Length, message.OrderId);
        return Task.CompletedTask;
    }
}

// A singleton: one long-lived confirm channel, reused for every publish.
public sealed class OrderEventPublisher(IConnection connection) : IAsyncDisposable
{
    private readonly SemaphoreSlim _gate = new(1, 1); // IChannel is not for concurrent publishes
    private IChannel? _channel;

    public async Task PublishAsync(Guid eventId, OrderPlaced message, CancellationToken ct)
    {
        var props = new BasicProperties
        {
            Persistent = true,
            MessageId = eventId.ToString(), // the event's own id (the outbox row id), never the order id
            ContentType = "application/json",
            Type = nameof(OrderPlaced),
        };
        var body = Encoding.UTF8.GetBytes(JsonSerializer.Serialize(message, JsonSerializerOptions.Web));

        await _gate.WaitAsync(ct);
        try
        {
            if (_channel is not { IsOpen: true })
            {
                if (_channel is not null) await _channel.DisposeAsync();
                _channel = await connection.CreateChannelAsync(
                    new CreateChannelOptions(publisherConfirmationsEnabled: true, publisherConfirmationTrackingEnabled: true), ct);
            }

            // Completes when the broker confirms; throws if it nacks or the message is unroutable.
            await _channel.BasicPublishAsync("orders", "order.placed", mandatory: true, props, body, ct);
        }
        finally
        {
            _gate.Release();
        }
    }

    public async ValueTask DisposeAsync()
    {
        if (_channel is not null) await _channel.DisposeAsync();
        _gate.Dispose();
    }
}