CREATE TABLE outbox (
id BIGSERIAL PRIMARY KEY,
aggregate_type TEXT NOT NULL,
aggregate_id TEXT NOT NULL,
event_type TEXT NOT NULL,
payload JSONB NOT NULL,
dedupe_key TEXT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
published_at TIMESTAMPTZ,
CONSTRAINT outbox_dedupe_key_uniq UNIQUE (dedupe_key)
);
-- Only unpublished rows are indexed, keeping the relay poll fast.
CREATE INDEX outbox_unpublished_idx
ON outbox (created_at)
WHERE published_at IS NULL;const { pool } = require('./db');
async function createOrder(order) {
const client = await pool.connect();
try {
await client.query('BEGIN');
const inserted = await client.query(
`INSERT INTO orders (customer_id, total_cents, status)
VALUES ($1, $2, 'pending')
RETURNING id, customer_id, total_cents, status, created_at`,
[order.customerId, order.totalCents]
);
const row = inserted.rows[0];
await client.query(
`INSERT INTO outbox
(aggregate_type, aggregate_id, event_type, payload, dedupe_key)
VALUES ('order', $1, 'OrderCreated', $2, $3)
ON CONFLICT (dedupe_key) DO NOTHING`,
[
String(row.id),
JSON.stringify(row),
`order:${row.id}:OrderCreated`,
]
);
await client.query('COMMIT');
return row;
} catch (err) {
await client.query('ROLLBACK');
throw err;
} finally {
client.release();
}
}
module.exports = { createOrder };const { pool } = require('./db');
const { publish } = require('./broker');
const BATCH_SIZE = 100;
const POLL_MS = 500;
async function drainBatch() {
const client = await pool.connect();
try {
await client.query('BEGIN');
const { rows } = await client.query(
`SELECT id, aggregate_id, event_type, payload
FROM outbox
WHERE published_at IS NULL
ORDER BY created_at
FOR UPDATE SKIP LOCKED
LIMIT $1`,
[BATCH_SIZE]
);
for (const evt of rows) {
await publish(evt.event_type, {
eventId: evt.id,
aggregateId: evt.aggregate_id,
payload: evt.payload,
});
await client.query(
`UPDATE outbox SET published_at = now() WHERE id = $1`,
[evt.id]
);
}
await client.query('COMMIT');
return rows.length;
} catch (err) {
await client.query('ROLLBACK');
throw err;
} finally {
client.release();
}
}
async function runRelay() {
for (;;) {
try {
const n = await drainBatch();
if (n === 0) await new Promise((r) => setTimeout(r, POLL_MS));
} catch (err) {
console.error('relay batch failed', err);
await new Promise((r) => setTimeout(r, POLL_MS));
}
}
}
module.exports = { runRelay, drainBatch };The transactional outbox pattern solves a subtle but common distributed-systems problem: a service needs to both change its own database and tell the outside world about that change, but there is no shared transaction spanning the database and the message broker. Writing the row and then publishing to Kafka in two steps means a crash between them either drops the event or, if reordered, publishes an event for a change that never committed. The fix is to write the event into an outbox table inside the same transaction as the business change, then relay those rows asynchronously.
The outbox schema.sql tab defines that table. Each row carries an aggregate_type and aggregate_id so consumers can partition by entity, a JSONB payload, and a dedupe_key with a unique constraint that makes idempotency enforceable at the storage layer. The partial index on published_at IS NULL keeps the relay's polling query cheap even as the table grows, since only unpublished rows are indexed.
In orderService.js, createOrder opens a transaction with BEGIN, inserts the order, and inserts the outbox event using the same client. Because both writes share one transaction, they commit or roll back together — the event can never exist without the order, and vice versa. The dedupe_key combines the aggregate id and event type, and ON CONFLICT DO NOTHING guards against duplicate inserts on retry.
The outboxRelay.js tab is the publisher side. It polls in a loop, claiming a batch with FOR UPDATE SKIP LOCKED so multiple relay instances can run concurrently without processing the same rows. Each claimed event is published to the broker and then marked with published_at. This is deliberately at-least-once: the process may publish and then crash before the update, so a row gets re-sent. That is why consumers must be idempotent, keyed off the event id.
The trade-off is added write amplification and eventual rather than immediate delivery, plus polling latency. In return the system gets a strong guarantee that no committed change silently fails to emit its event, which is usually the property that matters most.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
class SignupForm
include ActiveModel::Model
include ActiveModel::Attributes
attribute :account_name, :string
attribute :email, :stringShallow Controller, Deep Params: Form Object Pattern
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
import { randomBytes, createHash } from "crypto";
import jwt from "jsonwebtoken";
import { RefreshTokenStore } from "./store";
const ACCESS_SECRET = process.env.ACCESS_SECRET!;
const ACCESS_TTL = "15m";JWT access + refresh token rotation (conceptual)
Share this code
Here's the card — post it anywhere.
