Transactional Outbox و Inbox: الـ event ما يضيعش وما يتنفذش مرتين

// outbox.js: الـ transactional outbox والـ relay، والـ inbox للـ consumer، والـ stats والـ cleanup
export async function withTransaction(pool, fn) {
  const client = await pool.connect();
  try {
    await client.query("BEGIN");
    const result = await fn(client);
    await client.query("COMMIT");
    return result;
  } catch (e) {
    await client.query("ROLLBACK").catch(() => {});
    throw e;
  } finally {
    client.release();
  }
}

export async function enqueue(client, { aggregateId, type, payload }) {
  const { rows } = await client.query(
    "INSERT INTO outbox (aggregate_id, type, payload) VALUES ($1, $2, $3) RETURNING id",
    [aggregateId, type, JSON.stringify(payload)],
  );
  return String(rows[0].id);
}

export function placeOrder(pool, { id, customer, totalMinor }) {
  return withTransaction(pool, async (c) => {
    await c.query("INSERT INTO orders (id, customer, total_minor) VALUES ($1, $2, $3)", [id, customer, totalMinor]);
    return enqueue(c, { aggregateId: id, type: "OrderPlaced", payload: { orderId: id, customer, totalMinor } });
  });
}

export function relayOnce(pool, publish, { batchSize = 10, maxAttempts = 5 } = {}) {
  return withTransaction(pool, async (c) => {
    const { rows } = await c.query(
      `SELECT id, aggregate_id, type, payload FROM outbox
        WHERE published_at IS NULL AND attempts < $1
        ORDER BY id LIMIT $2 FOR UPDATE SKIP LOCKED`,
      [maxAttempts, batchSize],
    );
    const blocked = new Set();
    const stats = { published: 0, failed: 0, skipped: 0 };
    for (const r of rows) {
      if (blocked.has(r.aggregate_id)) { stats.skipped++; continue; }
      try {
        await publish({ id: String(r.id), aggregateId: r.aggregate_id, type: r.type, payload: r.payload });
        await c.query("UPDATE outbox SET published_at = now() WHERE id = $1", [r.id]);
        stats.published++;
      } catch (e) {
        await c.query("UPDATE outbox SET attempts = attempts + 1, last_error = $2 WHERE id = $1", [r.id, String(e.message)]);
        blocked.add(r.aggregate_id);
        stats.failed++;
      }
    }
    return stats;
  });
}

export function handleOnce(pool, consumer, message, effect) {
  return withTransaction(pool, async (c) => {
    const { rowCount } = await c.query(
      "INSERT INTO inbox (consumer, message_id) VALUES ($1, $2) ON CONFLICT DO NOTHING",
      [consumer, message.id],
    );
    if (!rowCount) return "duplicate";
    await effect(c, message);
    return "processed";
  });
}

