d
This commit is contained in:
parent
c3a6f5ff37
commit
4c4e767ec6
73 changed files with 3502 additions and 741 deletions
|
|
@ -7,8 +7,10 @@ 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_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;
|
||||
|
|
@ -58,13 +60,21 @@ function encodeFrame(opcode, payload = Buffer.alloc(0)) {
|
|||
return Buffer.concat([header, payload]);
|
||||
}
|
||||
function sendJson(client, value) {
|
||||
if (!client || client.closed || !client.socket.writable) return false;
|
||||
if (!client || client.closed) return false;
|
||||
if (client.transport === 'poll') {
|
||||
const sequence = ++client.eventSequence;
|
||||
client.events.push({sequence, message:value});
|
||||
if (client.events.length > 256) client.events.splice(0, client.events.length - 256);
|
||||
return true;
|
||||
}
|
||||
if (!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;
|
||||
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(); }
|
||||
|
|
@ -91,13 +101,13 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
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()};
|
||||
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, x:reaction.x, y:reaction.y, createdAt:reaction.createdAt, expiresAt:reaction.expiresAt}; }
|
||||
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);
|
||||
|
|
@ -142,25 +152,35 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
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 () => {
|
||||
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 sendJson(client, {type:'claim-result', requestId, ok:false, reason:board?.solved?'solved':'missing', serverTime:now()});
|
||||
if (!board || board.solved) return {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
|
||||
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); client.currentClaimBoardId = boardId;
|
||||
sendJson(client, {type:'claim-result', requestId, ok:true, claim:publicClaim(claim), serverTime:timestamp});
|
||||
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;
|
||||
|
|
@ -174,12 +194,14 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
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();
|
||||
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;
|
||||
client.lastReactionAt=timestamp;pruneReactions(timestamp);
|
||||
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,x,y,createdAt:timestamp,expiresAt:timestamp+REACTION_TTL_MS};reactions.set(id,reaction);pruneReactions(timestamp);
|
||||
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) {
|
||||
|
|
@ -200,7 +222,7 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
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.cursorStyle = cleanCursorStyle(message.cursorStyle); client.cursorAt = timestamp; broadcastCursor(client); 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();
|
||||
|
|
@ -245,18 +267,55 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
}
|
||||
}
|
||||
function removeClient(client) {
|
||||
if (!client || client.removed) return; client.removed = true; clearTimeout(client.authTimer); unregisterViewport(client); clients.delete(client.id);
|
||||
if (!client || client.removed) return; client.removed = true; client.closed = true; clearTimeout(client.authTimer); 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 sequence = events.length ? events[events.length - 1].sequence : Math.max(after, client.eventSequence || 0);
|
||||
if (events.length) client.events = client.events.filter(event => event.sequence > sequence);
|
||||
return {presenceId:client.id, sequence, messages:events.map(event => event.message), serverTime:now()};
|
||||
}
|
||||
function createPollingClient(identity) {
|
||||
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};
|
||||
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 pollingEnvelope(client, afterSequence);
|
||||
}
|
||||
function pollPollingClient(identity, presenceId, afterSequence = 0) {
|
||||
const client = pollingClientFor(identity, presenceId);
|
||||
return client ? pollingEnvelope(client, afterSequence) : null;
|
||||
}
|
||||
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) { socket.destroy(); return; }
|
||||
if (url.pathname !== path && !url.pathname.endsWith(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:[]};
|
||||
const 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:now(), lastSeenAt:now(), fragments:[], fragmentBytes:0, fragmentOpcode:0, currentClaimBoardId:null, lastReactionAt:0, authTimer:0, viewportBucketKeys:[], events:[], eventSequence: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));
|
||||
|
|
@ -267,6 +326,10 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
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(); }
|
||||
}
|
||||
|
|
@ -278,6 +341,11 @@ function createRealtimeHub({server, authenticate, getBoardInfo, path = '/api/rea
|
|||
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()}); },
|
||||
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}
|
||||
};
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue