CQRS و Event Sourcing: الـ events هي الحقيقة

// es.js: الـ account كـ decide و evolve، و event store على Postgres بالـ optimistic concurrency، و projections بـ checkpoints
export class DomainError extends Error {
  constructor(code, message = code) { super(message); this.code = code; }
}

export function evolve(state, e) {
  switch (e.type) {
    case "AccountOpened": return { owner: e.data.owner, balance: 0, status: "open" };
    case "MoneyDeposited": return { ...state, balance: state.balance + e.data.amount };
    case "MoneyWithdrawn": return { ...state, balance: state.balance - e.data.amount };
    case "AccountFrozen": return { ...state, status: "frozen" };
    case "AccountUnfrozen": return { ...state, status: "open" };
    default: return state;
  }
}

export function decide(cmd, state) {
  const fail = (code) => { throw new DomainError(code); };
  if (cmd.type === "Open") {
    if (state) fail("ALREADY_OPENED");
    if (!cmd.owner) fail("INVALID_OWNER");
    return [{ type: "AccountOpened", data: { owner: cmd.owner } }];
  }
  if (!["Deposit", "Withdraw", "Freeze", "Unfreeze"].includes(cmd.type)) fail("UNKNOWN_COMMAND");
  if (!state) fail("NOT_OPENED");
  if (cmd.type === "Freeze") return state.status === "frozen" ? [] : [{ type: "AccountFrozen", data: { reason: cmd.reason ?? null } }];
  if (cmd.type === "Unfreeze") return state.status === "open" ? [] : [{ type: "AccountUnfrozen", data: {} }];
  if (!Number.isInteger(cmd.amount) || cmd.amount <= 0) fail("INVALID_AMOUNT");
  if (state.status === "frozen") fail("ACCOUNT_FROZEN");
  if (cmd.type === "Deposit") return [{ type: "MoneyDeposited", data: { amount: cmd.amount } }];
  if (state.balance < cmd.amount) fail("INSUFFICIENT_FUNDS");
  return [{ type: "MoneyWithdrawn", data: { amount: cmd.amount } }];
}

async function tx(pool, fn) {
  const c = await pool.connect();
  try { await c.query("BEGIN"); const r = await fn(c); await c.query("COMMIT"); return r; }
  catch (e) { await c.query("ROLLBACK").catch(() => {}); throw e; }
  finally { c.release(); }
}

const conflict = (streamId, expected, actual) =>
  Object.assign(new Error(`${streamId}: expected ${expected} بس الـ version ${actual}`), { code: "WRONG_EXPECTED_VERSION", expected, actual });

