الـ Queues والـ Brokers: BullMQ و RabbitMQ و Kafka و SQS

أهلًا بيك. في الدروس اللي فاتت عملنا outbox و sagas و projections، وكلهم بيقولوا ابعت message. بس ابعتها فين؟ وإيه اللي يضمن إنها توصل؟ بص على jobs dot naive dot js. queue جُوّه الـ process، في array عادية. وده أول حاجة أي حد بيكتبها. take بتشيل الـ job من الـ array أول ما الـ worker ياخده، قبل ما يخلص. لو الـ worker وقع في النص، الـ job راح خالص. الـ process عمل restart؟ الـ array كلها راحت. عندك تلات instances؟ كل واحدة ليها queue لوحدها. job فيه data بايزة وبيوقّع الـ worker كل مرة؟ مفيش حد بيعدّ. علشان كده بنستخدم broker: process منفصل، أو service، الـ messages بتتخزّن فيه، وبيدّيها للـ workers بقواعد واضحة.

أول سؤال تسأله لأي broker: بيضمن إيه في التوصيل؟ at-most-once: الرسالة بتوصل مرة أو مبتوصلش خالص. زي الـ array بتاعتنا. سريع، بس ممكن تضيع. at-least-once: الرسالة مش هتتمسح غير لما الـ consumer يقول خلصت، يعني ack. لو وقع قبل الـ ack، الرسالة ترجع لحد تاني. مش هتضيع، بس ممكن توصل مرتين. وده الـ default في أغلب الـ brokers. exactly-once: الكل بيحلم بيه. في الحقيقة، على الشبكة، مفيش حاجة اسمها exactly-once delivery. اللي فيه effectively-once: at-least-once، ومعاه consumer idempotent، يعني لو الرسالة جت مرتين، المرة التانية ما تعملش حاجة. زي الـ inbox اللي عملناه قبل كده.

إزاي الـ queue بيعرف إن الـ consumer وقع؟ في SQS، اسمها visibility timeout: لما consumer ياخد رسالة، الرسالة بتستخبى عن الباقيين، تلاتين تانية على سبيل المتال. لو عمل ack، يعني delete، تتمسح. لو الوقت خلص من غير ack، ترجع تبان، وحد تاني ياخدها. ولو الشغل طويل، الـ consumer يطوّل الوقت، وده اسمه heartbeat أو extend. في RabbitMQ، الـ broker شايف الـ connection نفسه. لو الـ connection اتقفل والرسالة من غير ack، ترجع على طول. وفي BullMQ، الـ job ليه lock في Redis، والـ worker بيجدده. لو الـ lock خلص، الـ job بيبقى stalled، ويرجع. وكل مرة الرسالة بتتسلّم، العدّاد بتاعها بيزيد. لو وصل لحد معيّن، زي خمس مرات، يبقى دي poison message، وبتتنقل للـ DLQ، يعني dead letter queue. هناك بتستنى بني آدم يبص عليها، ولما يصلّح الـ bug، يعمل redrive يرجّعها.

والترتيب؟ أغلب الـ queues مش بتضمن ترتيب لما يبقى عندك أكتر من consumer. لأن consumer ممكن ياخد الرسالة الأولى ويتأخر، وتاني ياخد التانية ويخلّص قبله. بس غالبًا انت مش محتاج ترتيب كامل. انت محتاج ترتيب جُوّه الـ user الواحد: user اتسجّل، وبعدين غيّر الإيميل. فيه SQS FIFO بالـ message group: الرسايل اللي في نفس الـ group بتتسلّم واحدة واحدة بالترتيب، و groups مختلفة شغّالة بالتوازي. و SQS FIFO كمان فيه dedup: لو بعتّ نفس الـ deduplication id مرتين في خمس دقايق، التانية بتتشال. وده بيحمي من الـ producer اللي عمل retry. بس خلّي بالك، ده مش بيلغي إن الـ consumer لازم يبقى idempotent.

