Building Fault-Tolerant Event-Sourced Architectures: CQRS, Saga Orchestration, and Outbox Patterns

graph LR
    subgraph "Command Side"
        C1[API Gateway] -->|Command| SvcA[Service A (Command)]
        SvcA -->|Append| ES[Event Store (Kafka/CommitLog)]
        ES -->|Outbox| OutboxA[Outbox Table (Postgres)]
    end

    subgraph "Saga Orchestrator"
        Orc[Saga Coordinator] -->|Read| OutboxA
        Orc -->|Publish| Bus[Message Bus (NATS/Redis Streams)]
    end

    subgraph "Query Side"
        Bus -->|Event| SvcB[Service B (Projection)]
        SvcB -->|Update| View[Read Model (Cassandra/Elastic)]
        View -->|Query| API[Read API]
    end

BEGIN;
INSERT INTO events (aggregate_id, payload, version) VALUES (...);
INSERT INTO outbox (event_id, destination, payload, status) VALUES (...);
COMMIT;

sent

event_id + destination

type OutboxRow struct {
    ID          uuid.UUID
    Destination string
    Payload     []byte
    SentAt      sql.NullTime
}

func pollAndPublish(db *sql.DB, bus MessageBus) error {
    rows, err := db.Query(`
        SELECT id, destination, payload
        FROM outbox
        WHERE sent_at IS NULL
        ORDER BY created_at ASC
        LIMIT $1`, batchSize)
    if err != nil { return err }

    for rows.Next() {
        var o OutboxRow
        if err := rows.Scan(&o.ID, &o.Destination, &o.Payload); err != nil { return err }

        if err := bus.Publish(o.Destination, o.Payload); err != nil {
            // transient failure – break to retry later
            return err
        }

        // Mark as sent atomically
        _, err = db.Exec(`UPDATE outbox SET sent_at = NOW() WHERE id = $1`, o.ID)
        if err != nil { return err }
    }
    return nil
}

+-------------------+          +-------------------+          +-------------------+
|  Service Instance |          |   PostgreSQL      |          |   Kafka Cluster   |
| (Command Handler) |          |   (Outbox Table)  |          |   (Topic)         |
+--------+----------+          +--------+----------+          +--------+----------+
         |                               |                           |
         | 1. Begin DB transaction        |                           |
         |------------------------------->|                           |
         |                               |                           |
         | 2. Persist domain aggregate   |                           |
         |    (INSERT/UPDATE)            |                           |
         |------------------------------->|                           |
         |                               |                           |
         | 3. INSERT outbox row (event)  |                           |
         |------------------------------->|                           |
         |                               |                           |
         | 4. COMMIT transaction          |                           |
         |<-------------------------------|                           |
         |                               |                           |
         |                               | 5. Logical replication    |
         |                               |    (Debezium) reads WAL   |
         |                               |--------------------------->|
         |                               |                           |
         |                               | 6. Debezium transforms     |
         |                               |    to Kafka Connect record |
         |                               |--------------------------->|
         |                               |                           |
         |                               | 7. Kafka Connect writes    |
         |                               |    to topic                |
         |                               |<--------------------------|
         |                               |                           |

// cmd/service/main.go
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgxpool"
)

// -------------------------------------------------------------------
// Domain model – a simplistic "BankAccount" aggregate
// -------------------------------------------------------------------
type BankAccount struct {
	ID        string    // UUID
	Owner     string    // Human readable
	Balance   int64     // Cents
	UpdatedAt time.Time // Last mutation timestamp
}

// Command: Deposit money into an account
type DepositCmd struct {
	AccountID string
	Amount    int64 // cents, must be >0
}

// Event that will be persisted to the outbox
type MoneyDeposited struct {
	AccountID string `json:"account_id"`
	Amount    int64  `json:"amount"`
	At        int64  `json:"at_unix_ms"` // epoch ms
}

// -------------------------------------------------------------------
// Outbox schema (Postgres)
// CREATE TABLE outbox (
//   id          BIGSERIAL PRIMARY KEY,
//   aggregate_id UUID NOT NULL,
//   topic       TEXT NOT NULL,
//   key         BYTEA,
//   payload     BYTEA NOT NULL,
//   created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
//   UNIQUE (aggregate_id, topic, key)   -- idempotency guard
// );
// -------------------------------------------------------------------

