Building Resilient Event-Driven Architectures with Postgres and Serverless Queues

await pool.query('INSERT INTO orders (id, tenant_id, total_cents) VALUES ($1, $2, $3)', [id, tenantId, cents]);
await queue.publish('order.placed', { orderId: id });

-- migrations/0001_outbox.sql
CREATE TABLE outbox (
  id                bigserial    PRIMARY KEY,
  tenant_id         uuid         NOT NULL,
  aggregate_type    text         NOT NULL,
  aggregate_id      text         NOT NULL,
  event_type        text         NOT NULL,
  payload           jsonb        NOT NULL,
  headers           jsonb        NOT NULL DEFAULT '{}'::jsonb,
  idempotency_key   text         NOT NULL,
  created_at        timestamptz  NOT NULL DEFAULT now(),
  available_at      timestamptz  NOT NULL DEFAULT now(),
  locked_until      timestamptz,
  attempt_count     integer      NOT NULL DEFAULT 0,
  published_at      timestamptz,
  dead_lettered_at  timestamptz,
  last_error        text,
  CONSTRAINT outbox_idempotency_key_key UNIQUE (idempotency_key)
);

-- The only index the relay's claim query needs. Partial => stays small.
CREATE INDEX outbox_ready_idx
  ON outbox (available_at, id)
  WHERE published_at IS NULL AND dead_lettered_at IS NULL;

import type { Pool, PoolClient } from 'pg';

export type JsonObject = Record<string, unknown>;

export interface OutboxRow<T extends JsonObject = JsonObject> {
  readonly id: string;
  readonly tenant_id: string;
  readonly aggregate_type: string;
  readonly aggregate_id: string;
  readonly event_type: string;
  readonly payload: T;
  readonly headers: Record<string, string>;
  readonly idempotency_key: string;
  readonly attempt_count: number;
}

export async function withTransaction<T>(
  pool: Pool,
  fn: (tx: PoolClient) => Promise<T>,
): Promise<T> {
  const tx = await pool.connect();
  try {
    await tx.query('BEGIN');
    const result = await fn(tx);
    await tx.query('COMMIT');
    return result;
  } catch (err) {
    await tx.query('ROLLBACK').catch(() => undefined);
    throw err;
  } finally {
    tx.release();
  }
}

export interface NewEvent<T extends JsonObject> {
  tenantId: string;
  aggregateType: string;
  aggregateId: string;
  eventType: string;
  payload: T;
  headers?: Record<string, string>;
  idempotencyKey: string;
}

export async function enqueue<T extends JsonObject>(
  tx: PoolClient,
  event: NewEvent<T>,
): Promise<void> {
  await tx.query(
    `INSERT INTO outbox
       (tenant_id, aggregate_type, aggregate_id, event_type, payload, headers, idempotency_key)
     VALUES ($1, $2, $3, $4, $5::jsonb, $6::jsonb, $7)
     ON CONFLICT (idempotency_key) DO NOTHING`,
    [
      event.tenantId,
      event.aggregateType,
      event.aggregateId,
      event.eventType,
      JSON.stringify(event.payload),
      JSON.stringify(event.headers ?? {}),
      event.idempotencyKey,
    ],
  );
}

// --- call site: business state + event, atomically ---
export async function placeOrder(pool: Pool, orderId: string, tenantId: string, totalCents: number) {
  await withTransaction(pool, async (tx) => {
    await tx.query(
      `INSERT INTO orders (id, tenant_id, total_cents, status) VALUES ($1, $2, $3, 'placed')`,
      [orderId, tenantId, totalCents],
    );
    await enqueue(tx, {
      tenantId,
      aggregateType: 'order',
      aggregateId: orderId,
      eventType: 'order.placed',
      payload: { orderId, totalCents, currency: 'USD' },
      headers: { 'schema-version': '1' },
      idempotencyKey: `order.placed:${orderId}`,
    });
  });
}

WITH claimed AS (
  SELECT id
  FROM outbox
  WHERE published_at IS NULL
    AND dead_lettered_at IS NULL
    AND available_at <= now()
    AND (locked_until IS NULL OR locked_until < now())
  ORDER BY id
  LIMIT $1
  FOR UPDATE SKIP LOCKED
)
UPDATE outbox o
SET locked_until  = now() + ($2 * interval '1 second'),
    attempt_count = o.attempt_count + 1
FROM claimed
WHERE o.id = claimed.id
RETURNING o.id, o.tenant_id, o.aggregate_type, o.aggregate_id,
          o.event_type, o.payload, o.headers, o.idempotency_key, o.attempt_count;

import { Client as QStashClient } from '@upstash/qstash';
import type { Pool } from 'pg';

const CLAIM_SQL = /* the query above */;

