# Outbox and inbox — implementation recipes

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

```sql
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):

```sql
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.

```csharp
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

```csharp
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`

```ts
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

```ts
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.