Realtime على أكتر من instance: Redis pub/sub و presence و resume

// realtime.js: realtime على أكتر من instance. Redis pub/sub بيوصّل الرسايل بين الـ instances،
// و history علشان اللي فصل يكمّل من آخر id، و presence بـ TTL، و SSE endpoint
import Redis from "ioredis";
import { createServer } from "node:http";

// INCR و RPUSH و LTRIM و PUBLISH في خطوة واحدة atomic: الـ ids بتطلع بنفس ترتيب الـ publish
const PUBLISH_LUA = `
local id = redis.call('INCR', KEYS[1])
local msg = '{"id":' .. id .. ',' .. string.sub(ARGV[1], 2)
redis.call('RPUSH', KEYS[2], msg)
redis.call('LTRIM', KEYS[2], -tonumber(ARGV[2]), -1)
redis.call('PUBLISH', KEYS[3], msg)
return id`;

export async function createHub({ url = process.env.REDIS_URL, instanceId, prefix = "rt:", presenceTtl = 30000, historySize = 100 } = {}) {
  if (!instanceId) throw new Error("instanceId required");
  const pub = new Redis(url);
  const sub = new Redis(url);
  // استنى الاتنين يبقوا ready: subscribe قبل الـ ready check بيبوّز الـ connection
  await Promise.all([pub, sub].map((c) => new Promise((ok, fail) => { c.once("ready", ok); c.once("error", fail); })));
  const k = {
    seq: (room) => `${prefix}seq:${room}`,
    history: (room) => `${prefix}history:${room}`,
    channel: (room) => `${prefix}room:${room}`,
    presence: (room) => `${prefix}presence:${room}`,
  };
  const rooms = new Map(); // room -> Set(conn)
  let nextConn = 0;

  sub.on("message", (channel, raw) => {
    const room = channel.slice(`${prefix}room:`.length);
    const conns = rooms.get(room);
    if (!conns) return;
    const msg = JSON.parse(raw);
    for (const c of [...conns]) {
      try {
        c.send(msg);
      } catch {
        c.leave(); // client بايز ما يوقّعش الباقيين
      }
    }
  });

  const member = (userId) => `${instanceId}|${userId}`;
  const touch = (room, userId) => pub.zadd(k.presence(room), Date.now() + presenceTtl, member(userId));

  async function join(room, userId, send) {
    let conns = rooms.get(room);
    if (!conns) {
      conns = new Set();
      rooms.set(room, conns);
      await sub.subscribe(k.channel(room));
    }
    const conn = {
      id: ++nextConn, room, userId, send, closed: false,
      async leave() {
        if (conn.closed) return;
        conn.closed = true;
        conns.delete(conn);
        if (![...conns].some((c) => c.userId === userId)) await pub.zrem(k.presence(room), member(userId));
        if (conns.size === 0) {
          rooms.delete(room);
          await sub.unsubscribe(k.channel(room));
        }
      },
    };
    conns.add(conn);
    await touch(room, userId);
    return conn;
  }

  async function publish(room, type, data = null) {
    const body = JSON.stringify({ room, type, data, from: instanceId, at: Date.now() });
    const id = await pub.eval(PUBLISH_LUA, 3, k.seq(room), k.history(room), k.channel(room), body, historySize);
    return Number(id);
  }

  async function since(room, lastId) {
    const events = (await pub.lrange(k.history(room), 0, -1)).map((x) => JSON.parse(x));
    const gap = events.length > 0 
…

$ node lab.js
— 1) من غير Redis: كل instance في حاله —
  الـ client موصول بـ A، واستلم: [ 'وأنا على A' ]
  الرسالة اللي راحت B ما وصلتش

— 2) بـ Redis pub/sub —
  POST على B ← id 1
  POST على C ← id 2
  POST على A ← id 3
  sara على A: [ '1 أهلًا من B', '2 وده من C', '3 وده من A' ]
  omar على C: [ '1 أهلًا من B', '2 وده من C', '3 وده من A' ]

— 3) مين online؟ —
  presence من B: [ 'omar', 'sara' ]
  stats لكل instance: A 1 ، B 0 ، C 1

— 4) sara فصلت، ورجعت على instance تاني —
  رجعت على B بـ Last-Event-ID 3: [ '4 اتبعتت وهي فاصلة', '5 ودي كمان', '6 ودي live بعد ما رجعت' ]

— 5) instance C وقع فجأة —
  presence على طول: [ 'ghost', 'sara' ]
  بعد الـ TTL: [ 'sara' ]

// chat.naive.js: SSE chat، والـ clients في Map جُوّه الـ process
import { createServer } from "node:http";

export function createNaiveApp(name) {
  const clients = new Map(); // room -> Set(res)
  return createServer(async (req, res) => {
    const m = /^\/rooms\/([\w-]+)\/(events|messages)$/.exec(new URL(req.url, "http://x").pathname);
    if (!m) return res.writeHead(404).end();
    const [, room, what] = m;
    if (what === "events") {
      res.writeHead(200, { "content-type": "text/event-stream" });
      res.write(`: connected to ${name}\n\n`);
      if (!clients.has(room)) clients.set(room, new Set());
      clients.get(room).add(res);
      req.on("close", () => clients.get(room).delete(res));
      return;
    }
    let raw = "";
    for await (const c of req) raw += c;
    const { type, data } = JSON.parse(raw);
    // بيبعت للـ clients اللي موصولين بالـ instance ده بس
    for (const r of clients.get(room) ?? []) r.write(`event: ${type}\ndata: ${JSON.stringify(data)}\n\n`);
    res.writeHead(202, { "content-type": "application/json" }).end(JSON.stringify({ instance: name }));
  });
}
// والمشاكل:
// - الـ client على A والـ POST راح B؟ مش هيوصل
// - الـ connection فصل تانية؟ اللي اتبعت فيها راح
// - مين online؟ كل instance عارف اللي عنده بس

