WebSockets & Server-Sent Events

import { createServer } from 'node:http';
import { createHash } from 'node:crypto';

const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; // fixed by RFC 6455

const server = createServer((req, res) => {
  res.writeHead(426, { Upgrade: 'websocket' }).end('Upgrade required\n');
});

server.on('upgrade', (req, socket) => {
  if (req.headers.upgrade?.toLowerCase() !== 'websocket') {
    socket.end('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n');
    return;
  }
  const accept = createHash('sha1')
    .update(req.headers['sec-websocket-key'] + GUID)
    .digest('base64');
  socket.write(
    'HTTP/1.1 101 Switching Protocols\r\n' +
      'Upgrade: websocket\r\n' +
      'Connection: Upgrade\r\n' +
      `Sec-WebSocket-Accept: ${accept}\r\n\r\n`,
  );
  // One unmasked text frame (FIN + opcode 1, payload shorter than 126 bytes),
  // then a close frame (opcode 8, status 1000). Real code should use `ws`.
  const text = Buffer.from('hello from a hand-written server');
  socket.write(Buffer.concat([Buffer.from([0x81, text.length]), text]));
  socket.end(Buffer.from([0x88, 0x02, 0x03, 0xe8]));
});

server.listen(0, '127.0.0.1', () => {
  const { port } = server.address();
  const ws = new WebSocket(`ws://127.0.0.1:${port}`);
  ws.onmessage = (e) => console.log('client got:', e.data);
  ws.onclose = (e) => {
    console.log('client close:', e.code);
    server.close();
  };
});

client got: hello from a hand-written server
client close: 1000

import { createServer } from 'node:http';
import { WebSocketServer } from 'ws';

const TOKEN = process.env.FEED_TOKEN ?? 'dev-only-token'; // from --env-file in real life
const ALLOWED_ORIGINS = new Set(['https://bookshelf.example']);

const server = createServer((req, res) => res.writeHead(404).end());
// noServer: we decide, in the 'upgrade' handler, which requests may upgrade.
const wss = new WebSocketServer({ noServer: true, maxPayload: 64 * 1024 });

server.on('upgrade', (req, socket, head) => {
  const { pathname } = new URL(req.url, 'http://placeholder');
  const origin = req.headers.origin;
  const reject = (status, text) => {
    socket.write(`HTTP/1.1 ${status} ${text}\r\nConnection: close\r\n\r\n`);
    socket.destroy();
  };
  if (pathname !== '/ws/reviews') return reject(404, 'Not Found');
  // Browsers always send Origin; non-browser clients may omit it.
  if (origin && !ALLOWED_ORIGINS.has(origin)) return reject(403, 'Forbidden');
  if (req.headers.authorization !== `Bearer ${TOKEN}`) {
    return reject(401, 'Unauthorized');
  }
  wss.handleUpgrade(req, socket, head, (ws) => wss.emit('connection', ws, req));
});

// Heartbeat: ping every client, terminate those that never answered the last ping.
const alive = new WeakSet();
wss.on('connection', (ws) => {
  alive.add(ws);
  ws.on('pong', () => alive.add(ws));
  ws.on('error', console.error);
});
const heartbeat = setInterval(() => {
  for (const ws of wss.clients) {
    if (!alive.has(ws)) { ws.terminate(); continue; }
    alive.delete(ws);
    ws.ping();
  }
}, 30_000);
heartbeat.unref();

export function broadcast(message) {
  const json = JSON.stringify(message);
  for (const ws of wss.clients) {
    if (ws.readyState === ws.OPEN) ws.send(json);
  }
}

server.listen(0, '127.0.0.1', async () => {
  const { port } = server.address();
  const open = (headers) => new Promise((resolve) => {
    const c = new WebSocket(`ws://127.0.0.1:${port}/ws/reviews`, { headers });
    c.onopen = () => resolve(c);
    c.onerror = () => resolve(null);
  });
  console.log('no token  ->', (await open({})) ? 'connected' : 'refused');
  const client = await open({ authorization: `Bearer ${TOKEN}` });
  console.log('with token->', client ? 'connected' : 'refused');
  client.onmessage = (e) => {
    console.log('received', e.data);
    client.close();
    wss.close();
    server.close();
  };
  broadcast({ type: 'review', bookId: 7, stars: 5 });
});

no token  -> refused
with token-> connected
received {"type":"review","bookId":7,"stars":5}

import { server, publishReview } from './sse-server.mjs';

publishReview({ bookId: 1, stars: 4, text: 'published before anyone listened' });

