The Transactional Outbox

using System.Text.Json;
using Microsoft.EntityFrameworkCore;

var builder = WebApplication.CreateBuilder(args);
// AddNpgsqlDbContext in Aspire turns the same retrying strategy on.
builder.Services.AddDbContext<OrdersDb>(o => o.UseNpgsql(
    builder.Configuration.GetConnectionString("Orders"),
    npgsql => npgsql.EnableRetryOnFailure()));
builder.Services.AddSingleton<IMessagePublisher, LoggingPublisher>();
builder.Services.AddHostedService<OutboxRelay>();
var app = builder.Build();

app.MapPost("/orders", async (PlaceOrder cmd, OrdersDb db, CancellationToken ct) =>
{
    var order = new Order { Id = Guid.CreateVersion7(), CustomerId = cmd.CustomerId, Total = cmd.Total, Status = "Placed" };
    db.Orders.Add(order);
    db.OutboxMessages.Add(OutboxMessage.From(new OrderPlaced(order.Id, order.CustomerId, order.Total)));
    await db.SaveChangesAsync(ct); // order and event commit together
    return TypedResults.Created($"/orders/{order.Id}", new { order.Id });
});
app.Run();

public sealed record PlaceOrder(int CustomerId, decimal Total);
public sealed record OrderPlaced(Guid OrderId, int CustomerId, decimal Total);

public sealed class Order
{
    public Guid Id { get; set; }
    public int CustomerId { get; set; }
    public decimal Total { get; set; }
    public string Status { get; set; } = "";
}

public sealed class OutboxMessage
{
    public Guid Id { get; set; }
    public string Type { get; set; } = "";
    public string Payload { get; set; } = "";
    public DateTimeOffset OccurredAt { get; set; }
    public DateTimeOffset? ProcessedAt { get; set; }
    public int Attempts { get; set; }
    public DateTimeOffset? NextAttemptAt { get; set; }
    public string? LastError { get; set; }

    public static OutboxMessage From<T>(T message) => new()
    {
        Id = Guid.CreateVersion7(),
        Type = typeof(T).Name,
        Payload = JsonSerializer.Serialize(message, JsonSerializerOptions.Web),
        OccurredAt = DateTimeOffset.UtcNow,
    };
}

public sealed class OrdersDb(DbContextOptions<OrdersDb> options) : DbContext(options)
{
    public DbSet<Order> Orders => Set<Order>();
    public DbSet<OutboxMessage> OutboxMessages => Set<OutboxMessage>();

    protected override void OnModelCreating(ModelBuilder b)
    {
        b.Entity<Order>().ToTable("orders");
        b.Entity<OutboxMessage>(e =>
        {
            e.ToTable("outbox_messages");
            e.Property(x => x.Id).HasColumnName("id");
            e.Property(x => x.Type).HasColumnName("type").HasMaxLength(200);
            e.Property(x => x.Payload).HasColumnName("payload").HasColumnType("jsonb");
            e.Property(x => x.OccurredAt).HasColumnName("occurred_at");
            e.Property(x => x.ProcessedAt).HasColumnName("processed_at");
            e.Property(x => x.Attempts).HasColumnName("attempts");
            e.Property(x => x.NextAttemptAt).HasColumnName("next_attempt_at");
            e.Property(x => x.LastError).HasColumnName("last_error");
        });
    }
}

public interface IMessagePublisher
{
    Task PublishAsync(string messageId, string type, string payload, CancellationToken ct);
}

public sealed class LoggingPublisher(ILogger<LoggingPublisher> logger) : IMessagePublisher
{
    public Task PublishAsync(string messageId, string type, string payload, CancellationToken ct)
    {
        logger.LogInformation("Published {Type} {MessageId}", type, messageId);
        return Task.CompletedTask;
    }
}

// Thrown by the publisher when the broker itself is unreachable: not the message's fault.
public sealed class BrokerUnavailableException(string message, Exception? inner = null) : Exception(message, inner);

