Событийная архитектура и очереди сообщений

docker run -d --rm --name kb-pg-events -e POSTGRES_PASSWORD=test \
  -p 127.0.0.1:55432:5432 postgres:17-alpine
docker run -d --rm --name kb-rabbit-events \
  -p 127.0.0.1:55672:5672 -p 127.0.0.1:55673:15672 rabbitmq:4.3-management-alpine

mkdir outbox-demo && cd outbox-demo
npm init -y && npm pkg set type=module
npm install pg@8.23.1 amqplib@2.2.0
npm install -D typescript@7.0.2 @types/pg @types/node@24

{
  "compilerOptions": {
    "target": "es2024",
    "module": "nodenext",
    "strict": true,
    "noEmit": true,
    "allowImportingTsExtensions": true,
    "erasableSyntaxOnly": true,
    "verbatimModuleSyntax": true,
    "skipLibCheck": true
  },
  "include": ["src"]
}

-- Orders service: business table + outbox in the same database
CREATE SCHEMA orders;

CREATE TABLE orders.orders (
  id          bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id bigint  NOT NULL,
  total_kzt   integer NOT NULL CHECK (total_kzt > 0),
  created_at  timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE orders.outbox (
  id             bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  event_id       uuid   NOT NULL UNIQUE DEFAULT gen_random_uuid(),
  aggregate_type text   NOT NULL,
  aggregate_id   text   NOT NULL,
  event_type     text   NOT NULL,
  payload        jsonb  NOT NULL,
  created_at     timestamptz NOT NULL DEFAULT now(),
  published_at   timestamptz
);
-- the relay only looks for unpublished rows
CREATE INDEX outbox_unpublished ON orders.outbox (id) WHERE published_at IS NULL;

-- Loyalty service: its own tables (in production — its own database)
CREATE SCHEMA loyalty;

-- a ledger: every row is a points accrual, nothing stops a duplicate row
CREATE TABLE loyalty.ledger (
  id          bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
  customer_id bigint  NOT NULL,
  order_id    bigint  NOT NULL,
  points      integer NOT NULL
);

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

docker exec -i kb-pg-events psql -U postgres < sql/schema.sql

import { Pool } from 'pg';

export const pool = new Pool({
  connectionString: process.env.DATABASE_URL ?? 'postgres://postgres:test@localhost:55432/postgres',
});

import { pool } from './db.ts';

export type OrderPlaced = { orderId: number; customerId: number; totalKzt: number };

// The order and its event are written in ONE transaction:
// either both exist or neither does.
export async function placeOrder(customerId: number, totalKzt: number): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const { rows } = await client.query<{ id: string }>(
      'INSERT INTO orders.orders (customer_id, total_kzt) VALUES ($1, $2) RETURNING id',
      [customerId, totalKzt],
    );
    const orderId = Number(rows[0].id);
    const event: OrderPlaced = { orderId, customerId, totalKzt };
    await client.query(
      `INSERT INTO orders.outbox (aggregate_type, aggregate_id, event_type, payload)
       VALUES ('order', $1, 'order.placed', $2)`,
      [String(orderId), event],
    );
    await client.query('COMMIT');
    return orderId;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

// node src/place-order.ts <customerId> <totalKzt>
if (import.meta.main) {
  const [customerId, totalKzt] = process.argv.slice(2).map(Number);
  const orderId = await placeOrder(customerId, totalKzt);
  console.log(`[orders] order ${orderId} saved, event written to outbox`);
  await pool.end();
}

import { connect, type Channel } from 'amqplib';

export const EXCHANGE = 'orders'; // events of the orders service
export const DLX = 'orders.dlx'; // where rejected messages go
export const QUEUE = 'loyalty.order-placed'; // the loyalty service's queue
export const DLQ = 'loyalty.order-placed.dlq';

export function connectRabbit() {
  return connect(process.env.AMQP_URL ?? 'amqp://guest:guest@localhost:55672');
}

// Declarations are idempotent: every process may call this on start
export async function assertTopology(ch: Channel) {
  await ch.assertExchange(EXCHANGE, 'topic', { durable: true });
  await ch.assertExchange(DLX, 'fanout', { durable: true });
  await ch.assertQueue(DLQ, { durable: true, arguments: { 'x-queue-type': 'quorum' } });
  await ch.bindQueue(DLQ, DLX, '');
  await ch.assertQueue(QUEUE, {
    durable: true,
    arguments: {
      'x-queue-type': 'quorum', // replicated durable queue
      'x-dead-letter-exchange': DLX,
      'x-delivery-limit': 5, // more than 5 failed deliveries -> DLQ
      'x-delayed-retry-type': 'failed', // RabbitMQ 4.3+: pause before a retry
      'x-delayed-retry-min': 1000, // 1 s, 2 s, 3 s... (linear back-off)
      'x-delayed-retry-max': 10000,
    },
  });
  await ch.bindQueue(QUEUE, EXCHANGE, 'order.placed');
}

import { setTimeout as sleep } from 'node:timers/promises';
import { pool } from './db.ts';
import { assertTopology, connectRabbit, EXCHANGE } from './rabbit.ts';

type OutboxRow = { id: string; event_id: string; event_type: string; payload: unknown; created_at: Date };

const BATCH = 100;
const once = process.argv.includes('--once');

const conn = await connectRabbit();
const ch = await conn.createConfirmChannel(); // the broker confirms every publish
await assertTopology(ch);

async function publishBatch(): Promise<number> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    // SKIP LOCKED: several relay instances take different rows instead of waiting
    const { rows } = await client.query<OutboxRow>(
      `SELECT id, event_id, event_type, payload, created_at
         FROM orders.outbox
        WHERE published_at IS NULL
        ORDER BY id
        LIMIT $1
        FOR UPDATE SKIP LOCKED`,
      [BATCH],
    );
    if (rows.length === 0) {
      await client.query('COMMIT');
      return 0;
    }
    for (const row of rows) {
      ch.publish(EXCHANGE, row.event_type, Buffer.from(JSON.stringify(row.payload)), {
        messageId: row.event_id, // consumers deduplicate by this id
        type: row.event_type,
        contentType: 'application/json',
        persistent: true,
        timestamp: Math.floor(row.created_at.getTime() / 1000),
      });
    }
    await ch.waitForConfirms(); // rejects if the broker did not take a message
    console.log(`[relay] published ${rows.length}: ${rows.map((r) => r.event_id.slice(0, 8)).join(', ')}`);

    if (process.env.CRASH_AFTER_PUBLISH) {
      console.log('[relay] simulated crash before marking rows as published');
      process.exit(1); // the transaction is rolled back, the rows stay unpublished
    }

    await client.query('UPDATE orders.outbox SET published_at = now() WHERE id = ANY($1::bigint[])', [
      rows.map((r) => r.id),
    ]);
    await client.query('COMMIT');
    return rows.length;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

do {
  const sent = await publishBatch();
  if (!once && sent === 0) await sleep(1000); // polling interval when idle
} while (!once);

await ch.close();
await conn.close();
await pool.end();

import type { ConsumeMessage } from 'amqplib';
import { pool } from './db.ts';
import type { OrderPlaced } from './place-order.ts';
import { assertTopology, connectRabbit, QUEUE } from './rabbit.ts';

const CONSUMER = 'loyalty.points';

const conn = await connectRabbit();
const ch = await conn.createChannel();
await assertTopology(ch);
await ch.prefetch(10); // at most 10 unacknowledged messages per consumer

function parse(msg: ConsumeMessage): OrderPlaced | null {
  try {
    const e = JSON.parse(msg.content.toString());
    return Number.isInteger(e.orderId) && Number.isInteger(e.customerId) && e.totalKzt > 0 ? e : null;
  } catch {
    return null;
  }
}

async function handle(msg: ConsumeMessage) {
  const id: string | undefined = msg.properties.messageId;
  const tag = `${id?.slice(0, 8)}${msg.fields.redelivered ? ' (redelivered)' : ''}`;
  const event = parse(msg);
  if (!id || !event) {
    console.log(`[loyalty] ${tag} malformed -> DLQ`);
    ch.reject(msg, false); // requeue = false: a retry cannot fix bad data -> DLQ
    return;
  }

  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    // Idempotency: the message id and the effect are committed together
    const seen = await client.query(
      `INSERT INTO loyalty.processed_messages (consumer, message_id)
       VALUES ($1, $2) ON CONFLICT DO NOTHING`,
      [CONSUMER, id],
    );
    if (seen.rowCount === 0) {
      await client.query('ROLLBACK');
      console.log(`[loyalty] ${tag} duplicate, skipped`);
      ch.ack(msg);
      return;
    }
    const points = Math.floor(event.totalKzt / 100); // 1 point per 100 KZT
    await client.query('INSERT INTO loyalty.ledger (customer_id, order_id, points) VALUES ($1, $2, $3)', [
      event.customerId,
      event.orderId,
      points,
    ]);
    await client.query('COMMIT');
    console.log(`[loyalty] ${tag} order ${event.orderId}: +${points} points`);

    if (process.env.CRASH_BEFORE_ACK) {
      console.log('[loyalty] simulated crash after commit, before ack');
      process.exit(1); // the broker will redeliver the unacknowledged message
    }
    ch.ack(msg); // ack only after the effect is committed
  } catch (err) {
    await client.query('ROLLBACK').catch(() => {});
    console.error(`[loyalty] ${new Date().toISOString().slice(11, 19)} ${tag} failed: ${(err as Error).message}`);
    ch.reject(msg, true); // counts as a failed delivery: delayed retry, then DLQ
  } finally {
    client.release();
  }
}

