import { RULES_VERSION, TIMEOUT_CHANCES } from '../../constants/gameplay.js'; import { RECONNECT_GRACE_MS } from '../../constants/session.js'; import { PEER_PING_MS, PEER_SILENCE_MS, PROTOCOL_TIMEOUT_MS, PROTOCOL_VERSION, RESUME_START_DELAY_MS } from '../../constants/webrtc.js'; import type { MatchEngine } from '../shared/match-engine.js'; import { createTurnExchange, digestValue, type TurnExchange } from '../shared/peer-protocol.js'; import type { MatchCheckpoint, PeerMessage, PublicCheckpoint } from '../shared/p2p-types.js'; import type { ActionId, Pair } from '../shared/types.js'; import type { GameClock } from './match-controller.js'; import type { PeerConnection } from './peer.js'; import { matchIntroductionDuration, matchPreparationDuration } from '../shared/match-introduction.js'; export type GamePeer = Pick; interface FriendConfig { matchId: string; slot: 0 | 1; engine: MatchEngine; peer: GamePeer; clock: GameClock; setOffset(offset: number): void; changed(): void; checkpointChanged(): void; finished(): void; saved?: { confirmed?: MatchCheckpoint; epoch: number }; onEmote?(id: unknown, variant: unknown): void; } export function createFriendMatch(config: FriendConfig) { const { peer, engine, clock } = config; let stopped = false, recovering = false, offset = 0, localReady = false, peerReady = false, begun = false; let confirmed = config.saved?.confirmed ?? engine.checkpoint(); let exchange: TurnExchange | undefined, ownDigest = '', remoteDigest = '', remoteAck = '', sentAck = false, presented = false; let timeoutReports: Pair = [null, null]; let pendingTurnId: number | undefined, pendingCommit: Extract | undefined; let presentAt: number | undefined, privateChoice: { action: ActionId | null; nonce: string } | undefined; let resumedPeer = false, resumedAck = false, lastPeerAt = clock.now(), queue = Promise.resolve(); let leaveResolve: (() => void) | undefined; let leaving = false; const timers = new Map(); const clockReplies = new Map(); const now = () => clock.now() + offset; function clear(name: string) { clock.clearTimeout(timers.get(name)); timers.delete(name); } function later(name: string, fn: () => void, ms: number) { clear(name); timers.set(name, clock.setTimeout(() => { timers.delete(name); if (!stopped) fn(); }, Math.max(0, ms))); } function clearAll() { for (const name of [...timers.keys()]) clear(name); } function handleError() { if (stopped) return; if (!peer.connected) { void recover(); return; } abort('protocol_error'); } function run(fn: () => unknown | Promise) { void Promise.resolve().then(fn).catch(handleError); } function send(type: PeerMessage['type'], fields: Record = {}) { peer.send({ protocolVersion: PROTOCOL_VERSION, rulesVersion: RULES_VERSION, matchId: config.matchId, connectionEpoch: peer.epoch, turnId: engine.view().turnId, type, ...fields } as PeerMessage); } function abort(reason: 'connection_failed' | 'protocol_error') { if (stopped) return; clearAll(); recovering = false; // An unconfirmed calculation is not a final result and may be discarded. if (!presented && confirmed.state.phase !== 'finished') engine.restore(confirmed); engine.abort(reason); config.changed(); config.finished(); stopped = true; peer.close(); leaveResolve?.(); } function armProtocol() { later('protocol', () => { void recover(); }, PROTOCOL_TIMEOUT_MS); } function publicCheckpoint(): PublicCheckpoint { const state = confirmed.state; return { turnId: state.phase === 'revealing' ? state.turnId + 1 : Math.max(1, state.turnId), energy: [...state.energy], timeouts: state.timeouts ?? [0, 0], finished: state.phase === 'finished', winnerIndex: state.finished?.winnerIndex ?? null, ...(state.finished ? { reason: state.finished.reason } : {}) }; } function sameCheckpoint(a: PublicCheckpoint, b: PublicCheckpoint) { return a.turnId === b.turnId && a.energy?.[0] === b.energy[0] && a.energy?.[1] === b.energy[1] && a.finished === b.finished && a.winnerIndex === b.winnerIndex && a.reason === b.reason; } function heartbeat() { if (stopped || recovering) return; if (clock.now() - lastPeerAt >= PEER_SILENCE_MS) { void recover(); return; } try { send('ping', { sentAt: clock.now() }); } catch { void recover(); return; } later('heartbeat', heartbeat, PEER_PING_MS); } async function synchronize() { if (config.slot === 0) return; let bestRtt = Infinity, bestOffset = offset; for (let i = 0; i < 3; i++) { const sentAt = clock.now(); const response = new Promise((resolve, reject) => clockReplies.set(sentAt, { resolve, reject })); later('clock', () => { clockReplies.get(sentAt)?.reject(new Error('CLOCK_TIMEOUT')); clockReplies.delete(sentAt); }, PROTOCOL_TIMEOUT_MS); send('clock:ping', { sentAt }); const remoteAt = await response; clear('clock'); const rtt = clock.now() - sentAt; if (rtt < bestRtt) { bestRtt = rtt; bestOffset = remoteAt - (sentAt + clock.now()) / 2; } } offset = bestOffset; config.setOffset(offset); } function present(at: number) { if (presented) return; presented = true; presentAt = undefined; clear('protocol'); engine.present(at); confirmed = engine.checkpoint(); config.changed(); if (engine.view().phase === 'finished') config.finished(); else if (config.slot === 0) later('next', () => startTurn(engine.view().reveal!.revealEndsAt), engine.view().reveal!.revealEndsAt - now()); } function agree() { if (!ownDigest || !remoteDigest) return; if (ownDigest !== remoteDigest) { abort('protocol_error'); return; } if (!sentAck) { sentAck = true; send('turn:ack', { digest: ownDigest }); } if (remoteAck && remoteAck !== ownDigest) { abort('protocol_error'); return; } if (remoteAck !== ownDigest || presented) return; // Agreement survives loss of the presentation packet and a page reload. confirmed = engine.checkpoint(); config.changed(); if (confirmed.state.phase === 'finished') config.finished(); if (config.slot === 0) { const at = now(); send('turn:present', { revealedAt: at }); present(at); } else if (presentAt !== undefined) present(presentAt); } async function resolved(actions: Pair) { clear('deadline'); if (engine.view().phase !== 'finished') engine.resolve(actions); const state = engine.view(); ownDigest = await digestValue([config.matchId, peer.epoch, state.turnId, state.energy, state.timeouts, actions, state.finished?.winnerIndex ?? null, state.finished?.reason ?? null]); if (stopped || recovering) return; send('turn:resolved', { digest: ownDigest }); agree(); } function applyTimeout(deadline: number) { const state = engine.view(); if (state.phase !== 'selecting' || deadline !== state.deadline || timeoutReports.includes(null)) return; const missing = timeoutReports as Pair; if (!missing.some(Boolean)) { armProtocol(); return; } clear('protocol'); const extended = engine.expireSelection(missing, deadline); // Persist timeout evidence even before terminal digest agreement. Recovery // must complete the third timeout, never restore another selection window. confirmed.state.timeouts = [...engine.view().timeouts!]; config.checkpointChanged(); if (extended) { config.changed(); armDeadline(); } else if (engine.view().phase === 'finished') { run(() => resolved([null, null])); armProtocol(); } } function armDeadline() { const deadline = engine.view().deadline; timeoutReports = [null, null]; later('deadline', () => { if (!peer.connected) { void recover(); return; } armProtocol(); run(() => { const missing = engine.view().ownAction === null; timeoutReports[config.slot] = missing; // Ordered delivery puts a last-moment commitment before this report. send('turn:timeout', { deadline, missing, slot: config.slot }); applyTimeout(deadline); }); }, deadline - now()); } function resetAgreement() { privateChoice = undefined; ownDigest = remoteDigest = remoteAck = ''; sentAck = false; presented = false; presentAt = undefined; } function selection() { const earlyCommit = pendingCommit; pendingTurnId = undefined; pendingCommit = undefined; resetAgreement(); const state = engine.view(); exchange = createTurnExchange({ matchId: config.matchId, epoch: peer.epoch, turnId: state.turnId, slot: config.slot, energy: state.energy, send: message => peer.send(message), onResolved: actions => run(() => resolved(actions)), onPrivateState: value => { privateChoice = value; config.changed(); }, onPhase: phase => { if (phase !== 'commit' || now() >= engine.view().deadline) armProtocol(); } }); if (earlyCommit) void exchange.receive(earlyCommit).catch(handleError); config.changed(); armDeadline(); } function applyStart(startsAt: number, resume = false) { if (!Number.isFinite(startsAt)) { abort('protocol_error'); return; } if (resume) { engine.resumeTurn(confirmed, startsAt); recovering = false; clear('recovery'); engine.presence(true); if (engine.view().phase === 'finished') { resetAgreement(); run(() => resolved([null, null])); armProtocol(); } else selection(); heartbeat(); return; } pendingTurnId = engine.view().turnId + 1; pendingCommit = undefined; if (!begun) { begun = true; engine.start(startsAt - matchPreparationDuration(engine.view().settings)); confirmed = engine.checkpoint(); config.changed(); } later('start', () => { if (engine.beginTurn(startsAt)) selection(); }, startsAt - now()); } function startTurn(startsAt: number, resume = false) { const turnId = resume ? publicCheckpoint().turnId : engine.view().turnId + 1; send('turn:start', { startsAt, resume, turnId }); applyStart(startsAt, resume); } function maybeStart() { if (localReady && peerReady && config.slot === 0 && !begun && !recovering) startTurn(now() + matchPreparationDuration(engine.view().settings) + matchIntroductionDuration(engine.view().settings, false)); } function maybeResume() { if (!recovering || !resumedPeer || !resumedAck) return; if (confirmed.state.phase === 'finished') { engine.restore(confirmed); recovering = false; clear('recovery'); config.changed(); config.finished(); return; } if (config.slot === 0) startTurn(now() + RESUME_START_DELAY_MS, true); } async function recover(disconnectedAt = clock.now()) { if (stopped || recovering) return; if (confirmed.state.phase === 'finished') { clearAll(); if (engine.view().phase !== 'finished') engine.restore(confirmed); // A confirmed result keeps its original reveal window when the peer leaves. engine.presence(false); config.changed(); config.finished(); return; } recovering = true; resumedPeer = resumedAck = false; clearAll(); exchange = undefined; pendingTurnId = undefined; pendingCommit = undefined; engine.restore(confirmed); engine.presence(false); config.changed(); later('recovery', () => abort('connection_failed'), RECONNECT_GRACE_MS - (clock.now() - disconnectedAt)); try { await peer.restart(peer.epoch + 1); if (stopped) return; lastPeerAt = clock.now(); await synchronize(); if (stopped) return; send('resume', { checkpoint: publicCheckpoint() }); } catch { abort('connection_failed'); } } async function receive(message: PeerMessage) { if (stopped || !message || message.matchId !== config.matchId || message.protocolVersion !== PROTOCOL_VERSION || message.rulesVersion !== RULES_VERSION || message.connectionEpoch !== peer.epoch) return; lastPeerAt = clock.now(); if (message.type === 'ping' || message.type === 'clock:ping') { send(message.type === 'ping' ? 'pong' : 'clock:pong', { sentAt: message.sentAt, remoteAt: now() }); return; } if (message.type === 'clock:pong') { if (Number.isFinite(message.remoteAt)) clockReplies.get(message.sentAt)?.resolve(message.remoteAt); clockReplies.delete(message.sentAt); return; } if (message.type === 'pong') return; if (message.type === 'emote') { if (!recovering && engine.view().phase !== 'finished' && (message.variant === 0 || message.variant === 1)) config.onEmote?.(message.id, message.variant); return; } if (message.type === 'ready') { if (message.slot === config.slot) throw new Error('INVALID_SLOT'); peerReady = true; maybeStart(); return; } if (message.type === 'resume' || message.type === 'resume:ack') { if (!recovering) return; if (!sameCheckpoint(message.checkpoint, publicCheckpoint())) { abort('protocol_error'); return; } const counts = message.checkpoint.timeouts ?? [0, 0]; if (!Array.isArray(counts) || counts.length !== 2 || counts.some(count => !Number.isInteger(count) || count < 0 || count > TIMEOUT_CHANCES)) throw new Error('INVALID_TIMEOUT'); // A disconnect can lose one timeout report. Keep the highest observed count // while discarding unconfirmed choices, so reconnect never refunds a chance. confirmed.state.timeouts = counts.map((count, slot) => Math.max(count, confirmed.state.timeouts?.[slot] ?? 0)) as Pair; config.checkpointChanged(); if (message.type === 'resume') { resumedPeer = true; send('resume:ack', { checkpoint: publicCheckpoint() }); } else resumedAck = true; maybeResume(); return; } if (message.type === 'turn:start') { if (config.slot !== 1) throw new Error('INVALID_ROLE'); if (message.resume) { if (recovering && resumedPeer && resumedAck) applyStart(message.startsAt, true); return; } if (recovering || message.turnId !== engine.view().turnId + 1) return; applyStart(message.startsAt); return; } if (message.type === 'match:leave') { send('match:leave-ack'); clearAll(); engine.finish(leaving ? null : config.slot, 'forfeit'); confirmed = engine.checkpoint(); presented = true; config.changed(); config.finished(); return; } if (message.type === 'match:leave-ack') { leaveResolve?.(); return; } if (message.type === 'match:rematch') { engine.requestRematch(config.slot === 0 ? 1 : 0); config.changed(); return; } if (recovering) return; // A peer may select before this browser's scheduled turn-start callback runs. if (message.type === 'turn:commit' && message.turnId === pendingTurnId) { if (message.slot === config.slot) throw new Error('INVALID_SLOT'); if (pendingCommit && pendingCommit.digest !== message.digest) throw new Error('COMMITMENT_CHANGED'); pendingCommit = message; return; } if (message.turnId !== engine.view().turnId) return; if (message.type === 'turn:timeout') { if (message.slot !== (config.slot === 0 ? 1 : 0) || typeof message.missing !== 'boolean') throw new Error('INVALID_TIMEOUT'); if (message.deadline !== engine.view().deadline) return; if (timeoutReports[message.slot] !== null && timeoutReports[message.slot] !== message.missing) throw new Error('TIMEOUT_CHANGED'); timeoutReports[message.slot] = message.missing; applyTimeout(message.deadline); return; } if (message.type === 'turn:commit' || message.type === 'turn:reveal') { await exchange?.receive(message); return; } if (message.type === 'turn:resolved') { if (remoteDigest && remoteDigest !== message.digest) throw new Error('RESULT_CHANGED'); remoteDigest = message.digest; agree(); } else if (message.type === 'turn:ack') { remoteAck = message.digest; agree(); } else if (message.type === 'turn:present') { if (config.slot !== 1 || !Number.isFinite(message.revealedAt)) throw new Error('INVALID_PRESENT'); presentAt = message.revealedAt; agree(); } } const offMessage = peer.onMessage(message => { queue = queue.then(() => receive(message)).catch(handleError); }); const offState = peer.onState(state => { if (state === 'disconnected' || state === 'failed') { const disconnectedAt = clock.now(); void queue.then(() => recover(disconnectedAt)); } }); return { async start() { if (config.saved?.confirmed) { begun = true; engine.restore(confirmed); await recover(); return; } try { await peer.connect(); await synchronize(); if (stopped) return; localReady = true; send('ready', { slot: config.slot }); maybeStart(); lastPeerAt = clock.now(); heartbeat(); } catch { abort('connection_failed'); } }, get recovering() { return recovering; }, sendEmote(id: import('../../constants/emotes.js').EmoteId, variant: 0 | 1) { if (stopped || recovering || !peer.connected || engine.view().phase === 'finished') return false; try { send('emote', { id, variant }); return true; } catch { return false; } }, submitAction() { if (!recovering && !stopped) run(() => exchange!.commit(engine.view().ownAction)); }, requestRematch() { engine.requestRematch(config.slot); if (peer.connected) send('match:rematch'); config.changed(); }, snapshot: () => ({ confirmed: structuredClone(confirmed), epoch: peer.epoch, privateChoice }), async leave() { if (engine.view().phase === 'finished') return; leaving = true; try { await new Promise((resolve, reject) => { leaveResolve = resolve; later('leave', () => reject(new Error('LEAVE_TIMEOUT')), PROTOCOL_TIMEOUT_MS); send('match:leave'); }); clearAll(); engine.finish(config.slot === 0 ? 1 : 0, 'forfeit'); confirmed = engine.checkpoint(); presented = true; config.changed(); config.finished(); } catch { abort('connection_failed'); } finally { leaveResolve = undefined; } }, destroy() { stopped = true; clearAll(); offMessage(); offState(); for (const pending of clockReplies.values()) pending.reject(new Error('CLOSED')); clockReplies.clear(); peer.close(); }, }; }