import assert from 'node:assert/strict'; import { Buffer } from 'node:buffer'; import http from 'node:http'; import { createRequire } from 'node:module'; import { fileURLToPath } from 'node:url'; import path from 'node:path'; import { connectionCounts, clearRateLimitMaps } from '../server/rate-limiter.js'; import { enqueueQueuedEvent, materializeMediaIntent, reserveLatestMediaIntentSequence } from '../extension/offline-media-intent.js'; import { EPISODE_SYNC_V2_LOAD_TIMEOUT, FORCE_SYNC_TARGET_DELAY_WARNING, FORCE_SYNC_TIMEOUT } from '../shared/constants.js'; const __dirname = path.dirname(fileURLToPath(import.meta.url)); const require = createRequire(path.join(__dirname, '..', 'server', 'package.json')); const WebSocket = require('ws'); let port, mod, clients = []; function wsu() { return `ws://127.0.0.1:${port}/socket.io/?EIO=4&transport=websocket&version=2.4.0&token=62170b705234c4f4807a9b22420bb93cf1a2aacfa4c5d3b47804482babb8eb50`; } async function c() { const ws = new WebSocket(wsu()); clients.push(ws); ws._m = []; ws.on('message', d => ws._m.push(d.toString())); await new Promise((r, j) => { const t = setTimeout(() => j(Error('connect')), 5e3); ws.on('open', () => { clearTimeout(t); r(); }); }); ws.send('40'); const s = Date.now(); while (ws._m.length < 2 && Date.now()-s < 5e3) await new Promise(r => setTimeout(r, 50)); if (ws._m.length < 2) throw Error('handshake'); ws._m.length = 0; return ws; } function s(ws, evt, d={}) { ws.send(`42${JSON.stringify([evt,d])}`); } function a(ws) { if (ws._m.length) { const r=ws._m.shift(); return r.startsWith('42') ? JSON.parse(r.substring(2)) : r; } return new Promise((resolve, reject) => { const t=setTimeout(()=>reject(Error('timeout')),3e3); const h=(d)=>{clearTimeout(t);ws.removeListener('message',h);const r=d.toString();resolve(r.startsWith('42')?JSON.parse(r.substring(2)):r);};ws.on('message',h);}); } async function w(ws, evt, ms=3000) { const st=Date.now(); while(Date.now()-stsetTimeout(r,50));} throw Error(`wait:${evt}`); } async function j(ws, rid, pid, pw=null, clientCapabilities=undefined) { s(ws,'join_room',{roomId:rid,peerId:pid,password:pw,protocolVersion:'1.0.0',clientCapabilities}); const [event, data] = await a(ws); assert.equal(event,'room_data'); return data; } async function advanceEpisodeSyncToExecute(peerA, peerB, expectedTitle, expectedEpisodeId = undefined) { s(peerA, 'episode_sync_v2', { phase: 'start', expectedTitle, expectedEpisodeId }); const lobby = await w(peerA, 'episode_sync_v2'); await w(peerB, 'episode_sync_v2'); s(peerA, 'episode_sync_v2', { phase: 'loaded', transactionId: lobby.transactionId }); await w(peerA, 'episode_sync_v2'); await w(peerB, 'episode_sync_v2'); s(peerB, 'episode_sync_v2', { phase: 'loaded', transactionId: lobby.transactionId }); await w(peerA, 'episode_sync_v2'); await w(peerB, 'episode_sync_v2'); s(peerA, 'episode_sync_v2', { phase: 'prepared', transactionId: lobby.transactionId }); await w(peerA, 'episode_sync_v2'); await w(peerB, 'episode_sync_v2'); s(peerB, 'episode_sync_v2', { phase: 'prepared', transactionId: lobby.transactionId }); const executeA = await w(peerA, 'episode_sync_v2'); const executeB = await w(peerB, 'episode_sync_v2'); assert.equal(executeA.phase, 'execute'); assert.equal(executeB.phase, 'execute'); return { lobby, executeA, executeB }; } const delay = ms => new Promise(resolve => setTimeout(resolve, ms)); function close() { clients.forEach(w=>{try{w.close()}catch{/* ignore */}}); clients.length=0; } // Test suite opens >10 connections/min — clear the IP connection counter so the // connection rate limiter doesn't mask test failures (test-only, never at runtime). function resetConnectionRate() { connectionCounts.clear(); clearRateLimitMaps(); } try { process.env.ADMIN_METRICS_TOKEN = 'ws-integration-test-32chars-minimum!'; mod = await import('../server/index.js'); await mod.startServer(0,'127.0.0.1'); port = mod.httpServer.address().port; // --- Pool: 2 peers in 1 room, test everything --- const rid = 't-'+Date.now(); const p1 = await c(), p2 = await c(), p3 = await c(); // Room + join await j(p1, rid, 'a'); await j(p2, rid, 'b'); await j(p3, rid, 'c'); p1._m.length = p2._m.length = p3._m.length = 0; // Relay s(p1,'play',{currentTime:10}); await w(p2,'play'); s(p1,'pause',{currentTime:20}); await w(p2,'pause'); s(p1,'seek',{currentTime:30}); await w(p2,'seek'); // Force Sync s(p1,'force_sync_prepare',{targetTime:0}); await w(p2,'force_sync_prepare'); s(p1,'force_sync_ack',{}); await w(p2,'force_sync_ack'); s(p1,'force_sync_execute',{}); await w(p2,'force_sync_execute'); // EVENT_ACK s(p2,'event_ack',{targetId:'a',actionTimestamp:Date.now()}); await w(p1,'event_ack'); // Lobby s(p1,'episode_lobby',{expectedTitle:'S01E01'}); await w(p2,'episode_lobby'); await w(p3,'episode_lobby'); s(p3,'episode_lobby',{expectedTitle:'S01E02'}); const authoritativeLobby = await w(p3, 'episode_lobby'); assert.equal(authoritativeLobby.authoritative, true, 'competing initiator receives an authoritative lobby correction'); assert.equal(authoritativeLobby.expectedTitle, 'S01E01'); assert.deepEqual(authoritativeLobby.readyPeers, ['a']); let competingLobbyDropped = false; try { await w(p2, 'episode_lobby', 500); } catch { competingLobbyDropped = true; } assert.ok(competingLobbyDropped, 'relay drops a competing lobby while one is active'); assert.equal(mod.rooms.get(rid).activeLobby.expectedTitle, 'S01E01'); s(p3,'episode_ready',{expectedTitle:'S01E02',title:'S01E02'}); let staleReadyDropped = false; try { await w(p2, 'episode_ready', 500); } catch { staleReadyDropped = true; } assert.ok(staleReadyDropped, 'relay drops ready frames for an obsolete lobby'); assert.deepEqual(mod.rooms.get(rid).activeLobby.readyPeers, ['a']); // Missing expectedTitle remains accepted for old-extension compatibility. s(p2,'episode_ready',{title:'S01E01'}); await w(p1,'episode_ready'); assert.deepEqual(mod.rooms.get(rid).activeLobby.readyPeers, ['a', 'b']); s(p3,'leave_room',{}); await w(p1,'peer_status'); assert.equal(mod.rooms.get(rid).activeLobby.expectedTitle, 'S01E01', 'an unrelated departure does not dissolve a lobby with two peers left'); p1._m.length = p2._m.length = 0; // Leave s(p1,'leave_room',{}); const [ev,d]=await a(p2); assert.equal(ev,'peer_status');assert.equal(d.status,'left'); close(); resetConnectionRate(); // --- Combined reconnect model: compacted new-client intent over legacy wire --- // The queue is entirely client-side. Feed its materialized PLAY/PAUSE/SEEK // frames through the real relay and an old-client-like receiver to prove the // optimization needs no new event, capability or ACK. const coalescedRid = 'coalesced-media-'+Date.now(); const legacyReceiver = await c(), coalescingSender = await c(); await j(legacyReceiver, coalescedRid, 'legacy-receiver'); await j(coalescingSender, coalescedRid, 'coalesce-sender', null, ['chat-v1', 'media-state-v1']); legacyReceiver._m.length = coalescingSender._m.length = 0; s(legacyReceiver, 'play', { currentTime: 100, seq: 1, actionTimestamp: 1 }); await w(coalescingSender, 'play'); const canonicalBeforeReplay = { ...mod.rooms.get(coalescedRid).mediaState }; let compactedQueue = enqueueQueuedEvent([], 'play', { currentTime: 500, seq: 10, actionTimestamp: 10 }, { roomId: coalescedRid }).queue; compactedQueue = reserveLatestMediaIntentSequence(compactedQueue, coalescedRid, 11).queue; for (const [targetTime, seq] of [[540, 12], [600, 13]]) { compactedQueue = enqueueQueuedEvent(compactedQueue, 'seek', { currentTime: targetTime, targetTime, seq, actionTimestamp: seq }, { roomId: coalescedRid }).queue; } compactedQueue = enqueueQueuedEvent(compactedQueue, 'pause', { currentTime: 605, seq: 14, actionTimestamp: 14 }, { roomId: coalescedRid }).queue; assert.equal(compactedQueue.length, 1, 'offline playback burst is one logical queue entry'); const replayFrames = materializeMediaIntent(compactedQueue[0]); assert.deepEqual(replayFrames.map(frame => frame.event), ['seek', 'pause'], 'compacted intent uses only the minimum existing legacy events'); const legacyReplayPayloads = []; for (const frame of replayFrames) { s(coalescingSender, frame.event, frame.data); legacyReplayPayloads.push([frame.event, await w(legacyReceiver, frame.event)]); } assert.equal(legacyReplayPayloads[0][1].targetTime, 605); assert.equal(legacyReplayPayloads[1][1].currentTime, 605); for (const [event, payload] of legacyReplayPayloads) { assert.ok(event === 'seek' || event === 'pause'); assert.equal(payload.mediaState, undefined); assert.equal(payload.revision, undefined); } const canonicalAfterReplay = mod.rooms.get(coalescedRid).mediaState; assert.equal(canonicalAfterReplay.revision, canonicalBeforeReplay.revision + replayFrames.length); assert.equal(canonicalAfterReplay.playbackState, 'paused'); assert.equal(canonicalAfterReplay.currentTime, 605); assert.equal(canonicalAfterReplay.updatedBy, 'coalesce-sender'); const coalescedLateJoiner = await c(); const coalescedLateRoom = await j(coalescedLateJoiner, coalescedRid, 'coalesced-late'); assert.equal(coalescedLateRoom.mediaState.revision, canonicalAfterReplay.revision); assert.equal(coalescedLateRoom.mediaState.playbackState, 'paused'); assert.equal(coalescedLateRoom.mediaState.currentTime, 605); assert.equal(coalescedLateRoom.mediaState.updatedBy, 'coalesce-sender'); // --- Stale peer reaper: terminal timeout + clean rejoin --- const staleClient = await c(); const staleRoomId = 'stale-'+Date.now(); await j(staleClient, staleRoomId, 'stale-peer'); staleClient._m.length = 0; const staleRoom = mod.rooms.get(staleRoomId); staleRoom.peerData.values().next().value.lastSeen = 1; mod.cleanupInactiveRooms(Date.now()); const [staleEvent, staleData] = await a(staleClient); assert.equal(staleEvent, 'error'); assert.equal(staleData.code, 'peer_timed_out'); assert.equal(staleData.message, 'Removed from room after inactivity'); assert.equal(mod.rooms.has(staleRoomId), false, 'stale peer room is deleted'); staleClient._m.length = 0; await j(staleClient, staleRoomId, 'stale-peer'); assert.equal(mod.rooms.has(staleRoomId), true, 'stale peer can rejoin cleanly'); close(); resetConnectionRate(); // --- Capabilities: ROOM_DATA advertises server features for client detection --- const capClient = await c(); s(capClient, 'join_room', { roomId: 'cap-'+Date.now(), peerId: 'capp', protocolVersion: '1.0.0' }); const [capEv, capData] = await a(capClient); assert.equal(capEv, 'room_data'); assert.ok(Array.isArray(capData.capabilities) && capData.capabilities.includes('host-control'), 'ROOM_DATA advertises the host-control capability'); assert.ok(capData.capabilities.includes('chat'), 'ROOM_DATA advertises the chat capability'); assert.ok(capData.capabilities.includes('chat-v1'), 'ROOM_DATA advertises the versioned chat capability'); assert.ok(capData.capabilities.includes('media-state-v1'), 'ROOM_DATA advertises canonical media state v1'); assert.ok(capData.capabilities.includes('episode-sync-v2'), 'ROOM_DATA advertises Episode Sync v2'); assert.equal(capData.mediaState, null, 'a new room starts without invented canonical media state'); assert.equal(capData.chatHistory, undefined, 'ROOM_DATA never contains chat history'); close(); resetConnectionRate(); // --- Episode Sync v2: relay-owned transaction, exact participants, no timeout execute --- const episodeCaps = ['chat-v1', 'media-state-v1', 'episode-sync-v2']; const episodeRid = 'episode-v2-'+Date.now(); const episodeA = await c(), episodeB = await c(); await j(episodeA, episodeRid, 'episode-a', null, episodeCaps); await j(episodeB, episodeRid, 'episode-b', null, episodeCaps); episodeA._m.length = episodeB._m.length = 0; s(episodeA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S01E06 - Visiting Ours', expectedEpisodeId: 's01e06' }); const episodeLobbyA = await w(episodeA, 'episode_sync_v2'); const episodeLobbyB = await w(episodeB, 'episode_sync_v2'); assert.equal(episodeLobbyA.phase, 'lobby'); assert.equal(episodeLobbyA.transactionId, episodeLobbyB.transactionId); assert.deepEqual(episodeLobbyA.participants, ['episode-a', 'episode-b']); assert.deepEqual(episodeLobbyA.loadedPeers, [], 'initiator is not pre-marked loaded'); assert.equal(episodeLobbyA.expectedEpisodeId, 'S01E06'); assert.ok(episodeLobbyA.remainingMs > 0 && episodeLobbyA.remainingMs <= EPISODE_SYNC_V2_LOAD_TIMEOUT); assert.ok(episodeLobbyA.remainingMs > 119_000, 'v2 loading advertises the independent 120s deadline, not the 60s legacy deadline'); assert.equal(episodeLobbyA.deadlineAt, undefined, 'relay wall clock is not exposed to clients'); const episodeTxId = episodeLobbyA.transactionId; s(episodeB, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S99E99' }); const competingEpisodeStart = await w(episodeB, 'episode_sync_v2'); assert.equal(competingEpisodeStart.transactionId, episodeTxId); assert.equal(competingEpisodeStart.expectedTitle, 'S01E06 - Visiting Ours'); let competingEpisodeBroadcast = false; try { await w(episodeA, 'episode_sync_v2', 300); competingEpisodeBroadcast = true; } catch { /* expected */ } assert.equal(competingEpisodeBroadcast, false, 'competing start only receives authoritative correction'); s(episodeA, 'episode_sync_v2', { phase: 'loaded', transactionId: 'stale-transaction' }); await delay(100); assert.deepEqual(mod.rooms.get(episodeRid).episodeSyncV2.loadedPeers, [], 'stale transaction frame is ignored'); s(episodeA, 'episode_sync_v2', { phase: 'loaded', transactionId: episodeTxId }); const loadedA = await w(episodeA, 'episode_sync_v2'); await w(episodeB, 'episode_sync_v2'); assert.equal(loadedA.phase, 'lobby'); assert.deepEqual(loadedA.loadedPeers, ['episode-a']); s(episodeA, 'episode_sync_v2', { phase: 'loaded', transactionId: episodeTxId }); let duplicateLoadedRelayed = false; try { await w(episodeB, 'episode_sync_v2', 300); duplicateLoadedRelayed = true; } catch { /* expected */ } assert.equal(duplicateLoadedRelayed, false, 'duplicate loaded is idempotent'); s(episodeB, 'episode_sync_v2', { phase: 'loaded', transactionId: episodeTxId }); const prepareA = await w(episodeA, 'episode_sync_v2'); const prepareB = await w(episodeB, 'episode_sync_v2'); assert.equal(prepareA.phase, 'prepare'); assert.equal(prepareB.phase, 'prepare'); assert.deepEqual(prepareA.loadedPeers, ['episode-a', 'episode-b']); assert.deepEqual(prepareA.preparedPeers, []); s(episodeA, 'episode_sync_v2', { phase: 'prepared', transactionId: episodeTxId }); const onePrepared = await w(episodeB, 'episode_sync_v2'); await w(episodeA, 'episode_sync_v2'); assert.equal(onePrepared.phase, 'prepare'); assert.deepEqual(onePrepared.preparedPeers, ['episode-a']); s(episodeA, 'episode_sync_v2', { phase: 'prepared', transactionId: episodeTxId }); let duplicatePreparedRelayed = false; try { await w(episodeB, 'episode_sync_v2', 300); duplicatePreparedRelayed = true; } catch { /* expected */ } assert.equal(duplicatePreparedRelayed, false, 'duplicate prepared is idempotent'); s(episodeB, 'episode_sync_v2', { phase: 'prepared', transactionId: episodeTxId }); const executeA = await w(episodeA, 'episode_sync_v2'); const executeB = await w(episodeB, 'episode_sync_v2'); assert.equal(executeA.phase, 'execute'); assert.equal(executeB.phase, 'execute'); assert.equal(mod.rooms.get(episodeRid).episodeSyncV2.phase, 'executing'); assert.equal(mod.rooms.get(episodeRid).mediaState, null, 'canonical playing is not committed before every execute ACK'); let duplicateExecute = false; try { await w(episodeB, 'episode_sync_v2', 300); duplicateExecute = true; } catch { /* expected */ } assert.equal(duplicateExecute, false, 'execute is emitted exactly once'); s(episodeA, 'episode_sync_v2', { phase: 'executed', transactionId: episodeTxId }); await delay(50); assert.deepEqual(mod.rooms.get(episodeRid).episodeSyncV2.executedPeers, ['episode-a']); assert.equal(mod.rooms.get(episodeRid).mediaState, null, 'one execute ACK cannot complete the room'); s(episodeA, 'episode_sync_v2', { phase: 'executed', transactionId: episodeTxId }); await delay(50); assert.deepEqual(mod.rooms.get(episodeRid).episodeSyncV2.executedPeers, ['episode-a'], 'duplicate execute ACK is idempotent'); s(episodeB, 'episode_sync_v2', { phase: 'executed', transactionId: episodeTxId }); const completeA = await w(episodeA, 'episode_sync_v2'); const completeB = await w(episodeB, 'episode_sync_v2'); assert.equal(completeA.phase, 'complete'); assert.equal(completeB.phase, 'complete'); assert.deepEqual(completeA.executedPeers, ['episode-a', 'episode-b']); assert.equal(completeA.remainingMs, 0); assert.equal(mod.rooms.get(episodeRid).episodeSyncV2, null); assert.equal(mod.rooms.get(episodeRid).mediaState.playbackState, 'playing'); assert.equal(mod.rooms.get(episodeRid).mediaState.currentTime, 0); s(episodeA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S01E07', expectedEpisodeId: '