console.log(`[loyalty] waiting for messages in ${QUEUE}`);
await ch.consume(QUEUE, (msg) => {
  if (!msg) return; // null means the broker cancelled the consumer
  handle(msg).catch((err) => {
    // e.g. the database is unreachable: return the message to the queue
    console.error(`[loyalty] ${(err as Error).message}`);
    ch.reject(msg, true);
  });
});

$ node src/place-order.ts 101 12000
[orders] order 1 saved, event written to outbox
$ node src/place-order.ts 102 4500
[orders] order 2 saved, event written to outbox
$ docker exec kb-pg-events psql -U postgres -c "INSERT INTO orders.outbox (aggregate_type, aggregate_id, event_type, payload) VALUES ('order', '999', 'order.placed', '{\"orderId\": \"oops\"}')"
INSERT 0 1
$ CRASH_AFTER_PUBLISH=1 node src/relay.ts --once
[relay] published 3: a17a9f69, f883ea62, e1c7b7f3
[relay] simulated crash before marking rows as published
$ node src/relay.ts --once
[relay] published 3: a17a9f69, f883ea62, e1c7b7f3

# node src/consumer.ts (another terminal)
[loyalty] waiting for messages in loyalty.order-placed
[loyalty] e1c7b7f3 malformed -> DLQ
[loyalty] f883ea62 order 2: +45 points
[loyalty] a17a9f69 order 1: +120 points
[loyalty] e1c7b7f3 malformed -> DLQ
[loyalty] a17a9f69 duplicate, skipped
[loyalty] f883ea62 duplicate, skipped

