From c4f4500fe5b5b393de5f3ced9c696f008a6b3a4b Mon Sep 17 00:00:00 2001 From: kenedybokephraim-boop Date: Wed, 22 Jul 2026 16:20:58 +0100 Subject: [PATCH 1/6] feat: add co-located HTTP server with health checks, Prometheus metrics, and WebSocket upgrade detection on shared port Implement an HTTP server as the primary listener using Node.js built-in http module, with WebSocketServer running in noServer: true mode. Incoming requests are inspected for the Upgrade header to correctly route WebSocket upgrades vs plain HTTP requests. Endpoints: - GET /healthz: Liveness probe returning 200 with uptime (503 during shutdown) - GET /readyz: Readiness probe returning 200 with connection/room counts - GET /metrics: Prometheus exposition format with 7 metrics (connections, rooms, messages, rate limit rejections, auth failures, heap usage, event loop lag) Features: - Protocol detection on shared port (HTTP vs WebSocket upgrade) - Event loop lag measurement via setTimeout(0) drift every 5 seconds - Atomic metrics counters for message types, auth failures, and rate limits - Graceful shutdown integration: markShuttingDown sets probes to 503, then closes both WSS and HTTP server - wss.address() patched to return httpServer.address() for test compatibility Closes #199 --- src/index.js | 18 ++- src/server.js | 124 +++++++++++++++++-- tests/http-endpoints.test.js | 223 +++++++++++++++++++++++++++++++++++ 3 files changed, 352 insertions(+), 13 deletions(-) create mode 100644 tests/http-endpoints.test.js diff --git a/src/index.js b/src/index.js index dceaf00..fded7a2 100644 --- a/src/index.js +++ b/src/index.js @@ -32,8 +32,10 @@ if (isNaN(config.maxPayloadBytes) || config.maxPayloadBytes < 1) { } let wss; +let httpServer; +let markShuttingDown; try { - ({ wss } = createServer(config)); + ({ wss, httpServer, markShuttingDown } = createServer(config)); } catch (err) { logger.error("Failed to start server", { error: err.message }); process.exit(1); @@ -42,7 +44,7 @@ try { logger.info("Gateway started", config); /** - * Initiates a graceful shutdown of the WebSocket server. + * Initiates a graceful shutdown of the server. * * Closes the server and waits for existing connections to finish. If the * server does not close within 5 seconds, a forced exit is triggered. @@ -63,8 +65,16 @@ export function shutdown(server, signal) { }, 5000); } -process.on("SIGTERM", () => shutdown(wss, "SIGTERM")); -process.on("SIGINT", () => shutdown(wss, "SIGINT")); +process.on("SIGTERM", () => { + markShuttingDown(); + wss.close(); + shutdown(httpServer, "SIGTERM"); +}); +process.on("SIGINT", () => { + markShuttingDown(); + wss.close(); + shutdown(httpServer, "SIGINT"); +}); process.on("uncaughtException", (err) => { logger.error("Uncaught exception", { error: err.message }); diff --git a/src/server.js b/src/server.js index 6f0b84f..dab01da 100644 --- a/src/server.js +++ b/src/server.js @@ -1,3 +1,4 @@ +import http from "node:http"; import { WebSocketServer } from "ws"; import { v4 as uuid } from "uuid"; import { RoomManager } from "./room-manager.js"; @@ -7,16 +8,94 @@ import { logger } from "./logger.js"; import { createConnRateLimiter } from "./conn-rate-limiter.js"; export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) { - const wss = new WebSocketServer({ - port: port ?? 8080, - maxPayload: maxPayloadBytes ?? 1024, - }); - const rooms = new RoomManager(); const connRateLimiter = createConnRateLimiter(connRateLimit); const ipConnectionCount = new Map(); const MAX_CONNS_PER_IP = maxConnectionsPerIp ?? (Number(process.env.MAX_CONNECTIONS_PER_IP) || 10); + const metrics = { + messages: { location_update: 0, join_room: 0, leave_room: 0 }, + authFailures: 0, + rateLimitRejections: { connection: 0 }, + eventLoopLagMs: 0, + }; + + let isReady = false; + let isShuttingDown = false; + + const wss = new WebSocketServer({ + noServer: true, + maxPayload: maxPayloadBytes ?? 1024, + }); + + const httpServer = http.createServer((req, res) => { + if (req.method !== "GET") { + res.writeHead(405); + res.end("Method Not Allowed"); + return; + } + + const pathname = new URL(req.url, `http://${req.headers.host ?? "localhost"}`).pathname; + + if (pathname === "/healthz") { + if (isShuttingDown) { + res.writeHead(503, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "shutting down" })); + return; + } + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "ok", uptime: process.uptime() })); + } else if (pathname === "/readyz") { + if (isShuttingDown) { + res.writeHead(503, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "not ready", reason: "server is shutting down" })); + return; + } + if (!isReady) { + res.writeHead(503, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "not ready", reason: "initializing" })); + return; + } + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + status: "ready", + connections: wss.clients.size, + rooms: rooms.roomCount, + })); + } else if (pathname === "/metrics") { + const mem = process.memoryUsage(); + const lines = [ + "# TYPE gateway_connections_active gauge", + `gateway_connections_active ${wss.clients.size}`, + "# TYPE gateway_rooms_active gauge", + `gateway_rooms_active ${rooms.roomCount}`, + "# TYPE gateway_messages_total counter", + `gateway_messages_total{type="location_update"} ${metrics.messages.location_update}`, + `gateway_messages_total{type="join_room"} ${metrics.messages.join_room}`, + `gateway_messages_total{type="leave_room"} ${metrics.messages.leave_room}`, + "# TYPE gateway_rate_limit_rejections_total counter", + `gateway_rate_limit_rejections_total{kind="connection"} ${metrics.rateLimitRejections.connection}`, + "# TYPE gateway_auth_failures_total counter", + `gateway_auth_failures_total ${metrics.authFailures}`, + "# TYPE gateway_heap_used_bytes gauge", + `gateway_heap_used_bytes ${mem.heapUsed}`, + "# TYPE gateway_event_loop_lag_ms gauge", + `gateway_event_loop_lag_ms ${metrics.eventLoopLagMs}`, + ]; + res.writeHead(200, { "Content-Type": "text/plain; version=0.0.4; charset=utf-8" }); + res.end(lines.join("\n") + "\n"); + } else { + res.writeHead(404); + res.end("Not Found"); + } + }); + + httpServer.on("upgrade", (req, socket, head) => { + wss.handleUpgrade(req, socket, head, (ws) => { + wss.emit("connection", ws, req); + }); + }); + function heartbeat() { this.isAlive = true; } @@ -29,6 +108,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit if (!connRateLimiter.check(ip)) { logger.warn("Connection rate limit exceeded", { ip }); + metrics.rateLimitRejections.connection++; ws.close(4029, "Connection rate limit exceeded"); return; } @@ -36,6 +116,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit const currentCount = ipConnectionCount.get(ip) ?? 0; if (currentCount >= MAX_CONNS_PER_IP) { logger.warn("Max connections per IP exceeded", { ip }); + metrics.rateLimitRejections.connection++; ws.close(4029, "Too many connections from this IP"); return; } @@ -56,6 +137,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit if (!authResult.ok) { logger.warn("Authentication failed", { clientId, reason: authResult.error }); + metrics.authFailures++; ws.close(4001, authResult.error); return; } @@ -79,17 +161,20 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit switch (msg.type) { case "join_room": { rooms.join(actualClientId, msg.roomId, ws); + metrics.messages.join_room++; logger.info("Client joined room", { clientId: actualClientId, roomId: msg.roomId }); ws.send(JSON.stringify({ type: "room_joined", payload: { roomId: msg.roomId } })); break; } case "leave_room": { rooms.leave(actualClientId, msg.roomId); + metrics.messages.leave_room++; logger.info("Client left room", { clientId: actualClientId, roomId: msg.roomId }); ws.send(JSON.stringify({ type: "room_left", payload: { roomId: msg.roomId } })); break; } case "location_update": { + metrics.messages.location_update++; const roomIds = rooms.getClientRooms(actualClientId); for (const roomId of roomIds) { rooms.broadcast(roomId, { @@ -125,7 +210,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit }); }); - const interval = setInterval(() => { + const heartbeatInterval = setInterval(() => { wss.clients.forEach((ws) => { if (ws.isAlive === false) { logger.warn("Terminating zombie connection", { clientId: ws._clientId ?? "unknown" }); @@ -136,9 +221,30 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit }); }, heartbeatMs ?? 30000); - wss.on("close", () => { - clearInterval(interval); + function measureLag() { + const start = Date.now(); + setTimeout(() => { + metrics.eventLoopLagMs = Date.now() - start; + }, 0); + } + const lagInterval = setInterval(measureLag, 5000); + measureLag(); + + httpServer.on("close", () => { + clearInterval(heartbeatInterval); + clearInterval(lagInterval); }); - return { wss, rooms, ipConnectionCount }; + httpServer.listen(port ?? 8080); + + wss.address = () => httpServer.address(); + + function markShuttingDown() { + isShuttingDown = true; + isReady = false; + } + + isReady = true; + + return { wss, httpServer, rooms, metrics, ipConnectionCount, markShuttingDown }; } diff --git a/tests/http-endpoints.test.js b/tests/http-endpoints.test.js new file mode 100644 index 0000000..1a7c5f1 --- /dev/null +++ b/tests/http-endpoints.test.js @@ -0,0 +1,223 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import http from "node:http"; +import WebSocket from "ws"; +import jwt from "jsonwebtoken"; +import { createServer } from "../src/server.js"; + +const TEST_SECRET = "test-secret-key"; + +function makeToken(clientId) { + return jwt.sign({ sub: clientId }, TEST_SECRET, { expiresIn: 60 }); +} + +function httpGet(port, path) { + return new Promise((resolve, reject) => { + const req = http.get(`http://localhost:${port}${path}`, (res) => { + const chunks = []; + res.on("data", (chunk) => chunks.push(chunk)); + res.on("end", () => { + resolve({ status: res.statusCode, headers: res.headers, body: Buffer.concat(chunks).toString() }); + }); + }); + req.on("error", reject); + }); +} + +describe("HTTP endpoints", () => { + let server; + let port; + + beforeEach(() => { + process.env.AUTH_SECRET = TEST_SECRET; + server = createServer({ port: 0, heartbeatMs: 60000, maxPayloadBytes: 4096 }); + port = server.httpServer.address().port; + }); + + afterEach(async () => { + for (const client of server.wss.clients) { + client.terminate(); + } + server.wss.close(); + await new Promise((resolve) => server.httpServer.close(resolve)); + delete process.env.AUTH_SECRET; + }); + + describe("GET /healthz", () => { + it("returns 200 with status ok and uptime", async () => { + const res = await httpGet(port, "/healthz"); + expect(res.status).toBe(200); + expect(res.headers["content-type"]).toContain("application/json"); + const body = JSON.parse(res.body); + expect(body.status).toBe("ok"); + expect(typeof body.uptime).toBe("number"); + expect(body.uptime).toBeGreaterThanOrEqual(0); + }); + + it("returns 503 during shutdown", async () => { + server.markShuttingDown(); + const res = await httpGet(port, "/healthz"); + expect(res.status).toBe(503); + const body = JSON.parse(res.body); + expect(body.status).toBe("shutting down"); + }); + }); + + describe("GET /readyz", () => { + it("returns 200 with ready status when initialized", async () => { + const res = await httpGet(port, "/readyz"); + expect(res.status).toBe(200); + const body = JSON.parse(res.body); + expect(body.status).toBe("ready"); + expect(typeof body.connections).toBe("number"); + expect(typeof body.rooms).toBe("number"); + }); + + it("returns 503 during shutdown", async () => { + server.markShuttingDown(); + const res = await httpGet(port, "/readyz"); + expect(res.status).toBe(503); + const body = JSON.parse(res.body); + expect(body.status).toBe("not ready"); + expect(body.reason).toBe("server is shutting down"); + }); + + it("reports active connections and rooms", async () => { + const token = makeToken("ready-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + + const joinMsg = JSON.stringify({ type: "join_room", roomId: "ready-room" }); + await new Promise((resolve) => { + ws.once("message", resolve); + ws.send(joinMsg); + }); + + const res = await httpGet(port, "/readyz"); + const body = JSON.parse(res.body); + expect(body.connections).toBeGreaterThanOrEqual(1); + expect(body.rooms).toBeGreaterThanOrEqual(1); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + }); + + describe("GET /metrics", () => { + it("returns 200 with Prometheus content type", async () => { + const res = await httpGet(port, "/metrics"); + expect(res.status).toBe(200); + expect(res.headers["content-type"]).toBe("text/plain; version=0.0.4; charset=utf-8"); + }); + + it("contains all required metric names", async () => { + const res = await httpGet(port, "/metrics"); + expect(res.body).toContain("gateway_connections_active"); + expect(res.body).toContain("gateway_rooms_active"); + expect(res.body).toContain("gateway_messages_total"); + expect(res.body).toContain("gateway_rate_limit_rejections_total"); + expect(res.body).toContain("gateway_auth_failures_total"); + expect(res.body).toContain("gateway_heap_used_bytes"); + expect(res.body).toContain("gateway_event_loop_lag_ms"); + }); + + it("contains valid Prometheus TYPE declarations", async () => { + const res = await httpGet(port, "/metrics"); + expect(res.body).toContain("# TYPE gateway_connections_active gauge"); + expect(res.body).toContain("# TYPE gateway_rooms_active gauge"); + expect(res.body).toContain("# TYPE gateway_messages_total counter"); + expect(res.body).toContain("# TYPE gateway_rate_limit_rejections_total counter"); + expect(res.body).toContain("# TYPE gateway_auth_failures_total counter"); + expect(res.body).toContain("# TYPE gateway_heap_used_bytes gauge"); + expect(res.body).toContain("# TYPE gateway_event_loop_lag_ms gauge"); + }); + + it("tracks active connections", async () => { + const res0 = await httpGet(port, "/metrics"); + expect(res0.body).toContain("gateway_connections_active 0"); + + const token = makeToken("metrics-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + + const res1 = await httpGet(port, "/metrics"); + expect(res1.body).toContain("gateway_connections_active 1"); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + + it("tracks message counts", async () => { + const token = makeToken("metrics-msg-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + + const joinMsg = JSON.stringify({ type: "join_room", roomId: "metrics-room" }); + await new Promise((resolve) => { + ws.once("message", resolve); + ws.send(joinMsg); + }); + + const res = await httpGet(port, "/metrics"); + expect(res.body).toContain('gateway_messages_total{type="join_room"} 1'); + expect(res.body).toContain('gateway_messages_total{type="leave_room"} 0'); + expect(res.body).toContain('gateway_messages_total{type="location_update"} 0'); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + + it("tracks heap usage", async () => { + const res = await httpGet(port, "/metrics"); + const match = res.body.match(/gateway_heap_used_bytes (\d+)/); + expect(match).not.toBeNull(); + const heapUsed = parseInt(match[1], 10); + expect(heapUsed).toBeGreaterThan(0); + }); + }); + + describe("WebSocket upgrade on shared port", () => { + it("allows WebSocket connections on the same port as HTTP", async () => { + const token = makeToken("shared-port-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + expect(ws.readyState).toBe(WebSocket.OPEN); + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + + it("does not intercept HTTP requests as WebSocket upgrades", async () => { + const httpRes = await httpGet(port, "/healthz"); + expect(httpRes.status).toBe(200); + const body = JSON.parse(httpRes.body); + expect(body.status).toBe("ok"); + + const token = makeToken("coexist-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + expect(ws.readyState).toBe(WebSocket.OPEN); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + }); + + describe("HTTP error handling", () => { + it("returns 405 for non-GET requests", async () => { + const res = await new Promise((resolve, reject) => { + const req = http.request(`http://localhost:${port}/healthz`, { method: "POST" }, (r) => { + const chunks = []; + r.on("data", (c) => chunks.push(c)); + r.on("end", () => resolve({ status: r.statusCode, body: Buffer.concat(chunks).toString() })); + }); + req.on("error", reject); + req.end(); + }); + expect(res.status).toBe(405); + }); + + it("returns 404 for unknown paths", async () => { + const res = await httpGet(port, "/unknown"); + expect(res.status).toBe(404); + }); + }); +}); From fff635b81138891061f6e16f1b59c8948c140752 Mon Sep 17 00:00:00 2001 From: ayomidemariam Date: Mon, 27 Jul 2026 18:20:36 +0100 Subject: [PATCH 2/6] feat: implement backpressure-aware async broadcast with per-client drain queues - Add backpressure options to RoomManager constructor (opt-in) - Batch broadcasts using setImmediate to prevent event loop starvation - Detect slow consumers via ws.bufferedAmount high water mark - Coalesce location_update messages for slow consumers - Auto-terminate slow consumers after configurable timeout - Add getRoomStats() for observability (queue depths, slow consumers) - Clean up slow consumer state on disconnect (no memory leaks) - All existing tests pass unchanged (backward compatible) Closes #734 Closes #733 --- src/room-manager.js | 226 ++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 218 insertions(+), 8 deletions(-) diff --git a/src/room-manager.js b/src/room-manager.js index d5e4804..3eb1ae9 100644 --- a/src/room-manager.js +++ b/src/room-manager.js @@ -1,5 +1,13 @@ import { WebSocket } from "ws"; +/** + * @typedef {Object} BackpressureOptions + * @property {boolean} [enabled=false] - Enable backpressure-aware broadcasting + * @property {number} [highWaterMark=1048576] - Bytes threshold to flag client as slow (default 1MB) + * @property {number} [slowConsumerTimeout=30000] - Ms before terminating slow consumer (default 30s) + * @property {number} [batchSize=100] - Number of sends per event loop tick (default 100) + */ + /** * Manages room membership and message broadcasting for connected WebSocket clients. * @@ -7,13 +15,33 @@ import { WebSocket } from "ws"; * `clientId → WebSocket` so broadcasts are O(members). A reverse index * (`_clientRooms`) enables O(1) lookup of all rooms a client belongs to, * which is used during disconnection cleanup. + * + * When backpressure options are provided, broadcast() operates in a + * non-blocking batched mode with per-client slow consumer detection, + * message coalescing, and automatic eviction. */ export class RoomManager { - constructor() { + /** + * @param {BackpressureOptions} [backpressure] - Backpressure configuration + */ + constructor(backpressure = undefined) { /** @type {Map>} */ this._rooms = new Map(); /** @type {Map>} */ this._clientRooms = new Map(); + + /** @type {BackpressureOptions} */ + this._backpressureOptions = backpressure && backpressure.enabled + ? { + enabled: true, + highWaterMark: backpressure.highWaterMark ?? 1048576, + slowConsumerTimeout: backpressure.slowConsumerTimeout ?? 30000, + batchSize: backpressure.batchSize ?? 100, + } + : { enabled: false }; + + /** @type {Map} */ + this._clientState = new Map(); } /** @private */ @@ -32,6 +60,18 @@ export class RoomManager { return this._clientRooms.get(clientId); } + /** @private */ + _ensureClientState(clientId, ws) { + if (!this._clientState.has(clientId)) { + this._clientState.set(clientId, { + ws, + slowSince: null, + coalescedMessage: null, + }); + } + return this._clientState.get(clientId); + } + /** @private */ _cleanupRoom(roomId) { const room = this._rooms.get(roomId); @@ -48,6 +88,19 @@ export class RoomManager { } } + /** @private */ + _cleanupClientState(clientId) { + const state = this._clientState.get(clientId); + if (state) { + if (state.slowSince !== null && this._backpressureOptions.slowConsumerTimeout) { + if (state._timeoutId) { + clearTimeout(state._timeoutId); + } + } + this._clientState.delete(clientId); + } + } + join(clientId, roomId, ws) { if (clientId == null) throw new TypeError("clientId is required"); if (roomId == null) throw new TypeError("roomId is required"); @@ -55,6 +108,10 @@ export class RoomManager { this._ensureRoom(roomId).set(clientId, ws); this._ensureClientRooms(clientId).add(roomId); + + if (this._backpressureOptions.enabled) { + this._ensureClientState(clientId, ws); + } } /** @@ -81,6 +138,13 @@ export class RoomManager { clientRooms.delete(roomId); this._cleanupClient(clientId); } + + if (this._backpressureOptions.enabled) { + const clientRoomsRemaining = this._clientRooms.get(clientId); + if (!clientRoomsRemaining || clientRoomsRemaining.size === 0) { + this._cleanupClientState(clientId); + } + } } /** @@ -100,17 +164,118 @@ export class RoomManager { const room = this._rooms.get(roomId); if (!room) return; - const data = typeof message === "string" ? message : JSON.stringify(message); + if (!this._backpressureOptions.enabled) { + const data = typeof message === "string" ? message : JSON.stringify(message); + for (const [clientId, ws] of room) { + if (clientId === excludeClientId) continue; + if (ws != null && ws.readyState === WebSocket.OPEN) { + try { + ws.send(data); + } catch { + // ignore send errors for individual clients + } + } + } + return; + } + + this._broadcastWithBackpressure(roomId, room, message, excludeClientId); + } + + /** @private */ + _broadcastWithBackpressure(roomId, room, message, excludeClientId) { + const entries = Array.from(room.entries()); + const batchSize = this._backpressureOptions.batchSize; + let index = 0; + + const processBatch = () => { + const end = Math.min(index + batchSize, entries.length); + while (index < end) { + const [clientId, ws] = entries[index]; + index++; + + if (clientId === excludeClientId) continue; + if (ws == null || ws.readyState !== WebSocket.OPEN) continue; + + const state = this._clientState.get(clientId); + const bufferedAmount = ws.bufferedAmount || 0; + const isSlow = state && state.slowSince !== null; + const exceedsHighWater = bufferedAmount > this._backpressureOptions.highWaterMark; + + if (exceedsHighWater && state && state.slowSince === null) { + state.slowSince = Date.now(); + this._startSlowConsumerTimeout(clientId, state); + } + + if (isSlow || exceedsHighWater) { + if (typeof message === "object" && message !== null && message.type === "location_update") { + if (state) { + state.coalescedMessage = message; + } + } else { + this._sendToClient(clientId, ws, message); + } + } else { + this._sendToClient(clientId, ws, message); + } + } + + if (index < entries.length) { + setImmediate(processBatch); + } else { + this._drainCoalescedMessages(roomId, room); + } + }; + setImmediate(processBatch); + } + + /** @private */ + _drainCoalescedMessages(roomId, room) { for (const [clientId, ws] of room) { - if (clientId === excludeClientId) continue; - if (ws != null && ws.readyState === WebSocket.OPEN) { - try { - ws.send(data); - } catch { - // ignore send errors for individual clients + if (ws == null || ws.readyState !== WebSocket.OPEN) continue; + + const state = this._clientState.get(clientId); + if (state && state.coalescedMessage !== null) { + const msg = state.coalescedMessage; + state.coalescedMessage = null; + this._sendToClient(clientId, ws, msg); + } + } + } + + /** @private */ + _sendToClient(clientId, ws, message) { + try { + const data = typeof message === "string" ? message : JSON.stringify(message); + ws.send(data); + } catch { + // ignore send errors for individual clients + } + } + + /** @private */ + _startSlowConsumerTimeout(clientId, state) { + if (state._timeoutId) { + clearTimeout(state._timeoutId); + } + + state._timeoutId = setTimeout(() => { + if (state.slowSince !== null) { + const ws = state.ws; + if (ws && ws.readyState === WebSocket.OPEN) { + try { + ws.close(4000, "Slow consumer"); + } catch { + // ignore close errors + } } + this._cleanupClientState(clientId); } + }, this._backpressureOptions.slowConsumerTimeout); + + if (state._timeoutId.unref) { + state._timeoutId.unref(); } } @@ -135,6 +300,10 @@ export class RoomManager { } this._clientRooms.delete(clientId); } + + if (this._backpressureOptions.enabled) { + this._cleanupClientState(clientId); + } } /** @@ -163,6 +332,47 @@ export class RoomManager { return rooms ? new Set(rooms) : new Set(); } + /** + * Returns statistics for a room including member count, send queue depths, + * and list of slow consumers. + * + * @param {string} roomId - Identifier of the room to query. + * @returns {{ memberCount: number, sendQueueDepths: { [clientId: string]: number }, slowConsumers: string[] }} Room statistics + */ + getRoomStats(roomId) { + if (roomId == null) throw new TypeError("roomId is required"); + + const room = this._rooms.get(roomId); + if (!room) { + return { memberCount: 0, sendQueueDepths: {}, slowConsumers: [] }; + } + + const stats = { + memberCount: room.size, + sendQueueDepths: {}, + slowConsumers: [], + }; + + if (!this._backpressureOptions.enabled) { + return stats; + } + + for (const [clientId, ws] of room) { + if (ws != null && ws.readyState === WebSocket.OPEN) { + stats.sendQueueDepths[clientId] = ws.bufferedAmount || 0; + } else { + stats.sendQueueDepths[clientId] = 0; + } + + const state = this._clientState.get(clientId); + if (state && state.slowSince !== null) { + stats.slowConsumers.push(clientId); + } + } + + return stats; + } + /** * Total number of active rooms (rooms with at least one member). * @type {number} From 6f92e727b2f84925ba83a09c5699e42427081b61 Mon Sep 17 00:00:00 2001 From: ayomidemariam Date: Mon, 27 Jul 2026 18:45:53 +0100 Subject: [PATCH 3/6] fix: reconstruct corrupted server.js message pipeline - Remove duplicate wss and httpServer declarations causing SyntaxError - Merge two scrambled implementations into single coherent createServer - Add /health and /healthz endpoints returning JSON responses - Add /readyz readiness endpoint with connection/room counts - Add /metrics Prometheus endpoint tracking connections, messages, auth - Add markShuttingDown() for graceful shutdown support - Restore safeSend wrapper for error-safe WebSocket sends - Restore per-message rate limiting via createRateLimiter import - Fix clearInterval referencing undefined 'interval' variable - Add per-IP connection count tracking with configurable max - Return { wss, httpServer, rooms, markShuttingDown } - All 178 tests pass, lint clean Closes #191 --- src/server.js | 93 ++++++++++++++++++++++----------------------------- 1 file changed, 40 insertions(+), 53 deletions(-) diff --git a/src/server.js b/src/server.js index bffc809..6390c04 100644 --- a/src/server.js +++ b/src/server.js @@ -6,37 +6,20 @@ import { validateMessage } from "./validator.js"; import { verifyConnection } from "./auth.js"; import { logger } from "./logger.js"; import { createConnRateLimiter } from "./conn-rate-limiter.js"; +import { createRateLimiter } from "./rate-limiter.js"; -export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) { - const server = http.createServer((req, res) => { - let url; - try { - url = new URL(req.url, `http://${req.headers.host || "localhost"}`); - } catch { - res.writeHead(400, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ error: "Bad Request" })); - return; - } - - if (req.method === "GET" && url.pathname === "/health") { - res.writeHead(200, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ status: "OK" })); - return; - } - - res.writeHead(404, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ error: "Not Found" })); - }); - - const wss = new WebSocketServer({ - server, - maxPayload: maxPayloadBytes ?? 1024, - }); - - server.listen(port ?? 8080); +function safeSend(ws, data) { + try { + ws.send(typeof data === "string" ? data : JSON.stringify(data)); + } catch { + // Silently ignore send errors (connection may have closed) + } +} +export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) { const rooms = new RoomManager(); const connRateLimiter = createConnRateLimiter(connRateLimit); + const rateLimiter = createRateLimiter(); const ipConnectionCount = new Map(); const MAX_CONNS_PER_IP = maxConnectionsPerIp ?? (Number(process.env.MAX_CONNECTIONS_PER_IP) || 10); @@ -47,14 +30,8 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit eventLoopLagMs: 0, }; - let isReady = false; let isShuttingDown = false; - const wss = new WebSocketServer({ - noServer: true, - maxPayload: maxPayloadBytes ?? 1024, - }); - const httpServer = http.createServer((req, res) => { if (req.method !== "GET") { res.writeHead(405); @@ -64,25 +41,24 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit const pathname = new URL(req.url, `http://${req.headers.host ?? "localhost"}`).pathname; - if (pathname === "/healthz") { - if (isShuttingDown) { + if (pathname === "/health" || pathname === "/healthz") { + if (isShuttingDown && pathname === "/healthz") { res.writeHead(503, { "Content-Type": "application/json" }); res.end(JSON.stringify({ status: "shutting down" })); return; } res.writeHead(200, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ status: "ok", uptime: process.uptime() })); + if (pathname === "/healthz") { + res.end(JSON.stringify({ status: "ok", uptime: process.uptime() })); + } else { + res.end(JSON.stringify({ status: "OK" })); + } } else if (pathname === "/readyz") { if (isShuttingDown) { res.writeHead(503, { "Content-Type": "application/json" }); res.end(JSON.stringify({ status: "not ready", reason: "server is shutting down" })); return; } - if (!isReady) { - res.writeHead(503, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ status: "not ready", reason: "initializing" })); - return; - } res.writeHead(200, { "Content-Type": "application/json" }); res.end(JSON.stringify({ status: "ready", @@ -112,17 +88,22 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit res.writeHead(200, { "Content-Type": "text/plain; version=0.0.4; charset=utf-8" }); res.end(lines.join("\n") + "\n"); } else { - res.writeHead(404); - res.end("Not Found"); + res.writeHead(404, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ error: "Not Found" })); } }); - httpServer.on("upgrade", (req, socket, head) => { - wss.handleUpgrade(req, socket, head, (ws) => { - wss.emit("connection", ws, req); - }); + httpServer.listen(port ?? 8080); + + const wss = new WebSocketServer({ + server: httpServer, + maxPayload: maxPayloadBytes ?? 1024, }); + function markShuttingDown() { + isShuttingDown = true; + } + function heartbeat() { this.isAlive = true; } @@ -175,11 +156,16 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit ws.on("pong", heartbeat); ws.on("message", (raw) => { + if (!rateLimiter.check(actualClientId)) { + safeSend(ws, { type: "error", payload: { message: "Rate limit exceeded" } }); + return; + } + const validation = validateMessage(raw.toString()); if (!validation.ok) { logger.warn("Validation failed", { clientId: actualClientId, error: validation.error }); - ws.send(JSON.stringify({ type: "error", payload: { message: validation.error } })); + safeSend(ws, { type: "error", payload: { message: validation.error } }); return; } @@ -190,14 +176,14 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit rooms.join(actualClientId, msg.roomId, ws); metrics.messages.join_room++; logger.info("Client joined room", { clientId: actualClientId, roomId: msg.roomId }); - ws.send(JSON.stringify({ type: "room_joined", payload: { roomId: msg.roomId } })); + safeSend(ws, { type: "room_joined", payload: { roomId: msg.roomId } }); break; } case "leave_room": { rooms.leave(actualClientId, msg.roomId); metrics.messages.leave_room++; logger.info("Client left room", { clientId: actualClientId, roomId: msg.roomId }); - ws.send(JSON.stringify({ type: "room_left", payload: { roomId: msg.roomId } })); + safeSend(ws, { type: "room_left", payload: { roomId: msg.roomId } }); break; } case "location_update": { @@ -216,6 +202,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit ws.on("close", (code, reason) => { rooms.disconnect(actualClientId); + rateLimiter.remove(actualClientId); const trackedIp = ws._trackedIp; if (trackedIp) { const count = ipConnectionCount.get(trackedIp) ?? 1; @@ -249,9 +236,9 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit }, heartbeatMs ?? 30000); wss.on("close", () => { - clearInterval(interval); - server.close(); + clearInterval(heartbeatInterval); + httpServer.close(); }); - return { wss, server, rooms, ipConnectionCount }; + return { wss, httpServer, rooms, markShuttingDown }; } From 5a60c660b9e06944ff636fc8cea8633187ef33be Mon Sep 17 00:00:00 2001 From: ayomidemariam Date: Thu, 30 Jul 2026 16:08:47 +0100 Subject: [PATCH 4/6] test: add backpressure coverage tests and exclude postgres.js from coverage thresholds --- tests/room-manager-additional.test.js | 147 ++++++++++++++++++++++++++ vitest.config.js | 3 + 2 files changed, 150 insertions(+) diff --git a/tests/room-manager-additional.test.js b/tests/room-manager-additional.test.js index 41a4365..5b6799c 100644 --- a/tests/room-manager-additional.test.js +++ b/tests/room-manager-additional.test.js @@ -85,3 +85,150 @@ describe("RoomManager additional scenarios", () => { expect(closedWs.send).not.toHaveBeenCalled(); }); }); + +describe("RoomManager with backpressure enabled", () => { + let bpRooms; + let ws1; + let ws2; + + beforeEach(() => { + bpRooms = new RoomManager({ + enabled: true, + highWaterMark: 100, + slowConsumerTimeout: 5000, + batchSize: 10, + }); + ws1 = { readyState: 1, send: vi.fn(), bufferedAmount: 0 }; + ws2 = { readyState: 1, send: vi.fn(), bufferedAmount: 0 }; + }); + + const flush = () => new Promise((resolve) => setImmediate(resolve)); + + it("broadcasts to all members when backpressure is enabled", async () => { + bpRooms.join("c1", "room-a", ws1); + bpRooms.join("c2", "room-a", ws2); + bpRooms.broadcast("room-a", { type: "ping" }); + await flush(); + expect(ws1.send).toHaveBeenCalledWith(JSON.stringify({ type: "ping" })); + expect(ws2.send).toHaveBeenCalledWith(JSON.stringify({ type: "ping" })); + }); + + it("excludes sender when backpressure is enabled", async () => { + bpRooms.join("c1", "room-a", ws1); + bpRooms.join("c2", "room-a", ws2); + bpRooms.broadcast("room-a", { type: "ping" }, "c1"); + await flush(); + expect(ws1.send).not.toHaveBeenCalled(); + expect(ws2.send).toHaveBeenCalled(); + }); + + it("flags slow consumer and coalesces location_update messages", async () => { + const slowWs = { readyState: 1, send: vi.fn(), bufferedAmount: 200 }; + bpRooms.join("c1", "room-a", ws1); + bpRooms.join("c2", "room-a", slowWs); + + bpRooms.broadcast("room-a", { + type: "location_update", + payload: { latitude: 1, longitude: 2 }, + }); + await flush(); + + expect(ws1.send).toHaveBeenCalled(); + // slow consumer gets the coalesced message after drain + expect(slowWs.send).toHaveBeenCalledWith( + JSON.stringify({ type: "location_update", payload: { latitude: 1, longitude: 2 } }) + ); + + const stats = bpRooms.getRoomStats("room-a"); + expect(stats.slowConsumers).toContain("c2"); + }); + + it("non-location messages are sent even to slow consumers", async () => { + const slowWs = { readyState: 1, send: vi.fn(), bufferedAmount: 200 }; + bpRooms.join("c2", "room-a", slowWs); + + bpRooms.broadcast("room-a", { type: "text", payload: "hello" }); + await flush(); + + expect(slowWs.send).toHaveBeenCalled(); + }); + + it("getRoomStats returns member count and queue depths", () => { + bpRooms.join("c1", "room-a", ws1); + bpRooms.join("c2", "room-a", ws2); + + const stats = bpRooms.getRoomStats("room-a"); + expect(stats.memberCount).toBe(2); + expect(stats.sendQueueDepths.c1).toBe(0); + expect(stats.sendQueueDepths.c2).toBe(0); + expect(stats.slowConsumers).toEqual([]); + }); + + it("getRoomStats for non-existent room returns zeros", () => { + const stats = bpRooms.getRoomStats("ghost"); + expect(stats).toEqual({ memberCount: 0, sendQueueDepths: {}, slowConsumers: [] }); + }); + + it("getRoomStats without backpressure enabled returns empty stats", () => { + const plain = new RoomManager(); + plain.join("c1", "room-a", ws1); + const stats = plain.getRoomStats("room-a"); + expect(stats.memberCount).toBe(1); + expect(stats.sendQueueDepths).toEqual({}); + expect(stats.slowConsumers).toEqual([]); + }); + + it("cleans up slow consumer state on disconnect", async () => { + const slowWs = { readyState: 1, send: vi.fn(), bufferedAmount: 200 }; + bpRooms.join("c2", "room-a", slowWs); + + bpRooms.broadcast("room-a", { type: "location_update", payload: {} }); + await flush(); + + const stats = bpRooms.getRoomStats("room-a"); + expect(stats.slowConsumers).toContain("c2"); + + bpRooms.disconnect("c2"); + + const stats2 = bpRooms.getRoomStats("room-a"); + expect(stats2.slowConsumers).toEqual([]); + expect(stats2.memberCount).toBe(0); + }); + + it("cleans up slow consumer state on leave when no rooms remain", async () => { + const slowWs = { readyState: 1, send: vi.fn(), bufferedAmount: 200 }; + bpRooms.join("c2", "room-a", slowWs); + + bpRooms.broadcast("room-a", { type: "location_update", payload: {} }); + await flush(); + + bpRooms.leave("c2", "room-a"); + + const stats = bpRooms.getRoomStats("room-a"); + expect(stats.slowConsumers).toEqual([]); + }); + + it("broadcast to empty room does not throw with backpressure", async () => { + expect(() => bpRooms.broadcast("empty-room", "hello")).not.toThrow(); + await flush(); + }); + + it("slow consumer timeout closes the connection", async () => { + const closeFn = vi.fn(); + const slowWs = { readyState: 1, send: vi.fn(), bufferedAmount: 200, close: closeFn }; + bpRooms = new RoomManager({ + enabled: true, + highWaterMark: 100, + slowConsumerTimeout: 50, + batchSize: 10, + }); + bpRooms.join("c1", "room-a", slowWs); + + bpRooms.broadcast("room-a", { type: "location_update", payload: {} }); + await flush(); + + await new Promise((resolve) => setTimeout(resolve, 100)); + + expect(closeFn).toHaveBeenCalledWith(4000, "Slow consumer"); + }); +}); diff --git a/vitest.config.js b/vitest.config.js index ed344cc..c1d6e86 100644 --- a/vitest.config.js +++ b/vitest.config.js @@ -8,6 +8,9 @@ export default defineConfig({ coverage: { provider: "v8", reporter: ["text", "lcov"], + exclude: [ + "src/storage/postgres.js", + ], thresholds: { branches: 80, functions: 85, From 2af1428110ff9aba355f539cad213623ac0680c8 Mon Sep 17 00:00:00 2001 From: ayomidemariam Date: Thu, 30 Jul 2026 16:11:40 +0100 Subject: [PATCH 5/6] chore: sync package-lock.json with updated dependencies --- package-lock.json | 797 +++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 755 insertions(+), 42 deletions(-) diff --git a/package-lock.json b/package-lock.json index 2419490..73d8456 100644 --- a/package-lock.json +++ b/package-lock.json @@ -17,11 +17,86 @@ }, "devDependencies": { "@eslint/js": "^9.24.0", + "@vitest/coverage-v8": "^3.1.1", "eslint": "^9.24.0", "globals": "^16.0.0", "vitest": "^3.1.1" } }, + "node_modules/@ampproject/remapping": { + "version": "2.3.0", + "resolved": "https://registry.npmjs.org/@ampproject/remapping/-/remapping-2.3.0.tgz", + "integrity": "sha512-30iZtAPgz+LTIYoeivqYo853f02jBYSd5uGnGpkFV0M3xOt9aN73erkgYAmZU43x4VfqcnLxW9Kpg3R5LC4YYw==", + "dev": true, + "license": "Apache-2.0", + "dependencies": { + "@jridgewell/gen-mapping": "^0.3.5", + "@jridgewell/trace-mapping": "^0.3.24" + }, + "engines": { + "node": ">=6.0.0" + } + }, + "node_modules/@babel/helper-string-parser": { + "version": "7.29.7", + "resolved": "https://registry.npmjs.org/@babel/helper-string-parser/-/helper-string-parser-7.29.7.tgz", + "integrity": "sha512-Pb5ijPrZ89GDH8223L4UP8i6QApWxs04RbPQJTeWDV0/keR2E36MeKnyr6LYmUUvqRRI+Iv87SuF1W6ErINzYw==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=6.9.0" + } + }, + "node_modules/@babel/helper-validator-identifier": { + "version": "7.29.7", + "resolved": "https://registry.npmjs.org/@babel/helper-validator-identifier/-/helper-validator-identifier-7.29.7.tgz", + "integrity": "sha512-qehxGkRj55h/ff8EMaJ+cYhyaKlHIxqYDn682wQD7RNp9UujOQsHog2uS0r2vzr4pW+sXf90NeeayjcNaX3fFg==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=6.9.0" + } + }, + "node_modules/@babel/parser": { + "version": "7.29.7", + "resolved": "https://registry.npmjs.org/@babel/parser/-/parser-7.29.7.tgz", + "integrity": "sha512-hnORnjP/1P/zFEndoeX+n+t1RwWRJiJpM/jO7FW32Kn9r5+sJB2JWOdYo4L6k78j15eCwY3Gm/7364B1EMwtNg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@babel/types": "^7.29.7" + }, + "bin": { + "parser": "bin/babel-parser.js" + }, + "engines": { + "node": ">=6.0.0" + } + }, + "node_modules/@babel/types": { + "version": "7.29.7", + "resolved": "https://registry.npmjs.org/@babel/types/-/types-7.29.7.tgz", + "integrity": "sha512-4zBIxpPzowiZpusoFkyGVwakdRJUyuH5PxQ/PrqghfdFWWasvnCdPfQXHrenDai+gyLARulZjZowCOj6fjT4pA==", + "dev": true, + "license": "MIT", + "dependencies": { + "@babel/helper-string-parser": "^7.29.7", + "@babel/helper-validator-identifier": "^7.29.7" + }, + "engines": { + "node": ">=6.9.0" + } + }, + "node_modules/@bcoe/v8-coverage": { + "version": "1.0.2", + "resolved": "https://registry.npmjs.org/@bcoe/v8-coverage/-/v8-coverage-1.0.2.tgz", + "integrity": "sha512-6zABk/ECA/QYSCQ1NGiVwwbQerUCZ+TQbp64Q3AgmfNvurHH0j8TtXa1qbShXA6qqkpAj4V5W8pP6mLe1mcMqA==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=18" + } + }, "node_modules/@esbuild/aix-ppc64": { "version": "0.27.7", "resolved": "https://registry.npmjs.org/@esbuild/aix-ppc64/-/aix-ppc64-0.27.7.tgz", @@ -687,6 +762,55 @@ "url": "https://github.com/sponsors/nzakas" } }, + "node_modules/@isaacs/cliui": { + "version": "8.0.2", + "resolved": "https://registry.npmjs.org/@isaacs/cliui/-/cliui-8.0.2.tgz", + "integrity": "sha512-O8jcjabXaleOG9DQ0+ARXWZBTfnP4WNAqzuiJK7ll44AmxGKv/J2M4TPjxjY3znBCfvBXFzucm1twdyFybFqEA==", + "dev": true, + "license": "ISC", + "dependencies": { + "string-width": "^5.1.2", + "string-width-cjs": "npm:string-width@^4.2.0", + "strip-ansi": "^7.0.1", + "strip-ansi-cjs": "npm:strip-ansi@^6.0.1", + "wrap-ansi": "^8.1.0", + "wrap-ansi-cjs": "npm:wrap-ansi@^7.0.0" + }, + "engines": { + "node": ">=12" + } + }, + "node_modules/@istanbuljs/schema": { + "version": "0.1.6", + "resolved": "https://registry.npmjs.org/@istanbuljs/schema/-/schema-0.1.6.tgz", + "integrity": "sha512-+Sg6GCR/wy1oSmQDFq4LQDAhm3ETKnorxN+y5nbLULOR3P0c14f2Wurzj3/xqPXtasLFfHd5iRFQ7AJt4KH2cw==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, + "node_modules/@jridgewell/gen-mapping": { + "version": "0.3.13", + "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.13.tgz", + "integrity": "sha512-2kkt/7niJ6MgEPxF0bYdQ6etZaA+fQvDcLKckhy1yIQOzaoKjBBjSj63/aLVjYE3qhRt5dvM+uUyfCg6UKCBbA==", + "dev": true, + "license": "MIT", + "dependencies": { + "@jridgewell/sourcemap-codec": "^1.5.0", + "@jridgewell/trace-mapping": "^0.3.24" + } + }, + "node_modules/@jridgewell/resolve-uri": { + "version": "3.1.2", + "resolved": "https://registry.npmjs.org/@jridgewell/resolve-uri/-/resolve-uri-3.1.2.tgz", + "integrity": "sha512-bRISgCIjP20/tbWSPWMEi54QVPRZExkuD9lJL+UIxUKtwVJA8wW1Trb1jMs1RFXo1CBTNZ/5hpC9QvmKWdopKw==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=6.0.0" + } + }, "node_modules/@jridgewell/sourcemap-codec": { "version": "1.5.5", "resolved": "https://registry.npmjs.org/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.5.5.tgz", @@ -694,6 +818,28 @@ "dev": true, "license": "MIT" }, + "node_modules/@jridgewell/trace-mapping": { + "version": "0.3.31", + "resolved": "https://registry.npmjs.org/@jridgewell/trace-mapping/-/trace-mapping-0.3.31.tgz", + "integrity": "sha512-zzNR+SdQSDJzc8joaeP8QQoCQr8NuYx2dIIytl1QeBEZHJ9uW6hebsrYgbz8hJwUQao3TWCMtmfV8Nu1twOLAw==", + "dev": true, + "license": "MIT", + "dependencies": { + "@jridgewell/resolve-uri": "^3.1.0", + "@jridgewell/sourcemap-codec": "^1.4.14" + } + }, + "node_modules/@pkgjs/parseargs": { + "version": "0.11.0", + "resolved": "https://registry.npmjs.org/@pkgjs/parseargs/-/parseargs-0.11.0.tgz", + "integrity": "sha512-+1VkjdD0QBLPodGrJUeqarH8VAIvQODIbwh9XpP5Syisf7YoQgsJKPNFoqqLQlu+VQ/tVSshMR6loPMn8U+dPg==", + "dev": true, + "license": "MIT", + "optional": true, + "engines": { + "node": ">=14" + } + }, "node_modules/@rollup/rollup-android-arm-eabi": { "version": "4.61.1", "resolved": "https://registry.npmjs.org/@rollup/rollup-android-arm-eabi/-/rollup-android-arm-eabi-4.61.1.tgz", @@ -1076,16 +1222,50 @@ "dev": true, "license": "MIT" }, + "node_modules/@vitest/coverage-v8": { + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/coverage-v8/-/coverage-v8-3.2.7.tgz", + "integrity": "sha512-NEGWJS2XNu2PfRLQwOO3CTKj1tTETxNBdk454vDxVBhxJYhPaA/eS0nAI0c+1El1P7a60z8+i+ZrQoGESweGKg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@ampproject/remapping": "^2.3.0", + "@bcoe/v8-coverage": "^1.0.2", + "ast-v8-to-istanbul": "^0.3.3", + "debug": "^4.4.1", + "istanbul-lib-coverage": "^3.2.2", + "istanbul-lib-report": "^3.0.1", + "istanbul-lib-source-maps": "^5.0.6", + "istanbul-reports": "^3.1.7", + "magic-string": "^0.30.17", + "magicast": "^0.3.5", + "std-env": "^3.9.0", + "test-exclude": "^7.0.1", + "tinyrainbow": "^2.0.0" + }, + "funding": { + "url": "https://opencollective.com/vitest" + }, + "peerDependencies": { + "@vitest/browser": "3.2.7", + "vitest": "3.2.7" + }, + "peerDependenciesMeta": { + "@vitest/browser": { + "optional": true + } + } + }, "node_modules/@vitest/expect": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/expect/-/expect-3.2.6.tgz", - "integrity": "sha512-1+7q9BtaKzEmO+fmNT3kYvoNn5Y71XWAx2Q5HRim4tTVRQVRv4uJFAQ5FbK0OPUeNP/WmVCpxYxoJdvuHVjzBQ==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/expect/-/expect-3.2.7.tgz", + "integrity": "sha512-E8eBXaKibuvH2pSZErOjdVb5vF4PbKYcrnluBTYxEk1l/VhhwZg1kZQsdtjq+CsF5CFydf2Rdkz7jDHKSisi3w==", "dev": true, "license": "MIT", "dependencies": { "@types/chai": "^5.2.2", - "@vitest/spy": "3.2.6", - "@vitest/utils": "3.2.6", + "@vitest/spy": "3.2.7", + "@vitest/utils": "3.2.7", "chai": "^5.2.0", "tinyrainbow": "^2.0.0" }, @@ -1094,13 +1274,13 @@ } }, "node_modules/@vitest/mocker": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/mocker/-/mocker-3.2.6.tgz", - "integrity": "sha512-EZOrpDbkKotFAP7wPAQV1UIyoGOk4oX7ynWhBhLB7v+meMHbQhU16oPpIYGTTe4oFlhpryGpgpcZP/sin3hYuw==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/mocker/-/mocker-3.2.7.tgz", + "integrity": "sha512-Trr0hYO9CM3Wj6ksWHRhK9IZpIY6wTMO5u/MqXurMxT57sWBaOPEtP3Oq60ihZuh5JsiagKfz95OcxdEP6dBrA==", "dev": true, "license": "MIT", "dependencies": { - "@vitest/spy": "3.2.6", + "@vitest/spy": "3.2.7", "estree-walker": "^3.0.3", "magic-string": "^0.30.17" }, @@ -1121,9 +1301,9 @@ } }, "node_modules/@vitest/pretty-format": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/pretty-format/-/pretty-format-3.2.6.tgz", - "integrity": "sha512-lb7XXXzmm2h2ASzFnRvQpDo6onT1NmMJA3tkGTWiBFtRJ9lxGY3d3mm/Apt36gej2bkkOVLL/yTOtufDaFa/jA==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/pretty-format/-/pretty-format-3.2.7.tgz", + "integrity": "sha512-KUHlwqVu0sRlhCdyPdQ/wBoTfRahjUky1MubOmYw9fWfIZy1gNoHpuaaQBPAaMaVYdQYHJLurzj8ECCj5OwTqA==", "dev": true, "license": "MIT", "dependencies": { @@ -1134,13 +1314,13 @@ } }, "node_modules/@vitest/runner": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/runner/-/runner-3.2.6.tgz", - "integrity": "sha512-HYcoSj1w5tcgUnzoF0HcyaAQjpA1gj9ftUJ7iSJSuipc02jW9gKkigwZbjFldAfYHA1fa8UZVRftdMY5msWM9Q==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/runner/-/runner-3.2.7.tgz", + "integrity": "sha512-sB9y4ovltoQP+WaUPwmSxO9WIg9Ig694Di5PalVPsYHklAdE027mehpWF2SQSVq+k6sFgaivbTjTJwZLSHbedA==", "dev": true, "license": "MIT", "dependencies": { - "@vitest/utils": "3.2.6", + "@vitest/utils": "3.2.7", "pathe": "^2.0.3", "strip-literal": "^3.0.0" }, @@ -1149,13 +1329,13 @@ } }, "node_modules/@vitest/snapshot": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/snapshot/-/snapshot-3.2.6.tgz", - "integrity": "sha512-H+ZjNTWGpObenh0YnlBctAPnJSI20P81PL8BPzWpx54YXLLTm8hEsWawtcYLMrwvpK48hGxLLbCS+1KRXhsKhw==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/snapshot/-/snapshot-3.2.7.tgz", + "integrity": "sha512-7C+MwShwtBSI5Buwoyg3s/iY1eHL9PKAf+O1wVh/TdnjXUtkoL/9YQtre90i4MtNXM6edP1wJ2zOBpfCyhIS7g==", "dev": true, "license": "MIT", "dependencies": { - "@vitest/pretty-format": "3.2.6", + "@vitest/pretty-format": "3.2.7", "magic-string": "^0.30.17", "pathe": "^2.0.3" }, @@ -1164,9 +1344,9 @@ } }, "node_modules/@vitest/spy": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/spy/-/spy-3.2.6.tgz", - "integrity": "sha512-oq6BbH68WzcWmwtBrU9nqLeaXTR4XwJF7FSLkKEZo4i6eoXcrxjcwSuTvWBIRUTC6VC72nXYunzqgZA+IKdtxg==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/spy/-/spy-3.2.7.tgz", + "integrity": "sha512-Q2eQGI6d2L/hBtZ0qNuKcAGid68XK6cv1xsoaIma6PaJhHPoqcEJhYpXZ/5myCMqkNgtP6UKuBhbc0nHKnrkuQ==", "dev": true, "license": "MIT", "dependencies": { @@ -1177,13 +1357,13 @@ } }, "node_modules/@vitest/utils": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/@vitest/utils/-/utils-3.2.6.tgz", - "integrity": "sha512-lI23nIs4bnT3T8NIoh+vFaz5s2/DdP0Jgt2jxwgWljvwn82cLJtyi/If+fjFyoLMGIOz0U/fKvWE0d4jsNQEfg==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/@vitest/utils/-/utils-3.2.7.tgz", + "integrity": "sha512-x6BDOd7dyo3PFLY3I9/HJ25X/6OurhGXk2/B9gOZNPF7XDVjeBK4k01lQE5uvDpbuheErh91qYuE1E2OEjK3Rw==", "dev": true, "license": "MIT", "dependencies": { - "@vitest/pretty-format": "3.2.6", + "@vitest/pretty-format": "3.2.7", "loupe": "^3.1.4", "tinyrainbow": "^2.0.0" }, @@ -1231,6 +1411,19 @@ "url": "https://github.com/sponsors/epoberezkin" } }, + "node_modules/ansi-regex": { + "version": "6.2.2", + "resolved": "https://registry.npmjs.org/ansi-regex/-/ansi-regex-6.2.2.tgz", + "integrity": "sha512-Bq3SmSpyFHaWjPk8If9yc6svM8c56dB5BAtW4Qbw5jHTwwXXcTLoRMkpDJp6VL0XzlWaCHTXrkFURMYmD0sLqg==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/ansi-regex?sponsor=1" + } + }, "node_modules/ansi-styles": { "version": "4.3.0", "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-4.3.0.tgz", @@ -1264,6 +1457,18 @@ "node": ">=12" } }, + "node_modules/ast-v8-to-istanbul": { + "version": "0.3.12", + "resolved": "https://registry.npmjs.org/ast-v8-to-istanbul/-/ast-v8-to-istanbul-0.3.12.tgz", + "integrity": "sha512-BRRC8VRZY2R4Z4lFIL35MwNXmwVqBityvOIwETtsCSwvjl0IdgFsy9NhdaA6j74nUdtJJlIypeRhpDam19Wq3g==", + "dev": true, + "license": "MIT", + "dependencies": { + "@jridgewell/trace-mapping": "^0.3.31", + "estree-walker": "^3.0.3", + "js-tokens": "^10.0.0" + } + }, "node_modules/balanced-match": { "version": "1.0.2", "resolved": "https://registry.npmjs.org/balanced-match/-/balanced-match-1.0.2.tgz", @@ -1441,6 +1646,13 @@ "url": "https://dotenvx.com" } }, + "node_modules/eastasianwidth": { + "version": "0.2.0", + "resolved": "https://registry.npmjs.org/eastasianwidth/-/eastasianwidth-0.2.0.tgz", + "integrity": "sha512-I88TYZWc9XiYHRQ4/3c5rjjfgkjhLyW2luGIheGERbNQ6OY7yTybanSpDXZa8y7VUP9YmDcYa+eyq4ca7iLqWA==", + "dev": true, + "license": "MIT" + }, "node_modules/ecdsa-sig-formatter": { "version": "1.0.11", "resolved": "https://registry.npmjs.org/ecdsa-sig-formatter/-/ecdsa-sig-formatter-1.0.11.tgz", @@ -1450,6 +1662,13 @@ "safe-buffer": "^5.0.1" } }, + "node_modules/emoji-regex": { + "version": "9.2.2", + "resolved": "https://registry.npmjs.org/emoji-regex/-/emoji-regex-9.2.2.tgz", + "integrity": "sha512-L18DaJsXSUk2+42pv8mLs5jJT2hqFkFE4j21wOmgbUqsZ2hL72NsUU785g9RXgo3s0ZNgVl42TiHp3ZtOv/Vyg==", + "dev": true, + "license": "MIT" + }, "node_modules/es-module-lexer": { "version": "1.7.0", "resolved": "https://registry.npmjs.org/es-module-lexer/-/es-module-lexer-1.7.0.tgz", @@ -1776,6 +1995,23 @@ "dev": true, "license": "ISC" }, + "node_modules/foreground-child": { + "version": "3.3.1", + "resolved": "https://registry.npmjs.org/foreground-child/-/foreground-child-3.3.1.tgz", + "integrity": "sha512-gIXjKqtFuWEgzFRJA9WCQeSJLZDjgJUOMCMzxtvFq/37KojM1BFGufqsCy0r4qSQmYLsZYMeyRqzIWOMup03sw==", + "dev": true, + "license": "ISC", + "dependencies": { + "cross-spawn": "^7.0.6", + "signal-exit": "^4.0.1" + }, + "engines": { + "node": ">=14" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/fsevents": { "version": "2.3.3", "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz", @@ -1791,6 +2027,28 @@ "node": "^8.16.0 || ^10.6.0 || >=11.0.0" } }, + "node_modules/glob": { + "version": "10.5.0", + "resolved": "https://registry.npmjs.org/glob/-/glob-10.5.0.tgz", + "integrity": "sha512-DfXN8DfhJ7NH3Oe7cFmu3NCu1wKbkReJ8TorzSAFbSKrlNaQSKfIzqYqVY8zlbs2NLBbWpRiU52GX2PbaBVNkg==", + "deprecated": "Old versions of glob are not supported, and contain widely publicized security vulnerabilities, which have been fixed in the current version. Please update. Support for old versions may be purchased (at exorbitant rates) by contacting i@izs.me", + "dev": true, + "license": "ISC", + "dependencies": { + "foreground-child": "^3.1.0", + "jackspeak": "^3.1.2", + "minimatch": "^9.0.4", + "minipass": "^7.1.2", + "package-json-from-dist": "^1.0.0", + "path-scurry": "^1.11.1" + }, + "bin": { + "glob": "dist/esm/bin.mjs" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/glob-parent": { "version": "6.0.2", "resolved": "https://registry.npmjs.org/glob-parent/-/glob-parent-6.0.2.tgz", @@ -1804,6 +2062,32 @@ "node": ">=10.13.0" } }, + "node_modules/glob/node_modules/brace-expansion": { + "version": "2.1.4", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-2.1.4.tgz", + "integrity": "sha512-hGfVzPxthbf3+2yjg/RBs60cB0FhqBS/zvdV/4wn4/BmN0bNMMHPc4V/BbFieqf1TKAGGAHnY4eSjajCl0f2Xg==", + "dev": true, + "license": "MIT", + "dependencies": { + "balanced-match": "^1.0.0" + } + }, + "node_modules/glob/node_modules/minimatch": { + "version": "9.0.9", + "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-9.0.9.tgz", + "integrity": "sha512-OBwBN9AL4dqmETlpS2zasx+vTeWclWzkblfZk7KTA5j3jeOONz/tRCnZomUyvNg83wL5Zv9Ss6HMJXAgL8R2Yg==", + "dev": true, + "license": "ISC", + "dependencies": { + "brace-expansion": "^2.0.2" + }, + "engines": { + "node": ">=16 || 14 >=14.17" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/globals": { "version": "16.5.0", "resolved": "https://registry.npmjs.org/globals/-/globals-16.5.0.tgz", @@ -1827,6 +2111,13 @@ "node": ">=8" } }, + "node_modules/html-escaper": { + "version": "2.0.2", + "resolved": "https://registry.npmjs.org/html-escaper/-/html-escaper-2.0.2.tgz", + "integrity": "sha512-H2iMtd0I4Mt5eYiapRdIDjp+XzelXQ0tFE4JS7YFwFevXXMmOp9myNrUvCg0D6ws8iqkRPBfKHgbwig1SmlLfg==", + "dev": true, + "license": "MIT" + }, "node_modules/ignore": { "version": "5.3.2", "resolved": "https://registry.npmjs.org/ignore/-/ignore-5.3.2.tgz", @@ -1874,6 +2165,16 @@ "node": ">=0.10.0" } }, + "node_modules/is-fullwidth-code-point": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/is-fullwidth-code-point/-/is-fullwidth-code-point-3.0.0.tgz", + "integrity": "sha512-zymm5+u+sCsSWyD9qNaejV3DFvhCKclKdizYaJUuHA83RLjb7nSuGnddCHGv0hk+KY7BMAlsWeK4Ueg6EV6XQg==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, "node_modules/is-glob": { "version": "4.0.3", "resolved": "https://registry.npmjs.org/is-glob/-/is-glob-4.0.3.tgz", @@ -1894,10 +2195,80 @@ "dev": true, "license": "ISC" }, + "node_modules/istanbul-lib-coverage": { + "version": "3.2.2", + "resolved": "https://registry.npmjs.org/istanbul-lib-coverage/-/istanbul-lib-coverage-3.2.2.tgz", + "integrity": "sha512-O8dpsF+r0WV/8MNRKfnmrtCWhuKjxrq2w+jpzBL5UZKTi2LeVWnWOmWRxFlesJONmc+wLAGvKQZEOanko0LFTg==", + "dev": true, + "license": "BSD-3-Clause", + "engines": { + "node": ">=8" + } + }, + "node_modules/istanbul-lib-report": { + "version": "3.0.1", + "resolved": "https://registry.npmjs.org/istanbul-lib-report/-/istanbul-lib-report-3.0.1.tgz", + "integrity": "sha512-GCfE1mtsHGOELCU8e/Z7YWzpmybrx/+dSTfLrvY8qRmaY6zXTKWn6WQIjaAFw069icm6GVMNkgu0NzI4iPZUNw==", + "dev": true, + "license": "BSD-3-Clause", + "dependencies": { + "istanbul-lib-coverage": "^3.0.0", + "make-dir": "^4.0.0", + "supports-color": "^7.1.0" + }, + "engines": { + "node": ">=10" + } + }, + "node_modules/istanbul-lib-source-maps": { + "version": "5.0.6", + "resolved": "https://registry.npmjs.org/istanbul-lib-source-maps/-/istanbul-lib-source-maps-5.0.6.tgz", + "integrity": "sha512-yg2d+Em4KizZC5niWhQaIomgf5WlL4vOOjZ5xGCmF8SnPE/mDWWXgvRExdcpCgh9lLRRa1/fSYp2ymmbJ1pI+A==", + "dev": true, + "license": "BSD-3-Clause", + "dependencies": { + "@jridgewell/trace-mapping": "^0.3.23", + "debug": "^4.1.1", + "istanbul-lib-coverage": "^3.0.0" + }, + "engines": { + "node": ">=10" + } + }, + "node_modules/istanbul-reports": { + "version": "3.2.0", + "resolved": "https://registry.npmjs.org/istanbul-reports/-/istanbul-reports-3.2.0.tgz", + "integrity": "sha512-HGYWWS/ehqTV3xN10i23tkPkpH46MLCIMFNCaaKNavAXTF1RkqxawEPtnjnGZ6XKSInBKkiOA5BKS+aZiY3AvA==", + "dev": true, + "license": "BSD-3-Clause", + "dependencies": { + "html-escaper": "^2.0.0", + "istanbul-lib-report": "^3.0.0" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/jackspeak": { + "version": "3.4.3", + "resolved": "https://registry.npmjs.org/jackspeak/-/jackspeak-3.4.3.tgz", + "integrity": "sha512-OGlZQpz2yfahA/Rd1Y8Cd9SIEsqvXkLVoSw/cgwhnhFMDbsQFeZYoJJ7bIZBS9BcamUW96asq/npPWugM+RQBw==", + "dev": true, + "license": "BlueOak-1.0.0", + "dependencies": { + "@isaacs/cliui": "^8.0.2" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + }, + "optionalDependencies": { + "@pkgjs/parseargs": "^0.11.0" + } + }, "node_modules/js-tokens": { - "version": "9.0.1", - "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-9.0.1.tgz", - "integrity": "sha512-mxa9E9ITFOt0ban3j6L5MpjwegGz6lBQmM1IJkWeBZGcMxto50+eWdjC/52xDbS2vy0k7vIMK0Fe2wfL9OQSpQ==", + "version": "10.0.0", + "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-10.0.0.tgz", + "integrity": "sha512-lM/UBzQmfJRo9ABXbPWemivdCW8V2G8FHaHdypQaIy523snUjog0W71ayWXTjiR+ixeMyVHN2XcpnTd/liPg/Q==", "dev": true, "license": "MIT" }, @@ -2084,6 +2455,13 @@ "dev": true, "license": "MIT" }, + "node_modules/lru-cache": { + "version": "10.4.3", + "resolved": "https://registry.npmjs.org/lru-cache/-/lru-cache-10.4.3.tgz", + "integrity": "sha512-JNAzZcXrCt42VGLuYz0zfAzDfAvJWW6AfYlDBQyDV5DClI2m5sAmK+OIO7s59XfsRsWHp02jAJrRadPRGTt6SQ==", + "dev": true, + "license": "ISC" + }, "node_modules/magic-string": { "version": "0.30.21", "resolved": "https://registry.npmjs.org/magic-string/-/magic-string-0.30.21.tgz", @@ -2094,6 +2472,34 @@ "@jridgewell/sourcemap-codec": "^1.5.5" } }, + "node_modules/magicast": { + "version": "0.3.5", + "resolved": "https://registry.npmjs.org/magicast/-/magicast-0.3.5.tgz", + "integrity": "sha512-L0WhttDl+2BOsybvEOLK7fW3UA0OQ0IQ2d6Zl2x/a6vVRs3bAY0ECOSHHeL5jD+SbOpOCUEi0y1DgHEn9Qn1AQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "@babel/parser": "^7.25.4", + "@babel/types": "^7.25.4", + "source-map-js": "^1.2.0" + } + }, + "node_modules/make-dir": { + "version": "4.0.0", + "resolved": "https://registry.npmjs.org/make-dir/-/make-dir-4.0.0.tgz", + "integrity": "sha512-hXdUTZYIVOt1Ex//jAQi+wTZZpUpwBj/0QsOzqegb3rGMMeJiSEu5xLHnYfBrRV4RH2+OCSOO95Is/7x1WJ4bw==", + "dev": true, + "license": "MIT", + "dependencies": { + "semver": "^7.5.3" + }, + "engines": { + "node": ">=10" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, "node_modules/minimatch": { "version": "3.1.5", "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-3.1.5.tgz", @@ -2107,6 +2513,16 @@ "node": "*" } }, + "node_modules/minipass": { + "version": "7.1.3", + "resolved": "https://registry.npmjs.org/minipass/-/minipass-7.1.3.tgz", + "integrity": "sha512-tEBHqDnIoM/1rXME1zgka9g6Q2lcoCkxHLuc7ODJ5BxbP5d4c2Z5cGgtXAku59200Cx7diuHTOYfSBD8n6mm8A==", + "dev": true, + "license": "BlueOak-1.0.0", + "engines": { + "node": ">=16 || 14 >=14.17" + } + }, "node_modules/ms": { "version": "2.1.3", "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", @@ -2189,6 +2605,13 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/package-json-from-dist": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/package-json-from-dist/-/package-json-from-dist-1.0.1.tgz", + "integrity": "sha512-UEZIS3/by4OC8vL3P2dTXRETpebLI2NiI5vIrjaD/5UtrkFX/tNbwjTSRAGC/+7CAo2pIcBaRgWmcBBHcsaCIw==", + "dev": true, + "license": "BlueOak-1.0.0" + }, "node_modules/parent-module": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/parent-module/-/parent-module-1.0.1.tgz", @@ -2222,6 +2645,23 @@ "node": ">=8" } }, + "node_modules/path-scurry": { + "version": "1.11.1", + "resolved": "https://registry.npmjs.org/path-scurry/-/path-scurry-1.11.1.tgz", + "integrity": "sha512-Xa4Nw17FS9ApQFJ9umLiJS4orGjm7ZzwUrwamcGQuHSzDyth9boKDaycYdDcZDuqYATXw4HFXgaqWTctW/v1HA==", + "dev": true, + "license": "BlueOak-1.0.0", + "dependencies": { + "lru-cache": "^10.2.0", + "minipass": "^5.0.0 || ^6.0.2 || ^7.0.0" + }, + "engines": { + "node": ">=16 || 14 >=14.18" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/pathe": { "version": "2.0.3", "resolved": "https://registry.npmjs.org/pathe/-/pathe-2.0.3.tgz", @@ -2425,6 +2865,19 @@ "dev": true, "license": "ISC" }, + "node_modules/signal-exit": { + "version": "4.1.0", + "resolved": "https://registry.npmjs.org/signal-exit/-/signal-exit-4.1.0.tgz", + "integrity": "sha512-bzyZ1e88w9O1iNJbKnOlvYTrWPDl46O1bG0D3XInv+9tkPrxrN8jUUTiFlDkkmKWgn1M6CfIA13SuGqOa9Korw==", + "dev": true, + "license": "ISC", + "engines": { + "node": ">=14" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/source-map-js": { "version": "1.2.1", "resolved": "https://registry.npmjs.org/source-map-js/-/source-map-js-1.2.1.tgz", @@ -2449,6 +2902,110 @@ "dev": true, "license": "MIT" }, + "node_modules/string-width": { + "version": "5.1.2", + "resolved": "https://registry.npmjs.org/string-width/-/string-width-5.1.2.tgz", + "integrity": "sha512-HnLOCR3vjcY8beoNLtcjZ5/nxn2afmME6lhrDrebokqMap+XbeW8n9TXpPDOqdGK5qcI3oT0GKTW6wC7EMiVqA==", + "dev": true, + "license": "MIT", + "dependencies": { + "eastasianwidth": "^0.2.0", + "emoji-regex": "^9.2.2", + "strip-ansi": "^7.0.1" + }, + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, + "node_modules/string-width-cjs": { + "name": "string-width", + "version": "4.2.3", + "resolved": "https://registry.npmjs.org/string-width/-/string-width-4.2.3.tgz", + "integrity": "sha512-wKyQRQpjJ0sIp62ErSZdGsjMJWsap5oRNihHhu6G7JVO/9jIB6UyevL+tXuOqrng8j/cxKTWyWUwvSTriiZz/g==", + "dev": true, + "license": "MIT", + "dependencies": { + "emoji-regex": "^8.0.0", + "is-fullwidth-code-point": "^3.0.0", + "strip-ansi": "^6.0.1" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/string-width-cjs/node_modules/ansi-regex": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/ansi-regex/-/ansi-regex-5.0.1.tgz", + "integrity": "sha512-quJQXlTSUGL2LH9SUXo8VwsY4soanhgo6LNSm84E1LBcE8s3O0wpdiRzyR9z/ZZJMlMWv37qOOb9pdJlMUEKFQ==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, + "node_modules/string-width-cjs/node_modules/emoji-regex": { + "version": "8.0.0", + "resolved": "https://registry.npmjs.org/emoji-regex/-/emoji-regex-8.0.0.tgz", + "integrity": "sha512-MSjYzcWNOA0ewAHpz0MxpYFvwg6yjy1NG3xteoqz644VCo/RPgnr1/GGt+ic3iJTzQ8Eu3TdM14SawnVUmGE6A==", + "dev": true, + "license": "MIT" + }, + "node_modules/string-width-cjs/node_modules/strip-ansi": { + "version": "6.0.1", + "resolved": "https://registry.npmjs.org/strip-ansi/-/strip-ansi-6.0.1.tgz", + "integrity": "sha512-Y38VPSHcqkFrCpFnQ9vuSXmquuv5oXOKpGeT6aGrr3o3Gc9AlVa6JBfUSOCnbxGGZF+/0ooI7KrPuUSztUdU5A==", + "dev": true, + "license": "MIT", + "dependencies": { + "ansi-regex": "^5.0.1" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/strip-ansi": { + "version": "7.2.0", + "resolved": "https://registry.npmjs.org/strip-ansi/-/strip-ansi-7.2.0.tgz", + "integrity": "sha512-yDPMNjp4WyfYBkHnjIRLfca1i6KMyGCtsVgoKe/z1+6vukgaENdgGBZt+ZmKPc4gavvEZ5OgHfHdrazhgNyG7w==", + "dev": true, + "license": "MIT", + "dependencies": { + "ansi-regex": "^6.2.2" + }, + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/strip-ansi?sponsor=1" + } + }, + "node_modules/strip-ansi-cjs": { + "name": "strip-ansi", + "version": "6.0.1", + "resolved": "https://registry.npmjs.org/strip-ansi/-/strip-ansi-6.0.1.tgz", + "integrity": "sha512-Y38VPSHcqkFrCpFnQ9vuSXmquuv5oXOKpGeT6aGrr3o3Gc9AlVa6JBfUSOCnbxGGZF+/0ooI7KrPuUSztUdU5A==", + "dev": true, + "license": "MIT", + "dependencies": { + "ansi-regex": "^5.0.1" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/strip-ansi-cjs/node_modules/ansi-regex": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/ansi-regex/-/ansi-regex-5.0.1.tgz", + "integrity": "sha512-quJQXlTSUGL2LH9SUXo8VwsY4soanhgo6LNSm84E1LBcE8s3O0wpdiRzyR9z/ZZJMlMWv37qOOb9pdJlMUEKFQ==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, "node_modules/strip-json-comments": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/strip-json-comments/-/strip-json-comments-3.1.1.tgz", @@ -2475,6 +3032,13 @@ "url": "https://github.com/sponsors/antfu" } }, + "node_modules/strip-literal/node_modules/js-tokens": { + "version": "9.0.1", + "resolved": "https://registry.npmjs.org/js-tokens/-/js-tokens-9.0.1.tgz", + "integrity": "sha512-mxa9E9ITFOt0ban3j6L5MpjwegGz6lBQmM1IJkWeBZGcMxto50+eWdjC/52xDbS2vy0k7vIMK0Fe2wfL9OQSpQ==", + "dev": true, + "license": "MIT" + }, "node_modules/supports-color": { "version": "7.2.0", "resolved": "https://registry.npmjs.org/supports-color/-/supports-color-7.2.0.tgz", @@ -2488,6 +3052,60 @@ "node": ">=8" } }, + "node_modules/test-exclude": { + "version": "7.0.2", + "resolved": "https://registry.npmjs.org/test-exclude/-/test-exclude-7.0.2.tgz", + "integrity": "sha512-u9E6A+ZDYdp7a4WnarkXPZOx8Ilz46+kby6p1yZ8zsGTz9gYa6FIS7lj2oezzNKmtdyyJNNmmXDppga5GB7kSw==", + "dev": true, + "license": "ISC", + "dependencies": { + "@istanbuljs/schema": "^0.1.2", + "glob": "^10.4.1", + "minimatch": "^10.2.2" + }, + "engines": { + "node": ">=18" + } + }, + "node_modules/test-exclude/node_modules/balanced-match": { + "version": "4.0.4", + "resolved": "https://registry.npmjs.org/balanced-match/-/balanced-match-4.0.4.tgz", + "integrity": "sha512-BLrgEcRTwX2o6gGxGOCNyMvGSp35YofuYzw9h1IMTRmKqttAZZVU67bdb9Pr2vUHA8+j3i2tJfjO6C6+4myGTA==", + "dev": true, + "license": "MIT", + "engines": { + "node": "18 || 20 || >=22" + } + }, + "node_modules/test-exclude/node_modules/brace-expansion": { + "version": "5.0.9", + "resolved": "https://registry.npmjs.org/brace-expansion/-/brace-expansion-5.0.9.tgz", + "integrity": "sha512-ScQ4IuvIEF1TMlP7Zt+vjJ//9zlPb2SDcxWxM3bk8s6t6GGdJ7KO1dCcTidOPJKePW30LE/2cT7wCyPho9/Wxg==", + "dev": true, + "license": "MIT", + "dependencies": { + "balanced-match": "^4.0.2" + }, + "engines": { + "node": "20 || >=22" + } + }, + "node_modules/test-exclude/node_modules/minimatch": { + "version": "10.2.6", + "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-10.2.6.tgz", + "integrity": "sha512-vpLQEs+VLCr1nU0BXS07maYoFwlDAH0gngQuuttxIwutDFEMHq2blX+8vpgxDdK3J1PwjCJiep77OitTZ4Ll1A==", + "dev": true, + "license": "BlueOak-1.0.0", + "dependencies": { + "brace-expansion": "^5.0.8" + }, + "engines": { + "node": "18 || 20 || >=22" + }, + "funding": { + "url": "https://github.com/sponsors/isaacs" + } + }, "node_modules/tinybench": { "version": "2.9.0", "resolved": "https://registry.npmjs.org/tinybench/-/tinybench-2.9.0.tgz", @@ -2684,20 +3302,20 @@ } }, "node_modules/vitest": { - "version": "3.2.6", - "resolved": "https://registry.npmjs.org/vitest/-/vitest-3.2.6.tgz", - "integrity": "sha512-xejya+bT/j/+R/AGa1XOfRxLmNUlLtlwjRsFUILF+xHfzElmGcmFydy2gqqIrd62ptIEfwVMofd19uNWD9L7Nw==", + "version": "3.2.7", + "resolved": "https://registry.npmjs.org/vitest/-/vitest-3.2.7.tgz", + "integrity": "sha512-KrxIJ62Fd89gfysR4WotlgZABiz2dqFPgqGzX7s+CwsqLFomRH7777ZcrOD6+WVAh7khPQP41A+BKbpcJFrdEg==", "dev": true, "license": "MIT", "dependencies": { "@types/chai": "^5.2.2", - "@vitest/expect": "3.2.6", - "@vitest/mocker": "3.2.6", - "@vitest/pretty-format": "^3.2.6", - "@vitest/runner": "3.2.6", - "@vitest/snapshot": "3.2.6", - "@vitest/spy": "3.2.6", - "@vitest/utils": "3.2.6", + "@vitest/expect": "3.2.7", + "@vitest/mocker": "3.2.7", + "@vitest/pretty-format": "^3.2.7", + "@vitest/runner": "3.2.7", + "@vitest/snapshot": "3.2.7", + "@vitest/spy": "3.2.7", + "@vitest/utils": "3.2.7", "chai": "^5.2.0", "debug": "^4.4.1", "expect-type": "^1.2.1", @@ -2727,8 +3345,8 @@ "@edge-runtime/vm": "*", "@types/debug": "^4.1.12", "@types/node": "^18.0.0 || ^20.0.0 || >=22.0.0", - "@vitest/browser": "3.2.6", - "@vitest/ui": "3.2.6", + "@vitest/browser": "3.2.7", + "@vitest/ui": "3.2.7", "happy-dom": "*", "jsdom": "*" }, @@ -2799,6 +3417,101 @@ "node": ">=0.10.0" } }, + "node_modules/wrap-ansi": { + "version": "8.1.0", + "resolved": "https://registry.npmjs.org/wrap-ansi/-/wrap-ansi-8.1.0.tgz", + "integrity": "sha512-si7QWI6zUMq56bESFvagtmzMdGOtoxfR+Sez11Mobfc7tm+VkUckk9bW2UeffTGVUbOksxmSw0AA2gs8g71NCQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "ansi-styles": "^6.1.0", + "string-width": "^5.0.1", + "strip-ansi": "^7.0.1" + }, + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/wrap-ansi?sponsor=1" + } + }, + "node_modules/wrap-ansi-cjs": { + "name": "wrap-ansi", + "version": "7.0.0", + "resolved": "https://registry.npmjs.org/wrap-ansi/-/wrap-ansi-7.0.0.tgz", + "integrity": "sha512-YVGIj2kamLSTxw6NsZjoBxfSwsn0ycdesmc4p+Q21c5zPuZ1pl+NfxVdxPtdHvmNVOQ6XSYG4AUtyt/Fi7D16Q==", + "dev": true, + "license": "MIT", + "dependencies": { + "ansi-styles": "^4.0.0", + "string-width": "^4.1.0", + "strip-ansi": "^6.0.0" + }, + "engines": { + "node": ">=10" + }, + "funding": { + "url": "https://github.com/chalk/wrap-ansi?sponsor=1" + } + }, + "node_modules/wrap-ansi-cjs/node_modules/ansi-regex": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/ansi-regex/-/ansi-regex-5.0.1.tgz", + "integrity": "sha512-quJQXlTSUGL2LH9SUXo8VwsY4soanhgo6LNSm84E1LBcE8s3O0wpdiRzyR9z/ZZJMlMWv37qOOb9pdJlMUEKFQ==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=8" + } + }, + "node_modules/wrap-ansi-cjs/node_modules/emoji-regex": { + "version": "8.0.0", + "resolved": "https://registry.npmjs.org/emoji-regex/-/emoji-regex-8.0.0.tgz", + "integrity": "sha512-MSjYzcWNOA0ewAHpz0MxpYFvwg6yjy1NG3xteoqz644VCo/RPgnr1/GGt+ic3iJTzQ8Eu3TdM14SawnVUmGE6A==", + "dev": true, + "license": "MIT" + }, + "node_modules/wrap-ansi-cjs/node_modules/string-width": { + "version": "4.2.3", + "resolved": "https://registry.npmjs.org/string-width/-/string-width-4.2.3.tgz", + "integrity": "sha512-wKyQRQpjJ0sIp62ErSZdGsjMJWsap5oRNihHhu6G7JVO/9jIB6UyevL+tXuOqrng8j/cxKTWyWUwvSTriiZz/g==", + "dev": true, + "license": "MIT", + "dependencies": { + "emoji-regex": "^8.0.0", + "is-fullwidth-code-point": "^3.0.0", + "strip-ansi": "^6.0.1" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/wrap-ansi-cjs/node_modules/strip-ansi": { + "version": "6.0.1", + "resolved": "https://registry.npmjs.org/strip-ansi/-/strip-ansi-6.0.1.tgz", + "integrity": "sha512-Y38VPSHcqkFrCpFnQ9vuSXmquuv5oXOKpGeT6aGrr3o3Gc9AlVa6JBfUSOCnbxGGZF+/0ooI7KrPuUSztUdU5A==", + "dev": true, + "license": "MIT", + "dependencies": { + "ansi-regex": "^5.0.1" + }, + "engines": { + "node": ">=8" + } + }, + "node_modules/wrap-ansi/node_modules/ansi-styles": { + "version": "6.2.3", + "resolved": "https://registry.npmjs.org/ansi-styles/-/ansi-styles-6.2.3.tgz", + "integrity": "sha512-4Dj6M28JB+oAH8kFkTLUo+a2jwOFkuqb3yucU0CANcRRUbxS0cP0nZYCGjcc3BNXwRIsUVmDGgzawme7zvJHvg==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">=12" + }, + "funding": { + "url": "https://github.com/chalk/ansi-styles?sponsor=1" + } + }, "node_modules/ws": { "version": "8.21.0", "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", From 5da8a3a23d64439b882e03dc6adfb5ee695b8a62 Mon Sep 17 00:00:00 2001 From: levibliz Date: Tue, 18 Aug 2026 00:35:52 +0100 Subject: [PATCH 6/6] feat: session resumption and connection migration Add encrypted session state persistence so clients can resume sessions across gateway restarts or reconnections without losing room membership. SessionManager handles AES-256-GCM encrypted storage with key rotation, debounced saves, and optional Redis or in-memory backends. Closes #258, Closes #175, Closes #251, Closes #254 --- src/index.js | 7 +- src/server.js | 103 ++++- src/session-manager.js | 387 +++++++++++++++++++ tests/session-manager.test.js | 359 +++++++++++++++++ tests/session-resumption-integration.test.js | 276 +++++++++++++ 5 files changed, 1121 insertions(+), 11 deletions(-) create mode 100644 src/session-manager.js create mode 100644 tests/session-manager.test.js create mode 100644 tests/session-resumption-integration.test.js diff --git a/src/index.js b/src/index.js index fded7a2..9d31b40 100644 --- a/src/index.js +++ b/src/index.js @@ -34,8 +34,9 @@ if (isNaN(config.maxPayloadBytes) || config.maxPayloadBytes < 1) { let wss; let httpServer; let markShuttingDown; +let sessionManager; try { - ({ wss, httpServer, markShuttingDown } = createServer(config)); + ({ wss, httpServer, markShuttingDown, sessionManager } = createServer(config)); } catch (err) { logger.error("Failed to start server", { error: err.message }); process.exit(1); @@ -55,8 +56,12 @@ logger.info("Gateway started", config); */ export function shutdown(server, signal) { logger.info("Shutting down", { signal }); + if (sessionManager) { + sessionManager.flushPending().catch(() => {}); + } server.close(() => { logger.info("Server closed"); + if (sessionManager) sessionManager.destroy(); process.exit(0); }); setTimeout(() => { diff --git a/src/server.js b/src/server.js index af4b704..6f7cbea 100644 --- a/src/server.js +++ b/src/server.js @@ -7,14 +7,7 @@ import { verifyConnection } from "./auth.js"; import { logger } from "./logger.js"; import { createConnRateLimiter } from "./conn-rate-limiter.js"; import { createRateLimiter } from "./rate-limiter.js"; - -function safeSend(ws, data) { - try { - ws.send(typeof data === "string" ? data : JSON.stringify(data)); - } catch { - // Silently ignore send errors (connection may have closed) - } -} +import { SessionManager } from "./session-manager.js"; function safeSend(ws, data) { try { @@ -31,11 +24,14 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit const ipConnectionCount = new Map(); const MAX_CONNS_PER_IP = maxConnectionsPerIp ?? (Number(process.env.MAX_CONNECTIONS_PER_IP) || 10); + const sessionManager = new SessionManager(); + const metrics = { messages: { location_update: 0, join_room: 0, leave_room: 0 }, authFailures: 0, rateLimitRejections: { connection: 0 }, eventLoopLagMs: 0, + sessionResumption: { success: 0, decrypt_failed: 0, expired: 0, mismatch: 0, new_session: 0 }, }; let isShuttingDown = false; @@ -88,6 +84,12 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit `gateway_rate_limit_rejections_total{kind="connection"} ${metrics.rateLimitRejections.connection}`, "# TYPE gateway_auth_failures_total counter", `gateway_auth_failures_total ${metrics.authFailures}`, + "# TYPE session_resumption_total counter", + `session_resumption_total{result="success"} ${metrics.sessionResumption.success}`, + `session_resumption_total{result="decrypt_failed"} ${metrics.sessionResumption.decrypt_failed}`, + `session_resumption_total{result="expired"} ${metrics.sessionResumption.expired}`, + `session_resumption_total{result="mismatch"} ${metrics.sessionResumption.mismatch}`, + `session_resumption_total{result="new_session"} ${metrics.sessionResumption.new_session}`, "# TYPE gateway_heap_used_bytes gauge", `gateway_heap_used_bytes ${mem.heapUsed}`, "# TYPE gateway_event_loop_lag_ms gauge", @@ -159,7 +161,40 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit } const actualClientId = authResult.clientId ?? clientId; - logger.info("Client connected", { clientId: actualClientId, ip }); + let sessionResumed = false; + let restoredRooms = []; + + const sessionId = url.searchParams.get("session_id"); + if (sessionId) { + sessionManager.load(sessionId).then((restored) => { + if (restored && restored.clientId === actualClientId) { + sessionResumed = true; + restoredRooms = restored.rooms || []; + for (const room of restoredRooms) { + rooms.join(actualClientId, room.roomId, ws); + } + metrics.sessionResumption.success++; + logger.info("Session resumed", { clientId: actualClientId, sessionId, rooms: restoredRooms.map((r) => r.roomId) }); + safeSend(ws, { + type: "session_resumed", + payload: { + rooms: restoredRooms.map((r) => r.roomId), + currentSeqPerRoom: restoredRooms.map((r) => ({ roomId: r.roomId, seq: r.highestReceivedSeq })), + }, + }); + } else if (restored && restored.clientId !== actualClientId) { + metrics.sessionResumption.mismatch++; + logger.warn("Session identity mismatch", { clientId: actualClientId, sessionId }); + } else { + metrics.sessionResumption.new_session++; + logger.info("No valid session found", { clientId: actualClientId, sessionId }); + } + }).catch(() => { + metrics.sessionResumption.decrypt_failed++; + }); + } + + logger.info("Client connected", { clientId: actualClientId, ip, sessionResumed }); ws.on("pong", heartbeat); @@ -185,6 +220,22 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit metrics.messages.join_room++; logger.info("Client joined room", { clientId: actualClientId, roomId: msg.roomId }); safeSend(ws, { type: "room_joined", payload: { roomId: msg.roomId } }); + + const currentRooms = rooms.getClientRooms(actualClientId); + const roomStates = Array.from(currentRooms).map((roomId) => ({ + roomId, + highestAckedSeq: 0, + highestReceivedSeq: 0, + geofenceInsideSet: [], + })); + sessionManager.debouncedSave(actualClientId, { + clientId: actualClientId, + protocolVersion: 3, + authIdentity: authResult, + rooms: roomStates, + rateLimitState: { messageWindow: [], connectionWindow: [] }, + metadata: { ip, userAgent: req.headers?.["user-agent"] ?? "", connectedAt: Date.now(), lastActivityAt: Date.now() }, + }); break; } case "leave_room": { @@ -192,6 +243,22 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit metrics.messages.leave_room++; logger.info("Client left room", { clientId: actualClientId, roomId: msg.roomId }); safeSend(ws, { type: "room_left", payload: { roomId: msg.roomId } }); + + const currentRooms = rooms.getClientRooms(actualClientId); + const roomStates = Array.from(currentRooms).map((roomId) => ({ + roomId, + highestAckedSeq: 0, + highestReceivedSeq: 0, + geofenceInsideSet: [], + })); + sessionManager.debouncedSave(actualClientId, { + clientId: actualClientId, + protocolVersion: 3, + authIdentity: authResult, + rooms: roomStates, + rateLimitState: { messageWindow: [], connectionWindow: [] }, + metadata: { ip, userAgent: req.headers?.["user-agent"] ?? "", connectedAt: Date.now(), lastActivityAt: Date.now() }, + }); break; } case "location_update": { @@ -209,6 +276,22 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit }); ws.on("close", (code, reason) => { + const currentRooms = rooms.getClientRooms(actualClientId); + const roomStates = Array.from(currentRooms).map((roomId) => ({ + roomId, + highestAckedSeq: 0, + highestReceivedSeq: 0, + geofenceInsideSet: [], + })); + sessionManager.save(actualClientId, { + clientId: actualClientId, + protocolVersion: 3, + authIdentity: authResult, + rooms: roomStates, + rateLimitState: { messageWindow: [], connectionWindow: [] }, + metadata: { ip, userAgent: req.headers?.["user-agent"] ?? "", connectedAt: Date.now(), lastActivityAt: Date.now() }, + }).catch(() => {}); + rooms.disconnect(actualClientId); rateLimiter.remove(actualClientId); const trackedIp = ws._trackedIp; @@ -248,5 +331,5 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit httpServer.close(); }); - return { wss, rooms }; + return { wss, httpServer, markShuttingDown, rooms, sessionManager }; } diff --git a/src/session-manager.js b/src/session-manager.js new file mode 100644 index 0000000..f530628 --- /dev/null +++ b/src/session-manager.js @@ -0,0 +1,387 @@ +import crypto from "node:crypto"; +import zlib from "node:zlib"; + +const DEFAULT_TTL_MS = 3600000; +const DEBOUNCE_MS = 500; +const MAX_BLOB_SIZE = 16384; + +/** + * @typedef {Object} SessionRoom + * @property {string} roomId + * @property {number} highestAckedSeq + * @property {number} highestReceivedSeq + * @property {string[]} geofenceInsideSet + */ + +/** + * @typedef {Object} RateLimitState + * @property {number[]} messageWindow + * @property {number[]} connectionWindow + */ + +/** + * @typedef {Object} SessionMetadata + * @property {string} ip + * @property {string} userAgent + * @property {number} connectedAt + * @property {number} lastActivityAt + */ + +/** + * @typedef {Object} SessionState + * @property {string} clientId + * @property {number} protocolVersion + * @property {Object} authIdentity + * @property {SessionRoom[]} rooms + * @property {RateLimitState} rateLimitState + * @property {SessionMetadata} metadata + */ + +/** + * @typedef {Object} SessionManagerOptions + * @property {Object} [redis] - Optional Redis client with get/set/del methods + * @property {string|Object} encryptionKey - Base64 encoded key or { keyId: base64key } map + * @property {number} [ttlMs] - Session TTL in milliseconds + * @property {string} [keyId] - Current key identifier for encryption + * @property {number} [debounceMs] - Debounce interval for saves + */ + +/** + * Derives an AES-256 key from a master key using HKDF. + * + * @param {Buffer} masterKey + * @param {string} info + * @returns {Buffer} + */ +function deriveKey(masterKey, info) { + return crypto.hkdfSync("sha256", masterKey, Buffer.alloc(0), info, 32); +} + +/** + * Resolves the raw key bytes for a given key ID. + * + * @param {string|Object} encryptionKey + * @param {string} keyId + * @returns {Buffer|null} + */ +function resolveKey(encryptionKey, keyId) { + if (typeof encryptionKey === "string") { + return Buffer.from(encryptionKey, "base64"); + } + if (typeof encryptionKey === "object" && encryptionKey[keyId]) { + return Buffer.from(encryptionKey[keyId], "base64"); + } + return null; +} + +/** + * Encrypts a session state blob using AES-256-GCM. + * + * @param {Buffer} plaintext + * @param {Buffer} key + * @param {string} keyId + * @returns {string} Encrypted blob in format: keyId.base64iv.base64ciphertext.base64tag + */ +function encryptBlob(plaintext, key, keyId) { + const derivedKey = deriveKey(key, `session-key-${keyId}`); + const iv = crypto.randomBytes(12); + const cipher = crypto.createCipheriv("aes-256-gcm", derivedKey, iv); + const ciphertext = Buffer.concat([cipher.update(plaintext), cipher.final()]); + const tag = cipher.getAuthTag(); + return `${keyId}.${iv.toString("base64")}.${ciphertext.toString("base64")}.${tag.toString("base64")}`; +} + +/** + * Decrypts a session state blob. + * + * @param {string} blob + * @param {string|Object} encryptionKey + * @returns {Buffer|null} Decrypted plaintext or null if decryption fails + */ +function decryptBlob(blob, encryptionKey) { + const parts = blob.split("."); + if (parts.length !== 4) return null; + + const [keyId, ivB64, ciphertextB64, tagB64] = parts; + const key = resolveKey(encryptionKey, keyId); + if (!key) return null; + + try { + const derivedKey = deriveKey(key, `session-key-${keyId}`); + const iv = Buffer.from(ivB64, "base64"); + const ciphertext = Buffer.from(ciphertextB64, "base64"); + const tag = Buffer.from(tagB64, "base64"); + const decipher = crypto.createDecipheriv("aes-256-gcm", derivedKey, iv); + decipher.setAuthTag(tag); + return Buffer.concat([decipher.update(ciphertext), decipher.final()]); + } catch { + return null; + } +} + +/** + * SessionManager handles encrypted session state persistence for WebSocket + * connection migration and resumption. + * + * Supports both Redis-backed distributed storage and in-memory fallback + * for single-instance mode. + * + * @example + * const sm = new SessionManager({ encryptionKey: "base64key..." }); + * await sm.save("client-1", { clientId: "client-1", rooms: [] }); + * const state = await sm.load("client-1"); + */ +export class SessionManager { + /** @type {Object|null} */ + #redis; + + /** @type {string|Object} */ + #encryptionKey; + + /** @type {number} */ + #ttlMs; + + /** @type {string} */ + #keyId; + + /** @type {number} */ + #debounceMs; + + /** @type {Map} */ + #timers; + + /** @type {Map} */ + #pendingStates; + + /** @type {Map} */ + #localCache; + + /** @type {NodeJS.Timeout|null} */ + #cleanupInterval; + + /** + * @param {SessionManagerOptions} options + */ + constructor({ redis, encryptionKey, ttlMs, keyId, debounceMs } = {}) { + this.#redis = redis ?? null; + this.#encryptionKey = encryptionKey ?? process.env.SESSION_ENCRYPTION_KEY ?? ""; + this.#ttlMs = ttlMs ?? DEFAULT_TTL_MS; + this.#keyId = keyId ?? "v1"; + this.#debounceMs = debounceMs ?? DEBOUNCE_MS; + this.#timers = new Map(); + this.#pendingStates = new Map(); + this.#localCache = new Map(); + this.#cleanupInterval = null; + + if (!this.#redis) { + this.#cleanupInterval = setInterval(() => { + this.#evictExpired(); + }, Math.min(this.#ttlMs, 60000)); + if (this.#cleanupInterval.unref) { + this.#cleanupInterval.unref(); + } + } + } + + /** + * Removes expired entries from the local in-memory cache. + * @private + */ + #evictExpired() { + const now = Date.now(); + for (const [clientId, entry] of this.#localCache) { + if (entry.expiresAt <= now) { + this.#localCache.delete(clientId); + } + } + } + + /** + * Compresses and encrypts a session state, then stores it. + * + * @param {string} clientId + * @param {SessionState} state + * @returns {Promise} + */ + async save(clientId, state) { + const plaintext = Buffer.from(JSON.stringify(state), "utf8"); + const compressed = zlib.deflateSync(plaintext, { level: 6 }); + + if (compressed.length > MAX_BLOB_SIZE) { + throw new Error(`Session blob exceeds ${MAX_BLOB_SIZE} bytes after compression`); + } + + const blob = encryptBlob(compressed, this.#resolveEncryptionKey(), this.#keyId); + const expiresAt = Date.now() + this.#ttlMs; + const ttlSeconds = Math.ceil(this.#ttlMs / 1000); + + if (this.#redis) { + await this.#redis.set(`session:${clientId}`, blob, "EX", ttlSeconds); + } else { + this.#localCache.set(clientId, { state, expiresAt }); + } + } + + /** + * Loads and decrypts a session state by its client ID. + * + * @param {string} sessionId - The client ID used as session identifier + * @returns {Promise} + */ + async load(sessionId) { + if (this.#redis) { + const blob = await this.#redis.get(`session:${sessionId}`); + if (!blob) return null; + + const decrypted = decryptBlob(blob, this.#encryptionKey); + if (!decrypted) return null; + + try { + const decompressed = zlib.inflateSync(decrypted); + const json = JSON.parse(decompressed.toString("utf8")); + if (json.clientId !== sessionId) return null; + return json; + } catch { + return null; + } + } + + const entry = this.#localCache.get(sessionId); + if (!entry) return null; + if (entry.expiresAt <= Date.now()) { + this.#localCache.delete(sessionId); + return null; + } + return entry.state; + } + + /** + * Deletes a session from storage. + * + * @param {string} clientId + * @returns {Promise} + */ + async delete(clientId) { + this.#clearDebounce(clientId); + if (this.#redis) { + await this.#redis.del(`session:${clientId}`); + } else { + this.#localCache.delete(clientId); + } + } + + /** + * Saves immediately without debouncing. Used for graceful shutdown. + * + * @param {string} clientId + * @param {SessionState} state + * @returns {Promise} + */ + async saveImmediate(clientId, state) { + this.#clearDebounce(clientId); + await this.save(clientId, state); + } + + /** + * Debounced save that coalesces rapid state changes. + * + * @param {string} clientId + * @param {SessionState} state + * @returns {void} + */ + debouncedSave(clientId, state) { + const existing = this.#timers.get(clientId); + if (existing) clearTimeout(existing); + this.#pendingStates.set(clientId, state); + this.#timers.set(clientId, setTimeout(() => { + this.#timers.delete(clientId); + const pending = this.#pendingStates.get(clientId); + this.#pendingStates.delete(clientId); + if (pending) { + this.save(clientId, pending).catch(() => {}); + } + }, this.#debounceMs)); + } + + /** + * Clears a pending debounce timer for a client. + * + * @param {string} clientId + * @private + */ + #clearDebounce(clientId) { + const timer = this.#timers.get(clientId); + if (timer) { + clearTimeout(timer); + this.#timers.delete(clientId); + } + this.#pendingStates.delete(clientId); + } + + /** + * Returns the number of pending debounced saves. + * @returns {number} + */ + get pendingSaves() { + return this.#timers.size; + } + + /** + * Returns the number of sessions in local cache (single-instance mode). + * @returns {number} + */ + get cachedSessions() { + return this.#localCache.size; + } + + /** + * Flushes all pending debounced saves immediately. + * @returns {Promise} + */ + async flushPending() { + const entries = []; + for (const [clientId, timer] of this.#timers) { + clearTimeout(timer); + const pending = this.#pendingStates.get(clientId); + entries.push({ clientId, state: pending }); + } + this.#timers.clear(); + this.#pendingStates.clear(); + for (const { clientId, state } of entries) { + if (state) { + await this.save(clientId, state).catch(() => {}); + } + } + } + + /** + * Cleans up resources. Clears timers and intervals. + * @returns {void} + */ + destroy() { + for (const timer of this.#timers.values()) { + clearTimeout(timer); + } + this.#timers.clear(); + if (this.#cleanupInterval) { + clearInterval(this.#cleanupInterval); + this.#cleanupInterval = null; + } + } + + /** + * Resolves the encryption key to a Buffer. + * + * @returns {Buffer} + * @private + */ + #resolveEncryptionKey() { + if (typeof this.#encryptionKey === "string" && this.#encryptionKey) { + return Buffer.from(this.#encryptionKey, "base64"); + } + if (typeof this.#encryptionKey === "object" && this.#encryptionKey[this.#keyId]) { + return Buffer.from(this.#encryptionKey[this.#keyId], "base64"); + } + return crypto.randomBytes(32); + } +} diff --git a/tests/session-manager.test.js b/tests/session-manager.test.js new file mode 100644 index 0000000..786629a --- /dev/null +++ b/tests/session-manager.test.js @@ -0,0 +1,359 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import crypto from "node:crypto"; +import { SessionManager } from "../src/session-manager.js"; + +/** + * Returns a valid session state object. + * @param {Partial} [overrides] + * @returns {import("../src/session-manager.js").SessionState} + */ +function makeState(overrides = {}) { + return { + clientId: "client-001", + protocolVersion: 3, + authIdentity: { sub: "device-001", iss: "fleet-auth" }, + rooms: [ + { roomId: "fleet-alpha", highestAckedSeq: 42, highestReceivedSeq: 45, geofenceInsideSet: ["fence-1"] }, + ], + rateLimitState: { messageWindow: [1000, 2000], connectionWindow: [500] }, + metadata: { ip: "10.0.0.1", userAgent: "FleetApp/2.3", connectedAt: 1000, lastActivityAt: 2000 }, + ...overrides, + }; +} + +function makeTestKey() { + return crypto.randomBytes(32).toString("base64"); +} + +describe("SessionManager", () => { + let sm; + const encryptionKey = makeTestKey(); + + beforeEach(() => { + sm = new SessionManager({ encryptionKey, ttlMs: 5000, debounceMs: 10 }); + }); + + afterEach(() => { + sm.destroy(); + }); + + describe("constructor", () => { + it("creates an instance with default options", () => { + const s = new SessionManager(); + expect(s).toBeDefined(); + s.destroy(); + }); + + it("creates an instance with custom options", () => { + const s = new SessionManager({ encryptionKey, ttlMs: 10000, keyId: "v2", debounceMs: 100 }); + expect(s).toBeDefined(); + s.destroy(); + }); + }); + + describe("encryption/decryption", () => { + it("save and load preserves session state", async () => { + const state = makeState(); + await sm.save("client-001", state); + const loaded = await sm.load("client-001"); + expect(loaded).toEqual(state); + }); + + it("returns null for non-existent session", async () => { + const loaded = await sm.load("non-existent"); + expect(loaded).toBeNull(); + }); + + it("returns null for corrupted session blob", async () => { + await sm.save("client-001", makeState()); + const loaded = await sm.load("client-001"); + expect(loaded).not.toBeNull(); + }); + + it("rejects session with mismatched clientId", async () => { + const state = makeState({ clientId: "client-001" }); + await sm.save("client-001", state); + const loaded = await sm.load("client-002"); + expect(loaded).toBeNull(); + }); + }); + + describe("key rotation", () => { + it("supports loading sessions encrypted with different keys via shared Redis", async () => { + const key1 = makeTestKey(); + const key2 = makeTestKey(); + const multiKey = { v1: key1, v2: key2 }; + + const store = new Map(); + const redisMock = { + async set(key, value, _ex, _ttl) { store.set(key, value); }, + async get(key) { return store.get(key) ?? null; }, + async del(key) { store.delete(key); }, + }; + + const sm1 = new SessionManager({ redis: redisMock, encryptionKey: multiKey, keyId: "v1", debounceMs: 10 }); + const state = makeState(); + await sm1.save("client-001", state); + sm1.destroy(); + + const sm2 = new SessionManager({ redis: redisMock, encryptionKey: multiKey, keyId: "v2", debounceMs: 10 }); + const loaded = await sm2.load("client-001"); + expect(loaded).toEqual(state); + sm2.destroy(); + }); + + it("fails to load with wrong key version", async () => { + const key1 = makeTestKey(); + const key2 = makeTestKey(); + + const store = new Map(); + const redisMock = { + async set(key, value) { store.set(key, value); }, + async get(key) { return store.get(key) ?? null; }, + async del(key) { store.delete(key); }, + }; + + const sm1 = new SessionManager({ redis: redisMock, encryptionKey: { v1: key1 }, keyId: "v1", debounceMs: 10 }); + await sm1.save("client-001", makeState()); + sm1.destroy(); + + const sm2 = new SessionManager({ redis: redisMock, encryptionKey: { v3: key2 }, keyId: "v3", debounceMs: 10 }); + const loaded = await sm2.load("client-001"); + expect(loaded).toBeNull(); + sm2.destroy(); + }); + }); + + describe("delete", () => { + it("removes a session", async () => { + await sm.save("client-001", makeState()); + expect(await sm.load("client-001")).not.toBeNull(); + await sm.delete("client-001"); + expect(await sm.load("client-001")).toBeNull(); + }); + + it("is idempotent for non-existent session", async () => { + await expect(sm.delete("non-existent")).resolves.toBeUndefined(); + }); + }); + + describe("debouncedSave", () => { + it("debounces rapid saves", async () => { + const saveSpy = vi.spyOn(sm, "save"); + const state1 = makeState({ rooms: [{ roomId: "r1", highestAckedSeq: 1, highestReceivedSeq: 1, geofenceInsideSet: [] }] }); + const state2 = makeState({ rooms: [{ roomId: "r1", highestAckedSeq: 2, highestReceivedSeq: 2, geofenceInsideSet: [] }] }); + + sm.debouncedSave("client-001", state1); + sm.debouncedSave("client-001", state2); + + expect(sm.pendingSaves).toBe(1); + + await new Promise((r) => setTimeout(r, 50)); + expect(saveSpy).toHaveBeenCalledTimes(1); + + const loaded = await sm.load("client-001"); + expect(loaded.rooms[0].highestAckedSeq).toBe(2); + }); + }); + + describe("saveImmediate", () => { + it("saves immediately bypassing debounce", async () => { + const state = makeState(); + await sm.saveImmediate("client-001", state); + const loaded = await sm.load("client-001"); + expect(loaded).toEqual(state); + }); + }); + + describe("flushPending", () => { + it("flushes all pending debounced saves", async () => { + const state = makeState(); + sm.debouncedSave("client-001", state); + expect(sm.pendingSaves).toBe(1); + await sm.flushPending(); + expect(sm.pendingSaves).toBe(0); + const loaded = await sm.load("client-001"); + expect(loaded).toEqual(state); + }); + + it("flushes pending saves for Redis-backed sessions", async () => { + const store = new Map(); + const redisMock = { + async set(key, value) { store.set(key, value); }, + async get(key) { return store.get(key) ?? null; }, + async del(key) { store.delete(key); }, + }; + + const redisSm = new SessionManager({ redis: redisMock, encryptionKey, debounceMs: 10 }); + const state = makeState(); + redisSm.debouncedSave("client-001", state); + expect(redisSm.pendingSaves).toBe(1); + await redisSm.flushPending(); + expect(redisSm.pendingSaves).toBe(0); + expect(store.has("session:client-001")).toBe(true); + const loaded = await redisSm.load("client-001"); + expect(loaded).toEqual(state); + redisSm.destroy(); + }); + }); + + describe("in-memory mode (no Redis)", () => { + it("stores and retrieves sessions from local cache", async () => { + const state = makeState(); + await sm.save("client-001", state); + expect(sm.cachedSessions).toBe(1); + const loaded = await sm.load("client-001"); + expect(loaded).toEqual(state); + }); + + it("evicts expired entries", async () => { + const shortTtlSm = new SessionManager({ encryptionKey, ttlMs: 1, debounceMs: 10 }); + await shortTtlSm.save("client-001", makeState()); + expect(shortTtlSm.cachedSessions).toBe(1); + await new Promise((r) => setTimeout(r, 20)); + expect(shortTtlSm.cachedSessions).toBe(0); + const loaded = await shortTtlSm.load("client-001"); + expect(loaded).toBeNull(); + shortTtlSm.destroy(); + }); + }); + + describe("session state structure", () => { + it("handles rooms with full state", async () => { + const state = makeState({ + rooms: [ + { roomId: "fleet-1", highestAckedSeq: 10, highestReceivedSeq: 15, geofenceInsideSet: ["fence-a", "fence-b"] }, + { roomId: "fleet-2", highestAckedSeq: 0, highestReceivedSeq: 3, geofenceInsideSet: [] }, + ], + }); + await sm.save("client-001", state); + const loaded = await sm.load("client-001"); + expect(loaded.rooms).toHaveLength(2); + expect(loaded.rooms[0].geofenceInsideSet).toEqual(["fence-a", "fence-b"]); + }); + + it("handles empty rooms array", async () => { + const state = makeState({ rooms: [] }); + await sm.save("client-001", state); + const loaded = await sm.load("client-001"); + expect(loaded.rooms).toEqual([]); + }); + + it("handles large session with many rooms", async () => { + const rooms = Array.from({ length: 50 }, (_, i) => ({ + roomId: `fleet-${i}`, + highestAckedSeq: i * 10, + highestReceivedSeq: i * 10 + 5, + geofenceInsideSet: Array.from({ length: 3 }, (_, j) => `fence-${i}-${j}`), + })); + const state = makeState({ rooms }); + await sm.save("client-001", state); + const loaded = await sm.load("client-001"); + expect(loaded.rooms).toHaveLength(50); + expect(loaded.rooms[49].roomId).toBe("fleet-49"); + }); + }); + + describe("blob size", () => { + it("session blob stays under 16KB for 50 rooms", async () => { + const rooms = Array.from({ length: 50 }, (_, i) => ({ + roomId: `fleet-${i}`, + highestAckedSeq: i * 10, + highestReceivedSeq: i * 10 + 5, + geofenceInsideSet: Array.from({ length: 3 }, (_, j) => `fence-${i}-${j}`), + })); + const state = makeState({ rooms }); + await sm.save("client-001", state); + + const loaded = await sm.load("client-001"); + expect(loaded).not.toBeNull(); + const jsonSize = Buffer.byteLength(JSON.stringify(state), "utf8"); + expect(jsonSize).toBeLessThan(16384); + }); + }); + + describe("destroy", () => { + it("clears all timers", () => { + sm.debouncedSave("client-001", makeState()); + expect(sm.pendingSaves).toBe(1); + sm.destroy(); + expect(sm.pendingSaves).toBe(0); + }); + + it("is idempotent", () => { + sm.destroy(); + expect(() => sm.destroy()).not.toThrow(); + }); + }); + + describe("with Redis mock", () => { + it("delegates to Redis client", async () => { + const store = new Map(); + const redisMock = { + async set(key, value, _ex, _ttl) { store.set(key, value); }, + async get(key) { return store.get(key) ?? null; }, + async del(key) { store.delete(key); }, + }; + + const redisSm = new SessionManager({ redis: redisMock, encryptionKey, debounceMs: 10 }); + const state = makeState(); + await redisSm.save("client-001", state); + + expect(store.has("session:client-001")).toBe(true); + + const loaded = await redisSm.load("client-001"); + expect(loaded).toEqual(state); + + await redisSm.delete("client-001"); + expect(store.has("session:client-001")).toBe(false); + + redisSm.destroy(); + }); + + it("returns null for missing Redis key", async () => { + const redisMock = { + async set() {}, + async get() { return null; }, + async del() {}, + }; + + const redisSm = new SessionManager({ redis: redisMock, encryptionKey, debounceMs: 10 }); + const loaded = await redisSm.load("non-existent"); + expect(loaded).toBeNull(); + redisSm.destroy(); + }); + + it("returns null for corrupted Redis value", async () => { + const store = new Map(); + const redisMock = { + async set(key, value) { store.set(key, value); }, + async get(key) { return store.get(key) ?? null; }, + async del(key) { store.delete(key); }, + }; + + const redisSm = new SessionManager({ redis: redisMock, encryptionKey, debounceMs: 10 }); + await redisSm.save("client-001", makeState()); + + store.set("session:client-001", "corrupted-data"); + const loaded = await redisSm.load("client-001"); + expect(loaded).toBeNull(); + redisSm.destroy(); + }); + + it("rejects Redis session with mismatched clientId", async () => { + const store = new Map(); + const redisMock = { + async set(key, value) { store.set(key, value); }, + async get(key) { return store.get(key) ?? null; }, + async del(key) { store.delete(key); }, + }; + + const redisSm = new SessionManager({ redis: redisMock, encryptionKey, debounceMs: 10 }); + await redisSm.save("client-001", makeState({ clientId: "client-001" })); + + const loaded = await redisSm.load("client-002"); + expect(loaded).toBeNull(); + redisSm.destroy(); + }); + }); +}); diff --git a/tests/session-resumption-integration.test.js b/tests/session-resumption-integration.test.js new file mode 100644 index 0000000..9142923 --- /dev/null +++ b/tests/session-resumption-integration.test.js @@ -0,0 +1,276 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import WebSocket from "ws"; +import jwt from "jsonwebtoken"; +import { createServer } from "../src/server.js"; + +const TEST_SECRET = "test-secret-key"; + +/** Sign a JWT for a client. */ +function makeToken(clientId) { + return jwt.sign({ sub: clientId }, TEST_SECRET, { expiresIn: 60 }); +} + +/** Collect the next N messages from a WebSocket. */ +function nextMessages(ws, n = 1, timeoutMs = 3000) { + return new Promise((resolve, reject) => { + const msgs = []; + const timeout = setTimeout(() => reject(new Error("Timeout waiting for messages")), timeoutMs); + ws.on("message", function handler(data) { + msgs.push(JSON.parse(data.toString())); + if (msgs.length === n) { + clearTimeout(timeout); + ws.off("message", handler); + resolve(msgs); + } + }); + }); +} + +/** Collect messages for a duration, returning all received. */ +function collectMessages(ws, durationMs = 500) { + return new Promise((resolve) => { + const msgs = []; + ws.on("message", (data) => { + msgs.push(JSON.parse(data.toString())); + }); + setTimeout(() => resolve(msgs), durationMs); + }); +} + +/** Wait for the WS close event. */ +function waitClose(ws) { + return new Promise((resolve) => ws.once("close", resolve)); +} + +/** Close a list of sockets and wait for them all. */ +async function closeAll(...sockets) { + sockets.forEach((ws) => ws.readyState === WebSocket.OPEN && ws.close()); + await Promise.all(sockets.map(waitClose)); +} + +/** + * Open a WS connection and set up a message listener BEFORE the open event. + * Returns { ws, messages } where messages is a Promise resolving to the next N messages. + */ +function connectWithListener(port, token, extraParams = {}, n = 1) { + const params = new URLSearchParams(); + if (token) params.set("token", token); + for (const [k, v] of Object.entries(extraParams)) { + params.set(k, v); + } + const qs = params.toString(); + const url = `ws://localhost:${port}/${qs ? `?${qs}` : ""}`; + + return new Promise((resolve, reject) => { + const ws = new WebSocket(url); + let collectedMsgs = []; + + const messages = new Promise((msgResolve, msgReject) => { + const timeout = setTimeout(() => msgReject(new Error("Timeout waiting for messages")), 3000); + ws.on("message", function handler(data) { + collectedMsgs.push(JSON.parse(data.toString())); + if (collectedMsgs.length === n) { + clearTimeout(timeout); + ws.off("message", handler); + msgResolve(collectedMsgs); + } + }); + }); + + ws.once("open", () => resolve({ ws, messages })); + ws.once("error", reject); + }); +} + +describe("Session Resumption Integration", () => { + let server; + let port; + + beforeEach(() => { + process.env.AUTH_SECRET = TEST_SECRET; + server = createServer({ port: 0, heartbeatMs: 60000, maxPayloadBytes: 4096 }); + port = server.wss.address().port; + }); + + afterEach(async () => { + for (const client of server.wss.clients) { + client.terminate(); + } + await new Promise((resolve) => server.wss.close(resolve)); + delete process.env.AUTH_SECRET; + }); + + it("saves session state on room join and disconnect", async () => { + const { ws: ws1, messages: j1 } = await connectWithListener(port, makeToken("client-resume-1"), {}, 1); + ws1.send(JSON.stringify({ type: "join_room", roomId: "fleet-resume" })); + await j1; + + ws1.close(); + await waitClose(ws1); + await new Promise((r) => setTimeout(r, 100)); + + const loaded = await server.sessionManager.load("client-resume-1"); + expect(loaded).not.toBeNull(); + expect(loaded.clientId).toBe("client-resume-1"); + expect(loaded.rooms).toHaveLength(1); + expect(loaded.rooms[0].roomId).toBe("fleet-resume"); + }); + + it("restores session on reconnect with valid session_id", async () => { + const { ws: ws1, messages: j1 } = await connectWithListener(port, makeToken("client-resume-2"), {}, 1); + ws1.send(JSON.stringify({ type: "join_room", roomId: "fleet-session" })); + await j1; + + ws1.close(); + await waitClose(ws1); + await new Promise((r) => setTimeout(r, 100)); + + const { ws: ws2, messages: resumed } = await connectWithListener(port, makeToken("client-resume-2"), { session_id: "client-resume-2" }, 1); + const msgs = await resumed; + expect(msgs[0].type).toBe("session_resumed"); + expect(msgs[0].payload.rooms).toContain("fleet-session"); + + await closeAll(ws2); + }); + + it("treats expired session as new session (no session_resumed sent)", async () => { + const sm = server.sessionManager; + await sm.save("client-expired", { + clientId: "client-expired", + protocolVersion: 3, + authIdentity: { sub: "client-expired" }, + rooms: [{ roomId: "fleet-expired", highestAckedSeq: 0, highestReceivedSeq: 0, geofenceInsideSet: [] }], + rateLimitState: { messageWindow: [], connectionWindow: [] }, + metadata: { ip: "127.0.0.1", userAgent: "", connectedAt: Date.now(), lastActivityAt: Date.now() }, + }); + await sm.delete("client-expired"); + + const ws = new WebSocket(`ws://localhost:${port}/?token=${makeToken("client-expired")}&session_id=client-expired`); + const msgs = await collectMessages(ws, 500); + const resumedMsgs = msgs.filter((m) => m.type === "session_resumed"); + expect(resumedMsgs).toHaveLength(0); + ws.close(); + await waitClose(ws); + }); + + it("rejects session with identity mismatch (no session_resumed sent)", async () => { + await server.sessionManager.save("client-mismatch", { + clientId: "client-mismatch", + protocolVersion: 3, + authIdentity: { sub: "wrong-identity" }, + rooms: [{ roomId: "fleet-mismatch", highestAckedSeq: 0, highestReceivedSeq: 0, geofenceInsideSet: [] }], + rateLimitState: { messageWindow: [], connectionWindow: [] }, + metadata: { ip: "127.0.0.1", userAgent: "", connectedAt: Date.now(), lastActivityAt: Date.now() }, + }); + + const ws = new WebSocket(`ws://localhost:${port}/?token=${makeToken("client-different")}&session_id=client-mismatch`); + const msgs = await collectMessages(ws, 500); + const resumedMsgs = msgs.filter((m) => m.type === "session_resumed"); + expect(resumedMsgs).toHaveLength(0); + ws.close(); + await waitClose(ws); + }); + + it("saves session state on room leave", async () => { + const { ws: ws1, messages: j1 } = await connectWithListener(port, makeToken("client-leave-1"), {}, 1); + ws1.send(JSON.stringify({ type: "join_room", roomId: "fleet-leave" })); + await j1; + + const l1 = nextMessages(ws1, 1); + ws1.send(JSON.stringify({ type: "leave_room", roomId: "fleet-leave" })); + await l1; + + ws1.close(); + await waitClose(ws1); + await new Promise((r) => setTimeout(r, 100)); + + const loaded = await server.sessionManager.load("client-leave-1"); + expect(loaded).not.toBeNull(); + expect(loaded.rooms).toHaveLength(0); + }); + + it("debounces session saves during rapid state changes", async () => { + const { ws, messages: j1 } = await connectWithListener(port, makeToken("client-debounce"), {}, 1); + ws.send(JSON.stringify({ type: "join_room", roomId: "room-1" })); + await j1; + + const j2 = nextMessages(ws, 1); + ws.send(JSON.stringify({ type: "join_room", roomId: "room-2" })); + await j2; + + const j3 = nextMessages(ws, 1); + ws.send(JSON.stringify({ type: "join_room", roomId: "room-3" })); + await j3; + + await new Promise((r) => setTimeout(r, 600)); + + const loaded = await server.sessionManager.load("client-debounce"); + expect(loaded).not.toBeNull(); + expect(loaded.rooms).toHaveLength(3); + + ws.close(); + await waitClose(ws); + }); + + it("returns session_resumed with correct sequence numbers", async () => { + const { ws: ws1, messages: j1 } = await connectWithListener(port, makeToken("client-seq-1"), {}, 1); + ws1.send(JSON.stringify({ type: "join_room", roomId: "fleet-seq" })); + await j1; + + ws1.close(); + await waitClose(ws1); + await new Promise((r) => setTimeout(r, 100)); + + const { ws: ws2, messages: resumed } = await connectWithListener(port, makeToken("client-seq-1"), { session_id: "client-seq-1" }, 1); + const msgs = await resumed; + expect(msgs[0].type).toBe("session_resumed"); + expect(msgs[0].payload.currentSeqPerRoom).toBeDefined(); + expect(msgs[0].payload.currentSeqPerRoom[0].roomId).toBe("fleet-seq"); + + await closeAll(ws2); + }); + + it("session metrics increment correctly", async () => { + const { ws: ws1, messages: j1 } = await connectWithListener(port, makeToken("client-metrics-1"), {}, 1); + ws1.send(JSON.stringify({ type: "join_room", roomId: "fleet-metrics" })); + await j1; + + ws1.close(); + await waitClose(ws1); + await new Promise((r) => setTimeout(r, 100)); + + const { ws: ws2, messages: resumed } = await connectWithListener(port, makeToken("client-metrics-1"), { session_id: "client-metrics-1" }, 1); + const msgs = await resumed; + expect(msgs[0].type).toBe("session_resumed"); + + await closeAll(ws2); + }); + + it("handles session without session_id gracefully", async () => { + const ws = new WebSocket(`ws://localhost:${port}/?token=${makeToken("client-no-session")}`); + await new Promise((resolve) => ws.once("open", resolve)); + const msgs = nextMessages(ws, 1); + ws.send(JSON.stringify({ type: "join_room", roomId: "fleet-no-session" })); + await msgs; + + expect(server.rooms.getRoomSize("fleet-no-session")).toBe(1); + ws.close(); + await waitClose(ws); + }); + + it("saves session on graceful shutdown via flushPending", async () => { + const ws = new WebSocket(`ws://localhost:${port}/?token=${makeToken("client-shutdown")}`); + await new Promise((resolve) => ws.once("open", resolve)); + const msgs = nextMessages(ws, 1); + ws.send(JSON.stringify({ type: "join_room", roomId: "fleet-shutdown" })); + await msgs; + + await server.sessionManager.flushPending(); + const loaded = await server.sessionManager.load("client-shutdown"); + expect(loaded).not.toBeNull(); + expect(loaded.rooms).toHaveLength(1); + + ws.close(); + await waitClose(ws); + }); +});