Outbox and inbox — implementation recipes · messaging-patterns

Companion to messaging-patterns §2–§3. SQL is PostgreSQL; the SQL Server equivalent of FOR UPDATE SKIP LOCKED is WITH (UPDLOCK, READPAST). Add the tables with a normal migration (see the database-migrations skill).

Tables

CREATE TABLE outbox (
  id            uuid PRIMARY KEY,
  occurred_at   timestamptz NOT NULL DEFAULT now(),
  aggregate_id  text  NOT NULL,
  type          text  NOT NULL,
  payload       jsonb NOT NULL,
  headers       jsonb NOT NULL DEFAULT '{}',
  sent_at       timestamptz
);
CREATE INDEX outbox_unsent ON outbox (occurred_at) WHERE sent_at IS NULL;

CREATE TABLE inbox (
  message_id  uuid NOT NULL,
  consumer    text NOT NULL,
  received_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (message_id, consumer)
);

Cleanup (scheduled job, batches of ~10 000 rows so the delete does not hold locks for long):

DELETE FROM outbox WHERE id IN (SELECT id FROM outbox WHERE sent_at < now() - interval '7 days' LIMIT 10000);
DELETE FROM inbox  WHERE (message_id, consumer) IN (SELECT message_id, consumer FROM inbox WHERE received_at < now() - interval '30 days' LIMIT 10000);

.NET relay — EF Core 10 + `BackgroundService`

IMessagePublisher is your thin wrapper over the broker SDK (or MassTransit's IPublishEndpoint). One DI scope per batch, a transaction around select → publish → mark sent.

public sealed class OutboxMessage
{
    public Guid Id { get; set; }
    public DateTimeOffset OccurredAt { get; set; }
    public required string AggregateId { get; set; }
    public required string Type { get; set; }
    public required string Payload { get; set; }
    public required string Headers { get; set; }
    public DateTimeOffset? SentAt { get; set; }
}

public sealed class OutboxRelay(IServiceScopeFactory scopes, IMessagePublisher publisher, ILogger<OutboxRelay> log)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            var sent = 0;
            try { sent = await RelayBatchAsync(stoppingToken); }
            catch (Exception ex) when (ex is not OperationCanceledException)
            {
                log.LogError(ex, "Outbox relay batch failed");
            }
            if (sent == 0) await Task.Delay(TimeSpan.FromSeconds(1), stoppingToken);
        }
    }

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

        var batch = await db.Outbox
            .FromSql($"SELECT * FROM outbox WHERE sent_at IS NULL ORDER BY occurred_at LIMIT 100 FOR UPDATE SKIP LOCKED")
            .ToListAsync(ct);

        foreach (var m in batch)
        {
            await publisher.PublishAsync(m.Type, m.Payload, messageId: m.Id, partitionKey: m.AggregateId, headers: m.Headers, ct);
            m.SentAt = DateTimeOffset.UtcNow;
        }

        await db.SaveChangesAsync(ct);
        await tx.CommitAsync(ct);
        return batch.Count;
    }
}

Writing to the outbox in the same transaction as the aggregate is just db.Outbox.Add(new OutboxMessage { … }) before the single SaveChangesAsync. Libraries that ship this: MassTransit (x.AddEntityFrameworkOutbox<OrdersDbContext>(o => { o.UsePostgres(); o.UseBusOutbox(); })), NServiceBus (endpointConfiguration.EnableOutbox()).

.NET inbox consumer

public async Task HandleAsync(IncomingMessage msg, CancellationToken ct)
{
    await using var tx = await db.Database.BeginTransactionAsync(ct);

    var inserted = await db.Database.ExecuteSqlAsync(
        $"INSERT INTO inbox (message_id, consumer) VALUES ({msg.Id}, {ConsumerName}) ON CONFLICT DO NOTHING", ct);
    if (inserted == 0) { await tx.CommitAsync(ct); return; }   // duplicate: ack without re-applying

    await ApplyAsync(msg, ct);            // domain effect, same DbContext
    await db.SaveChangesAsync(ct);
    await tx.CommitAsync(ct);
}                                         // the broker ack happens after this method returns

Node relay — `pg`

import type pg from 'pg';

type OutboxRow = { id: string; type: string; aggregate_id: string; payload: unknown; headers: Record<string, string> };
type Publish = (m: { id: string; type: string; key: string; body: unknown; headers: Record<string, string> }) => Promise<void>;

export async function relayOnce(pool: pg.Pool, publish: Publish): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query<OutboxRow>(
      `SELECT id, type, aggregate_id, payload, headers FROM outbox
        WHERE sent_at IS NULL ORDER BY occurred_at LIMIT 100 FOR UPDATE SKIP LOCKED`,
    );
    for (const row of rows) {
      await publish({ id: row.id, type: row.type, key: row.aggregate_id, body: row.payload, headers: row.headers });
    }
    if (rows.length > 0) {
      await client.query('UPDATE outbox SET sent_at = now() WHERE id = ANY($1::uuid[])', [rows.map((r) => r.id)]);
    }
    await client.query('COMMIT');
    return rows.length;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

Run it from a worker process (not the API process) in a loop that sleeps one second when a batch is empty. Alternative: pg-boss already stores jobs in PostgreSQL, so boss.send() inside your own transaction (pass the same client) is an outbox.

Node inbox consumer

export async function handle(client: pg.PoolClient, msg: { id: string; body: unknown }, consumer: string) {
  await client.query('BEGIN');
  try {
    const { rowCount } = await client.query(
      'INSERT INTO inbox (message_id, consumer) VALUES ($1, $2) ON CONFLICT DO NOTHING', [msg.id, consumer]);
    if (rowCount === 0) { await client.query('COMMIT'); return; }     // duplicate
    await apply(client, msg.body);                                    // domain effect on the same client
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;                                                        // broker retries; inbox dedupes the replay
  }
}

Queue-library dedupe and retry options

Library	Message id for dedupe	Retry with backoff	Dead letter
BullMQ	`queue.add(name, data, { jobId: messageId })` — same `jobId` is ignored while the job exists	`{ attempts: 5, backoff: { type: 'exponential', delay: 1000 } }`	Failed jobs stay in the `failed` set; move to a DLQ queue in a `failed` listener
pg-boss	`boss.send(name, data, { singletonKey: messageId })`	`{ retryLimit: 5, retryBackoff: true }`	Jobs exceed `retryLimit` → `failed` state; query and requeue
Azure Service Bus	`MessageId` + duplicate detection window on the queue/topic	Delivery count + `MaxDeliveryCount`	Built-in `$DeadLetterQueue` sub-queue
SQS	`MessageDeduplicationId` (FIFO queues only)	Visibility timeout + `maxReceiveCount`	Redrive policy to a DLQ
Kafka	No broker dedupe: inbox table is mandatory	Consumer-side retry topic or pause/resume	Dead-letter topic written by the consumer

Broker dedupe windows are short (minutes); keep the inbox table for anything that must never be applied twice.