Sagas: عملية واحدة على تلات services من غير transaction واحدة

// saga.js: saga orchestrator بالـ steps والـ compensations، والـ retries، و state محفوظ يقدر يكمّل بعد crash
import { readFileSync, writeFileSync, renameSync, existsSync } from "node:fs";

export function defineSaga(name, steps) {
  if (!Array.isArray(steps) || !steps.length) throw new Error("الـ saga محتاجة step واحدة على الأقل");
  const seen = new Set();
  for (const s of steps) {
    if (!s?.name || seen.has(s.name)) throw new Error(`اسم step فاضي أو متكرر: ${s?.name}`);
    if (typeof s.action !== "function") throw new Error(`${s.name}: مفيش action`);
    if (s.compensate != null && typeof s.compensate !== "function") throw new Error(`${s.name}: الـ compensate لازم function`);
    seen.add(s.name);
  }
  return Object.freeze({ name, steps: Object.freeze([...steps]) });
}

export function createMemoryStore() {
  const m = new Map();
  return {
    get: (id) => (m.has(id) ? structuredClone(m.get(id)) : null),
    put: (r) => { m.set(r.sagaId, structuredClone(r)); },
    list: () => [...m.values()].map((r) => structuredClone(r)),
  };
}

export function createFileStore(path) {
  const read = () => (existsSync(path) ? JSON.parse(readFileSync(path, "utf8")) : {});
  return {
    get: (id) => read()[id] ?? null,
    put: (r) => {
      const all = read();
      all[r.sagaId] = r;
      writeFileSync(path + ".tmp", JSON.stringify(all, null, 2));
      renameSync(path + ".tmp", path);
    },
    list: () => Object.values(read()),
  };
}

const summary = (r) => ({
  sagaId: r.sagaId, status: r.status, failedStep: r.failedStep, error: r.error, compensationError: r.compensationError,
  completed: [...r.completed], compensated: [...r.compensated], results: structuredClone(r.results),
});

const ctxFor = (r, step, attempt) => ({
  sagaId: r.sagaId, input: structuredClone(r.input), results: structuredClone(r.results),
  idempotencyKey: `${r.sagaId}:${step.name}`, attempt,
});

