370 lines
27 KiB
JavaScript
370 lines
27 KiB
JavaScript
'use strict';
|
|
|
|
const crypto = require('crypto');
|
|
const { URL } = require('url');
|
|
|
|
const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';
|
|
const MAX_MESSAGE_BYTES = 64 * 1024;
|
|
const MAX_SOCKET_BUFFER_BYTES = 1024 * 1024;
|
|
const MAX_POLL_EVENTS = 256;
|
|
const MAX_VIEW_SPAN = 512;
|
|
const MAX_WORLD_COORD = 1_000_000_000;
|
|
const REACTION_EMOJIS = new Set(['👍','🤩','🙏','🧠','🎉']);
|
|
const REACTION_STYLES = new Set(['classic','giant','laser','orbit','firework','comet']);
|
|
const REACTION_TTL_MS = 4500;
|
|
function reactionTtlForStyle(style) { return style==='comet'?3000:style==='firework'?3400:style==='orbit'?3200:style==='giant'||style==='laser'?2700:1050; }
|
|
const MAX_REACTIONS = 1024;
|
|
const REACTION_MIN_INTERVAL_MS = 450;
|
|
const SUBSCRIPTION_BUCKET_SIZE = 64;
|
|
|
|
function finite(value, fallback = 0) {
|
|
const number = Number(value);
|
|
return Number.isFinite(number) ? number : fallback;
|
|
}
|
|
function clamp(value, min, max) { return Math.max(min, Math.min(max, value)); }
|
|
function cleanCursorStyle(value) {
|
|
const style = String(value || 'default').replace(/[^a-z0-9_-]/gi, '').slice(0, 80);
|
|
return style || 'default';
|
|
}
|
|
function cleanBoardId(value) {
|
|
const id = String(value || '');
|
|
return /^B(?:0|[1-9][0-9]*)$/.test(id) ? id : null;
|
|
}
|
|
function normalizeViewport(message) {
|
|
let minX = clamp(finite(message.minX), -MAX_WORLD_COORD, MAX_WORLD_COORD);
|
|
let minY = clamp(finite(message.minY), -MAX_WORLD_COORD, MAX_WORLD_COORD);
|
|
let maxX = clamp(finite(message.maxX), -MAX_WORLD_COORD, MAX_WORLD_COORD);
|
|
let maxY = clamp(finite(message.maxY), -MAX_WORLD_COORD, MAX_WORLD_COORD);
|
|
if (maxX < minX) [minX, maxX] = [maxX, minX];
|
|
if (maxY < minY) [minY, maxY] = [maxY, minY];
|
|
const centerX = (minX + maxX) / 2, centerY = (minY + maxY) / 2;
|
|
const spanX = Math.min(MAX_VIEW_SPAN, Math.max(1, maxX - minX));
|
|
const spanY = Math.min(MAX_VIEW_SPAN, Math.max(1, maxY - minY));
|
|
return {minX: centerX - spanX / 2, minY: centerY - spanY / 2, maxX: centerX + spanX / 2, maxY: centerY + spanY / 2};
|
|
}
|
|
function pointInViewport(viewport, x, y, margin = 4) {
|
|
return Boolean(viewport) && x >= viewport.minX - margin && x <= viewport.maxX + margin && y >= viewport.minY - margin && y <= viewport.maxY + margin;
|
|
}
|
|
function boundsInViewport(viewport, bounds, margin = 3) {
|
|
return Boolean(viewport && bounds) && bounds.maxX >= viewport.minX - margin && bounds.minX <= viewport.maxX + margin && bounds.maxY >= viewport.minY - margin && bounds.minY <= viewport.maxY + margin;
|
|
}
|
|
function encodeFrame(opcode, payload = Buffer.alloc(0)) {
|
|
if (!Buffer.isBuffer(payload)) payload = Buffer.from(payload);
|
|
const length = payload.length;
|
|
let header;
|
|
if (length < 126) {
|
|
header = Buffer.allocUnsafe(2); header[0] = 0x80 | opcode; header[1] = length;
|
|
} else if (length <= 0xffff) {
|
|
header = Buffer.allocUnsafe(4); header[0] = 0x80 | opcode; header[1] = 126; header.writeUInt16BE(length, 2);
|
|
} else {
|
|
header = Buffer.allocUnsafe(10); header[0] = 0x80 | opcode; header[1] = 127; header.writeBigUInt64BE(BigInt(length), 2);
|
|
}
|
|
return Buffer.concat([header, payload]);
|
|
}
|
|
function sendJson(client, value) {
|
|
if (!client || client.closed) return false;
|
|
if (client.transport === 'poll') {
|
|
const sequence = ++client.eventSequence;
|
|
client.events.push({sequence, message:value});
|
|
if (client.events.length > MAX_POLL_EVENTS) {const removed=client.events.splice(0, client.events.length - MAX_POLL_EVENTS);client.droppedThroughSequence=Math.max(client.droppedThroughSequence||0,...removed.map(event=>event.sequence))}
|
|
client.wakePoll?.();
|
|
return true;
|
|
}
|
|
if (!client.socket?.writable) return false;
|
|
if(client.backpressured)return false;
|
|
if((client.socket.writableLength||0)>MAX_SOCKET_BUFFER_BYTES){client.closed=true;client.socket.destroy();return false}
|
|
try {const accepted=client.socket.write(encodeFrame(1, Buffer.from(JSON.stringify(value))));if(!accepted){client.backpressured=true;client.socket.once('drain',()=>{client.backpressured=false})}return accepted;}
|
|
catch (_) { return false; }
|
|
}
|
|
function sendClose(client, code = 1000, reason = '') {
|
|
if (!client || client.closed) return;
|
|
client.closed = true;
|
|
if (client.transport === 'poll') return;
|
|
const text = Buffer.from(String(reason).slice(0, 100));
|
|
const payload = Buffer.allocUnsafe(2 + text.length); payload.writeUInt16BE(code, 0); text.copy(payload, 2);
|
|
try { client.socket.end(encodeFrame(8, payload)); } catch (_) { client.socket.destroy(); }
|
|
}
|
|
|
|
function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/realtime', claimTtlMs = 5 * 60 * 1000, now = Date.now, maxClients = 500, maxClientsPerPlayer = 5}) {
|
|
if (!server || typeof authenticate !== 'function' || typeof getBoardInfo !== 'function') throw new Error('Invalid realtime hub options');
|
|
const clients = new Map();
|
|
const claims = new Map();
|
|
const reactions = new Map();
|
|
const viewportBuckets = new Map();
|
|
let serial = Promise.resolve();
|
|
let nextConnection = 1;
|
|
let nextReaction = 1;
|
|
function playerClientCount(playerId){let count=0;for(const client of clients.values())if(!client.closed&&client.playerId===playerId)count++;return count}
|
|
|
|
function withSerial(task) {
|
|
const run = serial.then(task, task); serial = run.then(() => undefined, () => undefined); return run;
|
|
}
|
|
function viewportBucketKeys(viewport){
|
|
if(!viewport)return[];const keys=[];const minX=Math.floor((viewport.minX-4)/SUBSCRIPTION_BUCKET_SIZE),maxX=Math.floor((viewport.maxX+4)/SUBSCRIPTION_BUCKET_SIZE),minY=Math.floor((viewport.minY-4)/SUBSCRIPTION_BUCKET_SIZE),maxY=Math.floor((viewport.maxY+4)/SUBSCRIPTION_BUCKET_SIZE);
|
|
for(let y=minY;y<=maxY;y++)for(let x=minX;x<=maxX;x++)keys.push(`${x},${y}`);return keys;
|
|
}
|
|
function unregisterViewport(client){for(const key of client.viewportBucketKeys||[]){const set=viewportBuckets.get(key);if(!set)continue;set.delete(client);if(!set.size)viewportBuckets.delete(key)}client.viewportBucketKeys=[]}
|
|
function registerViewport(client,viewport){unregisterViewport(client);client.viewport=viewport;client.viewportBucketKeys=viewportBucketKeys(viewport);for(const key of client.viewportBucketKeys){let set=viewportBuckets.get(key);if(!set)viewportBuckets.set(key,set=new Set());set.add(client)}}
|
|
function subscribersAt(x,y){const key=`${Math.floor(x/SUBSCRIPTION_BUCKET_SIZE)},${Math.floor(y/SUBSCRIPTION_BUCKET_SIZE)}`;return viewportBuckets.get(key)||new Set()}
|
|
function publicPlayer(client) {
|
|
return {presenceId: client.id, playerId: client.playerId, name: client.name, x: client.x, y: client.y, vx:client.vx||0, vy:client.vy||0, cursorStyle: client.cursorStyle || 'default', sentAt:client.cursorAt || now(), at: client.cursorAt || now()};
|
|
}
|
|
function publicClaim(claim) {
|
|
return {boardId: claim.boardId, playerId: claim.ownerPlayerId, playerName: claim.ownerName, presenceId: claim.ownerPresenceId, expiresAt: claim.expiresAt, ...claim.bounds};
|
|
}
|
|
function claimActive(claim, timestamp = now()) { return Boolean(claim && claim.expiresAt > timestamp); }
|
|
function publicReaction(reaction) { return {id:reaction.id, playerId:reaction.playerId, playerName:reaction.playerName, emoji:reaction.emoji, style:reaction.style, x:reaction.x, y:reaction.y, createdAt:reaction.createdAt, expiresAt:reaction.expiresAt}; }
|
|
function pruneReactions(timestamp = now()) {
|
|
for (const [id, reaction] of reactions) if (reaction.expiresAt <= timestamp) reactions.delete(id);
|
|
while (reactions.size > MAX_REACTIONS) reactions.delete(reactions.keys().next().value);
|
|
}
|
|
function pruneClaims(timestamp = now()) {
|
|
const removed = [];
|
|
for (const [boardId, claim] of claims) if (!claimActive(claim, timestamp)) { claims.delete(boardId); removed.push(boardId); }
|
|
for (const boardId of removed) broadcastAll({type:'claim-release', boardId, reason:'expired', serverTime:timestamp});
|
|
return removed.length;
|
|
}
|
|
function snapshotFor(client) {
|
|
pruneClaims();
|
|
const players = [];
|
|
for (const other of clients.values()) {
|
|
if (!other.authenticated || other.id === client.id || !Number.isFinite(other.x) || !Number.isFinite(other.y)) continue;
|
|
if (pointInViewport(client.viewport, other.x, other.y)) players.push(publicPlayer(other));
|
|
}
|
|
const visibleClaims = [];
|
|
for (const claim of claims.values()) if (claim.ownerPlayerId === client.playerId || boundsInViewport(client.viewport, claim.bounds)) visibleClaims.push(publicClaim(claim));
|
|
pruneReactions(); const visibleReactions=[];
|
|
for (const reaction of reactions.values()) if (pointInViewport(client.viewport,reaction.x,reaction.y)) visibleReactions.push(publicReaction(reaction));
|
|
return {type:'snapshot', players, claims:visibleClaims, reactions:visibleReactions, serverTime:now()};
|
|
}
|
|
function sendSnapshot(client) { if (client.authenticated) sendJson(client, snapshotFor(client)); }
|
|
function broadcastAll(message, predicate = null) {
|
|
for (const client of clients.values()) if (client.authenticated && (!predicate || predicate(client))) sendJson(client, message);
|
|
}
|
|
function broadcastCursor(sender) {
|
|
const message = {type:'cursor', ...publicPlayer(sender), serverTime:now()};
|
|
for(const client of subscribersAt(sender.x,sender.y))if(client.authenticated&&client.id!==sender.id&&pointInViewport(client.viewport,sender.x,sender.y))sendJson(client,message);
|
|
}
|
|
function broadcastClaim(claim) {
|
|
const message = {type:'claim', claim:publicClaim(claim), serverTime:now()};
|
|
broadcastAll(message, client => client.playerId === claim.ownerPlayerId || boundsInViewport(client.viewport, claim.bounds));
|
|
}
|
|
function releaseBoardClaim(boardId, reason = 'released') {
|
|
const claim = claims.get(boardId); if (!claim) return false;
|
|
claims.delete(boardId);
|
|
broadcastAll({type:'claim-release', boardId, reason, serverTime:now()}, client => client.playerId === claim.ownerPlayerId || boundsInViewport(client.viewport, claim.bounds));
|
|
return true;
|
|
}
|
|
function releaseOtherClaimsForPlayer(playerId, keepBoardId) {
|
|
for (const [boardId, claim] of claims) if (boardId !== keepBoardId && claim.ownerPlayerId === playerId) releaseBoardClaim(boardId, 'moved');
|
|
}
|
|
async function claimBoard(identity, boardIdValue, presenceId = null) {
|
|
const boardId = cleanBoardId(boardIdValue), timestamp = now();
|
|
if (!boardId) return {ok:false, reason:'invalid-board', serverTime:timestamp};
|
|
return withSerial(async () => {
|
|
pruneClaims();
|
|
const board = await getBoardInfo(boardId);
|
|
if (!board || board.solved) return {ok:false, reason:board?.solved?'solved':'missing', serverTime:now()};
|
|
const existing = claims.get(boardId);
|
|
if (claimActive(existing) && existing.ownerPlayerId !== identity.playerId) {
|
|
return {ok:false, reason:'occupied', claim:publicClaim(existing), serverTime:now()};
|
|
}
|
|
releaseOtherClaimsForPlayer(identity.playerId, boardId);
|
|
const ownerClient = presenceId ? clients.get(String(presenceId)) : null;
|
|
const linkedClient = ownerClient && ownerClient.authenticated && ownerClient.playerId === identity.playerId ? ownerClient : null;
|
|
const claimedAt = now(), claim = {
|
|
boardId, ownerPlayerId:identity.playerId, ownerName:identity.name,
|
|
ownerPresenceId:linkedClient?.id || null, expiresAt:claimedAt + claimTtlMs, bounds:board.bounds
|
|
};
|
|
claims.set(boardId, claim);
|
|
if (linkedClient) linkedClient.currentClaimBoardId = boardId;
|
|
broadcastClaim(claim);
|
|
return {ok:true, claim:publicClaim(claim), serverTime:claimedAt};
|
|
});
|
|
}
|
|
async function requestClaim(client, message) {
|
|
const requestId = String(message.requestId || '').slice(0, 80);
|
|
const result = await claimBoard({playerId:client.playerId, name:client.name}, message.boardId, client.id);
|
|
sendJson(client, {type:'claim-result', requestId, ...result});
|
|
}
|
|
function touchClaim(client, message) {
|
|
const boardId = cleanBoardId(message.boardId), claim = boardId && claims.get(boardId);
|
|
if (!claimActive(claim) || claim.ownerPlayerId !== client.playerId) return false;
|
|
claim.expiresAt = now() + claimTtlMs; claim.ownerName = client.name; claim.ownerPresenceId = client.id;
|
|
broadcastClaim(claim);
|
|
return true;
|
|
}
|
|
function releaseOwnClaim(client, message) {
|
|
const boardId = cleanBoardId(message.boardId) || client.currentClaimBoardId, claim = boardId && claims.get(boardId);
|
|
if (!claim || claim.ownerPlayerId !== client.playerId) return false;
|
|
return releaseBoardClaim(boardId, 'released');
|
|
}
|
|
function publishReaction(client, message) {
|
|
const emoji=String(message.emoji||''),style=REACTION_STYLES.has(message.style)?message.style:'classic',x=finite(message.x,NaN),y=finite(message.y,NaN),timestamp=now();
|
|
if(!REACTION_EMOJIS.has(emoji)||!Number.isFinite(x)||!Number.isFinite(y)||Math.abs(x)>MAX_WORLD_COORD||Math.abs(y)>MAX_WORLD_COORD)return false;
|
|
if(timestamp-client.lastReactionAt<REACTION_MIN_INTERVAL_MS)return false;
|
|
pruneReactions(timestamp);
|
|
if(style!=='classic')for(const active of reactions.values())if(active.playerId===client.playerId&&active.style!=='classic'&&active.expiresAt>timestamp)return false;
|
|
client.lastReactionAt=timestamp;
|
|
const requested=String(message.id||'').replace(/[^a-zA-Z0-9_-]/g,'').slice(0,64),id=requested||`r${timestamp.toString(36)}-${nextReaction++}`;
|
|
const reaction={id,playerId:client.playerId,playerName:client.name,emoji,style,x,y,createdAt:timestamp,expiresAt:timestamp+reactionTtlForStyle(style)};reactions.set(id,reaction);pruneReactions(timestamp);
|
|
const event={type:'reaction',reaction:publicReaction(reaction),serverTime:timestamp};for(const other of subscribersAt(x,y))if(other.authenticated&&pointInViewport(other.viewport,x,y))sendJson(other,event);return true;
|
|
}
|
|
async function handleMessage(client, raw) {
|
|
const timestamp=now();if(!client.messageWindowStartedAt||timestamp-client.messageWindowStartedAt>=10_000){client.messageWindowStartedAt=timestamp;client.messageCount=0}client.messageCount=(client.messageCount||0)+1;if(client.messageCount>300){sendClose(client,1013,'Message rate exceeded');removeClient(client);return false}
|
|
let message; try { message = JSON.parse(raw); } catch (_) { return sendJson(client, {type:'error', error:'invalid-json'}); }
|
|
if (!message || typeof message !== 'object' || Array.isArray(message)) return;
|
|
if (!client.authenticated) {
|
|
if (message.type !== 'hello') return sendClose(client, 1008, 'Authentication required');
|
|
try {
|
|
const identity = await authenticate({playerId:message.playerId, token:message.token});
|
|
if(playerClientCount(identity.playerId)>=maxClientsPerPlayer)return sendClose(client,1013,'Too many player connections');
|
|
client.authenticated = true; client.playerId = identity.playerId; client.name = identity.name;
|
|
clearTimeout(client.authTimer); client.authTimer = 0;
|
|
sendJson(client, {type:'ready', presenceId:client.id, playerId:client.playerId, name:client.name, claimTtlMs, serverTime:now()});
|
|
} catch (_) { sendClose(client, 1008, 'Authentication failed'); }
|
|
return;
|
|
}
|
|
if (message.type === 'viewport') { registerViewport(client,normalizeViewport(message)); return sendSnapshot(client); }
|
|
if (message.type === 'cursor') {
|
|
const x = finite(message.x, NaN), y = finite(message.y, NaN);
|
|
if (!Number.isFinite(x) || !Number.isFinite(y) || Math.abs(x) > MAX_WORLD_COORD || Math.abs(y) > MAX_WORLD_COORD) return;
|
|
const timestamp = now(); if (timestamp - client.lastCursorMessageAt < 45) return;
|
|
client.lastCursorMessageAt = timestamp; client.x = x; client.y = y; client.vx=clamp(finite(message.vx,0),-2000,2000); client.vy=clamp(finite(message.vy,0),-2000,2000); client.cursorStyle = cleanCursorStyle(message.cursorStyle); client.cursorAt = timestamp; broadcastCursor(client); return;
|
|
}
|
|
if (message.type === 'cursor-hide') {
|
|
const x = client.x, y = client.y; client.x = NaN; client.y = NaN; client.cursorAt = now();
|
|
if (Number.isFinite(x) && Number.isFinite(y)) broadcastAll({type:'player-left', presenceId:client.id, playerId:client.playerId, serverTime:now()}, other => pointInViewport(other.viewport, x, y));
|
|
return;
|
|
}
|
|
if (message.type === 'claim') return requestClaim(client, message);
|
|
if (message.type === 'claim-touch') return touchClaim(client, message);
|
|
if (message.type === 'release') return releaseOwnClaim(client, message);
|
|
if (message.type === 'reaction') return publishReaction(client,message);
|
|
if (message.type === 'snapshot-request') return sendSnapshot(client);
|
|
}
|
|
function consumeFrames(client, chunk) {
|
|
if (client.closed) return;
|
|
client.buffer = client.buffer.length ? Buffer.concat([client.buffer, chunk]) : chunk;
|
|
while (client.buffer.length >= 2) {
|
|
const first = client.buffer[0], second = client.buffer[1], fin = Boolean(first & 0x80), opcode = first & 0x0f, masked = Boolean(second & 0x80);
|
|
let length = second & 0x7f, offset = 2;
|
|
if (!masked) return sendClose(client, 1002, 'Client frames must be masked');
|
|
if (length === 126) { if (client.buffer.length < 4) return; length = client.buffer.readUInt16BE(2); offset = 4; }
|
|
else if (length === 127) {
|
|
if (client.buffer.length < 10) return; const big = client.buffer.readBigUInt64BE(2); if (big > BigInt(MAX_MESSAGE_BYTES)) return sendClose(client, 1009, 'Message too large'); length = Number(big); offset = 10;
|
|
}
|
|
if (length > MAX_MESSAGE_BYTES || client.buffer.length < offset + 4 + length) { if (length > MAX_MESSAGE_BYTES) sendClose(client, 1009, 'Message too large'); return; }
|
|
const mask = client.buffer.subarray(offset, offset + 4); offset += 4;
|
|
const payload = Buffer.from(client.buffer.subarray(offset, offset + length)); client.buffer = client.buffer.subarray(offset + length);
|
|
for (let index = 0; index < payload.length; index++) payload[index] ^= mask[index & 3];
|
|
if (opcode === 8) return sendClose(client, 1000, '');
|
|
if (opcode === 9) { try { client.socket.write(encodeFrame(10, payload)); } catch (_) {} continue; }
|
|
if (opcode === 10) { client.lastPongAt = now(); continue; }
|
|
if (opcode === 1) {
|
|
if (fin) client.queue = client.queue.then(() => handleMessage(client, payload.toString('utf8'))).catch(() => sendClose(client, 1011, 'Message handling failed'));
|
|
else { client.fragmentOpcode = 1; client.fragments = [payload]; client.fragmentBytes = payload.length; }
|
|
continue;
|
|
}
|
|
if (opcode === 0 && client.fragmentOpcode === 1) {
|
|
client.fragmentBytes += payload.length; if (client.fragmentBytes > MAX_MESSAGE_BYTES) return sendClose(client, 1009, 'Message too large'); client.fragments.push(payload);
|
|
if (fin) { const text = Buffer.concat(client.fragments).toString('utf8'); client.fragmentOpcode = 0; client.fragments = []; client.fragmentBytes = 0; client.queue = client.queue.then(() => handleMessage(client, text)).catch(() => sendClose(client, 1011, 'Message handling failed')); }
|
|
continue;
|
|
}
|
|
if (opcode !== 2) return sendClose(client, 1003, 'Unsupported frame');
|
|
}
|
|
}
|
|
function removeClient(client) {
|
|
if (!client || client.removed) return; client.removed = true; client.closed = true; clearTimeout(client.authTimer);if(client.pollWaiter){clearTimeout(client.pollWaiter.timer);client.pollWaiter.resolve(null);client.pollWaiter=null} unregisterViewport(client); clients.delete(client.id);
|
|
for (const [boardId, claim] of claims) if (claim.ownerPresenceId === client.id) releaseBoardClaim(boardId, 'disconnected');
|
|
if (client.authenticated && Number.isFinite(client.x) && Number.isFinite(client.y)) broadcastAll({type:'player-left', presenceId:client.id, playerId:client.playerId, serverTime:now()}, other => pointInViewport(other.viewport, client.x, client.y));
|
|
}
|
|
function pollingClientFor(identity, presenceId) {
|
|
const client = clients.get(String(presenceId || ''));
|
|
if (!client || client.transport !== 'poll' || client.closed || client.playerId !== identity?.playerId) return null;
|
|
client.lastSeenAt = now();
|
|
return client;
|
|
}
|
|
function pollingEnvelope(client, afterSequence = 0) {
|
|
const after = Math.max(0, Math.floor(finite(afterSequence, 0)));
|
|
const events = client.events.filter(event => event.sequence > after);
|
|
const eventGap=after>client.eventSequence||after<(client.droppedThroughSequence||0);
|
|
const sequence = events.length ? events[events.length - 1].sequence : client.eventSequence || 0;
|
|
if (events.length) client.events = client.events.filter(event => event.sequence > sequence);
|
|
return {presenceId:client.id, sequence, eventGap, messages:events.map(event => event.message), serverTime:now()};
|
|
}
|
|
function createPollingClient(identity) {
|
|
if(clients.size>=maxClients||playerClientCount(identity.playerId)>=maxClientsPerPlayer)return null;
|
|
const timestamp = now();
|
|
const client = {id:`h${nextConnection++}-${crypto.randomBytes(4).toString('hex')}`, transport:'poll', socket:null, buffer:Buffer.alloc(0), queue:Promise.resolve(), authenticated:true, closed:false, removed:false, viewport:null, x:NaN, y:NaN, vx:0, vy:0,cursorStyle:'default',cursorAt:0,lastCursorMessageAt:0,lastPongAt:timestamp,lastSeenAt:timestamp,fragments:[],fragmentBytes:0,fragmentOpcode:0,currentClaimBoardId:null,lastReactionAt:0,authTimer:0,viewportBucketKeys:[],playerId:identity.playerId,name:identity.name,events:[],eventSequence:0,droppedThroughSequence:0,pollWaiter:null,wakePoll:null,messageWindowStartedAt:timestamp,messageCount:0};
|
|
clients.set(client.id, client);
|
|
sendJson(client, {type:'ready', presenceId:client.id, playerId:client.playerId, name:client.name, claimTtlMs, serverTime:timestamp});
|
|
return pollingEnvelope(client, 0);
|
|
}
|
|
async function handlePollingMessage(identity, presenceId, message, afterSequence = 0) {
|
|
const client = pollingClientFor(identity, presenceId);
|
|
if (!client) return null;
|
|
await handleMessage(client, JSON.stringify(message || {}));
|
|
return client.closed||client.removed?null:pollingEnvelope(client, afterSequence);
|
|
}
|
|
function pollPollingClient(identity, presenceId, afterSequence = 0, waitMs = 20_000) {
|
|
const client = pollingClientFor(identity, presenceId);
|
|
if(!client)return null;const after=Math.max(0,Math.floor(finite(afterSequence,0)));
|
|
if(client.events.some(event=>event.sequence>after)||after>client.eventSequence||after<(client.droppedThroughSequence||0))return pollingEnvelope(client,after);
|
|
if(client.pollWaiter){clearTimeout(client.pollWaiter.timer);client.pollWaiter.resolve(pollingEnvelope(client,client.pollWaiter.after));client.pollWaiter=null}
|
|
if(waitMs<=0)return pollingEnvelope(client,after);
|
|
return new Promise(resolve=>{const finish=()=>{if(client.pollWaiter?.resolve!==resolve)return;clearTimeout(client.pollWaiter.timer);client.pollWaiter=null;client.wakePoll=null;resolve(client.closed?null:pollingEnvelope(client,after))},timer=setTimeout(finish,Math.max(1000,Math.min(25_000,waitMs)));client.pollWaiter={resolve,after,timer};client.wakePoll=finish});
|
|
}
|
|
function disconnectPollingClient(identity, presenceId) {
|
|
const client = pollingClientFor(identity, presenceId);
|
|
if (!client) return false;
|
|
removeClient(client);
|
|
return true;
|
|
}
|
|
function handleUpgrade(req, socket, head) {
|
|
let url; try { url = new URL(req.url, `http://${req.headers.host || 'localhost'}`); } catch (_) { socket.destroy(); return; }
|
|
if (url.pathname !== path && !url.pathname.endsWith(path)) { socket.destroy(); return; }
|
|
if(clients.size>=maxClients){socket.write('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n');socket.destroy();return}
|
|
const key = req.headers['sec-websocket-key'], version = req.headers['sec-websocket-version'];
|
|
if (req.method !== 'GET' || typeof key !== 'string' || version !== '13') { socket.write('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n'); socket.destroy(); return; }
|
|
const accept = crypto.createHash('sha1').update(key + WS_GUID).digest('base64');
|
|
socket.write(`HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: ${accept}\r\n\r\n`);
|
|
socket.setNoDelay(true);
|
|
const timestamp=now(),client = {id:`p${nextConnection++}-${crypto.randomBytes(4).toString('hex')}`, transport:'websocket', socket, buffer:Buffer.alloc(0), queue:Promise.resolve(), authenticated:false, closed:false, removed:false, viewport:null, x:NaN, y:NaN, vx:0,vy:0,cursorStyle:'default',cursorAt:0,lastCursorMessageAt:0,lastPongAt:timestamp,lastSeenAt:timestamp,fragments:[],fragmentBytes:0,fragmentOpcode:0,currentClaimBoardId:null,lastReactionAt:0,authTimer:0,viewportBucketKeys:[],events:[],eventSequence:0,backpressured:false,messageWindowStartedAt:timestamp,messageCount:0};
|
|
clients.set(client.id, client);
|
|
client.authTimer = setTimeout(() => sendClose(client, 1008, 'Authentication timeout'), 5000);
|
|
socket.on('data', chunk => consumeFrames(client, chunk)); socket.on('error', () => removeClient(client)); socket.on('close', () => removeClient(client)); socket.on('end', () => removeClient(client));
|
|
if (head?.length) consumeFrames(client, head);
|
|
}
|
|
server.on('upgrade', handleUpgrade);
|
|
const cleanupTimer = setInterval(() => {pruneClaims();pruneReactions();}, 15_000); cleanupTimer.unref?.();
|
|
const heartbeatTimer = setInterval(() => {
|
|
const timestamp = now();
|
|
for (const client of clients.values()) {
|
|
if (client.transport === 'poll') {
|
|
if (timestamp - client.lastSeenAt > 45_000) removeClient(client);
|
|
continue;
|
|
}
|
|
if (timestamp - client.lastPongAt > 90_000) { client.socket.destroy(); continue; }
|
|
try { client.socket.write(encodeFrame(9, Buffer.from(String(timestamp)))); } catch (_) { client.socket.destroy(); }
|
|
}
|
|
}, 30_000); heartbeatTimer.unref?.();
|
|
|
|
return {
|
|
claimTtlMs,
|
|
hasClaim(playerId, boardId) { pruneClaims(); const claim = claims.get(boardId); return claimActive(claim) && claim.ownerPlayerId === playerId; },
|
|
releaseBoardClaim,
|
|
broadcastClearEvents(events) { for (const event of events || []) broadcastAll({type:'board-cleared', event, serverTime:now()}); },
|
|
broadcastWorldRevision(revision, boardIds = [], page = null) { const cleanRevision=Math.max(0,Math.floor(finite(revision,0))),ids=[...new Set([...boardIds].map(cleanBoardId).filter(Boolean))],message={type:'world-revision',revision:cleanRevision,boardIds:ids,serverTime:now()};if(page&&typeof page==='object')message.page=page;broadcastAll(message); },
|
|
notifyProfileChange(playerId, name) { for (const client of clients.values()) if (client.playerId === playerId) client.name = name; broadcastAll({type:'player-profile', playerId, name, serverTime:now()}); },
|
|
createPollingClient,
|
|
handlePollingMessage,
|
|
pollPollingClient,
|
|
disconnectPollingClient,
|
|
claimBoard,
|
|
close() { clearInterval(cleanupTimer); clearInterval(heartbeatTimer); server.off('upgrade', handleUpgrade); for (const client of clients.values()) sendClose(client, 1001, 'Server shutdown'); clients.clear(); claims.clear(); reactions.clear(); },
|
|
_debug: {clients, claims, reactions, snapshotFor}
|
|
};
|
|
}
|
|
|
|
module.exports = {createRealtimeHub};
|