diff --git a/packages/server/src/ws/handler.ts b/packages/server/src/ws/handler.ts index b0923b16..2a6e0dbb 100644 --- a/packages/server/src/ws/handler.ts +++ b/packages/server/src/ws/handler.ts @@ -18,6 +18,10 @@ import type { } from '@backspace/shared'; import { sanitizeUser } from '../utils/sanitize.js'; +// ─── Heartbeat State ────────────────────────────────────────────────────────── +const wsIsAlive: WeakMap = new WeakMap(); +let heartbeatInterval: ReturnType | null = null; + // SQLite's SQLITE_MAX_VARIABLE_NUMBER default is 999. // Chunk inArray() calls to stay safely under this limit. const BATCH_CHUNK_SIZE = 500; @@ -193,6 +197,9 @@ class ConnectionManager { status: 'offline', }); } + + // Clean up userSpaces (re-populated on next connect via setUserSpaces) + this.userSpaces.delete(userId); } getUserConnections(userId: string): Set { @@ -529,6 +536,10 @@ class ConnectionManager { return Array.from(this.connections.keys()); } + getAllConnections(): Map> { + return this.connections; + } + /** Push a fresh ready payload to a specific user, forcing full store re-sync. */ pushReadyPayload(userId: string): void { const connections = this.getUserConnections(userId); @@ -965,6 +976,9 @@ export async function registerWebSocket(app: FastifyInstance): Promise { return; } + // Any received message proves liveness + wsIsAlive.set(ws, true); + if (!authenticated) { // First message must be auth if (parsed.type !== 'auth' || typeof parsed.token !== 'string') { @@ -987,6 +1001,10 @@ export async function registerWebSocket(app: FastifyInstance): Promise { // Add connection connectionManager.addConnection(userId, ws); + // Mark alive for heartbeat detection; browsers auto-respond to ping frames (RFC 6455) + wsIsAlive.set(ws, true); + ws.on('pong', () => { wsIsAlive.set(ws, true); }); + // Build and send ready payload const readyData = buildReadyPayload(userId); ws.send(JSON.stringify({ @@ -1039,4 +1057,30 @@ export async function registerWebSocket(app: FastifyInstance): Promise { clearTimeout(authTimeout); }); }); + + // ─── Heartbeat Sweep ────────────────────────────────────────────────────── + // Detect dead connections (e.g. PC shut off without TCP FIN). + // Sends protocol-level ping frames; browsers auto-respond with pong (RFC 6455). + // Worst-case detection: 30s + 30s + 5s grace = ~65s. + const HEARTBEAT_INTERVAL_MS = 30_000; + + heartbeatInterval = setInterval(() => { + for (const [, userConnections] of connectionManager.getAllConnections()) { + for (const ws of userConnections) { + if (wsIsAlive.get(ws) === false) { + ws.terminate(); // Emits 'close' → removeConnection → scheduleDisconnect → finalizeDisconnect + continue; + } + wsIsAlive.set(ws, false); + if (ws.readyState === 1) ws.ping(); + } + } + }, HEARTBEAT_INTERVAL_MS); + + app.addHook('onClose', async () => { + if (heartbeatInterval) { + clearInterval(heartbeatInterval); + heartbeatInterval = null; + } + }); }