// OutboxWriter abstracts the INSERT operation. It is deliberately
// kept tiny to avoid pulling in heavy ORMs.
type OutboxWriter struct {
	tx pgx.Tx // Transaction that already holds the domain mutation
}

// NewOutboxWriter returns a writer bound to the current transaction.
func NewOutboxWriter(tx pgx.Tx) *OutboxWriter {
	return &OutboxWriter{tx: tx}
}

// Insert serialises the event as JSON and stores it in the outbox.
// Returns the generated outbox row ID (useful for debugging).
func (ow *OutboxWriter) Insert(ctx context.Context, aggregateID string, topic string, key []byte, event any) (int64, error) {
	// 1️⃣ Serialise payload – use json.Marshaler interface for flexibility.
	payload, err := json.Marshal(event)
	if err != nil {
		return 0, fmt.Errorf("outbox payload marshal: %w", err)
	}

	// 2️⃣ INSERT with ON CONFLICT to guarantee idempotency.
	const sql = `
		INSERT INTO outbox (aggregate_id, topic, key, payload)
		VALUES ($1, $2, $3, $4)
		ON CONFLICT (aggregate_id, topic, key) DO NOTHING
		RETURNING id;
	`
	var outboxID int64
	err = ow.tx.QueryRow(ctx, sql, aggregateID, topic, key, payload).Scan(&outboxID)
	if err != nil {
		// pgx returns pgx.ErrNoRows when the ON CONFLICT path fires.
		if err == pgx.ErrNoRows {
			// Event already present – treat as success but report ID = 0.
			return 0, nil
		}
		return 0, fmt.Errorf("outbox insert: %w", err)
	}
	return outboxID, nil
}

// -------------------------------------------------------------------
// Service layer – orchestrates the command handling
// -------------------------------------------------------------------
type AccountService struct {
	db *pgxpool.Pool
}

// NewAccountService constructs the service with a connection pool.
func NewAccountService(db *pgxpool.Pool) *AccountService {
	return &AccountService{db: db}
}

// HandleDeposit processes a DepositCmd atomically and writes an outbox event.
func (as *AccountService) HandleDeposit(ctx context.Context, cmd DepositCmd) error {
	if cmd.Amount <= 0 {
		return fmt.Errorf("deposit amount must be positive")
	}

	// 1️⃣ Acquire a transaction with a *reasonable* timeout.
	tx, err := as.db.BeginTx(ctx, pgx.TxOptions{
		AccessMode: pgx.ReadWrite,
		// Use a short statement timeout to avoid long‑running locks.
		// This also forces the client to back‑off early.
	})
	if err != nil {
		return fmt.Errorf("begin tx: %w", err)
	}
	// Ensure rollback on any early exit.
	defer func() {
		_ = tx.Rollback(context.Background())
	}()

	// 2️⃣ Load the aggregate FOR UPDATE to enforce serialisation.
	var acct BankAccount
	const selSQL = `SELECT id, owner, balance, updated_at FROM bank_account WHERE id=$1 FOR UPDATE`
	if err := tx.QueryRow(ctx, selSQL, cmd.AccountID).Scan(
		&acct.ID, &acct.Owner, &acct.Balance, &acct.UpdatedAt,
	); err != nil {
		return fmt.Errorf("load account: %w", err)
	}

	// 3️⃣ Apply business rule.
	newBal := acct.Balance + cmd.Amount
	if newBal < 0 {
		return fmt.Errorf("insufficient funds after deposit")
	}
	acct.Balance = newBal
	acct.UpdatedAt = time.Now().UTC()

	// 4️⃣ Persist the updated aggregate.
	const updSQL = `
		UPDATE bank_account
		SET balance=$1, updated_at=$2
		WHERE id=$3;
	`
	if _, err := tx.Exec(ctx, updSQL, acct.Balance, acct.UpdatedAt, acct.ID); err != nil {
		return fmt.Errorf("update account: %w", err)
	}

	// 5️⃣ Write to the outbox.
	ow := NewOutboxWriter(tx)
	event := MoneyDeposited{
		AccountID: acct.ID,
		Amount:    cmd.Amount,
		At:        acct.UpdatedAt.UnixMilli(),
	}
	if _, err := ow.Insert(ctx, acct.ID, "bank.account.events", []byte(acct.ID), event); err != nil {
		return fmt.Errorf("outbox write: %w", err)
	}

	// 6️⃣ Commit – if this succeeds, the outbox row is durably stored.
	if err := tx.Commit(ctx); err != nil {
		return fmt.Errorf("tx commit: %w", err)
	}
	// At this point Debezium will see the new WAL record and push to Kafka.
	return nil
}

