Outbox pattern — зачем нужен при работе с очередями и БД?

CREATE TABLE outbox_messages (
    id            uuid PRIMARY KEY,
    occurred_at   timestamptz NOT NULL,
    type          text        NOT NULL,
    payload       jsonb       NOT NULL,
    trace_parent  text,                       -- для сквозной трассировки
    processed_at  timestamptz,
    attempts      int NOT NULL DEFAULT 0,
    error         text
);
CREATE INDEX ix_outbox_unprocessed ON outbox_messages (occurred_at) WHERE processed_at IS NULL;

public async Task<Guid> Handle(PlaceOrder cmd, CancellationToken ct)
{
    var order = Order.Place(cmd.CustomerId, cmd.Items);
    db.Orders.Add(order);
    db.OutboxMessages.Add(OutboxMessage.From(new OrderPlaced(order.Id, order.CustomerId, order.Total)));
    await db.SaveChangesAsync(ct);          // заказ и сообщение — атомарно
    return order.Id;
}

public sealed class OutboxPublisher(IServiceScopeFactory scopes, IMessagePublisher bus, ILogger<OutboxPublisher> log) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        while (!ct.IsCancellationRequested)
        {
            await using var scope = scopes.CreateAsyncScope();
            var db = scope.ServiceProvider.GetRequiredService<ShopDbContext>();
            await using var tx = await db.Database.BeginTransactionAsync(ct);

            // SKIP LOCKED: несколько экземпляров сервиса не возьмут одни и те же сообщения
            var batch = await db.OutboxMessages
                .FromSql($"SELECT * FROM outbox_messages WHERE processed_at IS NULL ORDER BY occurred_at LIMIT 100 FOR UPDATE SKIP LOCKED")
                .ToListAsync(ct);

            foreach (var msg in batch)
            {
                try
                {
                    await bus.PublishAsync(msg.Type, msg.Payload, messageId: msg.Id, ct);
                    msg.ProcessedAt = DateTime.UtcNow;
                }
                catch (Exception ex)
                {
                    msg.Attempts++; msg.Error = ex.Message;
                    log.LogWarning(ex, "Не удалось опубликовать {MessageId}", msg.Id);
                }
            }
            await db.SaveChangesAsync(ct);
            await tx.CommitAsync(ct);

            if (batch.Count == 0) await Task.Delay(TimeSpan.FromSeconds(1), ct);
        }
    }
}

public async Task Consume(OrderPlaced message, Guid messageId, CancellationToken ct)
{
    await using var tx = await db.Database.BeginTransactionAsync(ct);

    // Уникальный ключ по message_id: повтор вызовет конфликт → пропускаем
    var inserted = await db.Database.ExecuteSqlAsync(
        $"INSERT INTO inbox_messages (message_id, processed_at) VALUES ({messageId}, now()) ON CONFLICT DO NOTHING", ct);
    if (inserted == 0) return;                                   // уже обработано

    await reservations.ReserveAsync(message.OrderId, message.Items, ct);   // бизнес-логика в той же транзакции
    await db.SaveChangesAsync(ct);
    await tx.CommitAsync(ct);
}

builder.Services.AddMassTransit(x =>
{
    x.AddEntityFrameworkOutbox<ShopDbContext>(o =>
    {
        o.UsePostgres();
        o.UseBusOutbox();                       // IPublishEndpoint пишет в outbox, а не в брокер
        o.QueryDelay = TimeSpan.FromSeconds(1);
    });
    x.AddConsumer<ReserveStockConsumer>();
    x.UsingRabbitMq((ctx, cfg) => cfg.ConfigureEndpoints(ctx));
});

// В обработчике: публикация «попадёт» в outbox и уйдёт после SaveChanges
await publishEndpoint.Publish(new OrderPlaced(order.Id), ct);
await db.SaveChangesAsync(ct);