export async function pendingStats(pool, { maxAttempts = 5 } = {}) {
  const { rows } = await pool.query(
    `SELECT count(*) FILTER (WHERE attempts < $1) AS pending,
            count(*) FILTER (WHERE attempts >= $1) AS dead,
            floor(extract(epoch FROM now() - min(created_at) FILTER (WHERE attempts < $1))) AS oldest
       FROM outbox WHERE published_at IS NULL`,
    [maxAttempts],
  );
  const r = rows[0];
  return { pending: Number(r.pending), dead: Number(r.dead), oldestPendingSecon
…

$ node lab.js
① dual write: احفظ وبعدين انشر، والـ process بتقع في النص:
   الـ process اتقفلت
   orders: 1 و events في الـ broker: 0 ← الـ billing عمره ما هيعرف

② انشر الأول وبعدين احفظ، والحفظ بيفشل:
   duplicate key value violates unique constraint "orders_pkey"
   events في الـ broker: 1 ← event لـ order سارة اللي عمره ما اتحفظ

③ outbox: الـ order والـ event في نفس الـ transaction:
   crash قبل الـ COMMIT: الـ process اتقفلت
   orders: 0 و outbox: 0 ← ولا حاجة، والاتنين متفقين
   بعد placeOrder: {"pending":1,"dead":0,"oldestPendingSeconds":0} والـ broker لسه فاضي: 0
   relayOnce: {"published":1,"failed":0,"skipped":0} ← الـ broker: OrderPlaced#2

④ عشرين order، والـ broker واقع تلات مرات، وأربع acks ضاعوا (الـ broker استلم والـ relay ما عرفش):
   الـ relay لفّ 6 مرات، والـ broker استلم 24 message لعشرين event
   الـ billing: {"processed":20,"duplicate":4} ← invoices: 20 ومجموعها 210000
   ومن غير inbox كان هيبقى فيه 24 invoice، يعني عميل اتحاسب مرتين

⑤ ترتيب الـ events لنفس الـ order، والأول فشل مرة:
   أول لفّة: {"published":1,"failed":1,"skipped":2} ← o8 اتنشر، و o7 كله استنى
   الـ broker: o8:OrderPlaced  o7:OrderPlaced  o7:OrderPaid  o7:OrderShipped

⑥ اتنين relays شغّال
…

// orders.naive.js: الطريقة اللي كلنا بنبدأ بيها، اسمها dual write
import { pool } from "./db.js";
import { broker } from "./broker.js";

export async function placeOrder(order) {
  await pool.query("INSERT INTO orders (id, customer, total_minor) VALUES ($1, $2, $3)", [order.id, order.customer, order.totalMinor]);
  // ← لو الـ process وقعت هنا، أو الـ broker واقع، الـ order اتحفظ والـ event ضاع
  await broker.publish({ type: "OrderPlaced", orderId: order.id, totalMinor: order.totalMinor });
}

// والـ consumer: كل message توصل، invoice جديدة
broker.subscribe("OrderPlaced", async (m) => {
  await pool.query("INSERT INTO invoices (order_id, total_minor) VALUES ($1, $2)", [m.orderId, m.totalMinor]);
});

-- schema.sql: الـ orders، والـ outbox جنبها في نفس الـ database، والـ inbox عند الـ consumer
DROP TABLE IF EXISTS invoices, inbox, outbox, orders;

CREATE TABLE orders (
  id          text PRIMARY KEY,
  customer    text NOT NULL,
  total_minor int  NOT NULL CHECK (total_minor > 0),
  created_at  timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE outbox (
  id           bigserial PRIMARY KEY,
  aggregate_id text  NOT NULL,
  type         text  NOT NULL,
  payload      jsonb NOT NULL,
  created_at   timestamptz NOT NULL DEFAULT now(),
  published_at timestamptz,
  attempts     int NOT NULL DEFAULT 0,
  last_error   text
);
-- الـ relay بيدوّر على اللي لسه ما اتنشرش بس، فالـ index ده صغير على طول
CREATE INDEX outbox_pending ON outbox (id) WHERE published_at IS NULL;

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

-- مفيش unique على order_id هنا عن قصد: الـ inbox هو اللي بيمنع التكرار
CREATE TABLE invoices (
  id          bigserial PRIMARY KEY,
  order_id    text NOT NULL,
  total_minor int  NOT NULL
);

// db.js: الـ pool، و reset بيشغّل schema.sql من الأول
import pg from "pg";
import { readFileSync } from "node:fs";

export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL, max: 10 });
export const reset = () => pool.query(readFileSync(new URL("./schema.sql", import.meta.url), "utf8"));
export const count = async (sql, params = []) => Number((await pool.query(sql, params)).rows[0].n);

// lab.js: dual write بيضيّع events، وبعدين outbox و relay و inbox على Postgres حقيقي. الـ broker هنا array في الـ memory
import { pool, reset, count } from "./db.js";
import { withTransaction, enqueue, placeOrder, relayOnce, handleOnce, pendingStats, cleanup } from "./outbox.js";

const broker = {
  log: [], down: 0, lostAcks: 0, delayMs: 0,
  async publish(m) {
    if (this.down > 0) { this.down--; throw new Error("broker unavailable"); }
    if (this.delayMs) await new Promise((r) => setTimeout(r, this.delayMs));
    this.log.push(m);
    if (this.lostAcks > 0) { this.lostAcks--; throw new Error("ack timeout"); }
  },
};
const publish = (m) => broker.publish(m);
const crash = () => { throw new Error("الـ process اتقفلت"); };
const drain = async (opts) => { let n = 0; for (let i = 0; i < 50 && (await pendingStats(pool)).pending; i++) n++, await relayOnce(pool, publish, opts); return n; };

await reset();
console.log("① dual write: احفظ وبعدين انشر، والـ process بتقع في النص:");
try { await pool.query("INSERT INTO orders VALUES ('n1', 'mahmoud', 500)"); crash(); await publish({ type: "OrderPlaced", orderId: "n1" }); }
catch (e) { console.log("  ", e.message); }
console.log("   orders:", await count("SELECT count(*) n FROM orders"), "و events في الـ broker:", broker.log.length, "← الـ billing عمره ما هيعرف");

console.log("\n② انشر الأول وبعدين احفظ، والحفظ بيفشل:");
try { await publish({ type: "OrderPlaced", orderId: "n1" }); await pool.query("INSERT INTO orders VALUES ('n1', 'sara', 700)"); }
catch (e) { console.log("  ", e.message); }
console.log("   events في الـ broker:", broker.log.length, "← event لـ order سارة اللي عمره ما اتحفظ");

await reset(); broker.log = [];
console.log("\n③ outbox: الـ order والـ event في نفس الـ transaction:");
try { await withTransaction(pool, async (c) => { await c.query("INSERT INTO orders VALUES ('o0', 'karim', 900)"); await enqueue(c, { aggregateId: "o0", type: "OrderPlaced", payload: {} }); crash(); }); }
catch (e) { console.log("   crash قبل الـ COMMIT:", e.message); }
console.log("   orders:", await count("SELECT count(*) n FROM orders"), "و outbox:", await count("SELECT count(*) n FROM outbox"), "← ولا حاجة، والاتنين متفقين");
await placeOrder(pool, { id: "o1", customer: "mahmoud", totalMinor: 44900 });
console.log("   بعد placeOrder:", JSON.stringify(await pendingStats(pool)), "والـ broker لسه فاضي:", broker.log.length);
console.log("   relayOnce:", JSON.stringify(await relayOnce(pool, publish)), "← الـ broker:", broker.log.map((m) => `${m.type}#${m.id}`).join());

await reset(); broker.log = [];
console.log("\n④ عشرين order، والـ broker واقع تلات مرات، وأربع acks ضاعوا (الـ broker استلم والـ relay ما عرفش):");
for (let i = 1; i <= 20; i++) await placeOrder(pool, { id: `o${i}`, customer: "u" + i, totalMinor: 1000 * i });
broker.down = 3; broker.lostAcks = 4;
console.log("   الـ relay لفّ", await drain({ batchSize: 5 }), "مرات، والـ broker استلم", broker.log.length, "message لعشرين event");
const results 
…