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

# In the extracted project directory, with Node.js 24+
node --version
npm ci
npm test
npm run check

import test from 'node:test';
import assert from 'node:assert/strict';
import { once } from 'node:events';
import { randomUUID, webcrypto } from 'node:crypto';
import { readFile } from 'node:fs/promises';
import vm from 'node:vm';
import WebSocket from 'ws';
import { createChatServer } from './app.mjs';
import { createRoom } from './room.mjs';
import { parsePacket, LIMITS } from './protocol.mjs';
import { sendBounded, heartbeat } from './connection.mjs';

const ORIGIN = 'http://127.0.0.1:3001';
const pause = ms => new Promise(resolve => setTimeout(resolve, ms));
async function fixture(run, options = {}) {
  const app = await createChatServer({ publicOrigin: ORIGIN, ...options });
  app.server.listen(0, '127.0.0.1');
  await once(app.server, 'listening');
  const base = `http://127.0.0.1:${app.server.address().port}`;
  try { await run({ ...app, base, url: base.replace('http:', 'ws:') + '/chat' }); }
  finally { await app.close(); }
}
async function client(url, options = {}) {
  const socket = new WebSocket(url, { origin: ORIGIN, ...options });
  const packets = [];
  const waiters = new Set();
  socket.on('error', () => {});
  socket.on('message', data => {
    const packet = JSON.parse(data.toString());
    packets.push(packet);
    for (const waiter of waiters) {
      if (waiter.match(packet)) { clearTimeout(waiter.timer); waiters.delete(waiter); waiter.resolve(packet); break; }
    }
  });
  await once(socket, 'open');
  function next(type, predicate = () => true) {
    return new Promise((resolve, reject) => {
      const waiter = { match: p => p.type === type && predicate(p), resolve };
      waiter.timer = setTimeout(() => { waiters.delete(waiter); reject(new Error(`No ${type} packet`)); }, 1500);
      waiters.add(waiter);
    });
  }
  async function join(name = 'سارا') {
    const waiting = next('history');
    socket.send(JSON.stringify({ type: 'join', name }));
    return waiting;
  }
  async function chat(text, clientMessageId = randomUUID()) {
    const waiting = next('ack', p => p.clientMessageId === clientMessageId);
    socket.send(JSON.stringify({ type: 'chat', clientMessageId, text }));
    return waiting;
  }
  return { socket, packets, next, join, chat };
}
async function rejected(url, options) {
  return new Promise((resolve, reject) => {
    const socket = new WebSocket(url, options);
    socket.on('error', () => {});
    socket.on('open', () => { socket.terminate(); reject(new Error('Unexpected upgrade')); });
    socket.on('unexpected-response', (req, res) => { const code = res.statusCode; res.resume(); req.destroy(); resolve(code); });
  });
}

test('HTTP page and exact assets are real, UTF-8 and self-hosted', async () => {
  await fixture(async ({ base }) => {
    const page = await fetch(base);
    assert.equal(page.status, 200);
    assert.match(await page.text(), /گفت‌وگوی <span>کوچک/);
    for (const path of ['/app.js', '/style.css']) assert.equal((await fetch(base + path)).status, 200);
    assert.equal((await fetch(base + '/server.mjs')).status, 404);
    assert.equal((await fetch(base + '/.env')).status, 404);
    const head = await fetch(base, { method: 'HEAD' });
    assert.equal(head.status, 200); assert.equal(await head.text(), '');
    const post = await fetch(base, { method: 'POST' });
    assert.equal(post.status, 405); assert.equal(post.headers.get('allow'), 'GET, HEAD');
  });
});

test('wrong, missing and null browser origins are rejected', async () => {
  await fixture(async ({ url }) => {
    assert.equal(await rejected(url, { origin: 'https://wrong.example' }), 403);
    assert.equal(await rejected(url, {}), 403);
    assert.equal(await rejected(url, { origin: 'null' }), 403);
  });
});

test('only the exact /chat upgrade path is accepted', async () => {
  await fixture(async ({ url }) => assert.equal(await rejected(url.replace('/chat', '/ws'), { origin: ORIGIN }), 404));
});

test('two real clients receive the same server-issued message and sender ACK', async () => {
  await fixture(async ({ url, base }) => {
    const a = await client(url); const b = await client(url);
    const first = await a.join('سارا'); const second = await b.join('آرمان');
    assert.equal(first.type, 'history'); assert.equal(first.latestSeq, 0); assert.deepEqual(first.messages, []);
    assert.equal(first.serverId, second.serverId); assert.notEqual(first.clientId, second.clientId);
    const broadcast = b.next('message');
    const ack = await a.chat('سلام از پنجرهٔ اول');
    const received = (await broadcast).message;
    assert.equal(received.id, ack.messageId); assert.equal(received.seq, ack.seq);
    assert.equal(received.text, 'سلام از پنجرهٔ اول'); assert.equal(received.name, 'سارا');
    assert.ok(Number.isFinite(Date.parse(received.at)));
    assert.deepEqual(await (await fetch(base + '/api/status')).json(), {
      ok: true, service: 'darsavand-chat', online: 2, historySize: 1
    });
  });
});

