Outbox e inbox

O problema

Um evento de domínio roda dentro da transação, e é isso que o torna consistente. Publicar num broker de dentro dela é o contrário: se o commit falhar depois de a mensagem sair, o mundo de fora viu algo que não aconteceu. Publicar depois do commit tem o problema inverso — o processo pode cair entre um e outro, e a mensagem nunca sai.

O outbox fecha as duas pontas. O evento é gravado numa tabela, na mesma transação das mudanças que o levantaram, e entregue por um serviço em background depois do commit. Desfeito o comando, o evento nunca existiu; confirmado, ele está lá até ser entregue.

Configuração

1. 1

   Ligue o outbox

   builder.Services.AddNexo<CatalogDbContext>(nexo => nexo
       .AddHandlersFromAssemblyOf<Program>()
       .UseOutbox());

2. 2

   Mapeie a tabela no contexto que os comandos mais escrevem

   O evento só comita junto com as mudanças se estiver no mesmo banco que elas. Com vários bancos, escolha o principal.

   protected override void OnModelCreating(ModelBuilder modelBuilder)
   {
       modelBuilder.ApplyNexoOutbox();          // tabela NexoOutbox; nome e schema são parâmetros
       modelBuilder.ApplyNexoConventions();
   }

3. 3

   Crie a tabela

   Numa migration, como qualquer outra mudança de modelo: dotnet ef migrations add Outbox.

Eventos de integração

Um evento de integração herda de IntegrationEvent (ou implementa IIntegrationEvent). Ele é levantado como qualquer evento de domínio — da entidade, com Raise — e o unit of work o manda para o outbox em vez de despachá-lo:

public sealed class ProductPublished : IntegrationEvent
{
    public Guid ProductId { get; init; }
    public string Sku { get; init; } = string.Empty;
}

public sealed class Product : AggregateRoot
{
    public void Publish()
    {
        IsPublished = true;
        Raise(new ProductPublished { ProductId = ExternalId, Sku = Sku });
    }
}

Quando o evento não nasce de uma entidade, grave-o direto pelo IOutbox, de dentro de um handler:

internal sealed class CloseMonthHandler(IOutbox outbox) : ICommandHandler<CloseMonth>
{
    public async Task HandleAsync(CloseMonth command, CancellationToken ct)
    {
        // … as mudanças do fechamento …
        await outbox.EnqueueAsync(new MonthClosed { Period = command.Period }, ct);
    }
}

Entrega

Por padrão, entregar é fazer BroadcastAsync do evento num escopo novo: os IEventHandler<ProductPublished> da aplicação rodam depois do commit, cada entrega com o seu próprio unit of work. Para mandar a um broker, registre o seu IOutboxPublisher:

public sealed class ServiceBusPublisher(ServiceBusSender sender) : IOutboxPublisher
{
    public Task PublishAsync(IIntegrationEvent @event, CancellationToken ct) =>
        sender.SendMessageAsync(
            new ServiceBusMessage(BinaryData.FromObjectAsJson(@event, @event.GetType()))
            {
                MessageId = @event.Id.ToString(),   // o broker descarta repetições por aqui
                Subject = @event.GetType().Name,
            },
            ct);
}

builder.Services.AddSingleton<IOutboxPublisher, ServiceBusPublisher>();

• Pelo menos uma vez. Uma entrega que falha é repetida, e uma que dá certo pode ser repetida se o processo cair antes de marcá-la. Consumidores idempotentes, sempre — o Id do evento atravessa o outbox intacto para isso.
• Em ordem de gravação, dentro de cada passada.
• Com back-off. Uma falha espera RetryDelay, dobrando a cada tentativa até MaxRetryDelay. Depois de MaxAttempts o evento fica na tabela com o último erro, para uma pessoa olhar.
• Várias instâncias sem entrega duplicada. Num banco relacional, cada evento é reservado com um UPDATE condicional antes de ser entregue, então duas réplicas da aplicação dividem o trabalho. No MongoDB não existe esse comando: rode um dispatcher só.

Inbox: tratar uma vez

"Pelo menos uma vez" empurra o problema para quem consome: ou o handler é idempotente, ou uma repetição faz o efeito duas vezes. O inbox resolve isso no lado do consumidor. Cada IIntegrationEvent que chega é registrado pelo Id, na mesma transação das mudanças que os handlers dele fazem, e uma repetição é pulada sem rodar handler nenhum.

builder.Services.AddNexo<CatalogDbContext>(nexo => nexo
    .AddHandlersFromAssemblyOf<Program>()
    .UseOutbox()
    .UseInbox());

// No contexto em que os handlers dos eventos escrevem:
modelBuilder.ApplyNexoInbox();   // tabela NexoInbox, índice único no id do evento

• O registro e o efeito andam juntos. Um handler que falha desfaz o registro com o resto, e a próxima entrega o roda de novo. Um que dá certo comita os dois, e a próxima entrega é pulada.
• Duas cópias ao mesmo tempo passam as duas pela checagem; o índice único deixa só uma comitar, e a outra falha com DuplicateKeyException — o broker a entrega de novo depois, quando já há registro.
• Vale para qualquer origem: o outbox desta aplicação, um consumidor de fila que faz BroadcastAsync do evento recebido, um webhook. O que importa é o Id do evento chegar intacto — o outbox preserva, e um publisher para broker deve mandá-lo como id da mensagem.
• Só eventos de integração são registrados. Eventos de domínio rodam dentro do comando que os levantou e não se repetem.

Rodando no seu agendador

O dispatcher em background emenda uma passada na outra enquanto há eventos pendentes e, quando não há, espera PollingInterval. Se o banco estiver fora, a passada falha, é registrada, e a seguinte encontra os eventos no mesmo lugar. Para entregar de um job seu — um Hangfire, um cron, um endpoint administrativo — desligue-o e chame o processador:

nexo.UseOutbox(outbox => outbox.RunDispatcher = false);

// No seu job:
var delivered = await serviceProvider
    .GetRequiredService<OutboxProcessor>()
    .ProcessAsync(ct);   // uma passada; devolve quantos tentou

As linhas entregues ficam na tabela, com ProcessedOn, para auditoria. Apague as antigas de tempos em tempos:

var cutoff = DateTime.UtcNow.AddDays(-7);

await db.Set<OutboxMessage>()
    .Where(m => m.ProcessedOn != null && m.ProcessedOn < cutoff)
    .ExecuteDeleteAsync(ct);