'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_VIEW_SPAN = 512; const MAX_WORLD_COORD = 1_000_000_000; const REACTION_EMOJIS = new Set(['👍','👉🏻','🙏','🧠','🎉']); const REACTION_TTL_MS = 4500; 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 || !client.socket.writable) return false; try { client.socket.write(encodeFrame(1, Buffer.from(JSON.stringify(value)))); return true; } catch (_) { return false; } } function sendClose(client, code = 1000, reason = '') { if (!client || client.closed) return; client.closed = true; 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}) { 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 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, cursorStyle: client.cursorStyle || 'default', 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, 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 requestClaim(client, message) { const requestId = String(message.requestId || '').slice(0, 80), boardId = cleanBoardId(message.boardId); if (!boardId) return sendJson(client, {type:'claim-result', requestId, ok:false, reason:'invalid-board', serverTime:now()}); await withSerial(async () => { pruneClaims(); const board = await getBoardInfo(boardId); if (!board || board.solved) return sendJson(client, {type:'claim-result', requestId, ok:false, reason:board?.solved?'solved':'missing', serverTime:now()}); const existing = claims.get(boardId); if (claimActive(existing) && existing.ownerPlayerId !== client.playerId) return sendJson(client, {type:'claim-result', requestId, ok:false, reason:'occupied', claim:publicClaim(existing), serverTime:now()}); releaseOtherClaimsForPlayer(client.playerId, boardId); const timestamp = now(), claim = { boardId, ownerPlayerId:client.playerId, ownerName:client.name, ownerPresenceId:client.id, expiresAt:timestamp + claimTtlMs, bounds:board.bounds }; claims.set(boardId, claim); client.currentClaimBoardId = boardId; sendJson(client, {type:'claim-result', requestId, ok:true, claim:publicClaim(claim), serverTime:timestamp}); broadcastClaim(claim); }); } 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||''),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 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.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; clearTimeout(client.authTimer); unregisterViewport(client); clients.delete(client.id); 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 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) { 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 client = {id:`p${nextConnection++}-${crypto.randomBytes(4).toString('hex')}`, socket, buffer:Buffer.alloc(0), queue:Promise.resolve(), authenticated:false, closed:false, removed:false, viewport:null, x:NaN, y:NaN, cursorStyle:'default', cursorAt:0, lastCursorMessageAt:0, lastPongAt:now(), fragments:[], fragmentBytes:0, fragmentOpcode:0, currentClaimBoardId:null, lastReactionAt:0, authTimer:0, viewportBucketKeys:[]}; 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 (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()}); }, notifyProfileChange(playerId, name) { for (const client of clients.values()) if (client.playerId === playerId) client.name = name; broadcastAll({type:'player-profile', playerId, name, serverTime:now()}); }, 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};