Outbox and inbox

The problem

A domain event runs inside the transaction, and that is what makes it consistent. Publishing to a broker from in there is the opposite: if the commit fails after the message has left, the outside world saw something that did not happen. Publishing after the commit has the inverse problem — the process can die in between, and the message never leaves.

The outbox closes both ends. The event is written to a table, in the same transaction as the changes that raised it, and delivered by a background service after the commit. If the command is undone, the event never existed; if it is committed, the event is there until it is delivered.

Configuration

1. 1

   Turn the outbox on

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

2. 2

   Map the table in the context your commands write to most

   The event only commits together with the changes if it lives in the same database as them. With several databases, pick the main one.

   protected override void OnModelCreating(ModelBuilder modelBuilder)
   {
       modelBuilder.ApplyNexoOutbox();          // the NexoOutbox table; name and schema are parameters
       modelBuilder.ApplyNexoConventions();
   }

3. 3

   Create the table

   In a migration, like any other model change: dotnet ef migrations add Outbox.

Integration events

An integration event derives from IntegrationEvent (or implements IIntegrationEvent). It is raised like any domain event — from the entity, with Raise — and the unit of work sends it to the outbox instead of dispatching it:

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 });
    }
}

When the event does not come from an entity, write it straight through IOutbox, from inside a handler:

internal sealed class CloseMonthHandler(IOutbox outbox) : ICommandHandler<CloseMonth>
{
    public async Task HandleAsync(CloseMonth command, CancellationToken ct)
    {
        // … the month-end changes …
        await outbox.EnqueueAsync(new MonthClosed { Period = command.Period }, ct);
    }
}

Delivery

By default, delivering means BroadcastAsync-ing the event in a fresh scope: the application's IEventHandler<ProductPublished>s run after the commit, each delivery with its own unit of work. To send to a broker, register your own 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(),   // the broker drops repeats by this
                Subject = @event.GetType().Name,
            },
            ct);
}

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

• At least once. A delivery that fails is retried, and one that succeeds may be retried if the process dies before marking it. Idempotent consumers, always — the event Id crosses the outbox intact for exactly that.
• In write order, within each pass.
• With back-off. A failure waits RetryDelay, doubling each attempt up to MaxRetryDelay. After MaxAttempts the event stays in the table with the last error, for a person to look at.
• Several instances without duplicate delivery. On a relational database, each event is claimed with a conditional UPDATE before being delivered, so two replicas of the application split the work. MongoDB has no such statement: run a single dispatcher.

Inbox: handling it once

"At least once" pushes the problem onto the consumer: either the handler is idempotent, or a repeat applies the effect twice. The inbox solves that on the consumer side. Every IIntegrationEvent that arrives is recorded by Id, in the same transaction as the changes its handlers make, and a repeat is skipped without running a single handler.

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

// In the context the event handlers write to:
modelBuilder.ApplyNexoInbox();   // the NexoInbox table, unique index on the event id

• The record and the effect travel together. A handler that fails undoes the record along with everything else, and the next delivery runs it again. One that succeeds commits both, and the next delivery is skipped.
• Two copies at once both pass the check; the unique index lets only one commit, and the other fails with DuplicateKeyException — the broker delivers it again later, when a record already exists.
• It works for any origin: this application's outbox, a queue consumer that BroadcastAsynces the event it received, a webhook. What matters is that the event Id arrives intact — the outbox preserves it, and a broker publisher should send it as the message id.
• Only integration events are recorded. Domain events run inside the command that raised them and do not repeat.

Running it on your own scheduler

The background dispatcher runs one pass straight into the next while events are pending and, when there are none, waits PollingInterval. If the database is down, the pass fails, is logged, and the next one finds the events in the same place. To deliver from a job of your own — Hangfire, a cron, an admin endpoint — turn it off and call the processor:

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

// In your job:
var delivered = await serviceProvider
    .GetRequiredService<OutboxProcessor>()
    .ProcessAsync(ct);   // one pass; returns how many it attempted

Delivered rows stay in the table, with ProcessedOn, for auditing. Clear out the old ones from time to time:

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

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