Drizzle transactional outbox

When a handler writes a row and emits a cross-service event, both must commit in the same database transaction. The transactional outbox pattern queues events locally; a flusher publishes them after commit.

1. Declare the outbox

import { defineOutbox } from "@nwire/forge";
import { OrderWasPlaced } from "./orders.events";

export const ordersOutbox = defineOutbox({
  name: "orders",
  description: "Miri's checkout row and the OrderWasPlaced fact share one commit.",
  publishes: [OrderWasPlaced],
  flushIntervalMs: 500,
  maxBatch: 50,
});

Register it on the forge plugin:

forgePlugins({
  // ... actors, projections, workflows ...
  outboxes: [ordersOutbox],
});

2. Outbox table (Drizzle)

import { pgTable, text, timestamp, jsonb, uuid } from "drizzle-orm/pg-core";

export const outboxEntries = pgTable("outbox_entries", {
  id: uuid("id").primaryKey().defaultRandom(),
  outboxName: text("outbox_name").notNull(),
  eventName: text("event_name").notNull(),
  payload: jsonb("payload").notNull(),
  envelope: jsonb("envelope").notNull(),
  createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
  flushedAt: timestamp("flushed_at", { withTimezone: true }),
});

3. Handler — one transaction

import { defineAction } from "@nwire/forge";
import { PlaceOrderInput } from "./place-order.action";
import { orders } from "./orders.schema";

export const placeOrder = defineAction({
  name: "orders.place-order",
  input: PlaceOrderInput,
  handler: async (ctx) => {
  const { input } = ctx;
  const db = ctx.resolve("db");

  await db.transaction(async (tx) => {
    await tx.insert(orders).values({
      id: input.orderId,
      tenantId: ctx.tenant,
      total: input.total,
    });

    await tx.insert(outboxEntries).values({
      outboxName: ordersOutbox.name,
      eventName: "orders.order-was-placed",
      payload: { orderId: input.orderId, total: input.total },
      envelope: ctx.envelope,
    });
  });

  return { orderId: input.orderId };
  },
});

If the transaction rolls back, neither the row nor the outbox entry survives — no ghost events.

4. Flusher (wire boundary)

The flusher runs at deployment time, not in domain code:

// outbox-flusher.ts
import { createCronWire } from "@nwire/cron";

// Poll pending rows, publish to @nwire/nats (or bus-amqp), mark flushedAt.
// Pair with defineInbox on consumers for dedup.

Production options:

• @nwire/cron poller in the same service
• @nwire/bullmq repeatable job
• Dedicated sidecar reading outbox_entries WHERE flushed_at IS NULL

5. Verify durability

Integration test checklist:

1. Insert row + outbox row in one txn → both visible.
2. Roll back txn → neither visible.
3. Flusher publishes → flushedAt set; bus consumer receives once.
4. Retry publish → consumer dedups via inbox.

See examples/station-mgmt-drizzle/ for Drizzle wiring and @nwire/trust-kit scenario C08 for the outbox primitive smoke.

See also

• Idempotency and dedup
• defineOutbox reference
• Cross-service events