// sse-client.js: client صغير للـ SSE بـ fetch، علشان lab.js و test.js
export function openSSE(url, { lastEventId } = {}) {
  const ctrl = new AbortController();
  const events = [];
  const waiters = [];
  let comment = null;
  const headers = lastEventId != null ? { "last-event-id": String(lastEventId) } : {};
  const done = fetch(url, { headers, signal: ctrl.signal }).then(async (res) => {
    const dec = new TextDecoder();
    let buf = "";
    for await (const chunk of res.body) {
      buf += dec.decode(chunk, { stream: true });
      let i;
      while ((i = buf.indexOf("\n\n")) >= 0) {
        const block = buf.slice(0, i);
        buf = buf.slice(i + 2);
        const ev = { id: null, event: "message", data: "" };
        for (const line of block.split("\n")) {
          if (line.startsWith(":")) comment = line.slice(1).trim();
          else if (line.startsWith("id: ")) ev.id = Number(line.slice(4));
          else if (line.startsWith("event: ")) ev.event = line.slice(7);
          else if (line.startsWith("data: ")) ev.data = JSON.parse(line.slice(6));
        }
        if (block.split("\n").every((l) => l.startsWith(":"))) continue;
        events.push(ev);
        waiters.splice(0).forEach((w) => w());
      }
    }
  }).catch(() => {});
  return {
    events,
    get server() { return comment; },
    async waitFor(n, ms = 2000) {
      const end = Date.now() + ms;
      while (events.length < n && Date.now() < end) await new Promise((r) => { waiters.push(r); setTimeout(r, 50); });
      return events.length >= n;
    },
    close() { ctrl.abort(); return done; },
  };
}
export const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
export const post = (base, room, type, data) =>
  fetch(`${base}/rooms/${room}/messages`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ type, data }) }).then(async (r) => ({ status: r.status, ...(await r.json()) }));

// lab.js: تلات instances لنفس الـ app، والـ client والـ POST على instances مختلفة
import Redis from "ioredis";
import { createNaiveApp } from "./chat.naive.js";
import { createHub, createApp } from "./realtime.js";
import { openSSE, sleep, post } from "./sse-client.js";

const inst = (c) => c.server?.replace("connected to ", "");
const listen = (srv) => new Promise((r) => srv.listen(0, () => r(`http://127.0.0.1:${srv.address().port}`)));
const admin = new Redis(process.env.REDIS_URL);
await admin.flushdb();
admin.disconnect();

console.log("— 1) من غير Redis: كل instance في حاله —");
const n1 = createNaiveApp("A"), n2 = createNaiveApp("B");
const [na, nb] = [await listen(n1), await listen(n2)];
const naiveClient = openSSE(`${na}/rooms/chat/events`);
await sleep(100);
await post(nb, "chat", "msg", { text: "حد سامعني؟" });
await post(na, "chat", "msg", { text: "وأنا على A" });
await sleep(200);
console.log(`  الـ client موصول بـ ${inst(naiveClient)}، واستلم:`, naiveClient.events.map((e) => e.data.text));
console.log("  الرسالة اللي راحت B ما وصلتش");
await naiveClient.close(); n1.close(); n2.close();

console.log("\n— 2) بـ Redis pub/sub —");
const hubs = await Promise.all(["A", "B", "C"].map((id) => createHub({ instanceId: id, presenceTtl: 1000 })));
const servers = hubs.map((h) => createApp(h));
const [A, B, C] = await Promise.all(servers.map(listen));
const sara = openSSE(`${A}/rooms/chat/events?user=sara`);
const omar = openSSE(`${C}/rooms/chat/events?user=omar`);
await sleep(150);
for (const [base, text] of [[B, "أهلًا من B"], [C, "وده من C"], [A, "وده من A"]]) {
  const r = await post(base, "chat", "msg", { text });
  console.log(`  POST على ${r.instance} ← id ${r.id}`);
}
await sara.waitFor(3); await omar.waitFor(3);
console.log(`  sara على ${inst(sara)}:`, sara.events.map((e) => `${e.id} ${e.data.text}`));
console.log(`  omar على ${inst(omar)}:`, omar.events.map((e) => `${e.id} ${e.data.text}`));

console.log("\n— 3) مين online؟ —");
console.log("  presence من B:", (await (await fetch(`${B}/rooms/chat/presence`)).json()).users);
console.log("  stats لكل instance:", hubs.map((h) => `${h.instanceId} ${h.stats().connections}`).join(" ، "));

console.log("\n— 4) sara فصلت، ورجعت على instance تاني —");
const last = sara.events.at(-1).id;
await sara.close();
await post(B, "chat", "msg", { text: "اتبعتت وهي فاصلة" });
await post(C, "chat", "msg", { text: "ودي كمان" });
const sara2 = openSSE(`${B}/rooms/chat/events?user=sara`, { lastEventId: last });
await sara2.waitFor(2);
await post(A, "chat", "msg", { text: "ودي live بعد ما رجعت" });
await sara2.waitFor(3);
console.log(`  رجعت على ${inst(sara2)} بـ Last-Event-ID ${last}:`, sara2.events.map((e) => `${e.id} ${e.data.text}`));

console.log("\n— 5) instance C وقع فجأة —");
await omar.close().catch(() => {});
await sleep(150);
await hubs[2].join("chat", "ghost", () => {});
await hubs[2].close({ graceful: false });
servers[2].close();
console.log("  presence على طول:", await hubs[0].presence("chat"))
…