public sealed class OutboxRelay(IServiceScopeFactory scopes, IMessagePublisher publisher, ILogger<OutboxRelay> logger) : BackgroundService
{
    private const int BatchSize = 50;
    private const int MaxAttempts = 10;
    private static readonly TimeSpan PublishTimeout = TimeSpan.FromSeconds(5);

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var brokerBackoff = TimeSpan.FromSeconds(1);
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
        while (await timer.WaitForNextTickAsync(stoppingToken))
        {
            try
            {
                // Drain a backlog without waiting, but only while whole batches publish.
                while (await RelayBatchAsync(stoppingToken) == BatchSize) { }
                brokerBackoff = TimeSpan.FromSeconds(1);
            }
            catch (BrokerUnavailableException ex)
            {
                // No attempts were counted: an outage must not turn the backlog into poison rows.
                logger.LogWarning(ex, "Broker unavailable; pausing the relay for {Delay}", brokerBackoff);
                await Task.Delay(brokerBackoff, stoppingToken);
                brokerBackoff = TimeSpan.FromTicks(Math.Min(brokerBackoff.Ticks * 2, TimeSpan.FromMinutes(1).Ticks));
            }
            catch (Exception ex) when (!stoppingToken.IsCancellationRequested)
            {
                logger.LogError(ex, "Outbox relay pass failed");
            }
        }
    }

    private async Task<int> RelayBatchAsync(CancellationToken ct)
    {
        await using var scope = scopes.CreateAsyncScope();
        var db = scope.ServiceProvider.GetRequiredService<OrdersDb>();

        // With EnableRetryOnFailure a transaction begun outside the strategy throws on its first command.
        // A retried batch reruns from the top, so rows published before the failure go out again:
        // the at-least-once delivery the consumers already de-duplicate.
        var strategy = db.Database.CreateExecutionStrategy();
        var (published, brokerDown) = await strategy.ExecuteAsync(async () =>
        {
            db.ChangeTracker.Clear();
            await using var tx = await db.Database.BeginTransactionAsync(ct);

            var batch = await db.OutboxMessages.FromSql($"""
                SELECT * FROM outbox_messages
                WHERE processed_at IS NULL AND attempts < {MaxAttempts}
                  AND (next_attempt_at IS NULL OR next_attempt_at <= now())
                ORDER BY occurred_at
                LIMIT {BatchSize}
                FOR UPDATE SKIP LOCKED
                """).ToListAsync(ct);

            var count = 0;
            BrokerUnavailableException? down = null;
            foreach (var message in batch)
            {
                // The row locks are held during the round trip, so a hung broker call must not wait forever.
                using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
                timeout.CancelAfter(PublishTimeout);
                try
                {
                    await publisher.PublishAsync(message.Id.ToString(), message.Type, message.Payload, timeout.Token);
                    message.ProcessedAt = DateTimeOffset.UtcNow;
                    count++;
                }
                catch (Exception ex) when (ex is BrokerUnavailableException || (ex is OperationCanceledException && !ct.IsCancellationRequested))
                {
                    // The broker is down or hung: stop the batch and leave the remaining rows untouched.
                    down = ex as BrokerUnavailableException ?? new BrokerUnavailableException("Publish timed out", ex);
                    break;
                }
                catch (Exception ex) when (!ct.IsCancellationRequested)
                {
                    // A problem with this message (rejected, unroutable): count it and back it off.
                    message.Attempts++;
                    message.LastError = ex.Message;
                    message.NextAttemptAt = DateTimeOffset.UtcNow.AddSeconds(Math.Min(Math.Pow(2, message.Attempts), 600));
                }
            }

            await db.SaveChangesAsync(ct);
            await tx.CommitAsync(ct);
            return (count, down);
        });

        // Thrown after the commit, outside the strategy: marks already made are kept and nothing is retried.
        if (brokerDown is not null) throw brokerDown;
        return published;
    }
}