server.listen(0, '127.0.0.1', async () => {
  const { port } = server.address();
  const ac = new AbortController();
  const res = await fetch(`http://127.0.0.1:${port}/reviews/stream`, {
    headers: { 'Last-Event-ID': '0' },
    signal: ac.signal,
  });
  console.log(res.status, res.headers.get('content-type'));
  setTimeout(() => publishReview({ bookId: 2, stars: 5, text: 'live' }), 100);

  // A tiny parser: split the byte stream into blank-line-separated events.
  let buffer = '';
  let events = 0;
  try {
    for await (const chunk of res.body.pipeThrough(new TextDecoderStream())) {
      buffer += chunk;
      let end;
      while ((end = buffer.indexOf('\n\n')) !== -1) {
        console.log(JSON.stringify(buffer.slice(0, end)));
        buffer = buffer.slice(end + 2);
        if (++events === 3) ac.abort();
      }
    }
  } catch (err) {
    console.log('stopped:', err.name);
  }
  server.close();
  server.closeAllConnections();
});

200 text/event-stream
"retry: 3000"
"id: 1\nevent: review\ndata: {\"bookId\":1,\"stars\":4,\"text\":\"published before anyone listened\"}"
"id: 2\nevent: review\ndata: {\"bookId\":2,\"stars\":5,\"text\":\"live\"}"
stopped: AbortError

import { server, publishReview } from './sse-server.mjs';

publishReview({ bookId: 1, stars: 4 });
server.listen(0, '127.0.0.1', () => {
  const { port } = server.address();
  const source = new EventSource(`http://127.0.0.1:${port}/reviews/stream`);
  source.addEventListener('review', (event) => {
    console.log(event.lastEventId, event.data);
    source.close(); // without close() the client reconnects forever
    server.close();
    server.closeAllConnections();
  });
});

import {
  BroadcastChannel, Worker, isMainThread, parentPort, threadId,
} from 'node:worker_threads';

if (isMainThread) {
  const bc = new BroadcastChannel('reviews'); // same name = same channel
  const workers = [new Worker(import.meta.filename), new Worker(import.meta.filename)];
  let ready = 0;
  for (const w of workers) {
    w.on('message', () => { // each worker says "listening" once
      if (++ready === workers.length) bc.postMessage({ bookId: 7, stars: 5 });
    });
    w.on('exit', () => { if (--ready === 0) bc.close(); });
  }
} else {
  const bc = new BroadcastChannel('reviews');
  bc.onmessage = (event) => {
    console.log(`thread ${threadId} received`, event.data);
    bc.close(); // an open channel keeps the thread alive
    parentPort.close();
  };
  parentPort.postMessage('listening');
}

thread 2 received { bookId: 7, stars: 5 }
thread 1 received { bookId: 7, stars: 5 }

import { createServer } from 'node:http';
import { createClient } from 'redis';

const CHANNEL = 'bookshelf:reviews';

// One "instance" of the app: it would normally be one process or container.
async function startInstance(redisUrl) {
  const publisher = createClient({ url: redisUrl });
  const subscriber = publisher.duplicate(); // a subscribed connection can do nothing else
  publisher.on('error', console.error);
  subscriber.on('error', console.error);
  await Promise.all([publisher.connect(), subscriber.connect()]);

  const streams = new Set(); // SSE responses connected to THIS instance
  await subscriber.subscribe(CHANNEL, (message) => {
    for (const res of streams) res.write(`event: review\ndata: ${message}\n\n`);
  });

  const server = createServer(async (req, res) => {
    if (req.method === 'GET' && req.url === '/reviews/stream') {
      res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' });
      res.write(': connected\n\n');
      streams.add(res);
      res.on('close', () => streams.delete(res));
    } else if (req.method === 'POST' && req.url === '/reviews') {
      let body = '';
      for await (const chunk of req.setEncoding('utf8')) body += chunk;
      // Publish to Redis, not to local clients: every instance (this one too)
      // receives the message through its subscription and fans it out.
      await publisher.publish(CHANNEL, JSON.stringify(JSON.parse(body)));
      res.writeHead(202).end();
    } else {
      res.writeHead(404).end();
    }
  });
  await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve));
  const close = async () => {
    for (const res of streams) res.end();
    server.close();
    await Promise.all([subscriber.close(), publisher.close()]);
  };
  return { port: server.address().port, close };
}

const redisUrl = process.env.REDIS_URL ?? 'redis://127.0.0.1:6379';
const [a, b] = [await startInstance(redisUrl), await startInstance(redisUrl)];

// A client listens on instance B...
const stream = await fetch(`http://127.0.0.1:${b.port}/reviews/stream`);
const reader = stream.body.pipeThrough(new TextDecoderStream()).getReader();
await reader.read(); // ": connected"

// ...and a review is posted to instance A.
await fetch(`http://127.0.0.1:${a.port}/reviews`, {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ bookId: 7, stars: 5 }),
});
console.log('instance B client got:', JSON.stringify((await reader.read()).value));

await reader.cancel();
await Promise.all([a.close(), b.close()]);

instance B client got: "event: review\ndata: {\"bookId\":7,\"stars\":5}\n\n"