دلوقتي النوع التاني خالص: الـ log. Kafka، وكمان Redis Streams و RabbitMQ Streams و Kinesis. الـ topic مقسوم partitions، وكل partition هو log بيتضاف في آخره بس. وكل رسالة ليها offset، رقمها في الـ partition. الـ producer بيختار الـ partition من الـ key. نفس الـ key، يعني نفس الـ user، دايمًا في نفس الـ partition، فبتترتب. والقراية مش بتمسح حاجة! الرسايل بتفضل لحد الـ retention، أيام أو للأبد. وكل consumer group بيحفظ هو وصل لحد أنهي offset في كل partition. billing ليه offsets، و analytics ليه offsets تانية. group جديد ممكن يقرا كل الـ history من الأول. ولو فيه bug، ترجع الـ offset لورا، وده اسمه replay. والـ scaling في Kafka ليه قاعدة مهمة: جُوّه الـ group الواحد، كل partition بيقراه consumer واحد بس. فلو عندك تلات partitions، أقصى حاجة تلات consumers شغّالين. الرابع هيقعد فاضي. يعني عدد الـ partitions هو حد الـ parallelism، فبتختاره من الأول بحساب. ولما consumer يدخل أو يخرج، بيحصل rebalance: الـ partitions بتتوزّع تاني. وهنا مشكلة: consumer قديم كان بيشتغل على partition، والـ partition راحت لحد تاني، فلو حاول يعمل commit، لازم يترفض. ده اسمه fencing. والـ commit بعد الشغل، مش قبله. لو عملت commit وبعدين وقعت، الرسالة ضاعت. ولو اشتغلت ووقعت قبل الـ commit، هتتقري تاني. يعني برضه at-least-once، وبرضه consumer idempotent.

نقارن بقى. BullMQ: library على Redis. ممتاز للـ background jobs في app بـ Node: delays، و retries بـ backoff، و priorities، و cron، و rate limit، و dashboard. ولو عندك Redis أصلًا، مفيش حاجة جديدة تشغّلها. بس الـ data في الـ memory بتاعة Redis، فلازم persistence كويسة، ومش معمول لـ history طويلة. RabbitMQ: broker كامل بالـ AMQP. قوته في الـ routing: exchanges من نوع direct و topic و fanout، والرسالة توصل للـ queues الصح. و ack لكل رسالة، و DLX، و priorities. كويس بين services بلغات مختلفة. SQS: managed من AWS. مفيش servers، بيكبر لوحده، وبتدفع على الـ request. Standard بيدّي at-least-once من غير ترتيب، و FIFO بالـ groups والـ dedup بس أبطأ. ومعاه SNS للـ fanout. Kafka: log موزّع. throughput عالي أوي، و retention، و replay، و consumer groups كتير على نفس الـ data. بس تشغيله تقيل، والـ per-message retries والـ delays مش طبيعية فيه.

طب أختار إزاي؟ اسأل الأسئلة دي بالترتيب: محتاج replay؟ يعني تقرا اللي فات تاني، أو service جديدة تتبني من الـ history؟ أو عندك teams كتير كل واحد عايز نفس الـ events؟ أو عشرات الآلاف في التانية؟ ده Kafka. لو لأ: انت على AWS ومش عايز تشغّل حاجة؟ SQS. لو لأ: محتاج routing معقّد بين services؟ RabbitMQ. لو لأ: jobs في app بـ Node، وعندك Redis؟ BullMQ. ومتنساش: أبسط حاجة تكفي هي الأحسن. كتير من الـ teams بتبدأ بـ Kafka علشان مشهور، وهي محتاجة queue بسيط. والعكس كمان: ناس بتحاول تعمل event sourcing على SQS، وده مش هينفع، لأن الرسالة بتتمسح. التحدي. عندك lab dot js، بيعيد نفس المواقف: الـ array اللي بتضيّع، والـ queue بالـ ack، والـ DLQ، والـ FIFO، والـ log بالـ partitions والـ groups والـ replay، والاختيار. والمطلوب broker dot js، بساعة now بتتحقن علشان الـ tests تتحكم في الوقت. كل التفاصيل في الكومنت: createQueue بالـ visibility timeout والـ handles والـ DLQ والـ dedup والـ FIFO، و partitionFor، و createLog بالـ consumer groups والـ commit والـ fencing والـ seek، و chooseBroker. اقرا الكومنت كويس، وخصوصًا الـ FIFO مع الـ DLQ، ونشغّل test dot js لحد ما تبقى أربعتاشر من أربعتاشر.

