Node.js تکمیلی؛ رویداد، جریان داده و سرور چت

node --version
npm --version
npm ci
npm run check
npm test
npm start

{
  "name": "darsavand-small-chat",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "engines": { "node": ">=24" },
  "scripts": {
    "start": "node server.mjs",
    "dev": "node --env-file-if-exists=.env server.mjs",
    "test": "node --test",
    "check": "node --check app.mjs && node --check server.mjs && node --check protocol.mjs && node --check room.mjs && node --check connection.mjs && node --check public/app.js && node --check app.test.mjs"
  },
  "dependencies": { "ws": "8.22.0" }
}

import { createServer } from 'node:http';
import { readFile } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import { WebSocketServer, WebSocket } from 'ws';
import { LIMITS, ProtocolError, parsePacket, errorPacket } from './protocol.mjs';
import { createRoom } from './room.mjs';
import { sendBounded, heartbeat } from './connection.mjs';

function originValue(value) {
  const url = new URL(value);
  if (!['http:', 'https:'].includes(url.protocol) || url.origin !== value) {
    throw new TypeError('PUBLIC_ORIGIN must be one exact http(s) origin without a path');
  }
  return value;
}

export async function createChatServer({
  publicOrigin = 'http://127.0.0.1:3001',
  heartbeatMs = LIMITS.heartbeatMs, joinTimeoutMs = LIMITS.joinDeadlineMs
} = {}) {
  const allowedOrigin = originValue(publicOrigin);
  if (!Number.isSafeInteger(heartbeatMs) || heartbeatMs < 20 ||
      !Number.isSafeInteger(joinTimeoutMs) || joinTimeoutMs < 20) throw new RangeError('Invalid timing');
  const assets = new Map();
  for (const [path, file, type] of [
    ['/', 'index.html', 'text/html; charset=utf-8'],
    ['/style.css', 'style.css', 'text/css; charset=utf-8'],
    ['/app.js', 'app.js', 'text/javascript; charset=utf-8']
  ]) assets.set(path, { body: await readFile(new URL(`./public/${file}`, import.meta.url)), type });
  const room = createRoom();
  const members = new Map();
  let shuttingDown = false;
  let closePromise;
  const online = () => [...members].filter(([socket, member]) => member.joined && socket.readyState === WebSocket.OPEN).length;
  const getStats = () => ({ ok: true, service: 'darsavand-chat', online: online(), historySize: room.size });
  const wsOrigin = allowedOrigin.replace(/^http/, 'ws');
  const headers = {
    'Cache-Control': 'no-store', 'X-Content-Type-Options': 'nosniff',
    'Referrer-Policy': 'same-origin',
    'Content-Security-Policy': `default-src 'self'; script-src 'self'; style-src 'self'; connect-src 'self' ${wsOrigin}; object-src 'none'; base-uri 'none'; frame-ancestors 'none'`
  };
  const server = createServer({ requestTimeout: 15000, headersTimeout: 10000, maxHeaderSize: 8192 }, (req, res) => {
    let path;
    try { path = new URL(req.url, 'http://127.0.0.1').pathname; }
    catch { res.writeHead(400); return res.end(); }
    const known = assets.has(path) || path === '/api/status';
    let status = known ? 200 : 404;
    let type = 'application/json; charset=utf-8';
    let body = Buffer.from(JSON.stringify(known ? getStats() : { error: 'NOT_FOUND' }));
    const extra = {};
    if (known && !['GET', 'HEAD'].includes(req.method)) {
      status = 405; extra.Allow = 'GET, HEAD';
      body = Buffer.from(JSON.stringify({ error: 'METHOD_NOT_ALLOWED' }));
    } else if (known && assets.has(path)) {
      ({ body, type } = assets.get(path));
    }
    req.resume();
    res.writeHead(status, { ...headers, ...extra, 'Content-Type': type, 'Content-Length': body.length });
    res.end(req.method === 'HEAD' ? undefined : body);
  });
  const wss = new WebSocketServer({ noServer: true, maxPayload: LIMITS.maxPayloadBytes, perMessageDeflate: false });
  function broadcast(packet) {
    for (const [socket, member] of members) if (member.joined) sendBounded(socket, packet);
  }
  function presence() { broadcast({ type: 'presence', online: online() }); }
  server.on('upgrade', (req, socket, head) => {
    socket.on('error', () => {});
    let path;
    try { path = new URL(req.url, 'http://127.0.0.1').pathname; }
    catch { path = ''; }
    const status = shuttingDown || wss.clients.size >= LIMITS.maxClients ? 503
      : path !== '/chat' ? 404 : req.headers.origin !== allowedOrigin ? 403 : 0;
    if (status) {
      socket.end(`HTTP/1.1 ${status} Rejected\r\nConnection: close\r\nContent-Length: 0\r\n\r\n`);
      return;
    }
    wss.handleUpgrade(req, socket, head, ws => wss.emit('connection', ws));
  });
  wss.on('connection', socket => {
    socket.isAlive = true;
    const member = { id: randomUUID(), joined: false, name: '', count: 0, windowStart: performance.now(), accepted: new Map() };
    members.set(socket, member);
    const joinTimer = setTimeout(() => socket.close(1008, 'Join timeout'), joinTimeoutMs);
    joinTimer.unref();
    member.joinTimer = joinTimer;
    socket.on('error', () => {});
    socket.on('pong', () => { socket.isAlive = true; });
    socket.on('message', (data, isBinary) => {
      if (shuttingDown || socket.readyState !== WebSocket.OPEN) return;
      const now = performance.now();
      if (now - member.windowStart >= LIMITS.windowMs) { member.count = 0; member.windowStart = now; }
      if (++member.count > LIMITS.packetsPerWindow) {
        sendBounded(socket, errorPacket('RATE_LIMIT'));
        return socket.close(1008, 'Rate limit');
      }
      if (isBinary) {
        sendBounded(socket, errorPacket('TEXT_ONLY'));
        return socket.close(1003, 'Text only');
      }
      let packet;
      try {
        packet = parsePacket(data);
        if (packet.type === 'join') {
          if (member.joined) throw new ProtocolError('ALREADY_JOINED');
          member.name = packet.name; member.joined = true; clearTimeout(joinTimer);
          sendBounded(socket, { type: 'history', clientId: member.id, ...room.snapshot() });
          return presence();
        }
        if (!member.joined) throw new ProtocolError('JOIN_REQUIRED');
        const previous = member.accepted.get(packet.clientMessageId);
        if (previous) {
          if (previous.text !== packet.text) throw new ProtocolError('DUPLICATE_ID');
          return sendBounded(socket, { type: 'ack', clientMessageId: packet.clientMessageId, messageId: previous.id, seq: previous.seq });
        }
        const message = room.accept(member.name, packet.text);
        member.accepted.set(packet.clientMessageId, message);
        if (member.accepted.size > LIMITS.historyMessages) member.accepted.delete(member.accepted.keys().next().value);
        broadcast({ type: 'message', message });
        sendBounded(socket, { type: 'ack', clientMessageId: packet.clientMessageId, messageId: message.id, seq: message.seq });
      } catch (error) {
        if (error instanceof ProtocolError) sendBounded(socket, errorPacket(error.code, packet?.clientMessageId));
        else socket.close(1011, 'Internal error');
      }
    });
    socket.on('close', () => { clearTimeout(joinTimer); members.delete(socket); if (!shuttingDown) presence(); });
  });
  const heartbeatTimer = setInterval(() => heartbeat(wss), heartbeatMs);
  heartbeatTimer.unref();
  async function close() {
    if (closePromise) return closePromise;
    shuttingDown = true; clearInterval(heartbeatTimer);
    for (const [socket, member] of members) { clearTimeout(member.joinTimer); socket.close(1001, 'Server shutdown'); }
    const deadline = setTimeout(() => { for (const socket of wss.clients) socket.terminate(); }, 3000);
    deadline.unref();
    closePromise = Promise.all([
      new Promise(resolve => wss.close(resolve)),
      new Promise(resolve => { server.close(() => resolve()); server.closeIdleConnections(); })
    ]).finally(() => { clearTimeout(deadline); members.clear(); });
    return closePromise;
  }
  return { server, wss, close, getStats };
}

import { createChatServer } from './app.mjs';

const raw = process.env.PORT ?? '3001';
if (!/^\d{1,5}$/.test(raw) || Number(raw) < 1 || Number(raw) > 65535) {
  console.error('PORT must be an integer from 1 to 65535');
  process.exitCode = 1;
} else {
  try {
    const port = Number(raw);
    const app = await createChatServer({ publicOrigin: process.env.PUBLIC_ORIGIN ?? `http://127.0.0.1:${port}` });
    app.server.on('error', error => {
      console.error(`Cannot listen: ${error.code ?? 'UNKNOWN'}`);
      process.exitCode = 1;
      void app.close();
    });
    app.server.listen(port, '127.0.0.1', () => console.log(`Small chat: http://127.0.0.1:${port}/`));
    const stop = () => { void app.close(); };
    process.on('SIGINT', stop);
    process.on('SIGTERM', stop);
  } catch {
    console.error('Cannot start: check PUBLIC_ORIGIN, dependencies and public files.');
    process.exitCode = 1;
  }
}