Pub/Sub و BullMQ: إشارات بين الـ instances، وطوابير شغل محترمة

import Redis from "ioredis";

const url = process.env.REDIS_URL;
const pub = new Redis(url);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// 3 instances من الـ app، كل واحد عنده cache في الـ memory
function instance(name) {
  const local = new Map([["course:1", "Node.js (قديم)"]]);
  const sub = new Redis(url); // connection في وضع subscribe ما تعملش أوامر تانية
  sub.subscribe("invalidate");
  sub.on("message", (channel, key) => {
    local.delete(key);
    console.log(`[${name}] مسح ${key}`);
  });
  return { name, local, sub };
}

const apps = ["app-1", "app-2"].map(instance);
await sleep(100);

console.log("subscribers:", await pub.publish("invalidate", "course:1"));
await sleep(100);

// instance اتولد بعد الرسالة: ما وصلتلوش
const late = instance("app-3");
await sleep(100);
console.log("app-3 لسه فاكر:", late.local.get("course:1"));

for (const a of [...apps, late]) a.sub.disconnect();
pub.disconnect();

$ node main.js
subscribers: 2
[app-2] مسح course:1
[app-1] مسح course:1
app-3 لسه فاكر: Node.js (قديم)

import { Queue, Worker } from "bullmq";

// BullMQ محتاج maxRetriesPerRequest: null للـ workers
const connection = { host: "localhost", port: 6379, maxRetriesPerRequest: null };
const emails = new Queue("emails", { connection });
await emails.obliterate({ force: true });

const worker = new Worker(
  "emails",
  async (job) => {
    if (job.data.to === "[email protected]") throw new Error("smtp 550");
    return `sent to ${job.data.to}`;
  },
  { connection, concurrency: 2 },
);
worker.on("completed", (job, result) => console.log("✓", job.name, result));
worker.on("failed", (job, err) => console.log("✗", job.name, `attempt ${job.attemptsMade}`, err.message));

await emails.add("welcome", { to: "[email protected]" });
await emails.add("receipt", { to: "[email protected]" }, { attempts: 3, backoff: { type: "exponential", delay: 100 } });
await emails.add("reminder", { to: "[email protected]" }, { delay: 300 });

await new Promise((r) => setTimeout(r, 1500));
console.log(await emails.getJobCounts("completed", "failed", "delayed", "waiting"));
await worker.close();
await emails.close();

$ node main.js
✗ receipt attempt 1 smtp 550
✓ welcome sent to [email protected]
✗ receipt attempt 2 smtp 550
✗ receipt attempt 3 smtp 550
✓ reminder sent to [email protected]
{ completed: 2, failed: 1, delayed: 0, waiting: 0 }

import { Queue, Worker } from "bullmq";

const connection = { host: "localhost", port: 6379, maxRetriesPerRequest: null };
const notify = new Queue("notify", { connection });
await notify.obliterate({ force: true });
const sent = [];
const flaky = new Set(["sara"]); // أول محاولتين بيفشلوا
const tries = {};


const worker = new Worker(
  "notify",
  async (job) => {
    const { user } = job.data;
    tries[user] = (tries[user] ?? 0) + 1;
    if (flaky.has(user) && tries[user] < 3) throw new Error("timeout");
    sent.push(user);
  },
  { connection, concurrency: 2, limiter: { max: 3, duration: 200 } },
);

for (const user of ["mona", "omar", "sara", "karim", "lila", "mona"]) {
  await notify.add(
    "like",
    { user },
    { jobId: `like-post7-${user}`, attempts: 4, backoff: { type: "exponential", delay: 50 } },
  );
}

for (let i = 0; i < 30; i++) {
  const c = await notify.getJobCounts("waiting", "active", "delayed");
  if (!c.waiting && !c.active && !c.delayed) break;
  await new Promise((r) => setTimeout(r, 100));
}
console.log("sent:", sent.sort());
console.log("tries.sara =", tries.sara);
console.log(await notify.getJobCounts("completed", "failed"));
await worker.close();
await notify.close();

$ node main.js
sent: [ 'karim', 'lila', 'mona', 'omar', 'sara' ]
tries.sara = 3
{ completed: 5, failed: 0 }