// -------------------------------------------------------------------
// Main entry point – wiring everything together
// -------------------------------------------------------------------
func main() {
	ctx := context.Background()
	dbURL := os.Getenv("POSTGRES_DSN")
	if dbURL == "" {
		log.Fatal("POSTGRES_DSN env var required")
	}
	cfg, err := pgxpool.ParseConfig(dbURL)
	if err != nil {
		log.Fatalf("parse dsn: %v", err)
	}
	// Tune the pool for high‑throughput event sourcing.
	cfg.MaxConns = 50
	cfg.MinConns = 5
	cfg.MaxConnLifetime = time.Hour
	cfg.HealthCheckPeriod = time.Minute

	pool, err := pgxpool.NewWithConfig(ctx, cfg)
	if err != nil {
		log.Fatalf("pgxpool init: %v", err)
	}
	defer pool.Close()

	svc := NewAccountService(pool)

	// Simulate a command received from an HTTP endpoint or message queue.
	cmd := DepositCmd{
		AccountID: "c3a1b2e4-5f67-8d90-1234-56789abcdef0",
		Amount:    10_00, // $10.00
	}
	if err := svc.HandleDeposit(ctx, cmd); err != nil {
		// In production we would push to a dead‑letter queue after N retries.
		log.Printf("[ERROR] deposit failed: %v", err)
		os.Exit(1)
	}
	log.Println("[INFO] deposit processed successfully")
}

# connectors/bank-outbox.yaml
name: bank-outbox-connector
config:
  connector.class: io.debezium.connector.postgresql.PostgresConnector
  database.hostname: postgres.internal
  database.port: 5432
  database.user: debezium
  database.password: ${DB_PASSWORD}
  database.dbname: bank
  database.server.name: bank-db
  slot.name: bank_outbox_slot
  publication.name: bank_outbox_pub
  table.include.list: public.outbox
  plugin.name: pgoutput
  # Transform to flatten the payload into a Kafka Connect record
  transforms: unwrap
  transforms.unwrap.type: io.debezium.transforms.ExtractNewRecordState
  transforms.unwrap.drop.tombstones: true
  # Optional: add a header with the aggregate_id for downstream routing
  transforms.addHeader.type: org.apache.kafka.connect.transforms.InsertHeader
  transforms.addHeader.header: aggregate_id
  transforms.addHeader.value: "${value.aggregate_id}"

max.poll.interval.ms

wal2json

-- events: immutable event stream
CREATE TABLE events (
    event_id      UUID PRIMARY KEY,
    aggregate_id  UUID NOT NULL,
    event_type    TEXT NOT NULL,
    payload       JSONB NOT NULL,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
    version       BIGINT NOT NULL,
    CONSTRAINT uq_aggregate_version UNIQUE (aggregate_id, version)
);

-- outbox: transient buffer for reliable delivery
CREATE TABLE outbox (
    outbox_id        BIGSERIAL PRIMARY KEY,
    event_id         UUID NOT NULL REFERENCES events(event_id) ON DELETE CASCADE,
    topic            TEXT NOT NULL,
    key              TEXT NOT NULL,
    payload          JSONB NOT NULL,
    idempotency_key  TEXT NOT NULL,
    processed_at     TIMESTAMPTZ,
    locked_until    TIMESTAMPTZ,
    retry_count      SMALLINT NOT NULL DEFAULT 0,
    CONSTRAINT uq_idempotency UNIQUE (idempotency_key)
);

import { createHash } from 'crypto';