export async function appendToStream(pool, streamId, expectedVersion, events) {
  if (!events.length) throw new Error("مفيش events");
  try {
    return await tx(pool, async (c) => {
      // lock واحد لكل الـ appends: الـ positions بتتعمل commit بالترتيب، فالـ projection عمره ما يعدّي واحدة
      await c.query("SELECT pg_advisory_xact_lock(4242)");
      const { rows } = await c.query("SELECT coalesce(max(version), 0)::int AS v FROM events WHERE stream_id = $1", [streamId]);
      if (rows[0].v !== expectedVersion) throw conflict(streamId, expectedVersion, rows[0].v);
      let v = expectedVersion;
      for (const e of events) await c.query("INSERT INTO events (stream_id, version, type, data) VALUES ($1, $2, $3, $4)", [streamId, ++v, e.type, JSON.stringify(e.data ?? {})]);
      return v;
…

$ node lab.js
① CRUD: اتنين سحبوا أربعمية من حساب فيه خمسمية في نفس الوقت (read، وبعدين احسب، وبعدين write):
   النتايج: تمام و تمام ← الرصيد: 100 ← تمانمية اتسحبوا من خمسمية، ومفيش أي أثر لده

② event sourcing: نفس الموقف:
   النتايج: تمام (version 3، محاولات 1) و INSUFFICIENT_FUNDS
   الـ stream: v1 AccountOpened ← v2 MoneyDeposited 500 ← v3 MoneyWithdrawn 400

③ CQRS: الـ commands بتكتب events، والـ queries بتقرا من read model:
   سحب من حساب frozen: ACCOUNT_FROZEN
   الـ read model قبل الـ projection: 0 rows، والـ lag: 8 events
   runProjection: 8 events، والـ lag: 0
     account-mahmoud  mahmoud    100  open  v3
     account-sara     sara      1200  frozen  v3
     account-youssef  youssef    300  open  v2
   يوسف سحب خمسين، والـ read model لسه: 300 ← eventual consistency. وبعد الـ projection: 250

④ read model جديد اتعمل النهارده، وبيتبني من كل الـ history:
   dailyTotals: 9 events ← [{"deposits":2000,"withdrawals":450}]
   حد غيّر الـ read model بإيده غلط. rebuild: 9 events، ورصيد سارة رجع: 1200

⑤ time travel: رصيد محمود عند كل version:
   v1: {"owner":"mahmoud","balance":0,"status":"open"}
   v2: {"owner":"mahmoud","balance":500,"status":"open"}
   v3: {"owner":"mahmoud","
…

// accounts.naive.js: CRUD عادي، الـ row فيها الرصيد الحالي وبس
import { pool } from "./db.js";

export async function withdraw(accountId, amount) {
  const { rows } = await pool.query("SELECT balance FROM accounts WHERE id = $1", [accountId]);
  if (rows[0].balance < amount) throw new Error("الرصيد مش كفاية");
  // ← بين الـ SELECT والـ UPDATE، request تاني ممكن يكون قرا نفس الرصيد
  await pool.query("UPDATE accounts SET balance = $1 WHERE id = $2", [rows[0].balance - amount, accountId]);
  // ← والرصيد القديم راح. مين سحب؟ إمتى؟ ليه؟ مفيش
}

// وصفحة الـ admin عايزة: مجموع الإيداعات كل يوم، و الحسابات اللي اتقفلت الشهر ده
// ← كل واحدة query تقيلة على نفس الـ table اللي الـ writes شغّالة عليه

-- schema.sql: الـ event store هو الحقيقة، والباقي read models بتتبني منه
DROP TABLE IF EXISTS events, projection_checkpoints, account_balances, daily_totals;

CREATE TABLE events (
  position   bigserial PRIMARY KEY,           -- ترتيب عام لكل الـ events
  stream_id  text  NOT NULL,                  -- الـ aggregate، زي account-mahmoud
  version    int   NOT NULL,                  -- ترتيب جُوّه الـ stream: واحد، اتنين، تلاتة
  type       text  NOT NULL,
  data       jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  UNIQUE (stream_id, version)                 -- مستحيل اتنين يكتبوا نفس الـ version
);

CREATE TABLE projection_checkpoints (
  name     text PRIMARY KEY,
  position bigint NOT NULL DEFAULT 0
);

-- read model للـ API: رصيد كل حساب في row واحدة
CREATE TABLE account_balances (
  account_id text PRIMARY KEY,
  owner      text NOT NULL,
  balance    int  NOT NULL,
  status     text NOT NULL,
  version    int  NOT NULL
);

-- read model للـ dashboard: مجموع كل يوم
CREATE TABLE daily_totals (
  day         date PRIMARY KEY,
  deposits    int NOT NULL DEFAULT 0,
  withdrawals int NOT NULL DEFAULT 0
);

// 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);

// projections.js: كل projection عبارة عن handlers لكل event type، و reset بيفضّي الـ read model بتاعه
export const balances = {
  name: "balances",
  reset: (c) => c.query("TRUNCATE account_balances"),
  handlers: {
    AccountOpened: (c, e) => c.query("INSERT INTO account_balances VALUES ($1, $2, 0, 'open', $3)", [e.streamId, e.data.owner, e.version]),
    MoneyDeposited: (c, e) => c.query("UPDATE account_balances SET balance = balance + $2, version = $3 WHERE account_id = $1", [e.streamId, e.data.amount, e.version]),
    MoneyWithdrawn: (c, e) => c.query("UPDATE account_balances SET balance = balance - $2, version = $3 WHERE account_id = $1", [e.streamId, e.data.amount, e.version]),
    AccountFrozen: (c, e) => c.query("UPDATE account_balances SET status = 'frozen', version = $2 WHERE account_id = $1", [e.streamId, e.version]),
    AccountUnfrozen: (c, e) => c.query("UPDATE account_balances SET status = 'open', version = $2 WHERE account_id = $1", [e.streamId, e.version]),
  },
};

const bump = (col) => (c, e) => c.query(
  `INSERT INTO daily_totals (day, ${col}) VALUES ($1::timestamptz::date, $2)
   ON CONFLICT (day) DO UPDATE SET ${col} = daily_totals.${col} + EXCLUDED.${col}`,
  [e.createdAt, e.data.amount],
);

export const dailyTotals = {
  name: "daily-totals",
  reset: (c) => c.query("TRUNCATE daily_totals"),
  handlers: { MoneyDeposited: bump("deposits"), MoneyWithdrawn: bump("withdrawals") },
};

// lab.js: CRUD بيضيّع update، وبعدين event sourcing و CQRS على Postgres حقيقي
import { pool, reset, count } from "./db.js";
import { handleCommand, readStream, loadAggregate, stateAt, runProjection, rebuildProjection, projectionLag, appendToStream } from "./es.js";
import { balances, dailyTotals } from "./projections.js";

await reset();
console.log("① CRUD: اتنين سحبوا أربعمية من حساب فيه خمسمية في نفس الوقت (read، وبعدين احسب، وبعدين write):");
await pool.query("DROP TABLE IF EXISTS crud_accounts; CREATE TABLE crud_accounts (id text PRIMARY KEY, balance int NOT NULL); INSERT INTO crud_accounts VALUES ('a', 500)");
const crudWithdraw = async (amount) => {
  const { rows } = await pool.query("SELECT balance FROM crud_accounts WHERE id = 'a'");
  await new Promise((r) => setTimeout(r, 20));
  if (rows[0].balance < amount) return "رفض";
  await pool.query("UPDATE crud_accounts SET balance = $1 WHERE id = 'a'", [rows[0].balance - amount]);
  return "تمام";
};
console.log("   النتايج:", (await Promise.all([crudWithdraw(400), crudWithdraw(400)])).join(" و "), "← الرصيد:", await count("SELECT balance n FROM crud_accounts"), "← تمانمية اتسحبوا من خمسمية، ومفيش أي أثر لده");

console.log("\n② event sourcing: نفس الموقف:");
const A = "account-mahmoud";
await handleCommand(pool, A, { type: "Open", owner: "mahmoud" });
await handleCommand(pool, A, { type: "Deposit", amount: 500 });
const res = await Promise.allSettled([handleCommand(pool, A, { type: "Withdraw", amount: 400 }), handleCommand(pool, A, { type: "Withdraw", amount: 400 })]);
console.log("   النتايج:", res.map((r) => (r.status === "fulfilled" ? `تمام (version ${r.value.version}، محاولات ${r.value.attempts})` : r.reason.code)).join(" و "));
console.log("   الـ stream:", (await readStream(pool, A)).map((e) => `v${e.version} ${e.type}${e.data.amount ? " " + e.data.amount : ""}`).join(" ← "));

console.log("\n③ CQRS: الـ commands بتكتب events، والـ queries بتقرا من read model:");
for (const [id, owner] of [["account-sara", "sara"], ["account-youssef", "youssef"]]) await handleCommand(pool, id, { type: "Open", owner });
await handleCommand(pool, "account-sara", { type: "Deposit", amount: 1200 });
await handleCommand(pool, "account-youssef", { type: "Deposit", amount: 300 });
await handleCommand(pool, "account-sara", { type: "Freeze", reason: "مراجعة" });
await handleCommand(pool, "account-sara", { type: "Freeze", reason: "تاني" });
try { await handleCommand(pool, "account-sara", { type: "Withdraw", amount: 100 }); } catch (e) { console.log("   سحب من حساب frozen:", e.code); }
console.log("   الـ read model قبل الـ projection:", await count("SELECT count(*) n FROM account_balances"), "rows، والـ lag:", await projectionLag(pool, "balances"), "events");
console.log("   runProjection:", await runProjection(pool, balances, { batchSize: 3 }), "events، والـ lag:", await projectionLag(pool, "balances"));
for (const r of (await pool.query("SELECT * FROM account_balances ORDER BY account_id")).rows) console.lo
…