الحل. الـ queue: مفيش timers خالص. كل رسالة فيها visibleAt، و in flight معناها إن الوقت ده لسه ما جاش. فالرسالة بترجع لوحدها لما الساعة تعدّي. الـ handle بيتغير مع كل receive، علشان consumer قديم الـ timeout بتاعه خلص ما يمسحش رسالة حد تاني شغّال عليها. الـ FIFO: Set بالـ groups، وأول رسالة من كل group هي اللي بتتحسب. والـ log: الرسايل ما بتتمسحش، والـ group ليه array من الـ committed offsets، والـ assignment بيتحسب من الـ members المترتبين. نشغّل lab dot js.

// broker.js: نموذجين للـ brokers. queue زي SQS و RabbitMQ (visibility timeout و ack و DLQ و FIFO groups)،
// و log زي Kafka (partitions و offsets و consumer groups و replay)، و chooseBroker بيختار حسب الـ requirements
export class BrokerError extends Error {
  constructor(code, message = code) {
    super(message);
    this.code = code;
  }
}

export function createQueue({ visibilityTimeout = 30000, maxReceives = 5, dedupWindow = 300000, fifo = false, now = Date.now } = {}) {
  let seq = 0;
  const messages = []; // بترتيب الـ send
  const dead = [];
  const dedup = new Map(); // dedupId -> { id, at }

  function send(body, { delay = 0, groupId = null, dedupId = null } = {}) {
    if (fifo && !groupId) throw new BrokerError("GROUP_REQUIRED");
    if (dedupId != null) {
      const prev = dedup.get(dedupId);
      if (prev && now() - prev.at < dedupWindow) return { id: prev.id, duplicate: true };
    }
    const id = `m${++seq}`;
    messages.push({ id, body, groupId, availableAt: now() + delay, receiveCount: 0, handle: null, visibleAt: 0 });
    if (dedupId != null) dedup.set(dedupId, { id, at: now() });
    return { id, duplicate: false };
  }

  const inFlight = (m) => m.handle !== null && m.visibleAt > now();
  const visible = (m) => m.availableAt <= now() && !inFlight(m);

  function receive({ max = 1 } = {}) {
    const out = [];
    const seenGroups = new Set();
    for (const m of [...messages]) {
      if (out.length >= max) break;
      if (fifo) {
        if (seenGroups.has(m.groupId)) continue; // أقدم رسالة بس في كل group
        seenGroups.add(m.groupId);
      }
      if (!visible(m)) continue;
      if (m.receiveCount >= maxReceives) {
        messages.splice(messages.indexOf(m), 1);
        dead.push({ id: m.id, body: m.body, receiveCount: m.receiveCount, groupId: m.groupId });
        if (fifo) seenGroups.delete(m.groupId); // الـ group ده يكمّل بالرسالة اللي بعدها
        continue;
      }
      m.receiveCount++;
      m.handle = `${m.id}:${m.receiveCount}`;
      m.visibleAt = now() + visibilityTimeout;
      out.push({ id: m.id, body: m.body, receiveCount: m.receiveCount, handle: m.handle, groupId: m.groupId });
    }
    return out;
  }

  const byHandle = (handle) => messages.find((m) => m.handle === handle && inFlight(m));

  function ack(handle) {
    const m = byHandle(handle);
    if (!m) return false;
    messages.splice(messages.indexOf(m), 1);
    return true;
  }
  function nack(handle, { delay = 0 } = {}) {
    const m = byHandle(handle);
    if (!m) return false;
    m.handle = null;
    m.visibleAt = 0;
    m.availableAt = now() + delay;
    return true;
  }
  function extend(handle, ms) {
    const m = byHandle(handle);
    if (!m) return false;
    m.visibleAt = now() + ms;
    return true;
  }
  function stats() {
    let ready = 0, flying = 0, delayed = 0;
    for (const m of messages) {
      if (inFlight(m)) flying++;
      else if (m.availableAt > now()) delayed++;
      else ready++;
    }
    return { re
…

$ node lab.js
— 1) queue في array: الـ worker وقع في النص —
  worker خد email-1 ووقع قبل ما يبعته. في الـ queue: 1، و email-1 راح خالص

— 2) queue بـ ack و visibility timeout: at-least-once —
  worker خد email-1 (receiveCount 1) ووقع من غير ack. stats: { ready: 0, inFlight: 1, delayed: 0, dlq: 0 }
  بعد الـ timeout رجعت: receiveCount 2
  الـ consumer: اتبعت، و ack: true
  ولو نفس الرسالة وصلت تاني (ده وارد في at-least-once): مكرر، اتعمله skip

— 3) رسالة بتوقّع الـ worker دايمًا: DLQ —
  receive رقم 1: اتسلّمت، receiveCount 1، والـ worker وقع
  receive رقم 2: اتسلّمت، receiveCount 2، والـ worker وقع
  receive رقم 3: اتسلّمت، receiveCount 3، والـ worker وقع
  receive رقم 4: مفيش، راحت الـ DLQ
  الـ DLQ: [ 'broken بعد 3 محاولات' ]

