Redis Streams والـ Consumer Groups: log بيتوزع على workers

import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL);
await redis.flushall();

// stream: log بيتزوّد في آخره بس، وكل entry ليه id بالوقت
const id1 = await redis.xadd("orders", "*", "orderId", "7", "total", "250");
const id2 = await redis.xadd("orders", "*", "orderId", "8", "total", "90");
console.log("ids:", id1, id2, "← <millisecond>-<رقم>");

const all = await redis.xrange("orders", "-", "+");
console.log(all.map(([, fields]) => fields));

// MAXLEN ~: خلي الـ stream في حدود معينة (الأقدم بيتشال)
for (let i = 0; i < 2000; i++) await redis.xadd("orders", "MAXLEN", "~", "1000", "*", "orderId", String(100 + i));
console.log("الطول:", (await redis.xlen("orders")) <= 1100 ? "حوالي 1000" : "كبير");
redis.disconnect();

$ node main.js
ids: 1792245377486-0 1792245377486-1 ← <millisecond>-<رقم>
[
  [ 'orderId', '7', 'total', '250' ],
  [ 'orderId', '8', 'total', '90' ]
]
الطول: حوالي 1000

import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL);
await redis.flushall();
for (let i = 1; i <= 6; i++) await redis.xadd("orders", "*", "orderId", String(i));

// consumer group: الـ entries بتتوزع على الـ consumers، وكل واحد بيعمل ACK بعد ما يخلص
await redis.xgroup("CREATE", "orders", "billing", "0");

async function consume(name, count) {
  const res = await redis.xreadgroup("GROUP", "billing", name, "COUNT", count, "STREAMS", "orders", ">");
  return res ? res[0][1] : [];
}

const a = await consume("worker-a", 2);
const b = await consume("worker-b", 2);
console.log("a أخد:", a.map(([, f]) => f[1]), "| b أخد:", b.map(([, f]) => f[1]));

// a خلّص وعمل ACK، و b مات من غير ACK
await redis.xack("orders", "billing", ...a.map(([id]) => id));
const pending = await redis.xpending("orders", "billing", "-", "+", 10);
console.log("pending:", pending.map(([id, consumer, idle, deliveries]) => `${consumer} (${deliveries})`));

// worker-c بيستلم اللي بقاله أكتر من 100ms مستني من غير ACK
await new Promise((r) => setTimeout(r, 150));
const [, claimed] = await redis.xautoclaim("orders", "billing", "worker-c", 100, "0", "COUNT", 10);
console.log("c استلم من b:", claimed.map(([, f]) => f[1]));
console.log("c الجديد:", (await consume("worker-c", 10)).map(([, f]) => f[1]));
redis.disconnect();

$ node main.js
a أخد: [ '1', '2' ] | b أخد: [ '3', '4' ]
pending: [ 'worker-b (1)', 'worker-b (1)' ]
c استلم من b: [ '3', '4' ]
c الجديد: [ '5', '6' ]

import Redis from "ioredis";

const redis = new Redis(process.env.REDIS_URL);
await redis.flushall();
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
for (let i = 1; i <= 8; i++) await redis.xadd("orders", "*", "orderId", String(i));
await redis.xgroup("CREATE", "orders", "billing", "0");

// order رقم 5 بايز: الـ handler بيقع عليه كل مرة
const processed = [];
async function handle(fields) {
  if (fields[1] === "5") throw new Error("payment provider رفض");
  processed.push(fields[1]);
}


async function loop(name) {
  for (;;) {
    const [, claimed] = await redis.xautoclaim("orders", "billing", name, 50, "0", "COUNT", 10);
    const res = await redis.xreadgroup("GROUP", "billing", name, "COUNT", 5, "STREAMS", "orders", ">");
    const fresh = res ? res[0][1] : [];
    for (const [id, fields] of [...claimed, ...fresh]) {
      try {
        await handle(fields);
        await redis.xack("orders", "billing", id);
      } catch (err) {
        const [[, , , deliveries]] = await redis.xpending("orders", "billing", id, id, 1);
        if (deliveries >= 3) {
          await redis.xadd("orders:dead", "*", ...fields, "error", err.message);
          await redis.xack("orders", "billing", id);
        }
      }
    }
    const [count] = await redis.xpending("orders", "billing");
    if (!fresh.length && !claimed.length && count === 0) return;
    await sleep(60);
  }
}

await Promise.all([loop("worker-a"), loop("worker-b")]);
console.log("processed:", processed.sort((x, y) => x - y));
console.log("dead:", (await redis.xrange("orders:dead", "-", "+")).map(([, f]) => f));
redis.disconnect();

$ node main.js
processed: [
  '1', '2', '3',
  '4', '6', '7',
  '8'
]
dead: [ [ 'orderId', '5', 'error', 'payment provider رفض' ] ]