diff --git a/docs/PROTOCOL.md b/docs/PROTOCOL.md index b26bcf8..229f69a 100644 --- a/docs/PROTOCOL.md +++ b/docs/PROTOCOL.md @@ -54,7 +54,7 @@ Payload: "password": "string, max 128, optional", "tabTitle": "string, max 100, optional", "mediaTitle": "string, max 100, optional", - "clientCapabilities": ["chat-v1", "media-state-v1"], + "clientCapabilities": ["chat-v1", "media-state-v1", "episode-sync-v2"], "protocolVersion": "string, max 16" } ``` @@ -84,7 +84,8 @@ Payload: "controlMode": "everyone | host-only", "controllers": ["peerId"], "mediaState": "canonical media state object or null", - "capabilities": ["host-control", "co-host", "chat", "chat-v1", "media-state-v1"] + "episodeSyncV2": "active transaction object for a frozen participant, or null", + "capabilities": ["host-control", "co-host", "chat", "chat-v1", "media-state-v1", "episode-sync-v2"] } ``` @@ -526,5 +527,42 @@ If sender and target are still in the same room, the relay emits: - `co-host` - `chat` - `chat-v1` +- `media-state-v1` +- `episode-sync-v2` Clients should treat a missing or unknown capabilities list as unsupported. + +## Episode Sync v2 + +Relays advertise `"episode-sync-v2"`. Updated clients use the additive +`episode_sync_v2` event without changing `PROTOCOL_VERSION`; automatic episode +sync is skipped safely when the relay omits the capability or any current room +peer did not announce it. Manual Force Sync remains on the legacy event family. + +The relay creates the transaction ID, freezes non-desynced participants, owns +deadlines, and is the only component that advances: + +```text +start -> lobby/loading -> prepare -> execute + \-> cancel +``` + +Client phases are `start`, `loaded`, `prepared`, `failed`, and `cancel`. Relay +phases are `lobby`, `prepare`, `execute`, and `cancel`. Every post-start frame is +correlated by `transactionId`; duplicate or stale frames are idempotently +ignored. No participant is pre-marked loaded. `execute` is emitted exactly once +only after every frozen participant reports `loaded`, then pauses, seeks to +0:00, reaches `readyState >= 3`, and remains on the same player/title in paused, +non-seeking state for `EPISODE_SYNC_V2_STABILITY_MS`. + +Any timeout, failed player action, participant departure, manual media command, +room/target change, or explicit cancellation produces `cancel`; v2 never +executes after a timeout. Peers resume only when that transaction paused a +previously playing player and no newer local user action superseded it. Peers +joining after `start` are excluded from the frozen barrier and receive no v2 +frames for that transaction. + +Legacy episode events remain accepted. A new relay binds their PREPARE, EXECUTE, +and CANCEL to the accepted lobby initiator, limiting duplicate old-client wire +actions; an unmodified old non-initiator can still run its own local timeout, so +full v2 guarantees require capable extensions on every peer. diff --git a/extension/async-session-guards.test.mjs b/extension/async-session-guards.test.mjs index 44eb2f4..1c60d80 100644 --- a/extension/async-session-guards.test.mjs +++ b/extension/async-session-guards.test.mjs @@ -29,7 +29,7 @@ describe('async room-session guards', () => { it('revalidates ROOM_DATA after every asynchronous join boundary', () => { const roomData = sourceBetween('case EVENTS.ROOM_DATA:', 'case EVENTS.CONTROL_MODE:'); expect(roomData.match(/currentRoom\?\.roomId !== data\.roomId/g)?.length).toBeGreaterThanOrEqual(2); - expect(roomData).toContain('const authoritativeLobby = normalizeEpisodeLobby('); + expect(roomData).toContain('const authoritativeLobby = authoritativeEpisodeSyncV2 ? null : normalizeEpisodeLobby('); }); it('does not return chat context after its room or target changed', () => { diff --git a/extension/background.js b/extension/background.js index b376def..632fc7b 100644 --- a/extension/background.js +++ b/extension/background.js @@ -1,7 +1,7 @@ import { EVENTS, ERROR_CODES, CONTROL_MODES, CAPABILITIES, PROTOCOL_VERSION, OFFICIAL_SERVER_URL, OFFICIAL_SERVER_TOKEN, EPISODE_LOBBY_TIMEOUT, FORCE_SYNC_TIMEOUT, HEARTBEAT_INTERVAL } from './shared/constants.js'; import { generateUsername } from './shared/names.js'; import { loadLocale, getMessage, getSystemLanguage } from './i18n.js'; -import { sameEpisode, extractEpisodeId } from './episode-utils.js'; +import { sameEpisode, sameEpisodeStrict, extractEpisodeId } from './episode-utils.js'; import { applyTitlePrivacyToPayload, sanitizeSharedTitle, sanitizeTabTitle, normalizeSendTabTitle, normalizeTitlePrivacyMode } from './title-privacy.js'; import { initTabManager } from './modules/tab-manager.js'; import { clearChatKeyCache, decryptChatMessage, encryptChatMessage, generateChatSecret, validateChatSecret } from './chat-crypto.js'; @@ -290,7 +290,8 @@ function serverSupportsChat() { } const CLIENT_CAPABILITIES = Object.freeze([ CAPABILITIES.CHAT_V1, - CAPABILITIES.MEDIA_STATE_V1 + CAPABILITIES.MEDIA_STATE_V1, + CAPABILITIES.EPISODE_SYNC_V2 ]); function persistCanonicalMediaRecovery() { @@ -462,7 +463,7 @@ function ensureState() { 'forceSyncDeadline', 'reconnectFailed', 'reconnectStartTime', 'reconnectAttempts', 'currentTabId', 'currentTabTitle', 'currentTargetFrameId', 'currentTargetDocumentId', 'currentTargetHasVideo', 'selectedTabId', 'selectedTabTitle', 'selectionErrorTabId', 'selectionErrorMessage', - 'episodeLobby', 'localSeq', 'lastSeqBySender', 'expectedAcksCount', 'roomIdleSince', 'lastContentHeartbeatAt', + 'episodeLobby', 'episodeSyncV2', 'localSeq', 'lastSeqBySender', 'expectedAcksCount', 'roomIdleSince', 'lastContentHeartbeatAt', 'hcmDesynced', 'chatActivityTimeline', 'canonicalMediaRecovery' ], (data) => { // A late callback must not resurrect a room, queue or canonical @@ -583,6 +584,20 @@ function ensureState() { } } + const restoredEpisodeSyncV2 = currentRoom + ? normalizeEpisodeSyncV2( + data.episodeSyncV2 || currentRoom.episodeSyncV2, + new Set(currentRoom.peers.map(candidate => candidate.peerId)) + ) + : null; + if (restoredEpisodeSyncV2) { + episodeSyncV2 = restoredEpisodeSyncV2; + if (episodeLobbyTimeout) clearTimeout(episodeLobbyTimeout); + episodeLobbyTimeout = null; + episodeLobby = null; + currentRoom.activeLobby = null; + } + if (Number.isSafeInteger(data.localSeq) && data.localSeq >= 0) localSeq = data.localSeq; eventQueue = normalizePersistedEventQueue( [...eventQueue, ...(Array.isArray(data.eventQueue) ? data.eventQueue : [])], @@ -661,6 +676,7 @@ let forceSyncTimeout = null; // Episode Auto-Sync Lobby let episodeLobby = null; // { expectedTitle, initiatorPeerId, readyPeers: [], createdAt } let episodeLobbyTimeout = null; +let episodeSyncV2 = null; // --- Storage Utils --- @@ -1192,6 +1208,46 @@ async function clearTargetSelectionForLifecycle({ return true; } +function normalizeEpisodeSyncV2(value, allowedPeerIds = null) { + if (!value || typeof value !== 'object' || Array.isArray(value)) return null; + const transactionId = typeof value.transactionId === 'string' + ? value.transactionId.substring(0, 64) + : ''; + const phase = value.phase === 'lobby' || value.phase === 'prepare' ? value.phase : ''; + const expectedTitle = typeof value.expectedTitle === 'string' + ? value.expectedTitle.substring(0, 100) + : ''; + const initiatorPeerId = typeof value.initiatorPeerId === 'string' + ? value.initiatorPeerId.substring(0, 16) + : ''; + if (!transactionId || !phase || !expectedTitle || !initiatorPeerId + || !Array.isArray(value.participants) + || !Array.isArray(value.loadedPeers) + || !Array.isArray(value.preparedPeers)) return null; + const normalizePeers = peers => [...new Set(peers + .filter(candidate => typeof candidate === 'string' && candidate) + .map(candidate => candidate.substring(0, 16)))]; + const participants = normalizePeers(value.participants); + const participantSet = new Set(participants); + const loadedPeers = normalizePeers(value.loadedPeers).filter(candidate => participantSet.has(candidate)); + const preparedPeers = normalizePeers(value.preparedPeers).filter(candidate => participantSet.has(candidate)); + if (participants.length < 2 + || !participantSet.has(initiatorPeerId) + || (allowedPeerIds && participants.some(candidate => !allowedPeerIds.has(candidate)))) return null; + return { + transactionId, + phase, + expectedTitle, + initiatorPeerId, + participants, + loadedPeers, + preparedPeers, + createdAt: Number.isFinite(value.createdAt) ? value.createdAt : Date.now(), + deadlineAt: Number.isFinite(value.deadlineAt) ? value.deadlineAt : null, + revision: Number.isSafeInteger(value.revision) && value.revision > 0 ? value.revision : 1 + }; +} + function clearTargetTabForIdle(expectedTabId = null, expectedGeneration = null) { return clearTargetSelectionForLifecycle({ expectedTabId, @@ -1212,6 +1268,8 @@ async function performRoomSessionTeardown({ notifyServer = false, reason = 'Left // Stop room-specific polling before the content script itself is removed. // Every terminal room exit must pass through the exact target identity while // it is still available, regardless of who initiated the exit. + if (notifyServer) cancelEpisodeSyncV2('room_exit'); + else clearEpisodeSyncV2State({ reason: 'room_exit' }); clearEpisodeLobbyState(); currentRoom = null; clearCanonicalMediaRecovery(); @@ -1249,6 +1307,7 @@ async function performRoomSessionTeardown({ notifyServer = false, reason = 'Left forceSyncDeadline: null, expectedAcksCount: 0, episodeLobby: null, + episodeSyncV2: null, hcmDesynced: false, reconnectFailed: false, reconnectAttempts: 0, @@ -1845,7 +1904,7 @@ async function flushEventQueue(replaySettingsOverride = undefined) { } applyQueuedRoomPolicy(flushRoomId, { canControl: !(controlMode === CONTROL_MODES.HOST_ONLY && hostPeerId && !amController()), - activeLobby: !!episodeLobby, + activeLobby: !!(episodeLobby || episodeSyncV2), desynced: hcmDesynced, authoritativeLobby: !!currentRoom?.activeLobby }, 'Queue replay authority changed'); @@ -2020,9 +2079,9 @@ async function performPendingCanonicalMediaStateApply() { addLog(`Canonical media state r${mediaState.revision} skipped: local guest is desynced`, 'info'); return { status: 'ignored_desynced' }; } - if (episodeLobby) { + if (episodeLobby || episodeSyncV2) { markCanonicalMediaStateHandled(roomId, mediaState.revision); - addLog(`Canonical media state r${mediaState.revision} skipped: episode lobby is active`, 'info'); + addLog(`Canonical media state r${mediaState.revision} skipped: episode sync is active`, 'info'); return { status: 'ignored_episode_lobby' }; } @@ -2193,6 +2252,29 @@ async function handleServerEvent(event, data, expectedConnectionGeneration = con currentRoom.peers = []; } + const roomPeerIds = new Set(currentRoom.peers.map(candidate => candidate.peerId)); + const authoritativeEpisodeSyncV2 = serverSupports(CAPABILITIES.EPISODE_SYNC_V2) + ? normalizeEpisodeSyncV2(data.episodeSyncV2, roomPeerIds) + : null; + if (data.episodeSyncV2 && !authoritativeEpisodeSyncV2) { + addLog('Ignored malformed Episode Sync v2 transaction in ROOM_DATA', 'warn'); + } + if (authoritativeEpisodeSyncV2) { + const shouldNotifyContent = !episodeSyncV2 + || episodeSyncV2.transactionId !== authoritativeEpisodeSyncV2.transactionId + || episodeSyncV2.phase !== authoritativeEpisodeSyncV2.phase; + if (episodeLobby) clearEpisodeLobbyState(); + episodeSyncV2 = authoritativeEpisodeSyncV2; + currentRoom.episodeSyncV2 = authoritativeEpisodeSyncV2; + persistEpisodeSyncV2(); + broadcastLobbyUpdate(); + if (shouldNotifyContent) sendEpisodeSyncV2ToContent().catch(() => {}); + } else if (episodeSyncV2) { + clearEpisodeSyncV2State({ reason: 'relay_state_ended' }); + } else { + currentRoom.episodeSyncV2 = null; + } + // ROOM_DATA is authoritative for an already-active server lobby, // but a locally-created offline lobby has not reached the relay // yet and must remain owned by its initiator until queued replay. @@ -2200,7 +2282,7 @@ async function handleServerEvent(event, data, expectedConnectionGeneration = con entry?.event === EVENTS.EPISODE_LOBBY && (!entry.roomId || entry.roomId === data.roomId) ); - const authoritativeLobby = normalizeEpisodeLobby( + const authoritativeLobby = authoritativeEpisodeSyncV2 ? null : normalizeEpisodeLobby( data.activeLobby, Date.now(), new Set(currentRoom.peers.map(candidate => candidate.peerId)) @@ -2274,12 +2356,12 @@ async function handleServerEvent(event, data, expectedConnectionGeneration = con && !amController(); const queuePolicy = applyQueuedRoomPolicy(data.roomId, { canControl: !lostRoomAuthority, - activeLobby: !!episodeLobby, + activeLobby: !!(episodeLobby || episodeSyncV2), desynced: hcmDesynced, - authoritativeLobby: !!authoritativeLobby + authoritativeLobby: !!(authoritativeLobby || authoritativeEpisodeSyncV2) }, lostRoomAuthority ? 'Host Control role changed while offline' - : (episodeLobby ? 'Active Episode Lobby takes precedence' : 'Reconnect queue policy')); + : ((episodeLobby || episodeSyncV2) ? 'Active Episode Sync takes precedence' : 'Reconnect queue policy')); await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent); await flushEventQueue(replaySettings); @@ -2608,6 +2690,76 @@ async function handleServerEvent(event, data, expectedConnectionGeneration = con } } break; + case EVENTS.EPISODE_SYNC_V2: { + if (!serverSupports(CAPABILITIES.EPISODE_SYNC_V2)) { + addLog('Ignored Episode Sync v2 event from relay without advertised capability', 'warn'); + break; + } + const phase = data.phase; + if (phase === 'cancel' && !data.transactionId) { + addLog(`Episode Sync v2 unavailable: ${data.reason || 'rejected'}`, 'warn'); + break; + } + if (phase === 'lobby' || phase === 'prepare') { + const activePeerIds = new Set((currentRoom?.peers || []).map(candidate => + typeof candidate === 'object' ? candidate.peerId : candidate + )); + const incoming = normalizeEpisodeSyncV2(data, activePeerIds); + if (!incoming + || !incoming.participants.includes(peerId) + || data.senderId !== incoming.initiatorPeerId) { + addLog('Ignored malformed Episode Sync v2 state', 'warn'); + break; + } + if (episodeSyncV2 + && episodeSyncV2.transactionId === incoming.transactionId + && incoming.revision < episodeSyncV2.revision) { + addLog(`Ignored stale Episode Sync v2 revision ${incoming.revision}`, 'warn'); + break; + } + const shouldNotifyContent = !episodeSyncV2 + || episodeSyncV2.transactionId !== incoming.transactionId + || episodeSyncV2.phase !== incoming.phase; + if (episodeSyncV2 && episodeSyncV2.transactionId !== incoming.transactionId) { + clearEpisodeSyncV2State({ reason: 'transaction_replaced' }); + } + if (episodeLobby) clearEpisodeLobbyState(); + episodeSyncV2 = incoming; + if (currentRoom) currentRoom.episodeSyncV2 = incoming; + persistEpisodeSyncV2(); + broadcastLobbyUpdate(); + if (shouldNotifyContent) sendEpisodeSyncV2ToContent().catch(() => {}); + addLog(`Episode Sync v2 ${incoming.phase}: "${incoming.expectedTitle}" (${incoming.transactionId.substring(0, 8)})`, 'info'); + break; + } + if ((phase === 'execute' || phase === 'cancel') + && episodeSyncV2 + && data.transactionId === episodeSyncV2.transactionId) { + const completed = episodeSyncV2; + if (phase === 'execute') { + sendMessageToCurrentContent({ + type: 'EPISODE_SYNC_V2', + transaction: { ...completed, phase: 'execute', targetTime: 0 } + }).catch(() => {}); + if (currentRoom && Array.isArray(currentRoom.peers)) { + currentRoom.peers.forEach(candidate => { + if (candidate && typeof candidate === 'object') { + candidate.playbackState = 'playing'; + candidate.currentTime = 0; + candidate.lastReactiveUpdate = Date.now(); + } + }); + chrome.runtime.sendMessage({ type: 'PEER_UPDATE', peers: currentRoom.peers }).catch(() => {}); + } + clearEpisodeSyncV2State({ notifyContent: false, reason: 'executed' }); + addLog(`Episode Sync v2 executed for "${completed.expectedTitle}"`, 'success'); + } else { + clearEpisodeSyncV2State({ reason: data.reason || 'cancelled' }); + addLog(`Episode Sync v2 cancelled: ${data.reason || 'cancelled'}`, 'warn'); + } + } + break; + } case EVENTS.EPISODE_LOBBY: if (typeof data.senderId === 'string' && typeof data.expectedTitle === 'string' @@ -2768,11 +2920,13 @@ function executeForceSync() { } function completeForceSyncBeforeTargetChange(nextTabId) { - if (!isForceSyncInitiator) return; const selectedTabId = normalizeTabId(currentTabId); const normalizedNextTabId = normalizeTabId(nextTabId); if (selectedTabId !== null && selectedTabId === normalizedNextTabId) return; + if (episodeSyncV2) cancelEpisodeSyncV2('target_changed'); + if (!isForceSyncInitiator) return; + addLog('Finishing Force Sync before target change', 'info'); executeForceSync(); } @@ -2782,8 +2936,66 @@ function persistEpisodeLobby() { if (storageInitialized) chrome.storage.session.set({ episodeLobby }); } +function episodeLobbyForUi() { + if (!episodeSyncV2) return episodeLobby; + return { + expectedTitle: episodeSyncV2.expectedTitle, + initiatorPeerId: episodeSyncV2.initiatorPeerId, + readyPeers: episodeSyncV2.phase === 'prepare' + ? [...episodeSyncV2.preparedPeers] + : [...episodeSyncV2.loadedPeers], + createdAt: episodeSyncV2.createdAt, + mode: 'v2', + phase: episodeSyncV2.phase, + transactionId: episodeSyncV2.transactionId + }; +} + function broadcastLobbyUpdate() { - chrome.runtime.sendMessage({ type: 'LOBBY_UPDATE', lobby: episodeLobby }).catch(() => {}); + chrome.runtime.sendMessage({ type: 'LOBBY_UPDATE', lobby: episodeLobbyForUi() }).catch(() => {}); +} + +function persistEpisodeSyncV2() { + if (!storageInitialized) return; + chrome.storage.session.set({ episodeSyncV2, currentRoom }).catch(() => {}); +} + +function sendEpisodeSyncV2ToContent(transaction = episodeSyncV2) { + if (!transaction || !currentTabId) return Promise.resolve(); + return sendMessageToCurrentContent({ + type: 'EPISODE_SYNC_V2', + transaction: { ...transaction } + }); +} + +function clearEpisodeSyncV2State({ notifyContent = true, reason = 'cancelled' } = {}) { + const previous = episodeSyncV2; + episodeSyncV2 = null; + if (currentRoom) currentRoom.episodeSyncV2 = null; + persistEpisodeSyncV2(); + broadcastLobbyUpdate(); + if (notifyContent && previous && currentTabId) { + sendMessageToCurrentContent({ + type: 'EPISODE_SYNC_V2', + transaction: { + ...previous, + phase: 'cancel', + reason + } + }).catch(() => {}); + } +} + +function cancelEpisodeSyncV2(reason = 'cancelled') { + if (!episodeSyncV2) return false; + const transaction = episodeSyncV2; + emitLive(EVENTS.EPISODE_SYNC_V2, { + phase: 'cancel', + transactionId: transaction.transactionId, + reason + }); + clearEpisodeSyncV2State({ reason }); + return true; } function clearEpisodeLobbyState() { @@ -2808,7 +3020,7 @@ function cancelEpisodeLobby(reason) { supersedeCanonicalMediaRecovery('local episode_lobby_cancel'); // Broadcast cancellation to room - emit(EVENTS.EPISODE_LOBBY_CANCEL, { peerId }); + emitLive(EVENTS.EPISODE_LOBBY_CANCEL, { peerId }); clearEpisodeLobbyState(); addLog(`Episode lobby cancelled: ${reason} for "${title}"`, 'warn'); @@ -2869,7 +3081,7 @@ function executeEpisodeLobby() { expectedAcksCount: expectedAcksCount }); - const syncPayload = { targetTime: 0.0 }; + const syncPayload = { targetTime: 0.0, mediaTitle: title }; localSeq++; chrome.storage.session.set({ localSeq }); emit(EVENTS.FORCE_SYNC_PREPARE, { ...syncPayload, peerId, actionTimestamp: timestamp, seq: localSeq }); @@ -2885,6 +3097,7 @@ function executeEpisodeLobby() { function checkEpisodeLobbyCompletion() { if (!episodeLobby || !currentRoom) return; + if (episodeLobby.initiatorPeerId !== peerId) return; const peers = Array.isArray(currentRoom.peers) ? currentRoom.peers : []; // M-3: desynced peers (watching on their own) sit out the lobby — their content // script ignores EPISODE_LOBBY and never reports ready. Don't let them block @@ -2895,7 +3108,7 @@ function checkEpisodeLobbyCompletion() { .filter(Boolean)); const readyParticipatingCount = episodeLobby.readyPeers .filter(candidate => participatingPeerIds.has(candidate)).length; - if (readyParticipatingCount >= participatingPeerIds.size) { + if (participatingPeerIds.size > 0 && readyParticipatingCount >= participatingPeerIds.size) { executeEpisodeLobby(); } } @@ -4456,6 +4669,7 @@ function leaveOldRoomIfSwitching(newRoomId) { if (currentRoom && currentRoom.roomId !== newRoomId) { if (currentTabId) sendMessageToChatOverlay({ type: 'CHAT_RESET' }).catch(() => {}); addLog(`Switching rooms: leaving ${currentRoom.roomId} to join ${newRoomId}`, 'info'); + cancelEpisodeSyncV2('room_switch'); forceDisconnect(); currentRoom = null; clearCanonicalMediaRecovery(); @@ -4680,7 +4894,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { pendingTargetHost: pendingTarget?.host ?? null, pendingTargetOriginPattern: pendingTarget?.originPattern ?? null, pendingTargetRequestId: pendingTarget?.requestId ?? null, - episodeLobby: episodeLobby, + episodeLobby: episodeLobbyForUi(), reconnectAttempts, reconnectSlowMode: reconnectFailed, queuedLogicalEvents: eventQueue.length, @@ -4914,6 +5128,9 @@ async function handleAsyncMessage(message, sender, sendResponse) { checkEpisodeLobbyCompletion(); } } + if (hcmDesynced && episodeSyncV2?.participants.includes(peerId)) { + cancelEpisodeSyncV2('desynced'); + } sendResponse({ status: 'ok' }); } else if (message.type === 'LEAVE_ROOM') { await endRoomSession({ notifyServer: true, reason: 'Left Room' }); @@ -5254,7 +5471,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { } else { localSeq++; chrome.storage.session.set({ localSeq }); - emit(EVENTS.FORCE_SYNC_ACK, { peerId, seq: localSeq }); + emitLive(EVENTS.FORCE_SYNC_ACK, { peerId, seq: localSeq }); } sendResponse({ status: 'ok' }); } else if (message.type === 'CMD_ACK') { @@ -5441,12 +5658,8 @@ async function handleAsyncMessage(message, sender, sendResponse) { return; } - // Host Control Mode: a gated guest must NOT initiate an episode lobby — the - // server drops the guest's EPISODE_LOBBY, so the lobby would never complete - // and the guest would self-pause (PAUSE_FOR_LOBBY) into a 60s freeze. In - // host-only the controllers (owner + co-hosts) drive episode sync; a plain - // guest just follows / snaps back. Use amController() for parity with the - // CONTENT_EVENT gate and the server's controllers-based check. + // Host Control Mode: only controllers may initiate the relay-owned + // transaction. Plain guests wait for the controller's accepted lobby. if (controlMode === CONTROL_MODES.HOST_ONLY && !amController()) { addLog(`Episode change ("${lobbyTitle}") — host-only guest, not creating a lobby (controller drives).`, 'info'); sendResponse({ status: 'host_only_guest_skip' }); @@ -5463,57 +5676,27 @@ async function handleAsyncMessage(message, sender, sendResponse) { return; } - // If lobby already exists for this title, just mark self ready - if (episodeLobby && sameEpisode(episodeLobby.expectedTitle, lobbyTitle)) { - if (!episodeLobby.readyPeers.includes(peerId)) { - episodeLobby.readyPeers.push(peerId); - persistEpisodeLobby(); - broadcastLobbyUpdate(); - emit(EVENTS.EPISODE_READY, { - peerId, - title: lobbyTitle, - expectedTitle: episodeLobby.expectedTitle - }); - checkEpisodeLobbyCompletion(); - } - sendResponse({ status: 'ready_sent' }); + // Automatic episode sync v2 is relay-owned. Never self-pause or fall + // back to the legacy client-owned lobby: on an old/mixed relay that path + // can create multiple Force Sync initiators. Manual Force Sync remains + // available and unchanged. + if (!serverSupports(CAPABILITIES.EPISODE_SYNC_V2)) { + addLog(`Episode change ("${lobbyTitle}") — relay lacks Episode Sync v2; automatic sync skipped safely.`, 'warn'); + sendResponse({ status: 'episode_sync_v2_unsupported' }); return; } - - // Cancel any existing lobby for a different episode - if (episodeLobby) clearEpisodeLobbyState(); - - // Create new lobby - supersedeCanonicalMediaRecovery('local episode_lobby', EVENTS.PAUSE); - episodeLobby = { - expectedTitle: lobbyTitle, - initiatorPeerId: peerId, - readyPeers: [peerId], // We are already ready - createdAt: Date.now() - }; - persistEpisodeLobby(); - broadcastLobbyUpdate(); - addLog(`Episode lobby created: "${lobbyTitle}"`, 'info'); - - // Tell content script to pause the video and start polling - // (This is the only place we pause — after confirming the feature is enabled) - if (sender.tab && sender.tab.id) { - sendMessageToFrame(sender.tab.id, sender.frameId, { - type: 'PAUSE_FOR_LOBBY', - expectedTitle: lobbyTitle - }, null, sender.documentId).catch(() => {}); + if (episodeSyncV2 && sameEpisode(episodeSyncV2.expectedTitle, lobbyTitle)) { + sendResponse({ status: 'transaction_active', transactionId: episodeSyncV2.transactionId }); + return; } - - // Broadcast to room - emit(EVENTS.EPISODE_LOBBY, { peerId, expectedTitle: lobbyTitle }); - - // Start timeout (Q1: Option B — cancel on timeout) - episodeLobbyTimeout = setTimeout(() => cancelEpisodeLobby('Timeout — not all peers loaded the episode'), EPISODE_LOBBY_TIMEOUT); - - // Immediate check — maybe we're the only one in the room - checkEpisodeLobbyCompletion(); - - sendResponse({ status: 'lobby_created' }); + if (episodeSyncV2) cancelEpisodeSyncV2('new_episode'); + if (!emitLive(EVENTS.EPISODE_SYNC_V2, { phase: 'start', expectedTitle: lobbyTitle })) { + addLog(`Episode change ("${lobbyTitle}") — not connected; automatic sync was not queued.`, 'warn'); + sendResponse({ status: 'episode_sync_v2_offline' }); + return; + } + addLog(`Episode Sync v2 requested: "${lobbyTitle}"`, 'info'); + sendResponse({ status: 'episode_sync_v2_requested' }); } else if (message.type === 'EPISODE_READY_LOCAL') { if (sender.tab) { if (!isCurrentContentSender(sender)) { @@ -5542,7 +5725,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { lobby.readyPeers.push(peerId); persistEpisodeLobby(); broadcastLobbyUpdate(); - emit(EVENTS.EPISODE_READY, { + emitLive(EVENTS.EPISODE_READY, { peerId, title: readyTitle, expectedTitle: lobby.expectedTitle @@ -5552,6 +5735,35 @@ async function handleAsyncMessage(message, sender, sendResponse) { } } sendResponse({ status: 'ok' }); + } else if (message.type === 'EPISODE_SYNC_V2_LOCAL') { + if (sender.tab && !isCurrentContentSender(sender)) { + sendResponse({ status: 'ignored_stale_target' }); + return; + } + const transaction = episodeSyncV2; + const localPhase = message.phase; + const expectedLocalPhase = localPhase === 'loaded' + ? 'lobby' + : (localPhase === 'prepared' ? 'prepare' : null); + if (!transaction + || message.transactionId !== transaction.transactionId + || (expectedLocalPhase && transaction.phase !== expectedLocalPhase) + || !['loaded', 'prepared', 'failed'].includes(localPhase)) { + sendResponse({ status: 'ignored_stale_transaction' }); + return; + } + const localTitle = message.payload?.title; + if (localPhase !== 'failed' + && (typeof localTitle !== 'string' || !sameEpisodeStrict(localTitle, transaction.expectedTitle))) { + sendResponse({ status: 'ignored_episode_mismatch' }); + return; + } + const sent = emitLive(EVENTS.EPISODE_SYNC_V2, { + phase: localPhase, + transactionId: transaction.transactionId, + reason: typeof message.reason === 'string' ? message.reason.substring(0, 32) : undefined + }); + sendResponse({ status: sent ? 'sent' : 'offline' }); } else if (message.type === 'TITLE_PRIVACY_CHANGED') { const privacyRoomId = currentRoom?.roomId || null; const privacyLobby = episodeLobby; @@ -5566,6 +5778,7 @@ async function handleAsyncMessage(message, sender, sendResponse) { cancelEpisodeLobby('Title privacy changed'); } } + if (episodeSyncV2) cancelEpisodeSyncV2('title_privacy_changed'); if (currentRoom) { const sharedTitles = getSharedTitleFields(settings); emit(EVENTS.PEER_STATUS, { @@ -5650,13 +5863,18 @@ async function handleAsyncMessage(message, sender, sendResponse) { } requestCanonicalMediaRecoveryAttempt(); // Content script re-injected, check if there's an active lobby - if (episodeLobby) { + if (episodeSyncV2) { + sendResponse({ episodeSyncV2: { ...episodeSyncV2 }, lobbyActive: false }); + } else if (episodeLobby) { sendResponse({ lobbyActive: true, expectedTitle: episodeLobby.expectedTitle }); } else { sendResponse({ lobbyActive: false }); } } else if (message.type === 'CANCEL_EPISODE_LOBBY') { - if (episodeLobby) { + if (episodeSyncV2) { + cancelEpisodeSyncV2('cancelled_by_user'); + sendResponse({ status: 'ok' }); + } else if (episodeLobby) { cancelEpisodeLobby('Cancelled by user'); sendResponse({ status: 'ok' }); } else { diff --git a/extension/canonical-media-state-background.test.mjs b/extension/canonical-media-state-background.test.mjs index 09af527..7eccb4d 100644 --- a/extension/canonical-media-state-background.test.mjs +++ b/extension/canonical-media-state-background.test.mjs @@ -81,7 +81,7 @@ describe('canonical ROOM_DATA recovery contract', () => { const apply = functionBody(backgroundSource, 'performPendingCanonicalMediaStateApply', 'tryApplyPendingCanonicalMediaState'); const roomData = functionBody(backgroundSource, 'handleCanonicalRoomData', 'handleServerEvent'); expect(apply).toContain('if (hcmDesynced)'); - expect(apply).toContain('if (episodeLobby)'); + expect(apply).toContain('if (episodeLobby || episodeSyncV2)'); expect(roomData).toContain('if (hasPendingLocalIntent)'); expect(backgroundSource).toContain('awaitingRoomData = true'); expect(backgroundSource).toContain('await handleCanonicalRoomData(data, queuePolicy.hasPendingLocalIntent)'); diff --git a/extension/content.js b/extension/content.js index b30eda3..c8b2a15 100644 --- a/extension/content.js +++ b/extension/content.js @@ -84,6 +84,7 @@ EPISODE_READY: "episode_ready" }; const MAX_MEDIA_TIME = 86400; + const EPISODE_SYNC_V2_STABILITY_MS = 1000; // --- SHARED_EVENTS_INJECT_END --- // Suppresses native event reporting after a programmatic action. @@ -334,6 +335,9 @@ let episodeTransitionDebounce = null; let _pendingLobbyTitle = null; // Title we're waiting to match (from remote lobby) let lobbyPollTimer = null; + let episodeSyncV2State = null; + let episodeSyncV2PollTimer = null; + let episodeSyncV2Generation = 0; let _autoSyncEnabled = true; // Cached setting, updated via storage.onChanged let _audioSettings = null; let _audioProcessingAllowed = true; @@ -1104,6 +1108,28 @@ if (idA || idB) return false; return titleA === titleB; } + + function episodeContext(title) { + if (!title || typeof title !== 'string') return ''; + return title + .replace(/S(?:eason\s*)?\d+[^a-zA-Z0-9]*E(?:pisode\s*)?\d+/ig, ' ') + .replace(/(?:Episode|Folge|Ep\.?)\s*\d+|#\s*\d+/ig, ' ') + .normalize('NFKC') + .toLowerCase() + .replace(/[^\p{L}\p{N}]+/gu, ' ') + .trim(); + } + + // Transactional episode sync is deliberately stricter than ordinary playback + // filtering: if both peers expose contextual text (episode/show name), require + // it to agree so two unrelated S01E06 videos cannot cross-sync. Privacy-reduced + // S/E-only titles still fall back to the canonical episode ID. + function sameEpisodeStrict(titleA, titleB) { + if (!sameEpisode(titleA, titleB)) return false; + const contextA = episodeContext(titleA); + const contextB = episodeContext(titleB); + return !contextA || !contextB || contextA === contextB; + } // --- SHARED_EPISODE_UTILS_INJECT_END --- // Returns true only when we are CERTAIN the episodes differ. @@ -1215,6 +1241,244 @@ } } + function isEpisodeSyncV2Current(state) { + return !destroyed + && episodeSyncV2State === state + && state.generation === episodeSyncV2Generation; + } + + function stopEpisodeSyncV2Poll() { + if (episodeSyncV2PollTimer) { + clearInterval(episodeSyncV2PollTimer); + episodeSyncV2PollTimer = null; + } + } + + async function reportEpisodeSyncV2Local(state, phase, reason = null) { + if (!isEpisodeSyncV2Current(state) || state.reportInFlight) return false; + state.reportInFlight = true; + try { + const response = await runtimeMessage({ + type: 'EPISODE_SYNC_V2_LOCAL', + transactionId: state.transactionId, + phase, + reason, + payload: { title: getMediaTitle() } + }); + if (!isEpisodeSyncV2Current(state)) return false; + if (response?.status === 'sent') { + if (phase === 'loaded') state.loadedReported = true; + if (phase === 'prepared') state.preparedReported = true; + return true; + } + return false; + } finally { + if (isEpisodeSyncV2Current(state)) state.reportInFlight = false; + } + } + + function retryEpisodeSyncV2Report(state, phase, reason = null) { + if (!isEpisodeSyncV2Current(state) || state.reportRetryScheduled) return; + if (Number.isFinite(state.deadlineAt) && Date.now() >= state.deadlineAt) return; + state.reportRetryScheduled = true; + scheduleLifecycleTimeout(async () => { + if (!isEpisodeSyncV2Current(state)) return; + state.reportRetryScheduled = false; + const sent = await reportEpisodeSyncV2Local(state, phase, reason); + if (!sent) retryEpisodeSyncV2Report(state, phase, reason); + }, 750); + } + + async function clearEpisodeSyncV2Content({ resume = false, manualAction = false } = {}) { + const state = episodeSyncV2State; + if (!state) return; + episodeSyncV2Generation++; + episodeSyncV2State = null; + stopEpisodeSyncV2Poll(); + state.manualAction = state.manualAction || manualAction; + const video = state.video; + const mayResume = resume + && state.pausedByTransaction + && state.wasPlayingBeforePrepare + && !state.manualAction + && video + && video === findVideo() + && video.isConnected !== false + && sameEpisodeStrict(getMediaTitle(), state.expectedTitle) + && video.paused; + if (mayResume) { + const resumed = await tryMediaAction(EVENTS.PLAY); + if (!resumed) reportLog('Episode Sync v2: could not restore pre-transaction playback', 'warn'); + } + } + + function checkEpisodeSyncV2Loaded(state) { + if (!isEpisodeSyncV2Current(state) || state.phase !== 'lobby' || state.loadedReported) return; + const video = findVideo(); + const title = getMediaTitle(); + const matches = video + && title + && sameEpisodeStrict(title, state.expectedTitle) + && video.readyState >= 1 + && getSyncCurrentTime(video) !== null; + if (!matches) { + state.loadCandidateVideo = null; + state.loadCandidateAt = 0; + return; + } + if (state.loadCandidateVideo !== video) { + state.loadCandidateVideo = video; + state.loadCandidateAt = Date.now(); + return; + } + if (Date.now() - state.loadCandidateAt < 500) return; + reportEpisodeSyncV2Local(state, 'loaded').then(sent => { + if (sent) reportLog(`Episode Sync v2: loaded "${title}"`, 'success'); + }).catch(() => {}); + } + + function startEpisodeSyncV2Lobby(transaction) { + if (!transaction?.transactionId || !transaction.expectedTitle) return; + if (episodeSyncV2State?.transactionId === transaction.transactionId) { + episodeSyncV2State.phase = 'lobby'; + episodeSyncV2State.deadlineAt = Number.isFinite(transaction.deadlineAt) + ? transaction.deadlineAt + : episodeSyncV2State.deadlineAt; + return; + } + clearEpisodeSyncV2Content({ resume: true }).catch(() => {}); + episodeSyncV2Generation++; + episodeSyncV2State = { + transactionId: transaction.transactionId, + expectedTitle: transaction.expectedTitle, + phase: 'lobby', + generation: episodeSyncV2Generation, + loadedReported: false, + preparedReported: false, + reportInFlight: false, + reportRetryScheduled: false, + deadlineAt: Number.isFinite(transaction.deadlineAt) ? transaction.deadlineAt : null, + loadCandidateVideo: null, + loadCandidateAt: 0, + video: null, + wasPlayingBeforePrepare: false, + pausedByTransaction: false, + manualAction: false, + programmaticPausePending: false, + prepareStarted: false + }; + stopEpisodeSyncV2Poll(); + checkEpisodeSyncV2Loaded(episodeSyncV2State); + episodeSyncV2PollTimer = setInterval(() => { + if (episodeSyncV2State) checkEpisodeSyncV2Loaded(episodeSyncV2State); + }, 500); + } + + function waitForEpisodeSyncV2Stable(state, video, targetTime = 0, timeoutMs = 12000) { + return new Promise(resolve => { + const startedAt = Date.now(); + let stableSince = 0; + const timer = setInterval(() => { + const current = getSyncCurrentTime(video); + const ready = isEpisodeSyncV2Current(state) + && state.phase === 'prepare' + && video === findVideo() + && video.isConnected !== false + && sameEpisodeStrict(getMediaTitle(), state.expectedTitle) + && video.paused + && !video.seeking + && video.readyState >= 3 + && current !== null + && Math.abs(current - targetTime) < 1; + if (ready) { + if (!stableSince) stableSince = Date.now(); + if (Date.now() - stableSince >= EPISODE_SYNC_V2_STABILITY_MS) { + clearInterval(timer); + seekPollTimers.delete(timer); + resolve(true); + } + } else { + stableSince = 0; + } + if (!isEpisodeSyncV2Current(state) || Date.now() - startedAt >= timeoutMs) { + clearInterval(timer); + seekPollTimers.delete(timer); + resolve(false); + } + }, 100); + seekPollTimers.add(timer); + }); + } + + async function prepareEpisodeSyncV2(transaction) { + if (!transaction?.transactionId || !transaction.expectedTitle) return; + if (!episodeSyncV2State || episodeSyncV2State.transactionId !== transaction.transactionId) { + startEpisodeSyncV2Lobby({ ...transaction, phase: 'lobby' }); + } + const state = episodeSyncV2State; + if (!state || state.transactionId !== transaction.transactionId) return; + state.deadlineAt = Number.isFinite(transaction.deadlineAt) ? transaction.deadlineAt : state.deadlineAt; + if (state.prepareStarted) return; + stopEpisodeSyncV2Poll(); + state.phase = 'prepare'; + state.prepareStarted = true; + const video = findVideo(); + const currentTitle = getMediaTitle(); + if (!video || !currentTitle || !sameEpisodeStrict(currentTitle, state.expectedTitle)) { + if (!await reportEpisodeSyncV2Local(state, 'failed', 'episode_mismatch')) { + retryEpisodeSyncV2Report(state, 'failed', 'episode_mismatch'); + } + return; + } + state.video = video; + state.wasPlayingBeforePrepare = !video.paused; + if (!video.paused) { + state.programmaticPausePending = true; + const paused = await tryMediaAction(EVENTS.PAUSE); + if (!isEpisodeSyncV2Current(state)) return; + state.pausedByTransaction = paused === true; + if (!paused) { + state.programmaticPausePending = false; + if (!await reportEpisodeSyncV2Local(state, 'failed', 'pause_failed')) { + retryEpisodeSyncV2Report(state, 'failed', 'pause_failed'); + } + return; + } + } + const current = getSyncCurrentTime(video); + if (current === null || Math.abs(current) >= 0.25) { + const sought = await tryMediaAction(EVENTS.SEEK, { targetTime: 0 }); + if (!isEpisodeSyncV2Current(state)) return; + if (!sought) { + if (!await reportEpisodeSyncV2Local(state, 'failed', 'seek_failed')) { + retryEpisodeSyncV2Report(state, 'failed', 'seek_failed'); + } + return; + } + } + const stable = await waitForEpisodeSyncV2Stable(state, video, 0); + if (!isEpisodeSyncV2Current(state)) return; + if (!stable) { + if (!await reportEpisodeSyncV2Local(state, 'failed', 'stability_timeout')) { + retryEpisodeSyncV2Report(state, 'failed', 'stability_timeout'); + } + return; + } + if (await reportEpisodeSyncV2Local(state, 'prepared')) { + reportLog(`Episode Sync v2: prepared "${currentTitle}"`, 'success'); + } else { + retryEpisodeSyncV2Report(state, 'prepared'); + } + } + + function failEpisodeSyncV2ForManualAction(action) { + const state = episodeSyncV2State; + if (!state || state.phase !== 'prepare') return; + state.manualAction = true; + reportEpisodeSyncV2Local(state, 'failed', `user_${action}`).catch(() => {}); + clearEpisodeSyncV2Content({ manualAction: true }).catch(() => {}); + } + function getPlayerActionFixes() { return [ { @@ -1702,7 +1966,44 @@ } } - // Episode Auto-Sync: Lobby notification from background + if (message.type === 'EPISODE_SYNC_V2') { + const transaction = message.transaction; + if (!transaction?.transactionId || !transaction.phase) { + sendResponse({ status: 'invalid_transaction' }); + return true; + } + if (transaction.phase === 'lobby') { + reportLog(`Episode Sync v2 lobby: waiting for "${transaction.expectedTitle}"`, 'info'); + startEpisodeSyncV2Lobby(transaction); + } else if (transaction.phase === 'prepare') { + prepareEpisodeSyncV2(transaction).catch(error => { + reportLog(`Episode Sync v2 prepare failed: ${error.message}`, 'warn'); + }); + } else if (transaction.phase === 'execute') { + if (episodeSyncV2State?.transactionId === transaction.transactionId) { + const state = episodeSyncV2State; + const video = state.video || findVideo(); + const canExecute = video + && video === findVideo() + && sameEpisodeStrict(getMediaTitle(), state.expectedTitle); + clearEpisodeSyncV2Content({ resume: false }).catch(() => {}); + if (canExecute) { + Promise.resolve(tryMediaAction(EVENTS.PLAY)).then(applied => { + if (applied) scheduleProactiveHeartbeat(); + else reportLog('Episode Sync v2 execute could not start playback', 'warn'); + }).catch(() => {}); + } + } + } else if (transaction.phase === 'cancel') { + if (episodeSyncV2State?.transactionId === transaction.transactionId) { + clearEpisodeSyncV2Content({ resume: true }).catch(() => {}); + } + } + sendResponse({ status: 'ok' }); + return true; + } + + // Episode Auto-Sync: Legacy lobby notification from background if (message.type === 'EPISODE_LOBBY') { // Host Control Mode: a desynced guest is watching on their own and must // not join the lobby flow. Otherwise they'd pause on title match, report @@ -1962,6 +2263,19 @@ return; } if (action === EVENTS.PLAY && consumeCanonicalRestorePlaySuppression()) return; + if (episodeSyncV2State?.phase === 'prepare' + && action === EVENTS.PAUSE + && episodeSyncV2State.programmaticPausePending + && video === episodeSyncV2State.video + && video.paused) { + episodeSyncV2State.programmaticPausePending = false; + return; + } + + // An unsuppressed native action during PREPARE is a newer user intent. + // Fail the automation before relaying that action; the relay broadcasts + // CANCEL first and then applies this manual PLAY/PAUSE/SEEK normally. + failEpisodeSyncV2ForManualAction(action); if (action === EVENTS.PLAY || action === EVENTS.PAUSE) { cancelCanonicalMediaApply(action, video); @@ -2037,6 +2351,7 @@ closeAudioContext(); if (keepAlivePort) { try { keepAlivePort.disconnect(); } catch (_e) { /* ignore */ } keepAlivePort = null; } if (lobbyPollTimer) { clearInterval(lobbyPollTimer); lobbyPollTimer = null; } + stopEpisodeSyncV2Poll(); if (heartbeatTimeout) { clearTimeout(heartbeatTimeout); heartbeatTimeout = null; } if (proactiveHeartbeatTimeout) { clearTimeout(proactiveHeartbeatTimeout); proactiveHeartbeatTimeout = null; } // Drop any pending coalesced play/pause: we are tearing down and will @@ -2064,6 +2379,12 @@ connectKeepAlivePort(); schedulePeriodicHeartbeat(); scheduleProactiveHeartbeat(); + if (episodeSyncV2State?.phase === 'lobby') { + checkEpisodeSyncV2Loaded(episodeSyncV2State); + episodeSyncV2PollTimer = setInterval(() => { + if (episodeSyncV2State) checkEpisodeSyncV2Loaded(episodeSyncV2State); + }, 500); + } reportLog(`Page restored from cache — suppressing seeks for ${VISIBILITY_GRACE_MS / 1000}s`, 'warn'); } document.addEventListener('visibilitychange', handleVisibilityChange); @@ -2445,6 +2766,9 @@ if (heartbeatTimeout) { clearTimeout(heartbeatTimeout); heartbeatTimeout = null; } if (proactiveHeartbeatTimeout) { clearTimeout(proactiveHeartbeatTimeout); proactiveHeartbeatTimeout = null; } stopLobbyPoll(); + stopEpisodeSyncV2Poll(); + episodeSyncV2Generation++; + episodeSyncV2State = null; hcmDeferredSnapPending = false; observer.disconnect(); @@ -2501,7 +2825,13 @@ // Episode Auto-Sync: Boot recovery — check if background has an active lobby runtimeMessage({ type: 'CONTENT_BOOT' }, (res) => { if (destroyed || chrome.runtime.lastError) return; - if (res && res.lobbyActive && res.expectedTitle) { + if (res?.episodeSyncV2) { + if (res.episodeSyncV2.phase === 'prepare') { + prepareEpisodeSyncV2(res.episodeSyncV2).catch(() => {}); + } else { + startEpisodeSyncV2Lobby(res.episodeSyncV2); + } + } else if (res && res.lobbyActive && res.expectedTitle) { reportLog(`Boot: Active lobby detected for "${res.expectedTitle}"`, 'info'); startLobbyPoll(res.expectedTitle); } diff --git a/extension/episode-lobby-background.test.mjs b/extension/episode-lobby-background.test.mjs index de4a02b..97bf405 100644 --- a/extension/episode-lobby-background.test.mjs +++ b/extension/episode-lobby-background.test.mjs @@ -53,7 +53,7 @@ describe('episode lobby completion races', () => { it('validates restored and authoritative lobby state before using readyPeers', () => { expect(backgroundSource).toContain('function normalizeEpisodeLobby(value, fallbackCreatedAt = Date.now(), allowedPeerIds = null)'); expect(backgroundSource).toContain('data.currentRoom.activeLobby,'); - expect(backgroundSource).toContain('const authoritativeLobby = normalizeEpisodeLobby('); + expect(backgroundSource).toContain('const authoritativeLobby = authoritativeEpisodeSyncV2 ? null : normalizeEpisodeLobby('); expect(backgroundSource).toContain('new Set(currentRoom.peers.map(candidate => candidate.peerId))'); }); @@ -74,9 +74,10 @@ describe('episode lobby completion races', () => { it('counts only ready peers who still participate in lobby completion', () => { const completion = sourceBetween('function checkEpisodeLobbyCompletion()', 'function checkEpisodeLobbyPeerDeparture()'); + expect(completion).toContain('episodeLobby.initiatorPeerId !== peerId'); expect(completion).toContain('const participatingPeerIds = new Set(peers'); expect(completion).toContain('participatingPeerIds.has(candidate)'); - expect(completion).toContain('readyParticipatingCount >= participatingPeerIds.size'); + expect(completion).toContain('participatingPeerIds.size > 0 && readyParticipatingCount >= participatingPeerIds.size'); }); it('re-evaluates or cancels a lobby when the local peer enters solo mode', () => { diff --git a/extension/episode-sync-v2.test.mjs b/extension/episode-sync-v2.test.mjs new file mode 100644 index 0000000..b2e3444 --- /dev/null +++ b/extension/episode-sync-v2.test.mjs @@ -0,0 +1,105 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +import { CAPABILITIES, EVENTS, EPISODE_SYNC_V2_STABILITY_MS } from '../shared/constants.js'; + +const extensionDir = path.dirname(fileURLToPath(import.meta.url)); +const backgroundSource = fs.readFileSync(path.join(extensionDir, 'background.js'), 'utf8'); +const contentSource = fs.readFileSync(path.join(extensionDir, 'content.js'), 'utf8'); +const offlineSource = fs.readFileSync(path.join(extensionDir, 'offline-media-intent.js'), 'utf8'); +const buildSource = fs.readFileSync(path.join(extensionDir, '..', 'scripts', 'build-extension.cjs'), 'utf8'); + +function between(source, startNeedle, endNeedle) { + const start = source.indexOf(startNeedle); + const end = source.indexOf(endNeedle, start + startNeedle.length); + expect(start).toBeGreaterThan(-1); + expect(end).toBeGreaterThan(start); + return source.slice(start, end); +} + +describe('Episode Sync v2 extension contract', () => { + it('advertises the additive capability without changing the protocol', () => { + expect(CAPABILITIES.EPISODE_SYNC_V2).toBe('episode-sync-v2'); + expect(EVENTS.EPISODE_SYNC_V2).toBe('episode_sync_v2'); + expect(backgroundSource).toContain('CAPABILITIES.EPISODE_SYNC_V2'); + expect(backgroundSource).toContain('EVENTS.EPISODE_SYNC_V2'); + }); + + it('never queues v2 coordination or auto-falls back to a client-owned legacy lobby', () => { + expect(offlineSource).toContain('EVENTS.EPISODE_SYNC_V2'); + const episodeChanged = between( + backgroundSource, + "message.type === 'EPISODE_CHANGED'", + "message.type === 'EPISODE_READY_LOCAL'" + ); + expect(episodeChanged).toContain('serverSupports(CAPABILITIES.EPISODE_SYNC_V2)'); + expect(episodeChanged).toContain("emitLive(EVENTS.EPISODE_SYNC_V2, { phase: 'start'"); + expect(episodeChanged).not.toContain('PAUSE_FOR_LOBBY'); + expect(episodeChanged).not.toContain('emit(EVENTS.EPISODE_LOBBY'); + }); + + it('correlates content reports to the current transaction, phase, target and episode', () => { + const handler = between( + backgroundSource, + "message.type === 'EPISODE_SYNC_V2_LOCAL'", + "message.type === 'TITLE_PRIVACY_CHANGED'" + ); + expect(handler).toContain('message.transactionId !== transaction.transactionId'); + expect(handler).toContain('transaction.phase !== expectedLocalPhase'); + expect(handler).toContain("!isCurrentContentSender(sender)"); + expect(handler).toContain('!sameEpisodeStrict(localTitle, transaction.expectedTitle)'); + expect(handler).toContain('emitLive(EVENTS.EPISODE_SYNC_V2'); + }); + + it('requires the same player and episode to remain paused, seeked, buffered and stable', () => { + const stable = between( + contentSource, + 'function waitForEpisodeSyncV2Stable(', + 'async function prepareEpisodeSyncV2(' + ); + expect(stable).toContain('video === findVideo()'); + expect(stable).toContain('sameEpisodeStrict(getMediaTitle(), state.expectedTitle)'); + expect(stable).toContain('video.paused'); + expect(stable).toContain('!video.seeking'); + expect(stable).toContain('video.readyState >= 3'); + expect(stable).toContain('Math.abs(current - targetTime) < 1'); + expect(stable).toContain('Date.now() - stableSince >= EPISODE_SYNC_V2_STABILITY_MS'); + expect(EPISODE_SYNC_V2_STABILITY_MS).toBe(1000); + }); + + it('reports prepared only after awaited pause, seek and stable verification', () => { + const prepare = between( + contentSource, + 'async function prepareEpisodeSyncV2(', + 'function failEpisodeSyncV2ForManualAction(' + ); + const pause = prepare.indexOf('await tryMediaAction(EVENTS.PAUSE)'); + const seek = prepare.indexOf('await tryMediaAction(EVENTS.SEEK'); + const stable = prepare.indexOf('await waitForEpisodeSyncV2Stable'); + const prepared = prepare.indexOf("reportEpisodeSyncV2Local(state, 'prepared')"); + expect(pause).toBeGreaterThan(-1); + expect(seek).toBeGreaterThan(pause); + expect(stable).toBeGreaterThan(seek); + expect(prepared).toBeGreaterThan(stable); + }); + + it('restores playback only when this transaction paused a playing unchanged player', () => { + const clear = between( + contentSource, + 'async function clearEpisodeSyncV2Content(', + 'function checkEpisodeSyncV2Loaded(' + ); + expect(clear).toContain('state.pausedByTransaction'); + expect(clear).toContain('state.wasPlayingBeforePrepare'); + expect(clear).toContain('!state.manualAction'); + expect(clear).toContain('video === findVideo()'); + expect(clear).toContain('sameEpisodeStrict(getMediaTitle(), state.expectedTitle)'); + expect(contentSource).toContain('failEpisodeSyncV2ForManualAction(action)'); + }); + + it('injects the shared stability window into packaged content scripts', () => { + expect(buildSource).toContain('EPISODE_SYNC_V2_STABILITY_MS'); + expect(buildSource).toContain('episodeSyncStabilityVal'); + }); +}); diff --git a/extension/episode-utils.js b/extension/episode-utils.js index 625ef8a..5a9ca99 100644 --- a/extension/episode-utils.js +++ b/extension/episode-utils.js @@ -22,3 +22,25 @@ export function sameEpisode(titleA, titleB) { if (idA || idB) return false; return titleA === titleB; } + +function episodeContext(title) { + if (!title || typeof title !== 'string') return ''; + return title + .replace(/S(?:eason\s*)?\d+[^a-zA-Z0-9]*E(?:pisode\s*)?\d+/ig, ' ') + .replace(/(?:Episode|Folge|Ep\.?)\s*\d+|#\s*\d+/ig, ' ') + .normalize('NFKC') + .toLowerCase() + .replace(/[^\p{L}\p{N}]+/gu, ' ') + .trim(); +} + +// Transactional episode sync is deliberately stricter than ordinary playback +// filtering: if both peers expose contextual text (episode/show name), require +// it to agree so two unrelated S01E06 videos cannot cross-sync. Privacy-reduced +// S/E-only titles still fall back to the canonical episode ID. +export function sameEpisodeStrict(titleA, titleB) { + if (!sameEpisode(titleA, titleB)) return false; + const contextA = episodeContext(titleA); + const contextB = episodeContext(titleB); + return !contextA || !contextB || contextA === contextB; +} diff --git a/extension/episode-utils.test.mjs b/extension/episode-utils.test.mjs index 32edd24..d56db18 100644 --- a/extension/episode-utils.test.mjs +++ b/extension/episode-utils.test.mjs @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest'; -import { extractEpisodeId, sameEpisode } from './episode-utils.js'; +import { extractEpisodeId, sameEpisode, sameEpisodeStrict } from './episode-utils.js'; describe('episode title matching', () => { it.each([ @@ -53,4 +53,10 @@ describe('episode title matching', () => { ])('rejects different titles %j and %j', (left, right) => { expect(sameEpisode(left, right)).toBe(false); }); + + it('keeps transactional matching strict when both titles expose context', () => { + expect(sameEpisodeStrict('S1:E6 - Visiting Ours', 'S01E06 - Visiting Ours')).toBe(true); + expect(sameEpisodeStrict('S1:E6 - Visiting Ours', 'S1:E6 - Another Show')).toBe(false); + expect(sameEpisodeStrict('S1:E6', 'S01E06 - Visiting Ours')).toBe(true); + }); }); diff --git a/extension/offline-media-intent-background.test.mjs b/extension/offline-media-intent-background.test.mjs index 2fe02a7..9eb8018 100644 --- a/extension/offline-media-intent-background.test.mjs +++ b/extension/offline-media-intent-background.test.mjs @@ -44,10 +44,10 @@ describe('offline media intent background integration', () => { expect(flushIndex).toBeGreaterThan(-1); expect(policyIndex).toBeLessThan(canonicalIndex); expect(canonicalIndex).toBeLessThan(flushIndex); - expect(roomData).toContain('activeLobby: !!episodeLobby'); + expect(roomData).toContain('activeLobby: !!(episodeLobby || episodeSyncV2)'); expect(roomData).toContain('desynced: hcmDesynced'); expect(roomData).toContain('if (!authoritativeLobby && episodeLobby && !hasQueuedLocalLobby)'); - expect(roomData).toContain('authoritativeLobby: !!authoritativeLobby'); + expect(roomData).toContain('authoritativeLobby: !!(authoritativeLobby || authoritativeEpisodeSyncV2)'); }); it('clears queued room intent on failed join, leave and room switch paths', () => { diff --git a/extension/offline-media-intent.js b/extension/offline-media-intent.js index ac8d12b..ac936e2 100644 --- a/extension/offline-media-intent.js +++ b/extension/offline-media-intent.js @@ -5,7 +5,13 @@ export const MAX_LOGICAL_QUEUE_SIZE = 50; const MEDIA_EVENTS = new Set([EVENTS.PLAY, EVENTS.PAUSE, EVENTS.SEEK]); const KNOWN_EVENTS = new Set(Object.values(EVENTS)); -const STALE_OFFLINE_EVENTS = new Set([EVENTS.PING, EVENTS.PONG, EVENTS.PEER_STATUS, EVENTS.EVENT_ACK]); +const STALE_OFFLINE_EVENTS = new Set([ + EVENTS.PING, + EVENTS.PONG, + EVENTS.PEER_STATUS, + EVENTS.EVENT_ACK, + EVENTS.EPISODE_SYNC_V2 +]); const FORCE_SYNC_EVENTS = new Set([EVENTS.FORCE_SYNC_PREPARE, EVENTS.FORCE_SYNC_EXECUTE]); const HOST_GATED_EVENTS = new Set([ ...MEDIA_EVENTS, diff --git a/scripts/build-extension.cjs b/scripts/build-extension.cjs index 537f06f..e21e241 100644 --- a/scripts/build-extension.cjs +++ b/scripts/build-extension.cjs @@ -108,6 +108,7 @@ function copyExtensionFiles(targetDir, browserName) { const eventsMatch = constantsContent.match(/export const EVENTS\s*=\s*({[\s\S]+?});/); const heartbeatMatch = constantsContent.match(/export const HEARTBEAT_INTERVAL\s*=\s*(\d+);/); const maxMediaTimeMatch = constantsContent.match(/export const MAX_MEDIA_TIME\s*=\s*(\d+);/); + const episodeSyncStabilityMatch = constantsContent.match(/export const EPISODE_SYNC_V2_STABILITY_MS\s*=\s*(\d+);/); if (!eventsMatch) { throw new Error('CRITICAL: Could not find EVENTS object in shared/constants.js'); @@ -118,10 +119,14 @@ function copyExtensionFiles(targetDir, browserName) { if (!maxMediaTimeMatch) { throw new Error('CRITICAL: Could not find MAX_MEDIA_TIME in shared/constants.js'); } + if (!episodeSyncStabilityMatch) { + throw new Error('CRITICAL: Could not find EPISODE_SYNC_V2_STABILITY_MS in shared/constants.js'); + } const eventsObject = eventsMatch[1]; const heartbeatVal = heartbeatMatch[1]; const maxMediaTimeVal = maxMediaTimeMatch[1]; + const episodeSyncStabilityVal = episodeSyncStabilityMatch[1]; const items = fs.readdirSync(extDir); for (const item of items) { @@ -141,7 +146,7 @@ function copyExtensionFiles(targetDir, browserName) { const eStart = '// --- SHARED_EVENTS_INJECT_START ---'; const eEnd = '// --- SHARED_EVENTS_INJECT_END ---'; const ePattern = new RegExp(`${eStart}[\\s\\S]+?${eEnd}`); - const eRep = `${eStart}\n // This block is automatically updated by /scripts/build-extension.cjs\n const EVENTS = ${eventsObject};\n const MAX_MEDIA_TIME = ${maxMediaTimeVal};\n ${eEnd}`; + const eRep = `${eStart}\n // This block is automatically updated by /scripts/build-extension.cjs\n const EVENTS = ${eventsObject};\n const MAX_MEDIA_TIME = ${maxMediaTimeVal};\n const EPISODE_SYNC_V2_STABILITY_MS = ${episodeSyncStabilityVal};\n ${eEnd}`; content = replaceRequiredBlock(content, ePattern, eRep, 'Event injection'); diff --git a/scripts/test-server-ws.mjs b/scripts/test-server-ws.mjs index ffffb7a..c137c22 100644 --- a/scripts/test-server-ws.mjs +++ b/scripts/test-server-ws.mjs @@ -194,11 +194,225 @@ try { 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' }); + 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'); + 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, null); + assert.equal(mod.rooms.get(episodeRid).mediaState.playbackState, 'playing'); + assert.equal(mod.rooms.get(episodeRid).mediaState.currentTime, 0); + let duplicateExecute = false; + try { await w(episodeB, 'episode_sync_v2', 300); duplicateExecute = true; } catch { /* expected */ } + assert.equal(duplicateExecute, false, 'execute is emitted exactly once'); + close(); + resetConnectionRate(); + + // Mixed rooms degrade safely: no v2 transaction and no unknown frame to legacy. + const mixedEpisodeRid = 'episode-mixed-'+Date.now(); + const mixedEpisodeNew = await c(), mixedEpisodeOld = await c(); + await j(mixedEpisodeNew, mixedEpisodeRid, 'episode-new', null, episodeCaps); + await j(mixedEpisodeOld, mixedEpisodeRid, 'episode-old'); + mixedEpisodeNew._m.length = mixedEpisodeOld._m.length = 0; + s(mixedEpisodeNew, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S01E07' }); + const mixedEpisodeCancel = await w(mixedEpisodeNew, 'episode_sync_v2'); + assert.equal(mixedEpisodeCancel.phase, 'cancel'); + assert.equal(mixedEpisodeCancel.reason, 'capability_mismatch'); + assert.equal(mod.rooms.get(mixedEpisodeRid).episodeSyncV2, null); + let legacySawV2 = false; + try { await w(mixedEpisodeOld, 'episode_sync_v2', 300); legacySawV2 = true; } catch { /* expected */ } + assert.equal(legacySawV2, false, 'legacy peer receives no v2 frame'); + close(); + resetConnectionRate(); + + // Failure/manual intent cancel the barrier; host-only authorization is + // evaluated at START and never delegated to clients. + const episodeAbortRid = 'episode-abort-'+Date.now(); + const episodeAbortA = await c(), episodeAbortB = await c(); + await j(episodeAbortA, episodeAbortRid, 'abort-a', null, episodeCaps); + await j(episodeAbortB, episodeAbortRid, 'abort-b', null, episodeCaps); + episodeAbortA._m.length = episodeAbortB._m.length = 0; + s(episodeAbortA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S02E01' }); + const failedLobby = await w(episodeAbortA, 'episode_sync_v2'); + await w(episodeAbortB, 'episode_sync_v2'); + s(episodeAbortB, 'episode_sync_v2', { + phase: 'failed', + transactionId: failedLobby.transactionId, + reason: 'pause_failed' + }); + const failedCancelA = await w(episodeAbortA, 'episode_sync_v2'); + const failedCancelB = await w(episodeAbortB, 'episode_sync_v2'); + assert.equal(failedCancelA.reason, 'peer_failed'); + assert.equal(failedCancelB.failedPeerId, 'abort-b'); + + s(episodeAbortA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S02E02' }); + await w(episodeAbortA, 'episode_sync_v2'); await w(episodeAbortB, 'episode_sync_v2'); + s(episodeAbortA, 'play', { currentTime: 2, seq: 1 }); + const manualCancelA = await w(episodeAbortA, 'episode_sync_v2'); + const manualCancelB = await w(episodeAbortB, 'episode_sync_v2'); + const manualPlayB = await w(episodeAbortB, 'play'); + assert.equal(manualCancelA.reason, 'superseded'); + assert.equal(manualCancelB.reason, 'superseded'); + assert.equal(manualPlayB.currentTime, 2); + assert.equal(mod.rooms.get(episodeAbortRid).episodeSyncV2, null); + + const abortRoom = mod.rooms.get(episodeAbortRid); + abortRoom.controlMode = 'host-only'; + abortRoom.controllers = new Set(['abort-a']); + s(episodeAbortB, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S02E03' }); + const guestStartCancel = await w(episodeAbortB, 'episode_sync_v2'); + assert.equal(guestStartCancel.reason, 'not_controller'); + assert.equal(abortRoom.episodeSyncV2, null); + close(); + resetConnectionRate(); + + // Frozen participants: a late legacy joiner is excluded and does not block completion. + const frozenRid = 'episode-frozen-'+Date.now(); + const frozenA = await c(), frozenB = await c(); + await j(frozenA, frozenRid, 'frozen-a', null, episodeCaps); + await j(frozenB, frozenRid, 'frozen-b', null, episodeCaps); + frozenA._m.length = frozenB._m.length = 0; + s(frozenA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S01E08' }); + const frozenLobbyA = await w(frozenA, 'episode_sync_v2'); + await w(frozenB, 'episode_sync_v2'); + const frozenLate = await c(); + const frozenLateRoom = await j(frozenLate, frozenRid, 'frozen-late'); + assert.equal(frozenLateRoom.episodeSyncV2, null); + assert.deepEqual(frozenLobbyA.participants, ['frozen-a', 'frozen-b']); + frozenLate._m.length = 0; + for (const peer of [frozenA, frozenB]) { + s(peer, 'episode_sync_v2', { phase: 'loaded', transactionId: frozenLobbyA.transactionId }); + if (peer === frozenA) { await w(frozenA, 'episode_sync_v2'); await w(frozenB, 'episode_sync_v2'); } + } + await w(frozenA, 'episode_sync_v2'); await w(frozenB, 'episode_sync_v2'); + for (const peer of [frozenA, frozenB]) { + s(peer, 'episode_sync_v2', { phase: 'prepared', transactionId: frozenLobbyA.transactionId }); + if (peer === frozenA) { await w(frozenA, 'episode_sync_v2'); await w(frozenB, 'episode_sync_v2'); } + } + await w(frozenA, 'episode_sync_v2'); await w(frozenB, 'episode_sync_v2'); + let lateReceivedV2 = false; + try { await w(frozenLate, 'episode_sync_v2', 300); lateReceivedV2 = true; } catch { /* expected */ } + assert.equal(lateReceivedV2, false); + close(); + resetConnectionRate(); + + // Departure and deadline are cancellation-only; neither can emit execute. + const cancelRid = 'episode-cancel-'+Date.now(); + const cancelA = await c(), cancelB = await c(); + await j(cancelA, cancelRid, 'cancel-a', null, episodeCaps); + await j(cancelB, cancelRid, 'cancel-b', null, episodeCaps); + cancelA._m.length = cancelB._m.length = 0; + s(cancelA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S01E09' }); + const cancelLobby = await w(cancelA, 'episode_sync_v2'); + await w(cancelB, 'episode_sync_v2'); + mod.rooms.get(cancelRid).episodeSyncV2.deadlineAt = 1; + mod.expireEpisodeSyncV2Transactions(Date.now()); + const timeoutCancelA = await w(cancelA, 'episode_sync_v2'); + const timeoutCancelB = await w(cancelB, 'episode_sync_v2'); + assert.equal(timeoutCancelA.phase, 'cancel'); + assert.equal(timeoutCancelA.reason, 'load_timeout'); + assert.equal(timeoutCancelB.phase, 'cancel'); + assert.equal(mod.rooms.get(cancelRid).episodeSyncV2, null); + + s(cancelA, 'episode_sync_v2', { phase: 'start', expectedTitle: 'S01E10' }); + const departureLobby = await w(cancelA, 'episode_sync_v2'); await w(cancelB, 'episode_sync_v2'); + assert.notEqual(cancelLobby.transactionId, departureLobby.transactionId, 'new transaction receives a fresh identity'); + s(cancelB, 'leave_room', {}); + const departureCancel = await w(cancelA, 'episode_sync_v2'); + assert.equal(departureCancel.phase, 'cancel'); + assert.equal(departureCancel.reason, 'participant_left'); + assert.equal(departureCancel.failedPeerId, 'cancel-b'); + assert.equal(mod.rooms.get(cancelRid).episodeSyncV2, null); + close(); + resetConnectionRate(); + + // New relay hardening for old extensions: only the accepted lobby owner may + // relay legacy PREPARE/EXECUTE/CANCEL. + const legacyEpisodeRid = 'episode-legacy-owner-'+Date.now(); + const legacyEpisodeA = await c(), legacyEpisodeB = await c(); + await j(legacyEpisodeA, legacyEpisodeRid, 'legacy-a'); + await j(legacyEpisodeB, legacyEpisodeRid, 'legacy-b'); + legacyEpisodeA._m.length = legacyEpisodeB._m.length = 0; + s(legacyEpisodeA, 'episode_lobby', { expectedTitle: 'S01E11' }); + await w(legacyEpisodeB, 'episode_lobby'); + s(legacyEpisodeB, 'force_sync_prepare', { targetTime: 0 }); + let nonOwnerPrepareRelayed = false; + try { await w(legacyEpisodeA, 'force_sync_prepare', 300); nonOwnerPrepareRelayed = true; } catch { /* expected */ } + assert.equal(nonOwnerPrepareRelayed, false); + s(legacyEpisodeA, 'force_sync_prepare', { targetTime: 0 }); + await w(legacyEpisodeB, 'force_sync_prepare'); + s(legacyEpisodeB, 'force_sync_execute', {}); + let nonOwnerExecuteRelayed = false; + try { await w(legacyEpisodeA, 'force_sync_execute', 300); nonOwnerExecuteRelayed = true; } catch { /* expected */ } + assert.equal(nonOwnerExecuteRelayed, false); + s(legacyEpisodeA, 'force_sync_execute', {}); + await w(legacyEpisodeB, 'force_sync_execute'); + close(); + resetConnectionRate(); + // --- Mixed-version rollout: pre-media-state extension + current extension --- // Legacy intentionally omits clientCapabilities entirely and uses only the // pre-feature JOIN/PLAY/PAUSE/SEEK/Force Sync wire contract. The current @@ -1024,9 +1238,12 @@ try { mxo._m.length = mxn._m.length = 0; s(mxn,'force_sync_execute',{}); await w(mxo,'force_sync_execute'); s(mxo,'episode_lobby',{expectedTitle:'S1E1'}); await w(mxn,'episode_lobby'); - s(mxn,'pause',{currentTime:2}); await w(mxo,'pause'); + s(mxn,'pause',{currentTime:2}); await w(mxo,'episode_lobby_cancel'); await w(mxo,'pause'); s(mxn,'seek',{currentTime:50}); await w(mxo,'seek'); - s(mxn,'episode_lobby_cancel',{}); await w(mxo,'episode_lobby_cancel'); + s(mxn,'episode_lobby_cancel',{}); + let staleMixedCancelDropped = false; + try { await w(mxo,'episode_lobby_cancel',300); } catch { staleMixedCancelDropped = true; } + assert.ok(staleMixedCancelDropped, 'stale legacy lobby cancel is dropped after manual playback supersedes it'); close(); resetConnectionRate(); diff --git a/server/index.js b/server/index.js index 9f3ffbf..7872ed8 100644 --- a/server/index.js +++ b/server/index.js @@ -4,7 +4,7 @@ import { fileURLToPath } from 'url'; import { Server } from 'socket.io'; import crypto from 'crypto'; import dotenv from 'dotenv'; -import { EVENTS, ERROR_CODES, OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION, CONTROL_MODES, CAPABILITIES, FORCE_SYNC_TARGET_DELAY_WARNING, MAX_MEDIA_TIME } from '../shared/constants.js'; +import { EVENTS, ERROR_CODES, OFFICIAL_SERVER_TOKEN, PROTOCOL_VERSION, CONTROL_MODES, CAPABILITIES, FORCE_SYNC_TARGET_DELAY_WARNING, MAX_MEDIA_TIME, EPISODE_LOBBY_TIMEOUT, EPISODE_SYNC_V2_PREPARE_TIMEOUT } from '../shared/constants.js'; import { createChatEnvelope } from './chat.js'; import { commitForceSyncMediaState, @@ -173,7 +173,8 @@ const HOST_ONLY_GATED_EVENTS = new Set([ EVENTS.FORCE_SYNC_PREPARE, EVENTS.FORCE_SYNC_EXECUTE, EVENTS.EPISODE_LOBBY, - EVENTS.EPISODE_LOBBY_CANCEL + EVENTS.EPISODE_LOBBY_CANCEL, + EVENTS.EPISODE_SYNC_V2 ]); // Current clients sequence room-moving media commands. The relay mirrors the @@ -188,6 +189,15 @@ const SEQUENCED_ROOM_EVENTS = new Set([ EVENTS.FORCE_SYNC_EXECUTE ]); +const EPISODE_SYNC_V2_SUPERSEDING_EVENTS = new Set([ + EVENTS.PLAY, + EVENTS.PAUSE, + EVENTS.SEEK, + EVENTS.FORCE_SYNC_PREPARE, + EVENTS.FORCE_SYNC_EXECUTE, + EVENTS.EPISODE_LOBBY +]); + // Features this relay supports, advertised to clients in ROOM_DATA so they can // enable matching UI/behavior only when the server actually backs it. Append a // flag here when a new server-gated feature ships (e.g. co-host promotion). @@ -196,7 +206,8 @@ const SERVER_CAPABILITIES = [ CAPABILITIES.CO_HOST, CAPABILITIES.CHAT, CAPABILITIES.CHAT_V1, - CAPABILITIES.MEDIA_STATE_V1 + CAPABILITIES.MEDIA_STATE_V1, + CAPABILITIES.EPISODE_SYNC_V2 ]; function normalizeClientCapabilities(value) { @@ -205,7 +216,8 @@ function normalizeClientCapabilities(value) { .filter(capability => typeof capability === 'string') .map(capability => capability.substring(0, 32)) .filter(capability => capability === CAPABILITIES.CHAT_V1 - || capability === CAPABILITIES.MEDIA_STATE_V1) + || capability === CAPABILITIES.MEDIA_STATE_V1 + || capability === CAPABILITIES.EPISODE_SYNC_V2) )]; } @@ -219,6 +231,86 @@ function clientSupportsMediaState(socket) { && socket.data.clientCapabilities.includes(CAPABILITIES.MEDIA_STATE_V1); } +function clientSupportsEpisodeSyncV2(socket) { + return Array.isArray(socket?.data?.clientCapabilities) + && socket.data.clientCapabilities.includes(CAPABILITIES.EPISODE_SYNC_V2); +} + +function publicEpisodeSyncV2(transaction, phase = null) { + if (!transaction) return null; + const publicPhase = phase || (transaction.phase === 'loading' ? 'lobby' : 'prepare'); + return { + transactionId: transaction.transactionId, + phase: publicPhase, + expectedTitle: transaction.expectedTitle, + initiatorPeerId: transaction.initiatorPeerId, + participants: [...transaction.participants], + loadedPeers: [...transaction.loadedPeers], + preparedPeers: [...transaction.preparedPeers], + createdAt: transaction.createdAt, + deadlineAt: transaction.deadlineAt, + revision: transaction.revision + }; +} + +function emitEpisodeSyncV2ToParticipants(roomId, room, transaction, payload) { + const participantIds = new Set(transaction.participants); + for (const socketId of room.peers) { + const participant = room.peerData.get(socketId); + const participantSocket = io.sockets.sockets.get(socketId); + if (!participantIds.has(participant?.peerId) || !clientSupportsEpisodeSyncV2(participantSocket)) continue; + participantSocket.emit(EVENTS.EPISODE_SYNC_V2, payload); + } +} + +function clearEpisodeSyncV2Timer(transaction) { + if (!transaction?.timeout) return; + clearTimeout(transaction.timeout); + transaction.timeout = null; +} + +function cancelEpisodeSyncV2(roomId, room, reason, failedPeerId = null) { + const transaction = room?.episodeSyncV2; + if (!transaction) return false; + clearEpisodeSyncV2Timer(transaction); + room.episodeSyncV2 = null; + room.lastEpisodeSyncV2Id = transaction.transactionId; + emitEpisodeSyncV2ToParticipants(roomId, room, transaction, { + ...publicEpisodeSyncV2(transaction, 'cancel'), + senderId: transaction.initiatorPeerId, + reason: typeof reason === 'string' ? reason.substring(0, 32) : 'cancelled', + failedPeerId: typeof failedPeerId === 'string' ? failedPeerId.substring(0, 16) : undefined + }); + return true; +} + +function scheduleEpisodeSyncV2Deadline(roomId, room, transaction, timeoutMs) { + clearEpisodeSyncV2Timer(transaction); + transaction.deadlineAt = Date.now() + timeoutMs; + transaction.timeout = setTimeout(() => { + if (room.episodeSyncV2 !== transaction) return; + cancelEpisodeSyncV2(roomId, room, transaction.phase === 'loading' ? 'load_timeout' : 'prepare_timeout'); + }, timeoutMs); + transaction.timeout.unref?.(); +} + +function roomFullySupportsEpisodeSyncV2(room) { + if (!room || room.peers.size === 0) return false; + for (const socketId of room.peers) { + if (!clientSupportsEpisodeSyncV2(io.sockets.sockets.get(socketId))) return false; + } + return true; +} + +export function expireEpisodeSyncV2Transactions(now = Date.now()) { + for (const [roomId, room] of rooms) { + const transaction = room.episodeSyncV2; + if (transaction && Number.isFinite(transaction.deadlineAt) && transaction.deadlineAt <= now) { + cancelEpisodeSyncV2(roomId, room, transaction.phase === 'loading' ? 'load_timeout' : 'prepare_timeout'); + } + } +} + // M-4: minimum interval between CONTROL_MODE changes per room. Stops a rapidly // toggling host from thrashing every guest's UI (locked/unlocked/locked...) and // from generating one broadcast per toggle across all peers. @@ -262,6 +354,13 @@ function removePeerFromRoom(socketId, roomId, reason, { notifyRemainingPeers = t const { peerId } = peerData; + // V2 freezes its participant set. Losing any participant invalidates the + // barrier; continuing with a smaller set could execute before a reconnecting + // player has actually prepared. + if (room.episodeSyncV2?.participants.includes(peerId) && !peerJoinLocks.has(peerId)) { + cancelEpisodeSyncV2(roomId, room, 'participant_left', peerId); + } + // 1. Remove from room data structures room.peers.delete(socketId); room.peerIds.delete(socketId); @@ -286,8 +385,10 @@ function removePeerFromRoom(socketId, roomId, reason, { notifyRemainingPeers = t room.activeLobby.readyPeers = room.activeLobby.readyPeers.filter(id => id !== peerId); if (room.peers.size <= 1 || room.activeLobby.initiatorPeerId === peerId) { room.activeLobby = null; // Dissolve lobby + room.legacyEpisodeSyncOwner = null; } } + if (room.legacyEpisodeSyncOwner === peerId) room.legacyEpisodeSyncOwner = null; // 3.6. Host Control Mode: if the host left (and isn't still connected via another // socket), fall back to 'everyone' so the room never gets stuck locked, and @@ -338,6 +439,7 @@ function removePeerFromRoom(socketId, roomId, reason, { notifyRemainingPeers = t // 4. Delete empty room if (room.peers.size === 0) { + clearEpisodeSyncV2Timer(room.episodeSyncV2); rooms.delete(roomId); log('ROOM', `Deleted empty room after ${reason}: ${roomId.substring(0, 3)}***`); } @@ -493,7 +595,11 @@ io.on('connection', (socket) => { forceSyncTarget: null, // Distinguishes an unknown target after relay restart from a // transaction explicitly replaced by newer room playback. - forceSyncSuperseded: false + forceSyncSuperseded: false, + activeLobby: null, + legacyEpisodeSyncOwner: null, + episodeSyncV2: null, + lastEpisodeSyncV2Id: null }; rooms.set(roomId, room); createdByMe = true; @@ -587,6 +693,10 @@ io.on('connection', (socket) => { controlMode: room.controlMode || CONTROL_MODES.EVERYONE, controllers: room.controllers ? Array.from(room.controllers) : [], mediaState: snapshotMediaState(room.mediaState, snapshotAt), + episodeSyncV2: room.episodeSyncV2?.participants.includes(peerId) + && clientSupportsEpisodeSyncV2(socket) + ? publicEpisodeSyncV2(room.episodeSyncV2) + : null, capabilities: SERVER_CAPABILITIES }); log('ROOM', `Peer ${peerId} joined: ${roomId.substring(0, 3)}***`); @@ -714,6 +824,13 @@ io.on('connection', (socket) => { // Strip undefined keys for clean wire format Object.keys(relayPayload).forEach(k => relayPayload[k] === undefined && delete relayPayload[k]); + if (room.episodeSyncV2 + && eventName === EVENTS.PEER_STATUS + && relayPayload.desynced === true + && room.episodeSyncV2.participants.includes(mapping.peerId)) { + cancelEpisodeSyncV2(mapping.roomId, room, 'participant_desynced', mapping.peerId); + } + // The first live lobby owns the room until completion or // cancellation. Drop concurrent lobby starts and stale ready // frames instead of letting clients build divergent lobbies. @@ -743,6 +860,29 @@ io.on('connection', (socket) => { return; } } + if (eventName === EVENTS.EPISODE_LOBBY_CANCEL && !room.activeLobby) { + log('ROOM', `Dropped stale episode lobby cancel from ${mapping.peerId}`); + return; + } + + // A user/manual legacy command wins over automation. Cancel + // the v2 barrier first so prepared peers can restore safely, + // then relay the newer command normally. + if (room.episodeSyncV2 && EPISODE_SYNC_V2_SUPERSEDING_EVENTS.has(eventName)) { + cancelEpisodeSyncV2(mapping.roomId, room, 'superseded', mapping.peerId); + } + + // Legacy clients all used to self-promote after lobby ready. + // Preserve their wire contract, but bind the room-wide PREPARE, + // EXECUTE and CANCEL to the accepted lobby owner. + if (room.legacyEpisodeSyncOwner + && (eventName === EVENTS.FORCE_SYNC_PREPARE + || eventName === EVENTS.FORCE_SYNC_EXECUTE + || eventName === EVENTS.EPISODE_LOBBY_CANCEL) + && mapping.peerId !== room.legacyEpisodeSyncOwner) { + log('ROOM', `Dropped legacy episode ${eventName} from non-owner ${mapping.peerId}`); + return; + } const mediaStateNow = Date.now(); @@ -771,6 +911,16 @@ io.on('connection', (socket) => { room.forceSyncSuperseded = true; room.forceSyncInitiator = null; room.forceSyncTarget = null; + if (canonicalStateUpdated) room.legacyEpisodeSyncOwner = null; + } + if (canonicalStateUpdated && room.activeLobby) { + room.activeLobby = null; + room.legacyEpisodeSyncOwner = null; + io.to(mapping.roomId).emit(EVENTS.EPISODE_LOBBY_CANCEL, { + senderId: mapping.peerId, + peerId: mapping.peerId, + reason: 'superseded' + }); } if (eventName === EVENTS.FORCE_SYNC_PREPARE) { // A malformed PREPARE must neither pause peers nor grant @@ -821,6 +971,7 @@ io.on('connection', (socket) => { room.forceSyncInitiator = null; room.forceSyncTarget = null; room.forceSyncSuperseded = false; + room.legacyEpisodeSyncOwner = null; } socket.to(mapping.roomId).emit(eventName, relayPayload); @@ -832,12 +983,14 @@ io.on('connection', (socket) => { initiatorPeerId: mapping.peerId, readyPeers: [mapping.peerId] }; + room.legacyEpisodeSyncOwner = mapping.peerId; } else if (eventName === EVENTS.EPISODE_READY && room.activeLobby) { if (!room.activeLobby.readyPeers.includes(mapping.peerId)) { room.activeLobby.readyPeers.push(mapping.peerId); } } else if ((eventName === EVENTS.FORCE_SYNC_PREPARE || eventName === EVENTS.FORCE_SYNC_EXECUTE || eventName === EVENTS.EPISODE_LOBBY_CANCEL) && room.activeLobby) { room.activeLobby = null; + if (eventName === EVENTS.EPISODE_LOBBY_CANCEL) room.legacyEpisodeSyncOwner = null; } } } @@ -847,6 +1000,159 @@ io.on('connection', (socket) => { }); }); + socket.on(EVENTS.EPISODE_SYNC_V2, (data) => { + try { + if (!checkEventRate(socket.id)) { + log('SECURITY', `Event rate limit exceeded for socket: ${socket.id}`); + socket.disconnect(true); + return; + } + if (!data || typeof data !== 'object') return; + const mapping = socketToRoom.get(socket.id); + const room = mapping ? rooms.get(mapping.roomId) : null; + if (!mapping || !room || !clientSupportsEpisodeSyncV2(socket)) return; + + const phase = typeof data.phase === 'string' ? data.phase.substring(0, 16) : ''; + const transactionId = typeof data.transactionId === 'string' + ? data.transactionId.substring(0, 64) + : ''; + room.lastActivity = Date.now(); + + if (phase === 'start') { + const expectedTitle = typeof data.expectedTitle === 'string' + ? data.expectedTitle.substring(0, 100) + : ''; + const isController = room.controlMode !== CONTROL_MODES.HOST_ONLY + || (room.controllers && room.controllers.has(mapping.peerId)); + const senderData = room.peerData.get(socket.id); + const rejectStart = (reason) => socket.emit(EVENTS.EPISODE_SYNC_V2, { + phase: 'cancel', + transactionId: null, + senderId: mapping.peerId, + expectedTitle, + reason + }); + if (!expectedTitle) return rejectStart('invalid_title'); + if (!isController) return rejectStart('not_controller'); + if (senderData?.desynced) return rejectStart('desynced'); + if (!roomFullySupportsEpisodeSyncV2(room)) return rejectStart('capability_mismatch'); + if (room.activeLobby || room.forceSyncTarget) return rejectStart('legacy_sync_active'); + if (room.episodeSyncV2) { + if (room.episodeSyncV2.participants.includes(mapping.peerId)) { + socket.emit(EVENTS.EPISODE_SYNC_V2, { + ...publicEpisodeSyncV2(room.episodeSyncV2), + senderId: room.episodeSyncV2.initiatorPeerId + }); + } else { + rejectStart('transaction_busy'); + } + return; + } + + const participants = [...new Set([...room.peerData.values()] + .filter(candidate => candidate && !candidate.desynced) + .map(candidate => candidate.peerId) + .filter(Boolean))]; + if (participants.length < 2 || !participants.includes(mapping.peerId)) { + return rejectStart('not_enough_participants'); + } + + const now = Date.now(); + const transaction = { + transactionId: crypto.randomUUID(), + phase: 'loading', + expectedTitle, + initiatorPeerId: mapping.peerId, + participants, + loadedPeers: [], + preparedPeers: [], + createdAt: now, + deadlineAt: now + EPISODE_LOBBY_TIMEOUT, + revision: 1, + timeout: null + }; + room.episodeSyncV2 = transaction; + scheduleEpisodeSyncV2Deadline(mapping.roomId, room, transaction, EPISODE_LOBBY_TIMEOUT); + emitEpisodeSyncV2ToParticipants(mapping.roomId, room, transaction, { + ...publicEpisodeSyncV2(transaction, 'lobby'), + senderId: transaction.initiatorPeerId + }); + return; + } + + const transaction = room.episodeSyncV2; + if (!transaction + || !transactionId + || transaction.transactionId !== transactionId + || !transaction.participants.includes(mapping.peerId)) { + return; + } + + if (phase === 'cancel' || phase === 'failed') { + cancelEpisodeSyncV2(mapping.roomId, room, phase === 'failed' ? 'peer_failed' : 'peer_cancelled', mapping.peerId); + return; + } + + if (phase === 'loaded' && transaction.phase === 'loading') { + if (transaction.loadedPeers.includes(mapping.peerId)) return; + transaction.loadedPeers.push(mapping.peerId); + transaction.revision++; + if (transaction.loadedPeers.length < transaction.participants.length) { + emitEpisodeSyncV2ToParticipants(mapping.roomId, room, transaction, { + ...publicEpisodeSyncV2(transaction, 'lobby'), + senderId: transaction.initiatorPeerId + }); + return; + } + + transaction.phase = 'preparing'; + transaction.revision++; + scheduleEpisodeSyncV2Deadline(mapping.roomId, room, transaction, EPISODE_SYNC_V2_PREPARE_TIMEOUT); + emitEpisodeSyncV2ToParticipants(mapping.roomId, room, transaction, { + ...publicEpisodeSyncV2(transaction, 'prepare'), + senderId: transaction.initiatorPeerId, + targetTime: 0 + }); + return; + } + + if (phase === 'prepared' && transaction.phase === 'preparing') { + if (transaction.preparedPeers.includes(mapping.peerId)) return; + transaction.preparedPeers.push(mapping.peerId); + transaction.revision++; + if (transaction.preparedPeers.length < transaction.participants.length) { + emitEpisodeSyncV2ToParticipants(mapping.roomId, room, transaction, { + ...publicEpisodeSyncV2(transaction, 'prepare'), + senderId: transaction.initiatorPeerId, + targetTime: 0 + }); + return; + } + + clearEpisodeSyncV2Timer(transaction); + room.episodeSyncV2 = null; + room.lastEpisodeSyncV2Id = transaction.transactionId; + room.forceSyncInitiator = null; + room.forceSyncTarget = null; + room.forceSyncSuperseded = false; + commitForceSyncMediaState( + room, + 0, + transaction.initiatorPeerId, + Date.now(), + transaction.expectedTitle + ); + emitEpisodeSyncV2ToParticipants(mapping.roomId, room, transaction, { + ...publicEpisodeSyncV2(transaction, 'execute'), + senderId: transaction.initiatorPeerId, + targetTime: 0 + }); + } + } catch (err) { + log('ERROR', `EPISODE_SYNC_V2 handler error: ${err.message}`); + } + }); + socket.on(EVENTS.GET_ROOMS, () => { if (!checkEventRate(socket.id)) { log('SECURITY', `Event rate limit exceeded for socket (GET_ROOMS): ${socket.id}`); @@ -1101,6 +1407,7 @@ io.on('connection', (socket) => { // Active Room & Dead Peer Cleanup (Every 2m) export function cleanupInactiveRooms(now = Date.now()) { + expireEpisodeSyncV2Transactions(now); const roomCutoff = now - (2 * 60 * 60 * 1000); // 2 hours const peerCutoff = now - (5 * 60 * 1000); // 5 minutes @@ -1137,6 +1444,7 @@ export function cleanupInactiveRooms(now = Date.now()) { // 2. Prune empty or inactive rooms const currentRoom = rooms.get(roomId); if (currentRoom && (currentRoom.peers.size === 0 || currentRoom.lastActivity < roomCutoff)) { + clearEpisodeSyncV2Timer(currentRoom.episodeSyncV2); io.to(roomId).emit(EVENTS.ERROR, { code: ERROR_CODES.ROOM_CLOSED, message: 'Room closed' diff --git a/shared/README.md b/shared/README.md index 79ff46f..dc6eb72 100644 --- a/shared/README.md +++ b/shared/README.md @@ -32,6 +32,7 @@ Browser extensions cannot import files outside their own root directory, so the - `CAPABILITIES.HOST_CONTROL`: relay supports host-only room authority. - `CAPABILITIES.CO_HOST`: relay supports promoted controller peers. - `CAPABILITIES.MEDIA_STATE_V1`: relay exposes canonical room playback recovery snapshots. +- `CAPABILITIES.EPISODE_SYNC_V2`: relay owns an ID-correlated load/prepare/execute episode barrier. Clients should enable capability-gated UI only when the relay advertises the matching flag in `room_data.capabilities`. @@ -57,6 +58,7 @@ For the complete event list, read the `EVENTS` object in [`constants.js`](consta | `EPISODE_LOBBY` | Bidirectional relay | Episode transition lobby started | | `EPISODE_READY` | Bidirectional relay | Peer has loaded the episode and is ready | | `EPISODE_LOBBY_CANCEL` | Bidirectional relay | Active episode lobby cancelled | +| `EPISODE_SYNC_V2` | Bidirectional, capability-gated | Relay-authoritative episode transaction (`start/lobby/loaded/prepare/prepared/execute/cancel`) | | `GET_ROOMS` / `ROOM_LIST` | Client <-> Server | Room discovery with server-side cooldown | | `PING` / `PONG` | Client <-> Server/Peer | Server RTT and peer latency checks | @@ -66,6 +68,8 @@ For the complete event list, read the `EVENTS` object in [`constants.js`](consta - `FORCE_SYNC_TIMEOUT`: max wait for force-sync ACKs. - `FORCE_SYNC_TARGET_DELAY_WARNING`: threshold for logging delayed Force Sync execution; an unsuperseded prepared target remains executable for receiver liveness. - `EPISODE_LOBBY_TIMEOUT`: max wait for episode-lobby readiness. +- `EPISODE_SYNC_V2_PREPARE_TIMEOUT`: max wait for every frozen participant to pause, seek, buffer, and verify stable state. +- `EPISODE_SYNC_V2_STABILITY_MS`: continuous ready-state window required before `prepared`. - `MAX_MEDIA_TIME`: shared relay/extension upper bound, in seconds, for synchronized media positions. ## Do Not Break diff --git a/shared/constants.js b/shared/constants.js index 56d6e46..14d7ca9 100644 --- a/shared/constants.js +++ b/shared/constants.js @@ -56,6 +56,7 @@ export const EVENTS = { EPISODE_LOBBY: "episode_lobby", // Broadcast: waiting for everyone on this episode EPISODE_READY: "episode_ready", // Response: loaded the episode and paused at 0:00 EPISODE_LOBBY_CANCEL: "episode_lobby_cancel", // Broadcast: cancel active lobby and resume + EPISODE_SYNC_V2: "episode_sync_v2", // Capability-gated, relay-authoritative episode transaction // Ephemeral end-to-end encrypted chat CHAT_MESSAGE: "chat_message", // Ciphertext relay; no server history @@ -91,7 +92,8 @@ export const CAPABILITIES = { CO_HOST: 'co-host', // owner promotes guests to additional controllers CHAT: 'chat', // legacy server capability used by the first chat beta CHAT_V1: 'chat-v1', // versioned client/server chat wire contract - MEDIA_STATE_V1: 'media-state-v1' // server-authoritative room playback recovery snapshot + MEDIA_STATE_V1: 'media-state-v1', // server-authoritative room playback recovery snapshot + EPISODE_SYNC_V2: 'episode-sync-v2' // transaction IDs + relay-owned load/prepare/execute phases }; // Relay and extension media-time validation must use the same upper bound. @@ -103,3 +105,5 @@ export const FORCE_SYNC_TIMEOUT = 8500; // 8.5s timeout for force sync ACKs (mus // transport grace. The target remains valid until newer room playback replaces it. export const FORCE_SYNC_TARGET_DELAY_WARNING = FORCE_SYNC_TIMEOUT + 2000; export const EPISODE_LOBBY_TIMEOUT = 60000; // 60s timeout for episode lobby +export const EPISODE_SYNC_V2_PREPARE_TIMEOUT = 15000; // pause, seek, buffer and 1s stable verification +export const EPISODE_SYNC_V2_STABILITY_MS = 1000; diff --git a/tests/e2e/extension.spec.mjs b/tests/e2e/extension.spec.mjs index 28574fb..77affe3 100644 --- a/tests/e2e/extension.spec.mjs +++ b/tests/e2e/extension.spec.mjs @@ -147,8 +147,39 @@ async function waitForLegacyRelayEvent(socket, event, timeoutMs = 10_000) { return JSON.parse(socket.messages.splice(index, 1)[0].substring(2))[1]; } -async function joinLegacyRelayRoom(socket, roomId, peerId) { - sendLegacyRelayEvent(socket, 'join_room', { roomId, peerId, protocolVersion: PROTOCOL_VERSION }); +async function waitForLegacyRelayPhase(socket, event, phase, transactionId, timeoutMs = 10_000) { + await expect.poll(() => socket.messages.some(message => { + if (!message.startsWith('42')) return false; + try { + const [candidateEvent, payload] = JSON.parse(message.substring(2)); + return candidateEvent === event + && payload?.phase === phase + && payload?.transactionId === transactionId; + } catch { + return false; + } + }), { timeout: timeoutMs }).toBe(true); + const index = socket.messages.findIndex(message => { + if (!message.startsWith('42')) return false; + try { + const [candidateEvent, payload] = JSON.parse(message.substring(2)); + return candidateEvent === event + && payload?.phase === phase + && payload?.transactionId === transactionId; + } catch { + return false; + } + }); + return JSON.parse(socket.messages.splice(index, 1)[0].substring(2))[1]; +} + +async function joinLegacyRelayRoom(socket, roomId, peerId, clientCapabilities = undefined) { + sendLegacyRelayEvent(socket, 'join_room', { + roomId, + peerId, + protocolVersion: PROTOCOL_VERSION, + clientCapabilities + }); return waitForLegacyRelayEvent(socket, 'room_data'); } @@ -1485,6 +1516,106 @@ test('drops the selection when the user clears it', async ({ context, extensionI }); }); +test('completes Episode Sync v2 only after the packed player is stably prepared', async ({ context, extensionId, baseURL }) => { + test.setTimeout(45_000); + const relay = await import('../../server/index.js'); + let coordinator = null; + try { + await relay.startServer(0, '127.0.0.1'); + const port = relay.httpServer.address().port; + const roomId = `e2e-episode-v2-${Date.now()}`; + const url = `${baseURL}/pages/simple-player.html`; + const page = await context.newPage(); + await page.goto(url); + await page.waitForFunction(() => window.__fixtureReady === true); + await page.evaluate(async () => { + navigator.mediaSession.metadata = new window.MediaMetadata({ title: 'S1:E6 - Visiting Ours' }); + const video = document.querySelector('#player'); + video.muted = true; + video.currentTime = 3; + await video.play(); + }); + const { tabId } = await selectTargetTab(context, extensionId, url); + + coordinator = await connectLegacyRelayClient(port); + const episodeCaps = ['chat-v1', 'media-state-v1', 'episode-sync-v2']; + const coordinatorPeerId = 'episode-coord'; + await joinLegacyRelayRoom(coordinator, roomId, coordinatorPeerId, episodeCaps); + coordinator.messages.length = 0; + + await withExtensionPage(context, extensionId, extensionPage => extensionPage.evaluate(async settings => { + await chrome.storage.local.set(settings); + return chrome.runtime.sendMessage({ type: 'CONNECT' }); + }, { + serverUrl: `ws://127.0.0.1:${port}`, + useCustomServer: true, + roomId, + password: '', + username: 'episode-player', + autoSyncNextEpisode: true + })); + await expectConnectedRoom(context, extensionId, roomId); + await expect.poll(() => relay.rooms.get(roomId)?.peers.size).toBe(2); + + sendLegacyRelayEvent(coordinator, 'episode_sync_v2', { + phase: 'start', + expectedTitle: 'S1:E6 - Visiting Ours' + }); + const lobby = await waitForLegacyRelayEvent(coordinator, 'episode_sync_v2'); + expect(lobby).toMatchObject({ phase: 'lobby', loadedPeers: [], preparedPeers: [] }); + sendLegacyRelayEvent(coordinator, 'episode_sync_v2', { + phase: 'loaded', + transactionId: lobby.transactionId + }); + + const prepare = await waitForLegacyRelayPhase( + coordinator, + 'episode_sync_v2', + 'prepare', + lobby.transactionId + ); + expect(prepare.loadedPeers).toEqual(expect.arrayContaining([coordinatorPeerId])); + const extensionPeerId = prepare.participants.find(peerId => peerId !== coordinatorPeerId); + expect(extensionPeerId).toBeTruthy(); + await expect.poll(() => page.locator('#player').evaluate(video => ({ + paused: video.paused, + currentTime: video.currentTime, + readyState: video.readyState, + seeking: video.seeking + }))).toMatchObject({ paused: true, readyState: 4, seeking: false }); + await expect.poll(() => page.locator('#player').evaluate(video => video.currentTime)).toBeLessThan(1); + await expect.poll(() => relay.rooms.get(roomId)?.episodeSyncV2?.preparedPeers || []) + .toContain(extensionPeerId); + + sendLegacyRelayEvent(coordinator, 'episode_sync_v2', { + phase: 'prepared', + transactionId: lobby.transactionId + }); + const execute = await waitForLegacyRelayPhase( + coordinator, + 'episode_sync_v2', + 'execute', + lobby.transactionId + ); + expect(execute).toMatchObject({ phase: 'execute', transactionId: lobby.transactionId, targetTime: 0 }); + await expect.poll(() => relay.rooms.get(roomId)?.episodeSyncV2).toBeNull(); + await expect.poll(() => page.locator('#player').evaluate(video => video.paused)).toBe(false); + await expect.poll(() => page.locator('#player').evaluate(video => video.currentTime)).toBeLessThan(3); + + const legacyForceFrames = coordinator.messages.filter(message => message.startsWith('42') && (() => { + try { return JSON.parse(message.substring(2))[0].startsWith('force_sync_'); } catch { return false; } + })()); + expect(legacyForceFrames).toEqual([]); + + const state = await getExtensionState(context, extensionId, { type: 'GET_STATUS' }); + expect(state.episodeLobby).toBeNull(); + expect(state.targetTabId).toBe(tabId); + } finally { + try { coordinator?.close(); } catch { /* already closed */ } + await relay.stopServerForTests(); + } +}); + test('clears the selected target after inactivity removal and room closure', async ({ context, extensionId, baseURL }) => { test.setTimeout(45_000); const relay = await import('../../server/index.js');