function backoffSeconds(attempt: number): number {
  const base = Math.min(300, 2 ** attempt * 2);          // 4s, 8s, 16s ... capped at 5 min
  return Math.round(base * (0.5 + Math.random() * 0.5)); // full jitter
}

async function publishOne(
  pool: Pool,
  qstash: QStashClient,
  row: OutboxRow,
  maxAttempts: number,
): Promise<'published' | 'retry' | 'dead'> {
  try {
    await qstash.publishJSON({
      url: `${process.env.CONSUMER_BASE_URL}/api/events`,
      body: {
        id: row.id,
        type: row.event_type,
        aggregateType: row.aggregate_type,
        aggregateId: row.aggregate_id,
        tenantId: row.tenant_id,
        payload: row.payload,
      },
      headers: { ...row.headers, 'x-event-id': String(row.id) },
      retries: 3,
      deduplicationId: row.idempotency_key,
    });

    await pool.query(
      `UPDATE outbox SET published_at = now(), locked_until = NULL, last_error = NULL WHERE id = $1`,
      [row.id],
    );
    return 'published';
  } catch (err) {
    const message = err instanceof Error ? err.message : String(err);
    const exhausted = row.attempt_count >= maxAttempts;

    await pool.query(
      `UPDATE outbox
          SET locked_until     = NULL,
              available_at     = now() + ($2 * interval '1 second'),
              dead_lettered_at = CASE WHEN $3 THEN now() ELSE dead_lettered_at END,
              last_error       = $4
        WHERE id = $1`,
      [row.id, exhausted ? 0 : backoffSeconds(row.attempt_count), exhausted, message.slice(0, 2000)],
    );
    return exhausted ? 'dead' : 'retry';
  }
}

interface RelayOptions { batchSize: number; leaseSeconds: number; maxAttempts: number }

export async function relayBatch(
  pool: Pool,
  qstash: QStashClient,
  opts: RelayOptions,
): Promise<{ published: number; retried: number; dead: number }> {
  const { rows } = await pool.query<OutboxRow>(CLAIM_SQL, [opts.batchSize, opts.leaseSeconds]);
  const counts = { published: 0, retried: 0, dead: 0 };

  const outcomes = await Promise.allSettled(
    rows.map((row) => publishOne(pool, qstash, row, opts.maxAttempts)),
  );
  for (const o of outcomes) {
    if (o.status === 'fulfilled') counts[o.value === 'published' ? 'published' : o.value === 'retry' ? 'retried' : 'dead']++;
  }
  return counts;
}

CREATE TABLE processed_messages (
  consumer     text        NOT NULL,
  message_id   text        NOT NULL,
  processed_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (consumer, message_id)
);

export async function consumeOnce(
  pool: Pool,
  consumer: string,
  messageId: string,
  work: (tx: PoolClient) => Promise<void>,
): Promise<'processed' | 'duplicate'> {
  return withTransaction(pool, async (tx) => {
    const { rowCount } = await tx.query(
      `INSERT INTO processed_messages (consumer, message_id) VALUES ($1, $2)
       ON CONFLICT DO NOTHING`,
      [consumer, messageId],
    );
    if (rowCount === 0) return 'duplicate' as const;
    await work(tx);           // side effects commit atomically with the dedupe row
    return 'processed' as const;
  });
}

import { Receiver } from '@upstash/qstash';

const receiver = new Receiver({
  currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
  nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
});

export async function POST(req: Request): Promise<Response> {
  const raw = await req.text();
  const ok = await receiver
    .verify({ signature: req.headers.get('upstash-signature') ?? '', body: raw })
    .catch(() => false);
  if (!ok) return new Response('invalid signature', { status: 401 });

  const event = JSON.parse(raw) as { id: string; type: string; payload: JsonObject };
  const status = await consumeOnce(pool, 'billing-service', event.id, async (tx) => {
    await tx.query(`UPDATE invoices SET status = 'paid' WHERE order_id = $1`, [event.payload.orderId]);
  });
  return Response.json({ status }, { status: 200 }); // 200 for duplicates: stop retrying
}

-- Backlog, lag, and poison depth in one query
SELECT
  count(*) FILTER (WHERE published_at IS NULL AND dead_lettered_at IS NULL) AS backlog,
  coalesce(max(now() - created_at) FILTER (WHERE published_at IS NULL AND dead_lettered_at IS NULL),
           interval '0') AS oldest_lag,
  count(*) FILTER (WHERE dead_lettered_at IS NOT NULL) AS dead_lettered
FROM outbox;

-- Retention: prune in batches so you never hold a long lock
DELETE FROM outbox
WHERE ctid IN (
  SELECT ctid FROM outbox
  WHERE published_at < now() - interval '7 days'
  ORDER BY id LIMIT 5000
);