test('chat before join gives a structured rejection', async () => {
  await fixture(async ({ url }) => {
    const a = await client(url); const id = randomUUID(); const error = a.next('error');
    a.socket.send(JSON.stringify({ type: 'chat', clientMessageId: id, text: 'سلام' }));
    const packet = await error; assert.equal(packet.code, 'JOIN_REQUIRED'); assert.equal(packet.clientMessageId, id);
  });
});

test('malformed JSON and extra fields are rejected without accepting history', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url);
    let waiting = a.next('error'); a.socket.send('{broken'); assert.equal((await waiting).code, 'INVALID_JSON');
    waiting = a.next('error'); a.socket.send(JSON.stringify({ type: 'join', name: 'سارا', admin: true }));
    assert.equal((await waiting).code, 'INVALID_PACKET'); assert.equal(getStats().historySize, 0);
  });
});

test('a joined connection cannot silently change its display name', async () => {
  await fixture(async ({ url }) => {
    const a = await client(url); await a.join(); const waiting = a.next('error');
    a.socket.send(JSON.stringify({ type: 'join', name: 'دیگری' }));
    assert.equal((await waiting).code, 'ALREADY_JOINED');
  });
});

test('Unicode normalization and code-point/UTF-8 bounds are explicit', () => {
  assert.equal(parsePacket(JSON.stringify({ type: 'join', name: ' e\u0301 ' })).name, 'é');
  assert.equal([...parsePacket(JSON.stringify({ type: 'join', name: '😀'.repeat(24) })).name].length, 24);
  assert.throws(() => parsePacket(JSON.stringify({ type: 'join', name: '😀'.repeat(25) })), /INVALID_NAME/);
  assert.equal([...parsePacket(JSON.stringify({ type: 'chat', clientMessageId: randomUUID(), text: '😀'.repeat(500) })).text].length, 500);
  assert.throws(() => parsePacket(JSON.stringify({ type: 'chat', clientMessageId: randomUUID(), text: '😀'.repeat(501) })), /INVALID_TEXT/);
  assert.throws(() => parsePacket(JSON.stringify({ type: 'chat', clientMessageId: randomUUID(), text: '\ud800' })), /INVALID_TEXT/);
  assert.throws(() => parsePacket(JSON.stringify({ type: 'chat', clientMessageId: 'fake', text: 'سلام' })), /INVALID_PACKET/);
});

test('invalid and binary messages never become chat messages', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url); await a.join();
    let waiting = a.next('error');
    a.socket.send(JSON.stringify({ type: 'chat', clientMessageId: randomUUID(), text: ' ' }));
    assert.equal((await waiting).code, 'INVALID_TEXT');
    waiting = a.next('error'); const closed = once(a.socket, 'close');
    a.socket.send(Buffer.from('binary'));
    assert.equal((await waiting).code, 'TEXT_ONLY'); assert.equal((await closed)[0], 1003);
    assert.equal(getStats().historySize, 0);
  });
});

test('the payload limit covers the aggregate of fragmented messages', async () => {
  await fixture(async ({ url }) => {
    const a = await client(url); const closed = once(a.socket, 'close');
    a.socket.send('x'.repeat(3000), { fin: false });
    a.socket.send('x'.repeat(2000), { fin: true });
    assert.equal((await closed)[0], 1009);
  });
});

test('duplicate request ID returns original ACK; different text rejects', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url); await a.join(); const id = randomUUID();
    const first = await a.chat('متن نخست', id); const duplicate = await a.chat('متن نخست', id);
    assert.equal(first.messageId, duplicate.messageId); assert.equal(first.seq, duplicate.seq);
    assert.equal(getStats().historySize, 1);
    const waiting = a.next('error'); a.socket.send(JSON.stringify({ type: 'chat', clientMessageId: id, text: 'متن دیگر' }));
    assert.equal((await waiting).code, 'DUPLICATE_ID'); assert.equal(getStats().historySize, 1);
    assert.equal(a.packets.filter(p => p.type === 'message').length, 1);
  });
});

test('rate bound counts join and invalid packets; 11th closes with 1008', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url); await a.join();
    for (let i = 0; i < 9; i++) {
      const waiting = a.next('error'); a.socket.send('{}'); assert.equal((await waiting).code, 'INVALID_PACKET');
    }
    const waiting = a.next('error', p => p.code === 'RATE_LIMIT'); const closed = once(a.socket, 'close');
    a.socket.send('{}'); assert.equal((await waiting).code, 'RATE_LIMIT'); assert.equal((await closed)[0], 1008);
    assert.equal(getStats().historySize, 0);
  });
});

test('room history is bounded to 50, sequenced, and snapshot copies cannot mutate it', () => {
  const room = createRoom();
  for (let i = 0; i < 60; i++) room.accept('سارا', `پیام ${i + 1}`);
  const state = room.snapshot(); assert.equal(state.messages.length, 50); assert.equal(state.latestSeq, 60);
  assert.equal(state.messages[0].seq, 11); assert.equal(state.messages.at(-1).seq, 60);
  state.messages[0].text = 'changed'; assert.notEqual(room.snapshot().messages[0].text, 'changed');
  assert.equal(new Set(state.messages.map(m => m.id)).size, 50);
});