— 4) FIFO: الترتيب جُوّه كل user —
  round 1: u1: اتسجّل | u2: اتسجّل
  round 2: u1: غيّر الإيميل

— 5) log زي Kafka: partitions و consumer groups —
  publish u1:created ← partition 0، offset 0
  publish u2:created ← partition 2، offset 0
  publish u1:paid ← partition 0، offset 1
  publish u3:created ← partition 1، offset 0
  publish u1:shipped ← partition 0، offset 2
  billing فيه consumer واحد: { b1: [ 0, 1, 2 ] }
  زوّدنا consumer تاني، rebalance: { b1:
…

الأول: الـ job راح، ومحدش هيعرف. التاني: الـ worker وقع، وبعد الـ timeout الرسالة رجعت بـ receiveCount اتنين. والـ consumer الـ idempotent عمل skip للمرة المكررة. التالت: الرسالة البايزة اتسلّمت تلات مرات، وفي الرابعة راحت الـ DLQ. الرابع: الـ round الأول فيه أول رسالة من كل user، وتغيير الإيميل استنى لحد ما التسجيل خلص. الخامس: كل events بتاعة u1 في نفس الـ partition بالترتيب. ولما زوّدنا consumer، الـ partitions اتوزعت، و b1 قرا كل حاجة u1 بالترتيب. و analytics، group جديد، قرا الخمسة من الأول، والـ seek رجّع billing لورا. والسادس: الاختيار لكل موقف. ونشغّل test dot js.

$ node test.js
✅ send بيرجّع ids مختلفة، و receive بالترتيب، و ack بيمسح {"ready":0,"inFlight":1,"delayed":0,"dlq":0}
✅ الرسالة مستخبية لحد الـ timeout، وبعده بترجع بـ receiveCount اتنين و handle جديد 
✅ nack بيرجّعها بعد الـ delay، و extend بيطوّل الـ timeout 
✅ send بـ delay ما بيظهرش قبل وقته 
✅ بعد تلات receives الرسالة بتروح الـ DLQ، و redrive بيرجّعها بعدّاد صفر {"counts":[1,2,3],"d":[{"id":"m1","body":"poison","receiveCount":3,"groupId":null}],"st":{"ready":0,"inFlight":0,"delayed":0,"dlq":1}}
✅ نفس الـ dedupId جُوّه الـ window بيرجّع نفس الـ id ومش بيضيف، وبعده بيضيف 
✅ FIFO بيطلب groupId، وبيدّي أقدم رسالة بس من كل group، والـ group مقفول وهي in flight {"r1":["a1","b1"],"r2":[]}
✅ الـ ack بيفتح الـ group، والرسالة اللي راحت DLQ ما بتقفلش الـ group ["a3"]
✅ نفس الـ key ← نفس الـ partition، والـ offsets بتزيد جُوّه الـ partition 
✅ كل partition لـ consumer واحد بس، بالترتيب، والـ rebalance بعد join و leave {"a1":{"c2":[0,1,2]},"a2":{"c1":[0,2],"c2":[1]},"a3":{"c1":[0,1,2]}}
✅ من غير commit الـ poll بيرجّع نفس الحاجة، وبعد الـ commit بيكمّل، و commit لورا ما بيعملش حاجة [[0,0,"v0"],[0,1,"v2"],[1,0,"v1"],[1,1,"v3"]]
✅ consumer برّه الـ group ← NOT_MEMBER، و commit على partitio
…

كله نجح. دلوقتي عندك messages بتتنقل بين الـ services. الدرس الجاي: الـ realtime على أكتر من instance. WebSockets، والـ client موصول بـ instance والرسالة طلعت من instance تاني، و Redis pub sub، والـ sticky sessions، والـ presence.

// jobs.naive.js: queue جُوّه الـ process، في array
export function createNaiveQueue() {
  const jobs = [];
  return {
    push: (job) => jobs.push(job),
    // الـ job بيتشال من الـ queue أول ما حد ياخده، قبل ما يخلص
    take: () => jobs.shift(),
    size: () => jobs.length,
  };
}
// والمشاكل:
// - الـ worker وقع في النص؟ الـ job راح، محدش هيعيده (at-most-once)
// - الـ process عمل restart؟ الـ array كلها راحت
// - instance تانية؟ كل واحدة ليها array لوحدها
// - job بيقع دايمًا؟ يا يتعاد للأبد، يا يتنسي

// lab.js: نفس المواقف على أكتر من نوع broker
import { createNaiveQueue } from "./jobs.naive.js";
import { createQueue, createLog, chooseBroker } from "./broker.js";

let t = 0;
const now = () => t;
const tick = (ms) => (t += ms);

console.log("— 1) queue في array: الـ worker وقع في النص —");
const naive = createNaiveQueue();
naive.push("email-1"); naive.push("email-2");
const taken = naive.take();
console.log(`  worker خد ${taken} ووقع قبل ما يبعته. في الـ queue: ${naive.size()}، و ${taken} راح خالص`);

console.log("\n— 2) queue بـ ack و visibility timeout: at-least-once —");
const q = createQueue({ visibilityTimeout: 30000, maxReceives: 3, now });
q.send({ email: "email-1" });
const [m] = q.receive();
console.log(`  worker خد ${m.body.email} (receiveCount ${m.receiveCount}) ووقع من غير ack. stats:`, q.stats());
tick(30000);
const [again] = q.receive();
console.log(`  بعد الـ timeout رجعت: receiveCount ${again.receiveCount}`);
const sent = new Set();
const handle = (msg) => {
  if (sent.has(msg.body.email)) return "مكرر، اتعمله skip";
  sent.add(msg.body.email);
  return "اتبعت";
};
console.log(`  الـ consumer: ${handle(again)}، و ack:`, q.ack(again.handle));
console.log(`  ولو نفس الرسالة وصلت تاني (ده وارد في at-least-once): ${handle(again)}`);

console.log("\n— 3) رسالة بتوقّع الـ worker دايمًا: DLQ —");
q.send({ email: "broken" });
for (let i = 1; i <= 4; i++) {
  const got = q.receive();
  console.log(`  receive رقم ${i}:`, got.length ? `اتسلّمت، receiveCount ${got[0].receiveCount}، والـ worker وقع` : "مفيش، راحت الـ DLQ");
  tick(30000);
}
console.log("  الـ DLQ:", q.dlq().map((d) => `${d.body.email} بعد ${d.receiveCount} محاولات`));

console.log("\n— 4) FIFO: الترتيب جُوّه كل user —");
const fifo = createQueue({ fifo: true, now });
fifo.send("u1: اتسجّل", { groupId: "u1" });
fifo.send("u1: غيّر الإيميل", { groupId: "u1" });
fifo.send("u2: اتسجّل", { groupId: "u2" });
let round = 1, got;
while ((got = fifo.receive({ max: 10 })).length) {
  console.log(`  round ${round++}:`, got.map((x) => x.body).join(" | "));
  got.forEach((x) => fifo.ack(x.handle));
}

console.log("\n— 5) log زي Kafka: partitions و consumer groups —");
const log = createLog({ partitions: 3 });
for (const [user, ev] of [["u1", "created"], ["u2", "created"], ["u1", "paid"], ["u3", "created"], ["u1", "shipped"]]) {
  const at = log.publish(user, `${user}:${ev}`);
  console.log(`  publish ${user}:${ev} ← partition ${at.partition}، offset ${at.offset}`);
}
console.log("  billing فيه consumer واحد:", log.join("billing", "b1"));
console.log("  زوّدنا consumer تاني، rebalance:", log.join("billing", "b2"));
for (const c of ["b1", "b2"]) {
  const recs = log.poll("billing", c);
  console.log(`  ${c} قرا:`, recs.map((r) => r.value).join(" , ") || "ولا حاجة");
  recs.forEach((r) => log.commit("billing", c, r.partition, r.offset));
}
console.log("  lag بتاع billing:", log.lag("billing"));
log.join("analytics", "a1");
console.log("  group جديد analytics بيقرا كل الـ history من الأول:"
…