async function drive(def, r, store, { compensationAttempts = 3 } = {}) {
  if (r.status === "running") {
    for (const step of def.steps) {
      if (r.completed.includes(step.name)) continue;
      let attempt = r.attempts[step.name] ?? 0;
      for (;;) {
        let out, failure = null;
        try { out = await step.action(ctxFor(r, step, attempt)); }
        catch (e) { failure = e instanceof Error ? e : new Error(String(e)); }
        // الـ put بره الـ try: لو الـ store نفسه وقع، ده crash مش فشل للـ step
        if (!failure) {
          r.results[step.name] = out ?? null;
          r.completed.push(step.name);
          store.put(r);
          break;
        }
        if (failure.retryable && attempt < (step.retries ?? 0)) {
          r.attempts[step.name] = ++attempt;
          store.put(r);
          continue;
        }
        r.status = "compensating"; r.failedStep = step.name; r.error = failure.message;
        store.put(r);
        break;
      }
      if (r.status !== "running") break;
    }
    if (r.status === "running") { r.status = "completed"; store.put(r); ret
…

$ node lab.js
البداية: stock mug 5 | balances mahmoud 1000، sara 300، youssef 800، karim 500 | bookings 0

s1: محمود، اتنين mug بأربعمية، للقاهرة:
   completed | completed [reserve,debit,ship] | compensated []
   stock mug 3 | balances mahmoud 600، sara 300، youssef 800، karim 500 | bookings 1

s2: سارة، الرصيد مش كفاية:
   compensated عند debit | completed [reserve] | compensated [reserve]
   error: /wallet/debit 402: الرصيد مش كفاية
   stock mug 3 | balances mahmoud 600، sara 300، youssef 800، karim 500 | bookings 1

s3: يوسف لسيوة، والشحن مش بيوصل هناك:
   compensated عند ship | completed [reserve,debit] | compensated [debit,reserve]
   error: /shipping/book 422: مفيش شحن لـ siwa
   stock mug 3 | balances mahmoud 600، sara 300، youssef 800، karim 500 | bookings 1

s4: كريم، والمحفظة بترجّع 503 مرتين:
   completed | completed [reserve,debit,ship] | compensated []
   calls على debit في s4: 3 و stock mug 2 | balances mahmoud 600، sara 300، youssef 800، karim 400 | bookings 2

s5: الشحن فشل، والـ refund نفسه واقع:
   stuck عند ship | completed [reserve,debit] | compensated []
   error: /shipping/book 422: مفيش شحن لـ siwa
   compensationError: debit: /wallet/refund 503: service unav
…

// checkout.naive.js: تلات services، وكل واحدة ليها database، والـ code بيكلّمهم ورا بعض وخلاص
export async function checkout(order) {
  const { reservationId } = await post("/inventory/reserve", { sku: order.sku, qty: order.qty });
  const { txId } = await post("/wallet/debit", { customer: order.customer, amount: order.amount });
  const { bookingId } = await post("/shipping/book", { city: order.city });
  // ← لو الشحن فشل: الـ stock محجوز والفلوس اتخصمت، ومحدش هيرجّعهم
  // ← لو الـ process وقعت بعد الـ debit: محدش فاكر إن فيه checkout كان شغّال أصلًا
  // ← ولو الـ debit رجّع timeout، تعيده؟ ممكن العميل يتخصم مرتين
  return { reservationId, txId, bookingId };
}
const post = async (path, body) => (await fetch("http://services" + path, { method: "POST", body: JSON.stringify(body) })).json();

// services.js: تلات services حقيقية على 127.0.0.1: inventory و wallet و shipping، وكل واحدة بتحترم الـ Idempotency-Key
import http from "node:http";

export async function startServices({ stock, balances, cities = ["cairo", "alex", "giza"] }) {
  const s = { stock: { ...stock }, balances: { ...balances }, reservations: {}, txs: {}, bookings: {}, calls: {}, fail: {} };
  const replay = new Map();
  let seq = 0;
  const id = (p) => `${p}${++seq}`;
  const routes = {
    "/inventory/reserve": ({ sku, qty }) => {
      if ((s.stock[sku] ?? 0) < qty) return [409, { error: `مفيش stock كفاية من ${sku}` }];
      s.stock[sku] -= qty;
      const r = id("r"); s.reservations[r] = { sku, qty };
      return [200, { reservationId: r }];
    },
    "/inventory/release": ({ reservationId }) => {
      const r = s.reservations[reservationId];
      if (r) { s.stock[r.sku] += r.qty; delete s.reservations[reservationId]; }
      return [200, { released: !!r }];
    },
    "/wallet/debit": ({ customer, amount }) => {
      if ((s.balances[customer] ?? 0) < amount) return [402, { error: "الرصيد مش كفاية" }];
      s.balances[customer] -= amount;
      const t = id("t"); s.txs[t] = { customer, amount };
      return [200, { txId: t }];
    },
    "/wallet/refund": ({ txId }) => {
      const t = s.txs[txId];
      if (t) { s.balances[t.customer] += t.amount; delete s.txs[txId]; }
      return [200, { refunded: !!t }];
    },
    "/shipping/book": ({ city }) => {
      if (!cities.includes(city)) return [422, { error: `مفيش شحن لـ ${city}` }];
      const b = id("b"); s.bookings[b] = { city };
      return [200, { bookingId: b }];
    },
    "/shipping/cancel": ({ bookingId }) => { const had = !!s.bookings[bookingId]; delete s.bookings[bookingId]; return [200, { cancelled: had }]; },
  };
  const server = http.createServer(async (req, res) => {
    let body = ""; for await (const c of req) body += c;
    const send = (code, obj) => res.writeHead(code, { "content-type": "application/json" }).end(JSON.stringify(obj));
    if (req.url === "/state") return send(200, s);
    const route = routes[req.url];
    if (!route) return send(404, { error: "not found" });
    s.calls[req.url] = (s.calls[req.url] ?? 0) + 1;
    if (s.fail[req.url] > 0) { s.fail[req.url]--; return send(503, { error: "service unavailable" }); }
    const key = req.headers["idempotency-key"];
    const k = `${req.url}:${key}`;
    if (key && replay.has(k)) return send(...replay.get(k));
    const out = route(JSON.parse(body || "{}"));
    if (key) replay.set(k, out);
    send(...out);
  });
  await new Promise((r) => server.listen(0, "127.0.0.1", r));
  const url = `http://127.0.0.1:${server.address().port}`;
  return {
    url,
    state: () => structuredClone(s),
    failNext: (path, n) => { s.fail[path] = n; },
    close: () => new Promise((r) => server.close(r)),
  };
}

// checkout.js: الـ checkout saga: احجز stock، واخصم من المحفظة، واحجز شحن. ولو اتشغّل لوحده بيبقى orchestrator ممكن يقع
import { fileURLToPath } from "node:url";
import { defineSaga, runSaga, createFileStore } from "./saga.js";

export function checkoutSaga(baseUrl) {
  const call = async (path, body, key) => {
    const res = await fetch(baseUrl + path, {
      method: "POST",
      headers: { "content-type": "application/json", ...(key && { "idempotency-key": key }) },
      body: JSON.stringify(body),
    });
    const data = await res.json();
    if (!res.ok) throw Object.assign(new Error(`${path} ${res.status}: ${data.error}`), { retryable: res.status >= 500 });
    return data;
  };
  return defineSaga("checkout", [
    {
      name: "reserve", retries: 2,
      action: ({ input, idempotencyKey }) => call("/inventory/reserve", { sku: input.sku, qty: input.qty }, idempotencyKey),
      compensate: ({ results, idempotencyKey }) => call("/inventory/release", { reservationId: results.reserve.reservationId }, idempotencyKey + ":undo"),
    },
    {
      name: "debit", retries: 3,
      action: ({ input, idempotencyKey }) => call("/wallet/debit", { customer: input.customer, amount: input.amount }, idempotencyKey),
      compensate: ({ results, idempotencyKey }) => call("/wallet/refund", { txId: results.debit.txId }, idempotencyKey + ":undo"),
    },
    {
      name: "ship", retries: 2,
      action: ({ input, idempotencyKey }) => call("/shipping/book", { city: input.city }, idempotencyKey),
      compensate: ({ results, idempotencyKey }) => call("/shipping/cancel", { bookingId: results.ship.bookingId }, idempotencyKey + ":undo"),
    },
  ]);
}

// node checkout.js <url> <store.json> <sagaId> <input json> <crashAfter>
if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const [url, file, sagaId, input, crashAfter] = process.argv.slice(2);
  const def = checkoutSaga(url);
  const crashing = { ...def, steps: def.steps.map((s) => ({ ...s, action: async (ctx) => { const out = await s.action(ctx); if (s.name === crashAfter) process.exit(137); return out; } })) };
  console.log(JSON.stringify(await runSaga(crashing, JSON.parse(input), { store: createFileStore(file), sagaId })));
}