@pg-prime/outbox

Transactional outbox for pg-prime: write a message in your transaction, and a relay delivers it at least once, in order per key.

Status: pre-alpha. The API may change before 1.0.

pnpm add @pg-prime/outbox

import { defineOutbox, functionSink, pgOutbox } from '@pg-prime/outbox'

const orderPlaced = defineOutbox<{ orderId: string }>('order-placed')
const outbox = pgOutbox({ db, define: { orderPlaced } })
await outbox.start() // once at boot: migrates the `pgprime_outbox` schema

await db.transaction(async (tx) => {
  // …insert the order with tx…
  await outbox.topics.orderPlaced.write({ orderId: '42' }, { tx, key: 'order:42' })
})

// In any number of processes:
const relay = outbox.topics.orderPlaced.relay(
  functionSink(async (wave, { epoch }) => {
    for (const r of wave) await publish(r.payload, { idempotencyKey: r.idem, epoch })
  }),
)

• Atomic: the record commits or rolls back with your data.
• Ordered per key: same-key writers serialize until commit; relays lease whole partitions, so a key has one relay at a time while its lease is valid, and a failing key blocks only itself. Key locks are sorted within one call only, so write a transaction's keys with one writeMany or in a consistent order, or crossing transactions can deadlock (40P01).
• Safe to redeliver: every record has a stable idempotency key and every delivery a fencing epoch, so a sink can reject a relay whose lease was taken over while its call was in flight.
• Operable: retries with backoff, parking with replay/discard, pause, stats, retention.
• Built-in sinks: functionSink and webhookSink (signed JSON over fetch).
• Works behind PgBouncer transaction mode (by polling) and on Bun. No runtime dependencies.

Docs: guide · operations · reference.