الـ Realtime: Server-Sent Events و WebSockets

import express from "express";

const app = express();
let nextId = 1;
const history = []; // آخر events علشان اللي يرجع يكمّل
const clients = new Set();

function publish(type, data) {
  const ev = { id: nextId++, type, data };
  history.push(ev);
  for (const res of clients) send(res, ev);
}
function send(res, ev) {
  res.write(`id: ${ev.id}\nevent: ${ev.type}\ndata: ${JSON.stringify(ev.data)}\n\n`);
}

app.get("/events", (req, res) => {
  res.set({
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-cache",
    Connection: "keep-alive",
  });
  res.flushHeaders();
  res.write("retry: 3000\n\n");
  // كمّل من آخر event الـ client شافه
  const last = Number(req.get("Last-Event-ID") ?? 0);
  for (const ev of history) if (ev.id > last) send(res, ev);
  clients.add(res);
  req.on("close", () => clients.delete(res));
});

const server = app.listen(3000, async () => {
  // client: نفس اللي EventSource بيعمله في المتصفح
  async function listen(name, lastId) {
    const r = await fetch("http://localhost:3000/events", { headers: lastId ? { "Last-Event-ID": lastId } : {} });
    const decoder = new TextDecoder();
    for await (const chunk of r.body) {
      for (const block of decoder.decode(chunk, { stream: true }).split("\n\n").filter(Boolean)) {
        console.log(`[${name}]`, block.replaceAll("\n", " | "));
      }
    }
  }
  listen("mona");
  await new Promise((r) => setTimeout(r, 100));
  publish("order", { id: 7, status: "paid" });
  publish("order", { id: 7, status: "shipped" });
  await new Promise((r) => setTimeout(r, 100));
  listen("omar", "1"); // راجع بعد ما فاته event رقم 1
  await new Promise((r) => setTimeout(r, 200));
  process.exit(0);
});

$ node server.js
[mona] retry: 3000
[mona] id: 1 | event: order | data: {"id":7,"status":"paid"}
[mona] id: 2 | event: order | data: {"id":7,"status":"shipped"}
[omar] retry: 3000
[omar] id: 2 | event: order | data: {"id":7,"status":"shipped"}

import http from "node:http";
import { WebSocketServer, WebSocket } from "ws";

const server = http.createServer();
const wss = new WebSocketServer({ noServer: true, maxPayload: 1024 });

// الـ auth بيحصل وقت الـ upgrade، قبل ما الـ connection يتفتح
server.on("upgrade", (req, socket, head) => {
  const token = new URL(req.url, "http://x").searchParams.get("token");
  if (token !== "secret-mona" && token !== "secret-omar") {
    socket.write("HTTP/1.1 401 Unauthorized\r\n\r\n");
    return socket.destroy();
  }
  wss.handleUpgrade(req, socket, head, (ws) => {
    ws.user = token.split("-")[1];
    wss.emit("connection", ws);
  });
});

wss.on("connection", (ws) => {
  ws.isAlive = true;
  ws.on("pong", () => (ws.isAlive = true));
  ws.on("message", (raw) => {
    for (const client of wss.clients) {
      if (client.readyState === WebSocket.OPEN) client.send(`${ws.user}: ${raw}`);
    }
  });
  // من غير error handler، أي رسالة كبيرة كانت هتوقّع الـ server كله
  ws.on("error", (err) => console.log(`[server] ${ws.user}: ${err.message}`));
  ws.on("close", (code) => console.log(`[server] ${ws.user} قفل (${code})`));
});

// heartbeat: اللي ما ردش على ping من آخر مرة يتقفل
setInterval(() => {
  for (const ws of wss.clients) {
    if (!ws.isAlive) ws.terminate();
    ws.isAlive = false;
    ws.ping();
  }
}, 30000).unref();

server.listen(3000, async () => {
  const open = (token) =>
    new Promise((resolve, reject) => {
      const ws = new WebSocket(`ws://localhost:3000/chat?token=${token}`);
      ws.on("open", () => resolve(ws));
      ws.on("unexpected-response", (req, res) => reject(new Error(`رفض: ${res.statusCode}`)));
    });

  await open("hacker").catch((e) => console.log("[hacker]", e.message));
  const mona = await open("secret-mona");
  const omar = await open("secret-omar");
  omar.on("message", (m) => console.log("[omar شاف]", String(m)));

  mona.send("أهلًا يا عمر");
  await new Promise((r) => setTimeout(r, 100));
  mona.send("x".repeat(5000)); // أكبر من maxPayload
  await new Promise((r) => setTimeout(r, 200));
  process.exit(0);
});

$ node server.js
[hacker] رفض: 401
[omar شاف] mona: أهلًا يا عمر
[server] mona: Max payload size exceeded
[server] mona قفل (1006)

import http from "node:http";
import { WebSocketServer, WebSocket } from "ws";

const server = http.createServer();
const wss = new WebSocketServer({ server });
const rooms = new Map(); // room ← Set of sockets


const send = (ws, obj) => ws.readyState === WebSocket.OPEN && ws.send(JSON.stringify(obj));

function leave(ws) {
  const set = rooms.get(ws.room);
  if (!set) return;
  set.delete(ws);
  if (!set.size) rooms.delete(ws.room);
}

wss.on("connection", (ws) => {
  ws.on("message", (raw) => {
    let msg;
    try {
      msg = JSON.parse(raw);
    } catch {
      return send(ws, { type: "error", error: "bad json" });
    }
    if (msg.type === "join") {
      leave(ws);
      ws.room = msg.room;
      rooms.set(ws.room, (rooms.get(ws.room) ?? new Set()).add(ws));
    } else if (msg.type === "say") {
      if (!ws.room) return send(ws, { type: "error", error: "join الأول" });
      for (const peer of rooms.get(ws.room)) send(peer, { type: "msg", room: ws.room, text: msg.text });
    }
  });
  ws.on("close", () => leave(ws));
});

server.listen(3000, async () => {
  const open = (name) =>
    new Promise((resolve) => {
      const ws = new WebSocket("ws://localhost:3000");
      ws.on("message", (m) => console.log(`${name} شاف:`, String(m)));
      ws.on("open", () => resolve(ws));
    });
  const wait = () => new Promise((r) => setTimeout(r, 80));
  const [a, b, c] = await Promise.all([open("a"), open("b"), open("c")]);
  a.send(JSON.stringify({ type: "join", room: "node" }));
  b.send(JSON.stringify({ type: "join", room: "node" }));
  c.send(JSON.stringify({ type: "join", room: "sql" }));
  await wait();
  a.send(JSON.stringify({ type: "say", text: "hi node" }));
  await wait();
  c.send(JSON.stringify({ type: "say", text: "hi sql" }));
  await wait();
  b.send("{bad");
  await wait();
  process.exit(0);
});

$ node server.js
a شاف: {"type":"msg","room":"node","text":"hi node"}
b شاف: {"type":"msg","room":"node","text":"hi node"}
c شاف: {"type":"msg","room":"sql","text":"hi sql"}
b شاف: {"type":"error","error":"bad json"}