Toju/toju-app/src/app/infrastructure/realtime/peer-connection-manager/connection/negotiation.ts

import {
  SIGNALING_TYPE_ANSWER,
  SIGNALING_TYPE_OFFER,
  TRACK_KIND_AUDIO,
  TRACK_KIND_VIDEO,
  TRANSCEIVER_SEND_RECV
} from '../../realtime.constants';
import { isPoliteOnOfferCollision } from '../../peer-role.rules';
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>
): 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<void> {
  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<boolean> {
  const signalingState = peerData.connection.signalingState;
  const hasCollision = signalingState === 'have-local-offer' || signalingState === 'have-local-pranswer';

  if (!hasCollision)
    return true;

  const localOderId = callbacks.getIdentifyCredentialsForPeer(fromUserId)?.oderId ?? null;
  const isPolite = isPoliteOnOfferCollision(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<void> {
  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<void> {
  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<void> {
  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<void> {
  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<void> {
  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<void> {
  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
    });
  }
}