$ node src/place-order.ts 103 8000
[orders] order 3 saved, event written to outbox
$ node src/relay.ts --once
[relay] published 1: 09d38984

# CRASH_BEFORE_ACK=1 node src/consumer.ts
[loyalty] waiting for messages in loyalty.order-placed
[loyalty] 09d38984 order 3: +80 points
[loyalty] simulated crash after commit, before ack

# node src/consumer.ts (restarted)
[loyalty] waiting for messages in loyalty.order-placed
[loyalty] 09d38984 (redelivered) duplicate, skipped

$ docker exec kb-pg-events psql -U postgres -c "SELECT customer_id, order_id, points FROM loyalty.ledger ORDER BY order_id"
 customer_id | order_id | points 
-------------+----------+--------
         101 |        1 |    120
         102 |        2 |     45
         103 |        3 |     80
(3 rows)

$ docker exec kb-pg-events psql -U postgres -c "ALTER TABLE loyalty.ledger RENAME TO ledger_broken"
ALTER TABLE
$ node src/place-order.ts 104 3000
[orders] order 4 saved, event written to outbox
$ node src/relay.ts --once
[relay] published 1: 670fd021

# node src/consumer.ts
[loyalty] waiting for messages in loyalty.order-placed
[loyalty] 02:28:57 670fd021 failed: relation "loyalty.ledger" does not exist
[loyalty] 02:28:58 670fd021 (redelivered) failed: relation "loyalty.ledger" does not exist
[loyalty] 02:29:00 670fd021 (redelivered) failed: relation "loyalty.ledger" does not exist
[loyalty] 02:29:03 670fd021 (redelivered) failed: relation "loyalty.ledger" does not exist
[loyalty] 02:29:07 670fd021 (redelivered) failed: relation "loyalty.ledger" does not exist
[loyalty] 02:29:12 670fd021 (redelivered) failed: relation "loyalty.ledger" does not exist

$ docker exec kb-rabbit-events rabbitmqctl -q list_queues name messages
name	messages
loyalty.order-placed	0
loyalty.order-placed.dlq	3

curl -s -u guest:guest -H 'content-type: application/json' \
  -X POST http://localhost:55673/api/queues/%2F/loyalty.order-placed.dlq/get \
  -d '{"count":5,"ackmode":"ack_requeue_true","encoding":"auto"}'

e1c7b7f3 {"orderId":"oops"} | reason: rejected | x-delivery-count: 1
e1c7b7f3 {"orderId":"oops"} | reason: rejected | x-delivery-count: 1
670fd021 {"orderId":4,"totalKzt":3000,"customerId":104} | reason: delivery_limit | x-delivery-count: 6