Somewhere in most event-driven backends, there's a method that looks like this:
async placeOrder(input: PlaceOrderInput) {
await this.db.insert(orders).values(input); // 1. write the row
await this.kafka.send({ topic: 'order.placed', … }); // 2. publish the event
}
It looks fine, and it has a bug. If the process crashes between 1 and 2, the order exists but the event never happened — downstream consumers silently miss it. Swap the order and you get the opposite failure: an event for an order that was rolled back. There is no try/catch arrangement that fixes this, because a database and a broker cannot commit atomically. This is the dual-write problem.
The boring, proven fix is the transactional outbox: don't publish in step 2. Instead, write the event into an outbox_events table in the same database transaction as the business row. A background worker then relays committed rows to the broker. The transaction is the only atomic boundary you have — so put both writes inside it.
That gives you at-least-once delivery, which means the consumer side needs the mirror-image pattern: an idempotent inbox that deduplicates redeliveries, so the side effect runs exactly once even when Kafka delivers twice.
@nest-native/messaging — part of nest-native, a set of NestJS integrations — packages this pair as a library, after the pattern was first built by hand inside a reference application. It's the outbox/inbox pattern for the Drizzle ORM + NestJS stack — a niche the existing NestJS outbox libraries (which target TypeORM and MikroORM — see nestjs-outbox and nestjs-inbox-outbox, both solid) don't cover.
The producer half: enqueue inside your transaction
The library ships the tables as Drizzle factories per dialect (SQLite, Postgres, MySQL). You add them to your schema and generate a migration like any other table:
// schema.ts
export { outboxEvents, inboxEvents } from '@nest-native/messaging/sqlite'; // or /postgres, /mysql
Transactions ride on @nestjs-cls/transactional with its Drizzle adapter — the same @Transactional() decorator you'd use anyway:
MessagingModule.forRoot({
drizzleInstanceToken: DRIZZLE, // your Drizzle DI token
outboxStore: new SqliteOutboxStore(), // or PostgresOutboxStore / MysqlOutboxStore
inboxStore: new SqliteInboxStore(),
transport, // where the claimer relays to — below
}),
Then the business code:
@Injectable()
export class OrderService {
constructor(
@InjectTransaction() private readonly db: AppDatabase,
private readonly producer: OutboxProducer<SqliteOutboxStore>,
) {}
@Transactional()
placeOrder(id: string, item: string) {
this.db.insert(orders).values({ id, item }).run();
this.producer.enqueue({
topic: 'order.placed',
payload: { id, item },
idempotencyKey: `order:${id}`,
});
}
}
The order row and the outbox row commit together, or roll back together. A throw after enqueue produces no phantom event; a crash after commit loses nothing, because the event is durably in your database.
(One nuance the library handles for you: better-sqlite3 transactions are synchronous while Postgres/MySQL are async. The per-dialect stores own that difference — on SQLite enqueue returns the row directly inside the sync transaction body; on Postgres you await it. Same code shape either way.)
Relaying: the claimer and the worker
A claimer polls for committed rows, publishes each through a transport, and applies retry-with-backoff on failure — including reclaiming rows from a worker that died mid-flight:
// scripts/start-worker.ts
const app = await NestFactory.createApplicationContext(AppModule);
await runWorkerLoop(app.get(OutboxClaimer), {
pollIntervalMs: 2_000,
signal: shutdownSignal, // AbortSignal wired to SIGTERM
});
A note on that pollIntervalMs, because a sharp question in the comments called it "a quiet latency knob": the loop is self-clocking. After a tick that claims a full batch it loops again immediately to drain the backlog; it only sleeps the interval when a tick claims nothing. So 2s is the worst case for a lone event landing in an otherwise-idle outbox — not a per-event tax — and under load, throughput is never gated by it.
When even that idle latency matters (something user-facing waiting behind an outbox event), lowering the interval works but has a floor and a DB-poll cost. As of 0.4.0 there's a better lever — an in-process wake:
const waker = new OutboxWaker();
runWorkerLoop(app.get(OutboxClaimer), { pollIntervalMs: 2_000, waker, signal });
// request path — right AFTER the enqueueing transaction commits:
waker.notify(); // the idle worker ticks now instead of on the next poll
Polling stays the backstop — a missed notify() only widens latency back to one interval, never drops an event — and wakes are latched, so a notify landing in the sliver between a tick and its sleep isn't lost.
Running the worker as a separate process on the same machine (the usual app + worker split)? The WakeSocketServer/WakeSocketClient pair carries the same wake over a unix domain socket — the worker feeds incoming connections into its waker, and the client's notify() is fire-and-forget, never throwing into the request path:
// worker process
await new WakeSocketServer({ path: env.wakeSocket, waker }).listen();
// app process — same path, right after the commit
new WakeSocketClient({ path: env.wakeSocket }).notify();
Workers on other machines (Postgres only, as of 0.5.0) get the wake through the one thing they already share — the database, via LISTEN/NOTIFY:
// producer side — one store option; pg_notify rides the enqueue transaction,
// so Postgres delivers the wake ON COMMIT and drops it on rollback: the signal
// is atomic with the event becoming visible, no post-commit discipline needed
new PostgresOutboxStore({ wakeChannel: 'outbox_wake' })
// worker side — a dedicated (non-pooled) LISTEN connection feeds the same waker
const listener = new PostgresWakeListener({
connect: () => new pg.Client({ connectionString, keepAlive: true }),
channel: 'outbox_wake',
waker,
});
listener.start(); // reconnects on drops; await listener.stop() on shutdown
Missed notifications aren't recovered — polling remains the backstop, as with every tier.
For Kafka the transport is one line, built on @nest-native/kafka (Confluent's official JS client underneath):
MessagingModule.forRootAsync({
drizzleInstanceToken: DRIZZLE,
outboxStore: new SqliteOutboxStore(),
inboxStore: new SqliteInboxStore(),
inject: [KafkaProducerService],
useTransport: (producer) => new KafkaOutboxTransport(producer),
}),
Don't have Kafka yet? There's an in-process transport (@nest-native/messaging/in-process) — a topic→handler registry with the same at-least-once semantics — so a modular monolith can adopt the pattern today and swap the transport for a broker later without touching a line of domain code.
The consumer half: exactly-once effects
Kafka is at-least-once by contract, so redelivery is a when, not an if. The inbox primitive is a single method:
const outcome = await inbox.runOnce(dedupKey, source, () => {
// your side effect — runs in the SAME transaction as the dedup row
this.audit.record({ … });
});
// 'processed' on the first delivery, 'duplicate' on any redelivery
runOnce inserts a (source, message_key) row protected by a unique index and runs your side effect in the same transaction. A redelivery violates the index → 'duplicate' → the side effect is skipped. If your side effect throws, the dedup row rolls back with it, so the retry reprocesses cleanly. That composition — unique index + shared transaction — is the entire trick, and it's provable in the database.
For Kafka consumers the library wraps the full delivery decision (validate → dedup → ack / dead-letter / redeliver) in an engine you delegate to from a thin @KafkaConsumer shell:
@KafkaConsumer('order.placed', { groupId: 'orders-service' })
export class OrderConsumer {
constructor(private readonly inbox: KafkaInboxConsumer, private readonly audit: OrderAuditService) {}
@KafkaHandler()
async handle(@KafkaMessage() payload: unknown, @KafkaHeaders() headers: Headers, @KafkaCtx() ctx: KafkaContext) {
await this.inbox.consume<OrderPlaced>({
source: 'order.placed:orders-service',
context: ctx, headers, payload,
validate: isOrderPlaced, // poison message → DLQ, then ack
sideEffect: (order, dedupKey) => this.audit.record(order, dedupKey),
dlqTopic: 'order.placed.DLQ',
});
}
}
Poison messages (unparseable, unkeyable) go to a dead-letter topic instead of redelivering forever; transient failures rethrow so the broker redelivers; duplicates ack silently.
Testing without a broker
Everything above runs in tests with no infrastructure: an in-memory outbox transport (@nest-native/messaging/testing) for the producer half, and @nest-native/kafka/testing's in-memory broker for the full pipeline — including redelivery:
await broker.emit('order.placed', publishedMessage); // redeliver the same message
await broker.idle(); // wait for handler pipelines to settle
expect(auditRows).toHaveLength(1); // side effect ran exactly once
The library itself is tested at 100% coverage against real SQLite, real in-process Postgres (pglite), and the in-memory broker — plus gated round-trips against real Postgres and MySQL containers.
See the whole thing running
The pattern is one chapter of a larger, runnable story: the nest-native reference app is a multi-tenant work-tracking SaaS where every task write emits task.created/assigned/completed through this outbox, a consumer builds an activity feed through this inbox, the event contracts are published as an AsyncAPI 3.0 catalog, and a streaming AI assistant summarizes the activity — eight libraries, one coherent journey, green tests, no Docker required for the default profile.
Honest scope
- This is the app-level outbox. At larger scale you may prefer CDC (Debezium + Kafka Connect) tailing the WAL — different trade-offs, no app code, more infrastructure.
- If you're on TypeORM or MikroORM, the libraries linked above already serve you well.
@nest-native/messagingexists specifically because nothing covered Drizzle. - Delivery is at-least-once end to end; the inbox gives you exactly-once effects, which is the guarantee that actually matters — exactly-once delivery across a network isn't something any tool can promise.
Docs: nest-native.dev/messaging · Source: github.com/nest-native/messaging. Feedback and issues welcome — especially if you hit a case the pattern doesn't cover.
Top comments (2)
The outbox is one of those patterns you don't really feel until the first time the DB commit lands and the publish doesn't. On a payments flow that's a customer charged with nothing downstream knowing. What I like here is you didn't skip the inbox half. Most people ship the outbox, call it done, then get bitten by the duplicate delivery they swore couldn't happen. One thing from running this in anger: that 2s poll interval is a quiet latency knob. Fine for most events, but the day someone puts a user-facing "we've got it" behind an outbox event, 2 seconds starts to feel long. Did you land on 2s for a reason, or is it a sane default for now?
Hi @mickyarun , you're right to call it a latency knob, and right that once something user-facing waits behind an outbox event, a poll interval is the wrong thing to be paying for it. Lowering it only trades latency for DB load. One detail that softens it: the loop only sleeps when a tick claims nothing (a full batch loops again immediately), so the interval is the worst case for a lone event on an idle outbox, not a cost on every event. But a lone event on a quiet system is exactly your scenario, so that doesn't really answer the concern.
This came directly out of your comment. 0.4.0 and 0.5.0 add an event-driven wake, with polling kept as the backstop (a missed wake costs one poll interval, never an event): an OutboxWaker you notify() right after the enqueueing transaction commits; a WakeSocketServer/WakeSocketClient pair for a worker running as a separate process on the same machine; and on Postgres, pg_notify riding the enqueue transaction — delivered on commit, dropped on rollback — with a PostgresWakeListener feeding workers on any machine.
I've updated the relaying section of the post with all of this.