import { SIGNALING_TYPE_ANSWER, SIGNALING_TYPE_OFFER, TRACK_KIND_AUDIO, TRACK_KIND_VIDEO, TRANSCEIVER_SEND_RECV } from '../../realtime.constants'; import { NegotiationHandlers, PeerConnectionManagerContext, PeerConnectionManagerState } from '../shared'; import type { PeerData } from '../../realtime.types'; /** * Queue a negotiation task so SDP operations for a single peer never overlap. */ export function enqueueNegotiation( state: PeerConnectionManagerState, peerId: string, task: () => Promise ): void { const previousTask = state.peerNegotiationQueue.get(peerId) ?? Promise.resolve(); const nextTask = previousTask.then(task, task); state.peerNegotiationQueue.set(peerId, nextTask); } export async function doCreateAndSendOffer( context: PeerConnectionManagerContext, remotePeerId: string ): Promise { const { callbacks, logger, state } = context; const peerData = state.activePeerConnections.get(remotePeerId); if (!peerData) return; try { const offer = await peerData.connection.createOffer(); await peerData.connection.setLocalDescription(offer); logger.info('Sending offer', { remotePeerId, type: offer.type, sdpLength: offer.sdp?.length }); callbacks.sendRawMessage({ type: SIGNALING_TYPE_OFFER, targetUserId: remotePeerId, payload: { sdp: offer } }); } catch (error) { logger.error('Failed to create offer', error, { localDescriptionType: peerData.connection.localDescription?.type ?? null, remotePeerId, signalingState: peerData.connection.signalingState }); } } /** * Replace a peer whose underlying connection is `failed` or `closed`. * Returns the existing peer if still usable, or `undefined` after cleanup. */ function replaceUnusablePeer( context: PeerConnectionManagerContext, peerId: string, reason: string ): void { const { logger, state } = context; const peerData = state.activePeerConnections.get(peerId); if (!peerData) return; const cs = peerData.connection.connectionState; if (cs !== 'failed' && cs !== 'closed') return; logger.info('Replacing unusable peer', { connectionState: cs, peerId, reason, signalingState: peerData.connection.signalingState }); try { peerData.connection.close(); } catch { /* already closing */ } state.activePeerConnections.delete(peerId); state.peerNegotiationQueue.delete(peerId); } function getOrCreatePeerForOffer( state: PeerConnectionManagerState, fromUserId: string, handlers: NegotiationHandlers ): PeerData { return state.activePeerConnections.get(fromUserId) ?? handlers.createPeerConnection(fromUserId, false); } async function resolveOfferCollision( peerData: PeerData, callbacks: PeerConnectionManagerContext['callbacks'], logger: PeerConnectionManagerContext['logger'], fromUserId: string ): Promise { const signalingState = peerData.connection.signalingState; const hasCollision = signalingState === 'have-local-offer' || signalingState === 'have-local-pranswer'; if (!hasCollision) return true; const localOderId = callbacks.getIdentifyCredentials()?.oderId ?? null; const isPolite = !localOderId || localOderId > fromUserId; if (!isPolite) { logger.info('Ignoring colliding offer (impolite side)', { fromUserId, localOderId }); return false; } logger.info('Rolling back local offer (polite side)', { fromUserId, localOderId }); await peerData.connection.setLocalDescription({ type: 'rollback' } as RTCSessionDescriptionInit); return true; } function syncPeerSendersFromTransceivers(peerData: PeerData): void { const transceivers = peerData.connection.getTransceivers(); for (const transceiver of transceivers) { const receiverKind = transceiver.receiver.track?.kind; if (receiverKind === TRACK_KIND_AUDIO) { if (!peerData.audioSender) { peerData.audioSender = transceiver.sender; } transceiver.direction = TRANSCEIVER_SEND_RECV; continue; } if (receiverKind === TRACK_KIND_VIDEO && !peerData.videoSender) { peerData.videoSender = transceiver.sender; } } } async function attachAnswererLocalTracks( peerData: PeerData, localStream: MediaStream | null, logger: PeerConnectionManagerContext['logger'], fromUserId: string ): Promise { if (!localStream) return; logger.logStream(`localStream->${fromUserId} (answerer)`, localStream); for (const track of localStream.getTracks()) { if (track.kind === TRACK_KIND_AUDIO && peerData.audioSender) { await peerData.audioSender.replaceTrack(track); logger.info('audio replaceTrack (answerer) ok', { fromUserId }); continue; } if (track.kind === TRACK_KIND_VIDEO && peerData.videoSender) { await peerData.videoSender.replaceTrack(track); logger.info('video replaceTrack (answerer) ok', { fromUserId }); } } } async function applyPendingIceCandidates(peerData: PeerData): Promise { for (const candidate of peerData.pendingIceCandidates) { await peerData.connection.addIceCandidate(new RTCIceCandidate(candidate)); } peerData.pendingIceCandidates = []; } export async function doHandleOffer( context: PeerConnectionManagerContext, fromUserId: string, sdp: RTCSessionDescriptionInit, handlers: NegotiationHandlers ): Promise { const { callbacks, logger, state } = context; logger.info('Handling offer', { fromUserId }); replaceUnusablePeer(context, fromUserId, 'incoming offer'); const peerData = getOrCreatePeerForOffer(state, fromUserId, handlers); try { const shouldApplyOffer = await resolveOfferCollision(peerData, callbacks, logger, fromUserId); if (!shouldApplyOffer) return; await peerData.connection.setRemoteDescription(new RTCSessionDescription(sdp)); syncPeerSendersFromTransceivers(peerData); await attachAnswererLocalTracks(peerData, callbacks.getLocalMediaStream(), logger, fromUserId); await applyPendingIceCandidates(peerData); const answer = await peerData.connection.createAnswer(); await peerData.connection.setLocalDescription(answer); logger.info('Sending answer', { to: fromUserId, type: answer.type, sdpLength: answer.sdp?.length }); callbacks.sendRawMessage({ type: SIGNALING_TYPE_ANSWER, targetUserId: fromUserId, payload: { sdp: answer } }); } catch (error) { logger.error('Failed to handle offer', error, { fromUserId, pendingIceCandidates: peerData.pendingIceCandidates.length, sdpLength: sdp.sdp?.length, signalingState: peerData.connection.signalingState }); } } export async function doHandleAnswer( context: PeerConnectionManagerContext, fromUserId: string, sdp: RTCSessionDescriptionInit ): Promise { const { logger, state } = context; logger.info('Handling answer', { fromUserId }); const peerData = state.activePeerConnections.get(fromUserId); if (!peerData) { logger.error('No peer for answer', new Error('Missing peer'), { fromUserId }); return; } try { if (peerData.connection.signalingState === 'have-local-offer') { await peerData.connection.setRemoteDescription(new RTCSessionDescription(sdp)); for (const candidate of peerData.pendingIceCandidates) { await peerData.connection.addIceCandidate(new RTCIceCandidate(candidate)); } peerData.pendingIceCandidates = []; } else { logger.warn('Ignoring answer - wrong signaling state', { state: peerData.connection.signalingState }); } } catch (error) { logger.error('Failed to handle answer', error, { fromUserId, pendingIceCandidates: peerData.pendingIceCandidates.length, sdpLength: sdp.sdp?.length, signalingState: peerData.connection.signalingState }); } } export async function doHandleIceCandidate( context: PeerConnectionManagerContext, fromUserId: string, candidate: RTCIceCandidateInit, handlers: NegotiationHandlers ): Promise { const { logger, state } = context; replaceUnusablePeer(context, fromUserId, 'early ICE'); let peerData = state.activePeerConnections.get(fromUserId); if (!peerData) { logger.info('Creating peer for early ICE', { fromUserId }); peerData = handlers.createPeerConnection(fromUserId, false); } try { if (peerData.connection.remoteDescription) { await peerData.connection.addIceCandidate(new RTCIceCandidate(candidate)); } else { logger.info('Queuing ICE candidate', { fromUserId }); peerData.pendingIceCandidates.push(candidate); } } catch (error) { logger.error('Failed to add ICE candidate', error, { candidateMid: candidate.sdpMid ?? null, candidateMLineIndex: candidate.sdpMLineIndex ?? null, fromUserId, hasRemoteDescription: !!peerData.connection.remoteDescription, pendingIceCandidates: peerData.pendingIceCandidates.length }); } } export async function doRenegotiate( context: PeerConnectionManagerContext, peerId: string ): Promise { const { callbacks, logger, state } = context; const peerData = state.activePeerConnections.get(peerId); if (!peerData) return; try { const offer = await peerData.connection.createOffer(); await peerData.connection.setLocalDescription(offer); logger.info('Renegotiate offer', { peerId, type: offer.type, sdpLength: offer.sdp?.length }); callbacks.sendRawMessage({ type: SIGNALING_TYPE_OFFER, targetUserId: peerId, payload: { sdp: offer } }); } catch (error) { logger.error('Failed to renegotiate', error, { peerId, signalingState: peerData.connection.signalingState }); } }