var test = require("node:test"); var assert = require("node:assert/strict"); var fs = require("node:fs"); var vm = require("node:vm"); var path = require("node:path"); var root = path.join(__dirname, ".."); function productionHarness() { var now = 0; var nextTimer = 1; var timers = new Map(); var listeners = {}; var sockets = []; var fetches = []; var socket = null; var state = { connected: false, wsPath: "/ws", socketPath: "/ws", currentSlug: "project-a", activeSessionId: 7 }; var context = { console: { warn: function () {}, log: function () {} }, Math: Object.assign(Object.create(Math), { random: function () { return 0.5; } }), Date: { now: function () { return now; } }, navigator: { onLine: true }, location: { protocol: "http:", host: "fixture", reload: function () {} }, localStorage: { removeItem: function () {} }, state: state, store: { get: function (key) { return state[key]; }, set: function (next) { Object.assign(state, next); }, subscribe: function () {} }, getWs: function () { return socket; }, setWs: function (next) { socket = next; }, window: { addEventListener: function (name, callback) { listeners[name] = callback; }, dispatchEvent: function () {} }, document: { hidden: false, getElementById: function () { return null; }, addEventListener: function (name, callback) { listeners[name] = callback; } }, fetch: function () { var success = null; var failure = null; var pending = { then: function (callback) { success = callback; return pending; }, catch: function (callback) { failure = callback; return pending; }, resolve: function (value) { if (value && value.status === 200 && !value.json) value.json = function () { return { then: function (callback) { callback({ projects: [{ slug: "project-a", contentAccess: true }] }); } }; }; if (success) success(value); }, reject: function () { if (failure) failure(); } }; fetches.push(pending); return pending; }, CustomEvent: function (type, init) { this.type = type; this.detail = init.detail; }, AbortController: function () { this.signal = {}; this.abort = function () {}; }, processMessage: function () {} }; context.paneOutgoing = function (data) { return data; }; var imported = fs.readFileSync(path.join(root, "lib/public/modules/app-connection.js"), "utf8"); Array.from(imported.matchAll(/import \{([^}]+)\}/g)).forEach(function (match) { match[1].split(",").forEach(function (name) { name = name.trim(); if (name && !context[name]) context[name] = function () {}; }); }); context.setTimeout = function (fn, delay) { var id = nextTimer++; timers.set(id, { fn: fn, at: now + delay }); return id; }; context.clearTimeout = function (id) { timers.delete(id); }; context.setInterval = function (fn, delay) { var id = nextTimer++; timers.set(id, { fn: fn, at: now + delay, interval: delay }); return id; }; context.clearInterval = context.clearTimeout; context.WebSocket = function () { var current = this; current.readyState = 0; current.send = function () {}; current.close = function () { current.readyState = 3; if (current.onclose) current.onclose({ code: 1000, wasClean: true }); }; sockets.push(current); }; vm.createContext(context); var lifecycle = fs.readFileSync(path.join(root, "lib/public/modules/websocket-lifecycle.js"), "utf8"); var watchdog = fs.readFileSync(path.join(root, "lib/public/modules/websocket-watchdog.js"), "utf8"); vm.runInContext(lifecycle.replace(/export \{[^}]+\};?/, ""), context); vm.runInContext(watchdog.replace(/export \{[^}]+\};?/, ""), context); var replay = fs.readFileSync(path.join(root, "lib/public/modules/history-replay-batch.js"), "utf8"); vm.runInContext(replay.replace(/^import .*;\n/gm, "").replace(/export /g, ""), context); var recovery = fs.readFileSync(path.join(root, "lib/public/modules/project-access-recovery.js"), "utf8"); vm.runInContext(recovery.replace(/^import .*;\n/gm, "").replace(/export /g, ""), context); vm.runInContext(imported.replace(/^import .*;\n/gm, "").replace(/export /g, ""), context); function tick(duration) { var end = now + duration; while (true) { var selected = null; timers.forEach(function (entry, id) { if (entry.at <= end && (!selected || entry.at < selected.entry.at)) selected = { id: id, entry: entry }; }); if (!selected) break; now = selected.entry.at; timers.delete(selected.id); if (selected.entry.interval) timers.set(selected.id, { fn: selected.entry.fn, at: now + selected.entry.interval, interval: selected.entry.interval }); selected.entry.fn(); } now = end; } function jump(duration) { now += duration; } function timerCallbacks() { return Array.from(timers.values()).map(function (entry) { return entry.fn; }); } return { context: context, listeners: listeners, sockets: sockets, fetches: fetches, timers: timers, tick: tick, jump: jump, timerCallbacks: timerCallbacks, getSocket: function () { return socket; } }; } test("production connection cancels close stability, stale messages, probes, and resume deadlines", function () { var f = productionHarness(); f.context.connect(); var first = f.sockets[0]; first.readyState = 1; first.onopen(); first.close(); f.tick(14999); var retryBeforeOldStableDeadline = f.context.lifecycle.getRetryAttempt(); f.tick(1); assert.equal(f.context.lifecycle.getRetryAttempt(), retryBeforeOldStableDeadline, "closed sockets cannot reset retry state through an old stable timer"); var g = productionHarness(); g.context.initConnection(); g.context.connect(); var live = g.sockets[0]; live.readyState = 1; live.send = function (data) { if (JSON.parse(data).type === "ping" && live.onmessage) live.onmessage({ data: JSON.stringify({ type: "pong" }) }); }; live.onopen(); g.listeners.visibilitychange(); g.context.cancelReconnect(); var sendsAfterCancel = 0; live.send = function () { sendsAfterCancel += 1; }; g.tick(10000); assert.equal(sendsAfterCancel, 0, "cancel prevents a pending probe from sending"); g.context.state.connected = false; live.onmessage({ data: JSON.stringify({ type: "pong" }) }); assert.equal(g.context.state.connected, false, "cancelled socket messages cannot restore connected state"); var h = productionHarness(); h.context.initConnection(); h.context.connect(); var offlineSocket = h.sockets[0]; offlineSocket.readyState = 1; offlineSocket.onopen(); h.listeners.offline(); assert.equal(h.getSocket(), null, "offline detaches the old OPEN socket"); assert.equal(offlineSocket.onmessage, null, "offline detaches stale message handlers"); var resumed = productionHarness(); resumed.context.initConnection(); resumed.context.connect(); var resumedSocket = resumed.sockets[0]; resumedSocket.readyState = 1; resumedSocket.send = function (data) { if (JSON.parse(data).type === "ping" && resumedSocket.onmessage) resumedSocket.onmessage({ data: JSON.stringify({ type: "pong" }) }); }; resumedSocket.onopen(); resumed.tick(25000); resumed.context.document.hidden = true; resumed.listeners.visibilitychange(); resumed.context.document.hidden = false; resumed.listeners.visibilitychange(); resumed.tick(10000); assert.equal(resumed.getSocket(), resumedSocket, "visible resume replaces the old heartbeat deadline"); }); test("hidden suspension invalidates queued watchdog callbacks before a fresh visible probe", function () { var f = productionHarness(); f.context.initConnection(); f.context.connect(); var socket = f.sockets[0]; var pings = 0; socket.readyState = 1; socket.send = function (data) { if (JSON.parse(data).type === "ping") pings += 1; }; socket.onopen(); f.tick(25000); f.listeners["clay-message-delivery-timeout"]({ detail: { socket: socket } }); var queuedCallbacks = f.timerCallbacks(); f.context.document.hidden = true; f.listeners.visibilitychange(); f.jump(60000); queuedCallbacks.forEach(function (callback) { callback(); }); assert.equal(f.getSocket(), socket, "callbacks queued before hidden cannot replace the retained socket"); socket.send = function (data) { if (JSON.parse(data).type !== "ping") return; pings += 1; socket.onmessage({ data: JSON.stringify({ type: "pong" }) }); }; f.context.document.hidden = false; f.listeners.visibilitychange(); f.tick(10000); assert.equal(f.getSocket(), socket, "the fresh visible probe proves the retained socket is healthy"); assert.equal(pings, 3, "one heartbeat, one old probe, and one fresh visible probe are sent"); }); test("freeze resume coalesces repeated events and accepts a queued pong", function () { var f = productionHarness(); f.context.initConnection(); f.context.connect(); var socket = f.sockets[0]; var pings = 0; socket.readyState = 1; socket.send = function (data) { if (JSON.parse(data).type === "ping") pings += 1; }; socket.onopen(); f.listeners.freeze(); f.jump(60000); f.listeners.resume(); f.listeners.resume(); assert.equal(pings, 1, "repeated resume events share one in-flight liveness probe"); socket.onmessage({ data: JSON.stringify({ type: "pong" }) }); f.tick(4000); assert.equal(f.getSocket(), socket, "a queued pong after resume clears the fresh probe"); }); test("online while hidden waits for visibility and hidden open starts no watchdog", function () { var hiddenOpen = productionHarness(); hiddenOpen.context.document.hidden = true; hiddenOpen.context.initConnection(); hiddenOpen.context.connect(); var hiddenSocket = hiddenOpen.sockets[0]; var hiddenPings = 0; hiddenSocket.readyState = 1; hiddenSocket.send = function (data) { if (JSON.parse(data).type === "ping") hiddenPings += 1; }; hiddenSocket.onopen(); hiddenOpen.tick(25000); assert.equal(hiddenPings, 0, "a socket opened while hidden has no background watchdog"); hiddenOpen.context.document.hidden = false; hiddenOpen.listeners.visibilitychange(); assert.equal(hiddenPings, 1, "visibility starts one fresh probe for a hidden-opened socket"); var offline = productionHarness(); offline.context.initConnection(); offline.context.connect(); var offlineSocket = offline.sockets[0]; offlineSocket.readyState = 1; offlineSocket.onopen(); offline.context.document.hidden = true; offline.listeners.visibilitychange(); offline.listeners.offline(); offline.listeners.online(); offline.tick(20000); assert.equal(offline.fetches.length, 0, "online while hidden does not start reconnect authentication"); assert.equal(offline.sockets.length, 1, "online while hidden does not create a socket"); offline.context.document.hidden = false; offline.listeners.visibilitychange(); offline.tick(1000); assert.equal(offline.fetches.length, 1, "visible recovery resumes the existing reconnect policy"); }); test("a dead resumed socket is replaced in four seconds and an old socket pong is ignored", function () { var f = productionHarness(); f.context.initConnection(); f.context.connect(); var first = f.sockets[0]; first.readyState = 1; first.send = function () {}; first.onopen(); var oldMessage = first.onmessage; f.context.document.hidden = true; f.listeners.visibilitychange(); f.context.document.hidden = false; f.listeners.visibilitychange(); f.tick(3999); assert.equal(f.getSocket(), first, "the resumed probe retains its full four-second budget"); f.tick(1); assert.equal(f.getSocket(), null, "an actually dead resumed socket is replaced at four seconds"); f.tick(1000); f.fetches[0].resolve({ status: 200 }); var second = f.sockets[1]; second.readyState = 1; second.send = function () {}; second.onopen(); f.listeners.visibilitychange(); oldMessage({ data: JSON.stringify({ type: "pong" }) }); f.tick(4000); assert.equal(f.getSocket(), null, "a late pong from the retired socket cannot clear the current probe"); }); test("production handshake timeout replaces a stalled socket and server health resets stability", function () { var f = productionHarness(); f.context.connect(); assert.equal(f.sockets.length, 1); f.tick(11999); assert.equal(f.sockets.length, 1, "the production handshake remains within its 12-second budget"); f.tick(1); assert.equal(f.getSocket(), null, "a stalled handshake is replaced without waiting for close"); var healthy = productionHarness(); healthy.context.lifecycle.schedule("prior", function () {}); healthy.context.connect(); var healthySocket = healthy.sockets[0]; healthySocket.readyState = 1; healthySocket.onopen(); healthy.tick(14999); assert.equal(healthy.context.lifecycle.getRetryAttempt(), 1, "open alone does not reset retry state"); healthySocket.onmessage({ data: JSON.stringify({ type: "bootstrap" }) }); healthy.tick(15000); assert.equal(healthy.context.lifecycle.getRetryAttempt(), 0, "server health proof resets retry state after stability"); }); test("production auth ignores late resolve and reject from a timed-out request", function () { ["resolve", "reject"].forEach(function (outcome) { var f = productionHarness(); f.context.scheduleReconnect("auth-race"); f.tick(1000); assert.equal(f.fetches.length, 1); f.tick(4000); f.tick(1500); assert.equal(f.fetches.length, 2, "a replacement auth request is pending in the same epoch"); f.fetches[0][outcome]({ status: 200 }); assert.equal(f.getSocket(), null, "the stale auth result cannot start a socket"); }); }); test("wakeups do not replace a CONNECTING handshake and errors replace directly", function () { var f = productionHarness(); f.context.initConnection(); f.context.connect(); var connecting = f.sockets[0]; f.listeners.online(); f.listeners.visibilitychange(); f.listeners["clay-message-delivery-timeout"]({ detail: { socket: connecting } }); f.tick(1000); assert.equal(f.sockets.length, 1, "wakeups do not create parallel sockets during the handshake"); assert.equal(f.getSocket(), connecting, "the allowed handshake remains live"); connecting.onerror(); assert.equal(f.getSocket(), null, "onerror replaces without waiting for close"); assert.equal(connecting.onclose, null, "replacement detaches the close callback"); }); test("a real ACK-timeout event probes a healthy OPEN socket without reconnecting", function () { var f = productionHarness(); f.context.initConnection(); f.context.connect(); var socket = f.sockets[0]; var pings = 0; socket.readyState = 1; socket.send = function (data) { if (JSON.parse(data).type === "ping") { pings++; socket.onmessage({ data: JSON.stringify({ type: "pong" }) }); } }; socket.onopen(); f.listeners["clay-message-delivery-timeout"]({ detail: { socket: socket, clientMessageId: "cm-health" } }); assert.equal(f.getSocket(), socket, "a healthy ACK-timeout probe does not replace the socket"); assert.equal(f.sockets.length, 1, "a healthy ACK-timeout probe does not create another socket"); assert.equal(pings, 1, "the ACK-timeout event sends one immediate ping"); }); test("returning to a background pane replaces queued output with an exact-session snapshot", function () { var f = productionHarness(); f.context.state.paneMode = true; f.context.state.paneSessionId = 42; f.context.initConnection(); f.context.connect(); var first = f.sockets[0]; first.readyState = 1; first.onopen(); var delivered = []; f.context.processMessage = function (message) { delivered.push(message); }; var queued = first.onmessage; f.context.state.historyReplayBatch = { items: [{ type: 'delta', text: 'interrupted history' }] }; f.context.document.hidden = true; f.listeners.visibilitychange(); queued({ data: JSON.stringify({ type: 'text_delta', text: 'old backlog' }) }); assert.equal(delivered.length, 0); f.context.document.hidden = false; // Browser delivery can run before visibilitychange; it must not drain backlog. queued({ data: JSON.stringify({ type: 'text_delta', text: 'old backlog' }) }); f.listeners.visibilitychange(); assert.equal(f.sockets.length, 2, 'resume creates exactly one replacement'); assert.equal(first.readyState, 3); assert.equal(f.context.state.historyReplayBatch, null, 'replacement drops partially received history'); queued({ data: JSON.stringify({ type: 'text_delta', text: 'stale socket' }) }); assert.equal(delivered.length, 0); var current = f.sockets[1]; current.readyState = 1; current.onopen(); assert.equal(f.context.state.panePinPending, true); assert.equal(f.context.state.paneSessionId, 42); current.onmessage({ data: JSON.stringify({ type: 'history_done' }) }); assert.equal(delivered.length, 1); }); test("frozen pane resumes once and offline pane waits for the network", function () { var f = productionHarness(); f.context.state.paneMode = true; f.context.state.paneSessionId = 42; f.context.initConnection(); f.context.connect(); var socket = f.sockets[0]; socket.readyState = 1; socket.onopen(); f.listeners.freeze(); f.listeners.resume(); f.listeners.resume(); assert.equal(f.sockets.length, 2); f.context.document.hidden = true; f.listeners.visibilitychange(); f.listeners.offline(); f.context.document.hidden = false; f.listeners.visibilitychange(); assert.equal(f.sockets.length, 2, 'visibility cannot reconnect while offline'); f.listeners.online(); f.tick(1000); assert.equal(f.fetches.length, 1, 'online uses authenticated reconnect recovery'); }); test("reconnect clears revoked content and enters administrator metadata instead of reopening the project socket", function () { var f = productionHarness(); var destination = null; var cleared = false; f.context.document.body = { replaceChildren: function () { cleared = true; } }; f.context.window.location = { replace: function (url) { destination = url; } }; f.context.initConnection(); f.context.scheduleReconnect("close"); f.tick(1000); f.fetches[0].resolve({ status: 200, json: function () { return { then: function (callback) { callback({ projects: [{ slug: "project-a", contentAccess: false }] }); } }; } }); assert.equal(destination, "/p/project-a/"); assert.equal(cleared, true); assert.equal(f.sockets.length, 0); assert.equal(f.context.state.activeSessionId, null); });