test('real new connection receives an ordered 50-message snapshot; join adds no chat', async () => {
  await fixture(async ({ url }) => {
    let previousServerId;
    for (let group = 0; group < 10; group++) {
      const a = await client(url); const history = await a.join(`کاربر ${group}`); previousServerId = history.serverId;
      for (let i = 0; i < 6; i++) await a.chat(`پیام ${group * 6 + i + 1}`);
      const closed = once(a.socket, 'close'); a.socket.close(1000); await closed;
    }
    const b = await client(url); const history = await b.join('ناظر');
    assert.equal(history.serverId, previousServerId); assert.equal(history.messages.length, 50);
    assert.equal(history.messages[0].seq, 11); assert.equal(history.latestSeq, 60);
    assert.equal(history.messages.at(-1).seq, 60);
  });
});

test('a new server process model has a new identity and empty history', () => {
  const a = createRoom(); a.accept('سارا', 'سلام'); const b = createRoom();
  assert.notEqual(a.snapshot().serverId, b.snapshot().serverId);
  assert.equal(b.snapshot().latestSeq, 0); assert.deepEqual(b.snapshot().messages, []);
});

test('unjoined connections are closed at the join deadline', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url); assert.equal((await once(a.socket, 'close'))[0], 1008);
    assert.equal(getStats().online, 0);
  }, { joinTimeoutMs: 50 });
});

test('responsive clients answer real Ping/Pong and stay connected', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url); await a.join(); let pings = 0;
    a.socket.on('ping', () => pings++); await pause(150);
    assert.ok(pings >= 2); assert.equal(a.socket.readyState, WebSocket.OPEN); assert.equal(getStats().online, 1);
  }, { heartbeatMs: 40 });
});

test('a real client that disables automatic Pong is terminated', async () => {
  await fixture(async ({ url, getStats }) => {
    const a = await client(url, { autoPong: false }); await a.join();
    assert.equal((await once(a.socket, 'close'))[0], 1006); assert.equal(getStats().online, 0);
  }, { heartbeatMs: 40 });
});

test('controlled slow-consumer policy terminates instead of growing the queue', () => {
  let terminated = 0; let sent = 0;
  const slow = { readyState: WebSocket.OPEN, bufferedAmount: LIMITS.maxBufferedBytes, terminate() { terminated++; }, send() { sent++; } };
  assert.equal(sendBounded(slow, { type: 'presence', online: 1 }), false);
  assert.equal(terminated, 1); assert.equal(sent, 0);
  const healthy = { readyState: WebSocket.OPEN, bufferedAmount: 0, terminate() { terminated++; }, send(text, callback) { sent++; assert.equal(JSON.parse(text).online, 1); callback(); } };
  assert.equal(sendBounded(healthy, { type: 'presence', online: 1 }), true); assert.equal(sent, 1);
});

test('heartbeat controlled policy sets liveness then terminates a missing-Pong peer', () => {
  let pinged = 0; let terminated = 0;
  const socket = { readyState: WebSocket.OPEN, isAlive: true, ping() { pinged++; }, terminate() { terminated++; } };
  const wss = { clients: new Set([socket]) };
  heartbeat(wss); assert.equal(socket.isAlive, false); assert.equal(pinged, 1);
  heartbeat(wss); assert.equal(terminated, 1);
});

test('capacity rejects the 21st concurrent client without weakening limits', async () => {
  await fixture(async ({ url }) => {
    const clients = [];
    for (let i = 0; i < 20; i++) clients.push(await client(url));
    assert.equal(await rejected(url, { origin: ORIGIN }), 503);
    for (const a of clients) a.socket.close(1000);
  });
});

test('shutdown closes real clients with 1001 and cleans connections', async () => {
  await fixture(async ({ url, close, wss, getStats }) => {
    const a = await client(url); await a.join(); const closed = once(a.socket, 'close');
    await close(); assert.equal((await closed)[0], 1001); assert.equal(wss.clients.size, 0); assert.equal(getStats().online, 0);
    await close();
  });
});

test('browser request IDs use secure UUID fallback on ordinary HTTP origins', async () => {
  const source = await readFile(new URL('./public/app.js', import.meta.url), 'utf8');
  const start = source.indexOf('function createRequestId()');
  const end = source.indexOf('\nfunction setStatus', start);
  assert.ok(start >= 0 && end > start);
  const code = source.slice(start, end) + '\ncreateRequestId();';
  const context = { crypto: { getRandomValues: bytes => webcrypto.getRandomValues(bytes) }, Uint8Array };
  const ids = Array.from({ length: 100 }, () => vm.runInNewContext(code, context));
  assert.equal(new Set(ids).size, 100);
  for (const id of ids) assert.match(id, /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/);
  const native = vm.runInNewContext(code, { crypto: { randomUUID: () => 'native-uuid' } });
  assert.equal(native, 'native-uuid');
  assert.throws(() => vm.runInNewContext(code, {}), /Secure random unavailable/);
});