import type { Socket } from 'socket.io-client'; import { RULES_VERSION } from '../../constants/gameplay.js'; import { CONNECT_TIMEOUT_MS, MAX_PEER_MESSAGE_BYTES, PROTOCOL_VERSION } from '../../constants/webrtc.js'; import type { PeerMessage, SignalingPacket } from '../shared/p2p-types.js'; export type PeerState = 'connecting' | 'connected' | 'disconnected' | 'failed' | 'closed'; export interface PeerConfig { matchId: string; roomId: string; slot: 0 | 1; epoch?: number; iceServers: RTCIceServer[]; createSignaling(): Socket; createConnection?: (config: RTCConfiguration) => RTCPeerConnection; refreshIce?: () => Promise; } export function createPeer(config: PeerConfig) { let pc: RTCPeerConnection | undefined, channel: RTCDataChannel | undefined, signal: Socket | undefined; let epoch = config.epoch ?? 0, disposed = false, ready = false, remoteReady = false, offering = false; let connectionPromise: Promise | undefined, yes: (() => void) | undefined, no: ((e: Error) => void) | undefined; let timer: ReturnType | undefined, signalQueue = Promise.resolve(); let candidates: RTCIceCandidateInit[] = []; let generation = 0; const messages = new Set<(message: PeerMessage) => void>(), states = new Set<(state: PeerState) => void>(); const emitState = (state: PeerState) => { for (const handler of states) handler(state); }; const stopSignaling = () => { signal?.removeAllListeners(); signal?.disconnect(); signal = undefined; }; function discardConnection() { const oldChannel = channel, oldPc = pc; channel = undefined; pc = undefined; if (oldChannel) { oldChannel.onopen = oldChannel.onclose = oldChannel.onerror = oldChannel.onmessage = null; oldChannel.close(); } if (oldPc) { oldPc.ondatachannel = oldPc.onicecandidate = oldPc.onconnectionstatechange = null; oldPc.close(); } } function failure(error: unknown, attempt = generation) { // In-flight ICE acknowledgements can arrive after both sides close signaling. if (disposed || attempt !== generation || (ready && channel?.readyState === 'open')) return; clearTimeout(timer); stopSignaling(); no?.(error instanceof Error ? error : new Error('PEER_FAILED')); yes = undefined; no = undefined; connectionPromise = undefined; ready = false; emitState('failed'); } function complete() { if (!remoteReady || channel?.readyState !== 'open' || ready) return; ready = true; clearTimeout(timer); stopSignaling(); yes?.(); yes = undefined; no = undefined; emitState('connected'); } function announceReady() { if (channel?.readyState === 'open') { channel.send(JSON.stringify({ type: 'transport:ready', matchId: config.matchId, epoch, slot: config.slot, protocolVersion: PROTOCOL_VERSION, rulesVersion: RULES_VERSION })); complete(); } } function installChannel(next: RTCDataChannel) { channel = next; channel.onopen = announceReady; channel.onclose = () => { if (!disposed) { ready = false; emitState('disconnected'); } }; channel.onerror = () => { if (!disposed) emitState('disconnected'); }; channel.onmessage = event => { if (disposed || typeof event.data !== 'string' || new TextEncoder().encode(event.data).byteLength > MAX_PEER_MESSAGE_BYTES) return; let message: any; try { message = JSON.parse(event.data); } catch { return; } if (!message || message.matchId !== config.matchId || message.protocolVersion !== PROTOCOL_VERSION || message.rulesVersion !== RULES_VERSION) return; if (message.type === 'transport:ready') { if (message.epoch !== epoch || message.slot === config.slot) return; remoteReady = true; complete(); return; } if (!ready || message.connectionEpoch !== epoch) return; for (const handler of messages) handler(message as PeerMessage); }; if (channel.readyState === 'open') announceReady(); } async function sendSignal(type: SignalingPacket['type'], payload: SignalingPacket['payload'], attempt = generation) { if (disposed || attempt !== generation) return; if (!signal?.connected) throw new Error('SIGNAL_DISCONNECTED'); const ack = await signal.timeout(5000).emitWithAck('signal', { type, payload, matchId: config.matchId, epoch } satisfies SignalingPacket); if (disposed || attempt !== generation) return; if (!ack?.ok) throw new Error(ack?.code ?? 'SIGNAL_FAILED'); } function installConnection() { const current = pc = (config.createConnection ?? (settings => new RTCPeerConnection(settings)))({ iceServers: config.iceServers }); current.ondatachannel = event => installChannel(event.channel); current.onicecandidate = event => { const attempt = generation; if (current === pc && event.candidate && signal?.connected) void sendSignal('ice', event.candidate.toJSON(), attempt).catch(error => failure(error, attempt)); }; current.onconnectionstatechange = () => { if (disposed || current !== pc) return; if (current.connectionState === 'failed' || current.connectionState === 'disconnected') { ready = false; emitState('disconnected'); } else if (current.connectionState === 'connected' && !ready) announceReady(); }; if (config.slot === 0) installChannel(current.createDataChannel('game', { ordered: true })); } async function received(packet: SignalingPacket, attempt: number) { if (disposed || attempt !== generation || !pc || packet.matchId !== config.matchId || packet.epoch !== epoch) return; const current = pc; if (packet.type === 'ice') { if (!current.remoteDescription) candidates.push(packet.payload as RTCIceCandidateInit); else await current.addIceCandidate(packet.payload as RTCIceCandidateInit); return; } if ((packet.type === 'offer' && config.slot !== 1) || (packet.type === 'answer' && config.slot !== 0)) return; await current.setRemoteDescription(packet.payload as RTCSessionDescriptionInit); if (disposed || attempt !== generation) return; for (const candidate of candidates.splice(0)) await current.addIceCandidate(candidate); if (packet.type === 'offer') { const answer = await current.createAnswer(); if (disposed || attempt !== generation) return; await current.setLocalDescription(answer); await sendSignal('answer', current.localDescription!, attempt); } if (attempt === generation && channel?.readyState === 'open') announceReady(); } function connect(): Promise { if (disposed) return Promise.reject(new Error('PEER_CLOSED')); if (ready) return Promise.resolve(); if (connectionPromise) return connectionPromise; const attempt = ++generation; remoteReady = false; offering = false; connectionPromise = new Promise((resolve, reject) => { yes = resolve; no = reject; }); const currentPromise = connectionPromise; timer = setTimeout(() => failure(new Error('PEER_TIMEOUT'), attempt), CONNECT_TIMEOUT_MS); if (!pc || pc.connectionState === 'closed' || channel?.readyState === 'closed') { discardConnection(); installConnection(); } signal = config.createSignaling(); signalQueue = Promise.resolve(); signal.on('signal', (packet: SignalingPacket) => { signalQueue = signalQueue.then(() => received(packet, attempt)).catch(error => failure(error, attempt)); }); signal.on('signal:peers', ({ count }: { count: number }) => { if (attempt !== generation || count < 2 || config.slot !== 0 || offering || !pc) return; offering = true; const current = pc; void (async () => { const offer = await current.createOffer(); if (disposed || attempt !== generation) return; await current.setLocalDescription(offer); await sendSignal('offer', current.localDescription!, attempt); })().catch(error => failure(error, attempt)); }); signal.on('connect_error', error => failure(error, attempt)); signal.on('disconnect', reason => { if (!ready && reason === 'io server disconnect') failure(new Error('SIGNAL_CLOSED'), attempt); }); signal.connect(); emitState('connecting'); return currentPromise; } return { connect, get epoch() { return epoch; }, get connected() { return ready && channel?.readyState === 'open'; }, send(message: PeerMessage) { if (!ready || channel?.readyState !== 'open') throw new Error('PEER_NOT_CONNECTED'); const body = JSON.stringify(message); if (new TextEncoder().encode(body).byteLength > MAX_PEER_MESSAGE_BYTES) throw new Error('MESSAGE_TOO_LARGE'); channel.send(body); }, onMessage(handler: (message: PeerMessage) => void) { messages.add(handler); return () => { messages.delete(handler); }; }, onState(handler: (state: PeerState) => void) { states.add(handler); return () => { states.delete(handler); }; }, async restart(nextEpoch = epoch + 1) { generation++; stopSignaling(); clearTimeout(timer); no?.(new Error('PEER_RESTARTED')); ready = false; remoteReady = false; yes = undefined; no = undefined; connectionPromise = undefined; epoch = nextEpoch; candidates = []; if (channel?.readyState === 'closed' || channel?.readyState === 'closing' || pc?.connectionState === 'closed') discardConnection(); if (config.refreshIce) { config.iceServers = await config.refreshIce(); if (disposed) throw new Error('PEER_CLOSED'); if (pc) { try { pc.setConfiguration({ ...pc.getConfiguration(), iceServers: config.iceServers }); } catch { discardConnection(); } } } pc?.restartIce(); return connect(); }, stats: async () => pc?.getStats(), close() { disposed = true; generation++; ready = false; clearTimeout(timer); stopSignaling(); no?.(new Error('PEER_CLOSED')); no = undefined; yes = undefined; channel?.close(); pc?.close(); emitState('closed'); messages.clear(); states.clear(); }, }; } export type PeerConnection = ReturnType;