Event-driven architecture: events, sagas, outbox and read models

HOW IT FITS TOGETHER

KEY TERMS

Messages, commands and events

A message is the envelope on the wire. A command (ReserveStock) asks one recipient to do something and may be rejected; an event (StockReserved) states a fact that already happened, is named in the past tense and may have zero or many subscribers. Fowler warns against "passive-aggressive" events: if the producer expects a specific action, send a command.

Pub/sub and topics

Producers publish to a topic without knowing subscribers; the broker fans out to each subscription. Ordering is usually per partition or key, so key by aggregate id when per-entity order matters. Delivery is at least once in practice; acks, redelivery and dead-letter queues are covered in the message queue delivery explainer.

Sagas: choreography vs orchestration

A saga replaces a distributed transaction with a sequence of local transactions plus compensating actions. In choreography each service reacts to others' events: no coordinator, but the flow lives in nobody's code and cycles creep in. In orchestration a coordinator (a state machine or a workflow engine such as Temporal) issues commands and runs compensations: explicit and testable, at the cost of a central component. Sagas have no isolation, so concurrent sagas need countermeasures such as semantic locks (a PENDING status).

Transactional outbox and CDC

You cannot atomically commit to a database and publish to a broker, so insert the event into an outbox table in the same transaction and let a relay publish it. Polling the table is simple; transaction log tailing (CDC, e.g. Debezium's outbox event router) has lower latency and no polling load but is database-specific. Both deliver at least once.

CQRS, read models and eventual consistency

Commands go to a write model that enforces invariants; queries hit read models built by projecting events. Separate stores mean the read side lags. UIs cope by returning the result of the command itself, showing a pending state, applying the change optimistically, or waiting until the read model reaches the version the command returned. Fowler's caution applies: use CQRS on a specific bounded context, not as a system-wide default.

Schema evolution and contracts

Events outlive their producers, especially in event-sourced stores. Additive changes (new optional fields) are safe if consumers ignore unknown fields; renames and type changes need a new version (OrderPlaced.v2) or upcasting on read. CloudEvents standardises the envelope (id, source, type, time); AsyncAPI documents channels, operations and message schemas the way OpenAPI documents HTTP.

Idempotent consumers

Because relays and brokers redeliver, processing the same event twice must have the effect of processing it once. Store (consumer, event id) in a processed-events table inside the same transaction as the side effect, or make the effect naturally idempotent with upserts and version guards. CloudEvents lets consumers treat identical source plus id as a duplicate.

IN YOUR STACK

-- Column names match Debezium's outbox event router defaults
CREATE TABLE outbox (id uuid PRIMARY KEY, aggregatetype text NOT NULL, aggregateid text NOT NULL,
  type text NOT NULL, payload jsonb NOT NULL, created_at timestamptz NOT NULL DEFAULT now());

// place-order.ts
import { randomUUID } from "node:crypto";
import type { Pool } from "pg";

export async function placeOrder(pool: Pool, cmd: { orderId: string; customerId: string; totalCents: number }) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    await client.query("INSERT INTO orders (id, customer_id, total_cents, status) VALUES ($1, $2, $3, 'pending')",
      [cmd.orderId, cmd.customerId, cmd.totalCents]);
    await client.query("INSERT INTO outbox (id, aggregatetype, aggregateid, type, payload) VALUES ($1, 'order', $2, 'OrderPlaced.v1', $3)",
      [randomUUID(), cmd.orderId, JSON.stringify(cmd)]);
    await client.query("COMMIT"); // both rows or neither; no broker call inside the transaction
  } catch (err) {
    await client.query("ROLLBACK");
    throw err;
  } finally {
    client.release();
  }
}

CREATE TABLE processed_events (consumer text NOT NULL, event_id uuid NOT NULL,
  processed_at timestamptz NOT NULL DEFAULT now(), PRIMARY KEY (consumer, event_id));

// shipping/on-payment-captured.ts
import type { Pool } from "pg";

export async function onPaymentCaptured(pool: Pool, event: { id: string; data: { orderId: string } }) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const claim = await client.query(
      "INSERT INTO processed_events (consumer, event_id) VALUES ('shipping', $1) ON CONFLICT DO NOTHING",
      [event.id]);
    if (claim.rowCount === 0) { await client.query("ROLLBACK"); return; } // duplicate: already applied
    await client.query("INSERT INTO shipments (order_id, status) VALUES ($1, 'ready')", [event.data.orderId]);
    await client.query("COMMIT"); // the dedupe record and the effect commit together
  } catch (err) {
    await client.query("ROLLBACK");
    throw err; // do not ack; the broker redelivers
  } finally {
    client.release();
  }
}

type AccountEvent =
  | { type: "Opened"; accountId: string; overdraftLimit: number }
  | { type: "Deposited" | "Withdrawn"; amount: number }
  | { type: "Closed" };