export function buildIdempotencyKey(
  eventId: string,
  topic: string,
  key: string
): string {
  // SHA‑256 provides a 256‑bit collision‑resistant digest.
  const hash = createHash('sha256');
  hash.update(eventId);
  hash.update('|');
  hash.update(topic);
  hash.update('|');
  hash.update(key);
  return hash.digest('hex'); // 64‑char hex string
}

import { Pool, QueryResult } from 'pg';
import { v4 as uuidv4 } from 'uuid';
import { buildIdempotencyKey } from './idempotency';

interface DomainEvent {
  aggregateId: string;
  type: string;
  payload: any;
  version: number;
}

export async function persistEventAndEnqueue(
  pg: Pool,
  event: DomainEvent,
  topic: string,
  kafkaKey: string
): Promise<void> {
  const client = await pg.connect();
  try {
    await client.query('BEGIN');

    const eventId = uuidv4();
    const insertEvent = `
      INSERT INTO events (event_id, aggregate_id, event_type, payload, version)
      VALUES ($1, $2, $3, $4, $5)
    `;
    await client.query(insertEvent, [
      eventId,
      event.aggregateId,
      event.type,
      event.payload,
      event.version,
    ]);

    const idempotencyKey = buildIdempotencyKey(eventId, topic, kafkaKey);
    const insertOutbox = `
      INSERT INTO outbox (
        event_id, topic, key, payload, idempotency_key
      ) VALUES ($1, $2, $3, $4, $5)
      ON CONFLICT (idempotency_key) DO NOTHING
    `;
    await client.query(insertOutbox, [
      eventId,
      topic,
      kafkaKey,
      event.payload,
      idempotencyKey,
    ]);

    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

locked_until

import { Pool } from 'pg';
import { Kafka, Producer, Message } from 'kafkajs';
import { buildIdempotencyKey } from './idempotency';

const pg = new Pool({ connectionString: process.env.DATABASE_URL });
const kafka = new Kafka({ brokers: [process.env.KAFKA_BROKER!] });
const producer: Producer = kafka.producer({ 
  // Enable exactly‑once semantics (EOS) if broker supports it.
  idempotent: true,
  // Transactional ID must be stable per worker instance.
  transactionTimeout: 60000,
});

const WORKER_ID = `outbox-worker-${process.pid}`;
const LOCK_TTL_MS = 30_000; // 30 s lease

async function acquireLock(rowId: number): Promise<boolean> {
  const now = new Date();
  const lockUntil = new Date(now.getTime() + LOCK_TTL_MS);
  const res = await pg.query(
    `UPDATE outbox
     SET locked_until = $1
     WHERE outbox_id = $2
       AND (locked_until IS NULL OR locked_until < $3)
     RETURNING outbox_id`,
    [lockUntil, rowId, now]
  );
  return res.rowCount === 1;
}

async function processBatch(batchSize = 100): Promise<void> {
  const client = await pg.connect();
  try {
    // 1️⃣ Pull a batch of unlocked rows.
    const rows = await client.query(
      `SELECT *
       FROM outbox
       WHERE processed_at IS NULL
         AND (locked_until IS NULL OR locked_until < now())
       ORDER BY outbox_id
       LIMIT $1`,
      [batchSize]
    );

    for (const row of rows.rows) {
      const lockAcquired = await acquireLock(row.outbox_id);
      if (!lockAcquired) continue; // Another worker beat us.

      // 2️⃣ Begin a Kafka transaction.
      const tx = await producer.transaction();
      try {
        const kafkaMsg: Message = {
          key: row.key,
          value: JSON.stringify(row.payload),
          headers: {
            // Propagate the idempotency key to the consumer side.
            'idempotency-key': row.idempotency_key,
          },
        };

        await tx.send({
          topic: row.topic,
          messages: [kafkaMsg],
        });

        // 3️⃣ Commit transaction and mark row processed.
        await tx.commit();

        await client.query(
          `UPDATE outbox
           SET processed_at = now(),
               locked_until = NULL
           WHERE outbox_id = $1`,
          [row.outbox_id]
        );
      } catch (err) {
        // 4️⃣ Abort transaction, bump retry counter, release lock.
        await tx.abort();
        await client.query(
          `UPDATE outbox
           SET retry_count = retry_count + 1,
               locked_until = NULL
           WHERE outbox_id = $1`,
          [row.outbox_id]
        );
        // Propagate for higher‑level alerting.
        throw err;
      }
    }
  } finally {
    client.release();
  }
}

// ------------------------------------------------------------------
// Main daemon loop
// ------------------------------------------------------------------
(async () => {
  await producer.connect();
  console.log(`[${WORKER_ID}] Outbox worker started`);

  while (true) {
    try {
      await processBatch();
      // Back‑off when queue is empty.
      await new Promise(r => setTimeout(r, 200));
    } catch (e) {
      console.error(`[${WORKER_ID}] Fatal error:`, e);
      // Exponential back‑off on unrecoverable errors.
      await new Promise(r => setTimeout(r, 5_000));
    }
  }
})();

event_type

payload

OrderCreated_v2

event_version

ALTER TABLE events ADD COLUMN event_version SMALLINT NOT NULL DEFAULT 1;

// upcaster.ts
export type UpcastFn = (payload: any) => any;

/**
 * Registry mapping (type, version) → upcast function that
 * transforms payload to the **latest** version.
 */
export const upcasterRegistry = new Map<string, UpcastFn>();

/**
 * Helper to generate the registry key.
 */
function key(eventType: string, version: number): string {
  return `${eventType}_v${version}`;
}

/**
 * Register an upcaster that knows how to go from `fromVersion` → `toVersion`.
 */
export function registerUpcaster(
  eventType: string,
  fromVersion: number,
  toVersion: number,
  fn: UpcastFn
): void {
  if (toVersion !== fromVersion + 1) {
    throw new Error('Upcasters must be chained one version step at a time');
  }
  upcasterRegistry.set(key(eventType, fromVersion), fn);
}

/**
 * Resolve the full upcast chain for a given event.
 */
export function upcast(eventType: string, version: number, payload: any): any {
  let currentVersion = version;
  let currentPayload = payload;

  while (currentVersion < LATEST_VERSIONS[eventType]) {
    const fn = upcasterRegistry.get(key(eventType, currentVersion));
    if (!fn) {
      throw new Error(`Missing upcaster for ${eventType} v${currentVersion}`);
    }
    currentPayload = fn(currentPayload);
    currentVersion += 1;
  }
  return currentPayload;
}

/**
 * Define the latest version per event type (single source of truth).
 */
export const LATEST_VERSIONS: Record<string, number> = {
  OrderCreated: 2,
  OrderPaid: 2,
};

// orderCreated_v1_to_v2.ts
import { registerUpcaster } from './upcaster';

registerUpcaster('OrderCreated', 1, 2, (payload) => ({
  ...payload,
  // Default to USD when missing.
  currency: payload.currency ?? 'USD',
}));

// orderPaid_v1_to_v2.ts
import { registerUpcaster } from './upcaster';

registerUpcaster('OrderPaid', 1, 2, (payload) => ({
  ...payload,
  method: payload.method ?? 'unknown',
}));

import { Pool } from 'pg';
import { upcast } from './upcaster';

export async function loadAggregate(
  pg: Pool,
  aggregateId: string
): Promise<OrderAggregate> {
  const { rows } = await pg.query(
    `SELECT event_type, event_version, payload
     FROM events
     WHERE aggregate_id = $1
     ORDER BY version ASC`,
    [aggregateId]
  );

  const aggregate = new OrderAggregate(aggregateId);
  for (const { event_type, event_version, payload } of rows) {
    const latestPayload = upcast(event_type, event_version, payload);
    aggregate.apply(event_type, latestPayload);
  }
  return aggregate;
}

order_summary

-- staging table for upcasted events
CREATE TABLE events_upcasted (
    event_id      UUID PRIMARY KEY,
    aggregate_id  UUID NOT NULL,
    event_type    TEXT NOT NULL,
    payload       JSONB NOT NULL,
    created_at    TIMESTAMPTZ NOT NULL
);

-- Populate via a server‑side function (plpgsql) that calls the upcaster via PL/v