type Account = { balance: number; limit: number; open: boolean };
interface EventStore {
  read(stream: string): Promise<AccountEvent[]>;
  append(stream: string, events: AccountEvent[], opts: { expectedVersion: number }): Promise<void>;
}
const initial: Account = { balance: 0, limit: 0, open: false };
function evolve(s: Account, e: AccountEvent): Account {
  switch (e.type) {
    case "Opened": return { ...s, limit: e.overdraftLimit, open: true };
    case "Deposited": return { ...s, balance: s.balance + e.amount };
    case "Withdrawn": return { ...s, balance: s.balance - e.amount };
    case "Closed": return { ...s, open: false };
  }
}
export async function withdraw(store: EventStore, id: string, amount: number) {
  const history = await store.read(`account-${id}`);
  const state = history.reduce(evolve, initial);
  if (!state.open) throw new Error("account closed");
  if (state.balance - amount < -state.limit) throw new Error("insufficient funds");
  await store.append(`account-${id}`, [{ type: "Withdrawn", amount }], { expectedVersion: history.length });
}

import { proxyActivities } from "@temporalio/workflow";
import type * as activities from "./activities";
import type { OrderRequest } from "./types";

const { authorizePayment, voidPayment, reserveStock, releaseStock, confirmOrder, rejectOrder } =
  proxyActivities<typeof activities>({
    startToCloseTimeout: "30 seconds",
    retry: { maximumAttempts: 5, nonRetryableErrorTypes: ["CardDeclined", "OutOfStock"] },
  });

export async function placeOrderSaga(order: OrderRequest): Promise<"confirmed" | "rejected"> {
  const compensations: Array<() => Promise<void>> = [];
  try {
    compensations.unshift(() => voidPayment(order.id)); // register before: the call may half-succeed
    await authorizePayment(order.id, order.totalCents);
    compensations.unshift(() => releaseStock(order.id));
    await reserveStock(order.id, order.items);
    await confirmOrder(order.id); // pivot: after this, only retry forward
    return "confirmed";
  } catch (err) {
    for (const compensate of compensations) await compensate(); // reverse order via unshift
    await rejectOrder(order.id, String(err));
    return "rejected";
  }
}

POST /events HTTP/1.1
Host: shipping.internal.example.com
Content-Type: application/cloudevents+json; charset=utf-8

{
  "specversion": "1.0",
  "id": "4f6c1a52-7c1e-4d55-9a5e-0b0f1c2d3e4f",
  "source": "https://orders.example.com/orders",
  "type": "com.example.orders.order.placed.v1",
  "subject": "orders/42",
  "time": "2026-10-11T09:30:00Z",
  "datacontenttype": "application/json",
  "dataschema": "https://schemas.example.com/orders/order-placed/v1.json",
  "data": {
    "orderId": "42",
    "customerId": "c-981",
    "totalCents": 12900,
    "currency": "EUR"
  }
}

// Read model shaped for the order list screen; rebuildable by replaying events.
// CREATE TABLE order_summaries (order_id text PRIMARY KEY, customer_name text, status text,
//   total_cents int, item_count int, version int NOT NULL);
import type { Pool } from "pg";

type OrderEvent =
  | { type: "OrderPlaced.v1"; orderId: string; version: number; customerName: string; totalCents: number; itemCount: number }
  | { type: "OrderShipped.v1" | "OrderCancelled.v1"; orderId: string; version: number };

export async function project(db: Pool, e: OrderEvent): Promise<void> {
  if (e.type === "OrderPlaced.v1") {
    await db.query(
      `INSERT INTO order_summaries (order_id, customer_name, status, total_cents, item_count, version)
       VALUES ($1, $2, 'placed', $3, $4, $5) ON CONFLICT (order_id) DO NOTHING`,
      [e.orderId, e.customerName, e.totalCents, e.itemCount, e.version]);
    return;
  }
  const status = e.type === "OrderShipped.v1" ? "shipped" : "cancelled";
  // The version guard turns duplicates and replays into no-ops and drops stale updates.
  await db.query(
    "UPDATE order_summaries SET status = $2, version = $3 WHERE order_id = $1 AND version < $3",
    [e.orderId, status, e.version]);
}

WHEN TO REACH FOR WHAT

WHERE IT BITES

• Dual writes: committing to the database and then publishing to the broker (or the reverse) from application code. A crash in between loses an event or publishes one for a rolled-back change. Use an outbox or the database's change stream.
• Non-idempotent consumers. Relays, brokers and retries all redeliver; a consumer that charges, emails or increments without a dedupe key will do it twice.
• Assuming global ordering. Order is usually per partition key at best; events for different aggregates, or after a consumer rebalance, can arrive in any order. Guard updates with versions.
• Breaking event schemas in place. Renaming a field breaks every consumer you do not know about and every stored event you replay. Add fields, version types, and run consumer contract tests.
• Publishing event-sourcing internals as the integration contract, which freezes your aggregate design. Map to explicit integration events at the boundary.
• UIs that read their own write from a lagging read model and show the old state. Return the command result or version, show pending states, or wait for the projection to catch up.

WHEN IT BREAKS IN PRODUCTION

• Webhooks arrive twice, late, or out of order
• One bad message blocks or floods the queue
• The UI shows old data after a save
• A retry creates two payments
• A brief blip turns into a long outage

SOURCES

• Martin Fowler: What do you mean by "Event-Driven"? (opens in new tab)Checked 2026-10-11
• microservices.io: Transactional outbox (opens in new tab)Checked 2026-10-11
• Azure Architecture Center: Saga design pattern (opens in new tab)Checked 2026-10-11
• Debezium: Outbox event router (opens in new tab)Checked 2026-10-11
• CloudEvents specification 1.0 (opens in new tab)Checked 2026-10-11
• Martin Fowler: CQRS (opens in new tab)Checked 2026-10-11