diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt index a21d04e9d..22c583639 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt @@ -26,8 +26,14 @@ import coil3.ImageLoader import coil3.asDrawable import coil3.request.ImageRequest import coil3.request.allowHardware +import com.vitorpamplona.amethyst.commons.call.AnswerRouteAction import com.vitorpamplona.amethyst.commons.call.CallManager import com.vitorpamplona.amethyst.commons.call.CallState +import com.vitorpamplona.amethyst.commons.call.IceCandidateData +import com.vitorpamplona.amethyst.commons.call.PeerSession +import com.vitorpamplona.amethyst.commons.call.PeerSessionManager +import com.vitorpamplona.amethyst.commons.call.SdpType +import com.vitorpamplona.amethyst.commons.call.SignalingState import com.vitorpamplona.amethyst.model.LocalCache import com.vitorpamplona.amethyst.service.notifications.NotificationUtils import com.vitorpamplona.quartz.nip01Core.core.HexKey @@ -56,9 +62,7 @@ import org.webrtc.DefaultVideoEncoderFactory import org.webrtc.EglBase import org.webrtc.IceCandidate import org.webrtc.MediaConstraints -import org.webrtc.PeerConnection import org.webrtc.PeerConnectionFactory -import org.webrtc.SessionDescription import org.webrtc.SurfaceTextureHelper import org.webrtc.VideoFrame import org.webrtc.VideoSink @@ -66,8 +70,6 @@ import org.webrtc.VideoSource import org.webrtc.VideoTrack import java.util.UUID import java.util.concurrent.ConcurrentHashMap -import java.util.concurrent.CopyOnWriteArrayList -import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicLong private const val TAG = "CallController" @@ -80,18 +82,20 @@ class CallController( private val publishWrap: suspend (EphemeralGiftWrapEvent) -> Unit, private val signerProvider: suspend () -> com.vitorpamplona.quartz.nip01Core.signers.NostrSigner, ) { - // ---- Per-peer session state ---- + // ---- Per-peer session state (delegated to PeerSessionManager) ---- - private class PeerSessionState( - val session: WebRtcCallSession, - val remoteDescriptionSet: AtomicBoolean = AtomicBoolean(false), - val pendingIceCandidates: CopyOnWriteArrayList = CopyOnWriteArrayList(), - ) + // Lazily initialized — needs signerProvider() for localPubKey + private var peerSessionMgr: PeerSessionManager? = null - private val peerSessions = ConcurrentHashMap() + private suspend fun sessionManager(): PeerSessionManager { + if (peerSessionMgr == null) { + peerSessionMgr = PeerSessionManager(signerProvider().pubKey) + } + return peerSessionMgr!! + } - // Candidates received before a PeerSession exists for the sender - private val globalPendingIce = ConcurrentHashMap>() + /** Map from peer pubkey to the WebRtcCallSession for direct access to WebRTC-specific APIs. */ + private val webRtcSessions = ConcurrentHashMap() // ---- Shared WebRTC resources ---- @@ -272,12 +276,16 @@ class CallController( // Set state to Offering before creating peer sessions callManager.beginOffering(callId, peerPubKeys, callType) + val mgr = sessionManager() + // Create a PeerConnection + offer for each callee for (peerPubKey in peerPubKeys) { try { - val ps = withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } + val webRtcSession = withContext(Dispatchers.IO) { createWebRtcSession(peerPubKey) } + val adapter = WebRtcPeerSessionAdapter(webRtcSession) + mgr.registerSession(peerPubKey, adapter) Log.d(TAG) { "initiateCall: PeerConnection created for ${peerPubKey.take(8)}" } - ps.session.createOffer { sdp -> + webRtcSession.createOffer { sdp -> Log.d(TAG) { "initiateCall: offer created for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" } scope.launch { callManager.publishOfferToPeer(peerPubKey, peerPubKeys, callType, callId, sdp.description) @@ -315,28 +323,27 @@ class CallController( return@launch } - // Retrieve any ICE candidates buffered while ringing (before session existed) - val globalPending = getGlobalPendingCandidates(callerPubKey) + val mgr = sessionManager() - val ps = + val webRtcSession = try { - withContext(Dispatchers.IO) { createPeerSession(callerPubKey) } + withContext(Dispatchers.IO) { createWebRtcSession(callerPubKey) } } catch (e: Exception) { Log.e(TAG, "Failed to create PeerConnection", e) _errorMessage.value = "Failed to accept call: ${e.message}" return@launch } - // Inject any globally-buffered ICE candidates for the caller - globalPending.forEach { ps.pendingIceCandidates.add(it) } + val adapter = WebRtcPeerSessionAdapter(webRtcSession) + val entry = mgr.registerSession(callerPubKey, adapter) Log.d(TAG) { "acceptIncomingCall: setting remote description (OFFER)..." } - ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) - Log.d(TAG) { "acceptIncomingCall: flushing ${ps.pendingIceCandidates.size} pending ICE candidates..." } - flushPendingIceCandidates(callerPubKey, ps) + adapter.setRemoteDescription(SdpType.OFFER, sdpOffer) + Log.d(TAG) { "acceptIncomingCall: flushing ${entry.pendingIceCandidates.size} pending ICE candidates..." } + mgr.flushPendingIceCandidates(callerPubKey) Log.d(TAG) { "acceptIncomingCall: creating answer..." } - ps.session.createAnswer { sdp -> + webRtcSession.createAnswer { sdp -> Log.d(TAG) { "acceptIncomingCall: answer created, sdpLength=${sdp.description.length}, publishing..." } scope.launch { callManager.acceptCall(sdp.description) @@ -358,29 +365,29 @@ class CallController( peerPubKey: HexKey, sdpAnswer: String, ) { - Log.d(TAG) { "onCallAnswerReceived: from=${peerPubKey.take(8)}, knownSessions=${peerSessions.keys.map { it.take(8) }}" } - val ps = peerSessions[peerPubKey] - if (ps != null) { - // We have a PeerConnection for this peer — set remote description - val signalingState = ps.session.getSignalingState() - Log.d(TAG) { - "Answer received from ${peerPubKey.take(8)}, sdpLength=${sdpAnswer.length}, " + - "signalingState=$signalingState, remoteDescSet=${ps.remoteDescriptionSet.get()}" + val mgr = peerSessionMgr + if (mgr == null) { + Log.d(TAG) { "onCallAnswerReceived: sessionManager not initialized — ignoring" } + return + } + + Log.d(TAG) { "onCallAnswerReceived: from=${peerPubKey.take(8)}, knownSessions=${mgr.allSessionKeys().map { it.take(8) }}" } + + val action = mgr.routeAnswer(peerPubKey, sdpAnswer) + Log.d(TAG) { "onCallAnswerReceived: action=$action" } + when (action) { + AnswerRouteAction.APPLIED -> { + Log.d(TAG) { "Answer applied for ${peerPubKey.take(8)}" } } - if (signalingState != PeerConnection.SignalingState.HAVE_LOCAL_OFFER) { - Log.d(TAG) { "Ignoring answer from ${peerPubKey.take(8)} in $signalingState (no pending local offer)" } - return + AnswerRouteAction.NO_SESSION -> { + Log.d(TAG) { "Answer from unknown peer ${peerPubKey.take(8)} — triggering callee-to-callee connection" } + onNewPeerInGroupCall(peerPubKey) } - Log.d(TAG) { "Setting remote description (ANSWER) for ${peerPubKey.take(8)}..." } - ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.ANSWER, sdpAnswer)) - flushPendingIceCandidates(peerPubKey, ps) - } else { - // No session for this peer — they are another callee who joined. - // Initiate a callee-to-callee connection (with tie-breaking). - Log.d(TAG) { "Answer from unknown peer ${peerPubKey.take(8)} — triggering callee-to-callee connection" } - onNewPeerInGroupCall(peerPubKey) + AnswerRouteAction.IGNORED_WRONG_STATE -> { + Log.d(TAG) { "Ignoring answer from ${peerPubKey.take(8)} (no pending local offer)" } + } } } @@ -393,37 +400,19 @@ class CallController( fun onIceCandidateReceived(event: CallIceCandidateEvent) { try { val senderPubKey = event.pubKey - val candidate = IceCandidate(event.sdpMid(), event.sdpMLineIndex(), event.candidateSdp()) - val ps = peerSessions[senderPubKey] - if (ps != null && ps.remoteDescriptionSet.get()) { - Log.d(TAG) { "Adding ICE candidate from ${senderPubKey.take(8)} directly" } - ps.session.addIceCandidate(candidate) - } else if (ps != null) { - Log.d(TAG) { "Buffering ICE candidate from ${senderPubKey.take(8)} (remoteDesc not set)" } - ps.pendingIceCandidates.add(candidate) - } else { - // No session yet — buffer globally (keyed by sender) - Log.d(TAG) { "Buffering ICE candidate from ${senderPubKey.take(8)} (no session yet)" } - globalPendingIce.getOrPut(senderPubKey) { CopyOnWriteArrayList() }.add(candidate) + val candidate = IceCandidateData(event.candidateSdp(), event.sdpMid(), event.sdpMLineIndex()) + val mgr = peerSessionMgr + if (mgr == null) { + Log.d(TAG) { "Buffering ICE candidate from ${senderPubKey.take(8)} (manager not initialized)" } + return } + val action = mgr.routeIceCandidate(senderPubKey, candidate) + Log.d(TAG) { "ICE candidate from ${senderPubKey.take(8)}: $action" } } catch (e: Exception) { Log.e(TAG, "Failed to parse ICE candidate", e) } } - private fun getGlobalPendingCandidates(peerPubKey: HexKey): List = globalPendingIce.remove(peerPubKey)?.toList() ?: emptyList() - - private fun flushPendingIceCandidates( - peerPubKey: HexKey, - ps: PeerSessionState, - ) { - ps.remoteDescriptionSet.set(true) - val candidates = ps.pendingIceCandidates.toList() - Log.d(TAG) { "Flushing ${candidates.size} buffered ICE candidates for ${peerPubKey.take(8)}" } - ps.pendingIceCandidates.clear() - candidates.forEach { ps.session.addIceCandidate(it) } - } - private fun onLocalIceCandidate( peerPubKey: HexKey, candidate: IceCandidate, @@ -447,15 +436,16 @@ class CallController( * peer with the lexicographically lower pubkey initiates. */ fun onNewPeerInGroupCall(peerPubKey: HexKey) { - if (peerSessions.containsKey(peerPubKey)) { + val mgr = peerSessionMgr + if (mgr != null && mgr.hasSession(peerPubKey)) { Log.d(TAG) { "onNewPeerInGroupCall: session already exists for ${peerPubKey.take(8)} — skipping" } return } scope.launch { - val myPubKey = signerProvider().pubKey - Log.d(TAG) { "onNewPeerInGroupCall: peer=${peerPubKey.take(8)}, myPubKey=${myPubKey.take(8)}, iAmLower=${myPubKey < peerPubKey}" } - if (myPubKey < peerPubKey) { + val sm = sessionManager() + Log.d(TAG) { "onNewPeerInGroupCall: peer=${peerPubKey.take(8)}, shouldInitiate=${sm.shouldInitiateOffer(peerPubKey)}" } + if (sm.shouldInitiateOffer(peerPubKey)) { Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" } createAndOfferToPeer(peerPubKey) } else { @@ -466,19 +456,20 @@ class CallController( private suspend fun createAndOfferToPeer(peerPubKey: HexKey) { if (peerConnectionFactory == null) return - val globalPending = getGlobalPendingCandidates(peerPubKey) + val mgr = sessionManager() - val ps = + val webRtcSession = try { - withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } + withContext(Dispatchers.IO) { createWebRtcSession(peerPubKey) } } catch (e: Exception) { Log.e(TAG, "Failed to create PeerConnection for ${peerPubKey.take(8)}", e) return } - globalPending.forEach { ps.pendingIceCandidates.add(it) } + val adapter = WebRtcPeerSessionAdapter(webRtcSession) + mgr.registerSession(peerPubKey, adapter) - ps.session.createOffer { sdp -> + webRtcSession.createOffer { sdp -> Log.d(TAG) { "Callee-to-callee offer created for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" } scope.launch { callManager.publishOfferToPeer(peerPubKey, sdp.description) @@ -494,7 +485,8 @@ class CallController( peerPubKey: HexKey, sdpOffer: String, ) { - if (peerSessions.containsKey(peerPubKey)) { + val mgr = peerSessionMgr + if (mgr != null && mgr.hasSession(peerPubKey)) { Log.d(TAG) { "Mid-call offer from ${peerPubKey.take(8)} but session already exists — ignoring" } return } @@ -506,22 +498,22 @@ class CallController( return@launch } - val globalPending = getGlobalPendingCandidates(peerPubKey) + val sm = sessionManager() - val ps = + val webRtcSession = try { - withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } + withContext(Dispatchers.IO) { createWebRtcSession(peerPubKey) } } catch (e: Exception) { Log.e(TAG, "Failed to create PeerConnection for mid-call offer from ${peerPubKey.take(8)}", e) return@launch } - globalPending.forEach { ps.pendingIceCandidates.add(it) } + val adapter = WebRtcPeerSessionAdapter(webRtcSession) + sm.registerSession(peerPubKey, adapter) + adapter.setRemoteDescription(SdpType.OFFER, sdpOffer) + sm.flushPendingIceCandidates(peerPubKey) - ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) - flushPendingIceCandidates(peerPubKey, ps) - - ps.session.createAnswer { sdp -> + webRtcSession.createAnswer { sdp -> Log.d(TAG) { "Callee-to-callee answer for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" } scope.launch { callManager.publishAnswerToPeer(peerPubKey, sdp.description) @@ -534,53 +526,39 @@ class CallController( private fun onRenegotiationOfferReceived(event: CallRenegotiateEvent) { val peerPubKey = event.pubKey - val ps = peerSessions[peerPubKey] ?: return + val mgr = peerSessionMgr ?: return val sdpOffer = event.sdpOffer() Log.d(TAG) { "Renegotiation offer from ${peerPubKey.take(8)}, sdpLength=${sdpOffer.length}" } scope.launch { - val signalingState = ps.session.getSignalingState() - - if (signalingState == PeerConnection.SignalingState.HAVE_LOCAL_OFFER) { - // Glare: both sides sent offers simultaneously. - // Use pubkey comparison as tiebreaker — higher pubkey wins. - val myPubKey = signerProvider().pubKey - if (myPubKey > peerPubKey) { - Log.d(TAG) { "Glare with ${peerPubKey.take(8)}: I win (my offer takes priority), ignoring remote offer" } - return@launch + val resolution = + mgr.resolveRenegotiationGlare(peerPubKey, sdpOffer) { entry -> + applyRenegotiationOffer(entry.session, peerPubKey, sdpOffer) } - Log.d(TAG) { "Glare with ${peerPubKey.take(8)}: I lose, rolling back my offer" } - ps.session.rollback { - scope.launch { - applyRenegotiationOffer(ps, peerPubKey, sdpOffer) - } - } - } else { - applyRenegotiationOffer(ps, peerPubKey, sdpOffer) - } + Log.d(TAG) { "Renegotiation glare resolution with ${peerPubKey.take(8)}: $resolution" } } } private fun applyRenegotiationOffer( - ps: PeerSessionState, + session: PeerSession, peerPubKey: HexKey, sdpOffer: String, ) { - ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) - ps.session.createAnswer { sdp -> + session.setRemoteDescription(SdpType.OFFER, sdpOffer) + session.createAnswer { sdpAnswer -> scope.launch { - callManager.sendRenegotiationAnswer(sdp.description, peerPubKey) + callManager.sendRenegotiationAnswer(sdpAnswer, peerPubKey) } } } private fun performRenegotiation(peerPubKey: HexKey) { - val ps = peerSessions[peerPubKey] ?: return + val webRtcSession = webRtcSessions[peerPubKey] ?: return val state = callManager.state.value if (state !is CallState.Connected && state !is CallState.Connecting) return Log.d(TAG) { "Starting renegotiation with ${peerPubKey.take(8)}" } - ps.session.createOffer { sdp -> + webRtcSession.createOffer { sdp -> scope.launch { callManager.sendRenegotiation(sdp.description, peerPubKey) } @@ -602,8 +580,8 @@ class CallController( if (localVideoTrackInternal == null) { // Voice → video upgrade: create video source/track and add to all sessions createVideoResources() - peerSessions.values.forEach { ps -> - localVideoTrackInternal?.let { ps.session.addTrack(it, VIDEO_MAX_BITRATE_BPS) } + webRtcSessions.values.forEach { session -> + localVideoTrackInternal?.let { session.addTrack(it, VIDEO_MAX_BITRATE_BPS) } } } else { localVideoTrackInternal?.setEnabled(true) @@ -710,8 +688,8 @@ class CallController( // ---- Per-peer PeerConnection creation ---- - private fun createPeerSession(peerPubKey: HexKey): PeerSessionState { - Log.d(TAG) { "createPeerSession: ${peerPubKey.take(8)}, existing sessions=${peerSessions.keys.map { it.take(8) }}" } + private fun createWebRtcSession(peerPubKey: HexKey): WebRtcCallSession { + Log.d(TAG) { "createWebRtcSession: ${peerPubKey.take(8)}, existing sessions=${webRtcSessions.keys.map { it.take(8) }}" } val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized") val session = @@ -744,9 +722,8 @@ class CallController( localAudioTrackInternal?.let { session.addTrack(it) } localVideoTrackInternal?.let { session.addTrack(it, VIDEO_MAX_BITRATE_BPS) } - val ps = PeerSessionState(session) - peerSessions[peerPubKey] = ps - return ps + webRtcSessions[peerPubKey] = session + return session } private fun onRemoteVideoTrack( @@ -814,17 +791,17 @@ class CallController( private fun onPeerDisconnected(peerPubKey: HexKey) { Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected (ICE FAILED)" } - // If all peers disconnected, hang up. - // A session counts as "not active" if it is the one that just failed, - // if it was already closed, or if it never received a remote description - // (e.g. the peer rejected before answering). - val sessionStates = peerSessions.map { (key, ps) -> "${key.take(8)}=${ps.session.getSignalingState()},remoteSet=${ps.remoteDescriptionSet.get()}" } - Log.d(TAG) { "onPeerDisconnected: checking remaining sessions: $sessionStates" } + val mgr = peerSessionMgr + if (mgr == null) { + scope.launch { callManager.hangup() } + return + } + val allDisconnected = - peerSessions.keys.all { key -> + mgr.allSessionKeys().all { key -> key == peerPubKey || - peerSessions[key]?.session?.getSignalingState() == PeerConnection.SignalingState.CLOSED || - peerSessions[key]?.remoteDescriptionSet?.get() != true + mgr.getSession(key)?.session?.getSignalingState() == SignalingState.CLOSED || + mgr.getSession(key)?.remoteDescriptionSet != true } if (allDisconnected) { Log.d(TAG) { "onPeerDisconnected: all peers disconnected, hanging up" } @@ -841,11 +818,12 @@ class CallController( * but the call continues with remaining peers. */ fun disposePeerSession(peerPubKey: HexKey) { - val ps = peerSessions.remove(peerPubKey) - if (ps != null) { - Log.d(TAG) { "disposePeerSession: closing session for ${peerPubKey.take(8)}, remaining sessions=${peerSessions.keys.map { it.take(8) }}" } + val entry = peerSessionMgr?.removeSession(peerPubKey) + val webRtcSession = webRtcSessions.remove(peerPubKey) + if (entry != null || webRtcSession != null) { + Log.d(TAG) { "disposePeerSession: closing session for ${peerPubKey.take(8)}" } try { - ps.session.dispose() + entry?.session?.dispose() } catch (e: Exception) { Log.e(TAG, "disposePeerSession: dispose() failed for ${peerPubKey.take(8)}", e) } @@ -868,14 +846,12 @@ class CallController( } else { Log.d(TAG) { "disposePeerSession: no session for ${peerPubKey.take(8)} (already disposed or never created)" } } - // Clean up any buffered ICE candidates for this peer - globalPendingIce.remove(peerPubKey) } // ---- Cleanup ---- fun cleanup() { - Log.d(TAG) { "cleanup: disposing ${peerSessions.size} peer sessions, state=${callManager.state.value::class.simpleName}" } + Log.d(TAG) { "cleanup: disposing ${webRtcSessions.size} peer sessions, state=${callManager.state.value::class.simpleName}" } // Each block is wrapped individually so that a failure in one // (e.g. a WebRTC native crash) does not prevent the rest from // running. Without this, a single exception could leave the @@ -900,14 +876,13 @@ class CallController( } // Dispose all peer sessions - for (ps in peerSessions.values) { - try { - ps.session.dispose() - } catch (e: Exception) { - Log.e(TAG, "cleanup: PeerSession.dispose() failed", e) - } + try { + peerSessionMgr?.disposeAll() + } catch (e: Exception) { + Log.e(TAG, "cleanup: sessionManager.disposeAll() failed", e) } - peerSessions.clear() + peerSessionMgr = null + webRtcSessions.clear() // Dispose shared resources — each in its own try-catch so one // failure does not prevent the others from being released. @@ -954,8 +929,6 @@ class CallController( peerConnectionFactory = null sharedEglBase = null - globalPendingIce.clear() - _remoteVideoTrack.value = null _remoteVideoTracks.value = emptyMap() _localVideoTrack.value = null diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcPeerSessionAdapter.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcPeerSessionAdapter.kt new file mode 100644 index 000000000..942db4ba1 --- /dev/null +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcPeerSessionAdapter.kt @@ -0,0 +1,83 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.service.call + +import com.vitorpamplona.amethyst.commons.call.IceCandidateData +import com.vitorpamplona.amethyst.commons.call.PeerSession +import com.vitorpamplona.amethyst.commons.call.SdpType +import com.vitorpamplona.amethyst.commons.call.SignalingState +import org.webrtc.IceCandidate +import org.webrtc.PeerConnection +import org.webrtc.SessionDescription + +/** + * Adapts a [WebRtcCallSession] to the platform-independent [PeerSession] interface, + * allowing [com.vitorpamplona.amethyst.commons.call.PeerSessionManager] to manage + * ICE buffering, answer routing, and renegotiation glare without depending on org.webrtc. + */ +class WebRtcPeerSessionAdapter( + val webRtcSession: WebRtcCallSession, +) : PeerSession { + override fun getSignalingState(): SignalingState? = + when (webRtcSession.getSignalingState()) { + PeerConnection.SignalingState.STABLE -> SignalingState.STABLE + PeerConnection.SignalingState.HAVE_LOCAL_OFFER -> SignalingState.HAVE_LOCAL_OFFER + PeerConnection.SignalingState.HAVE_REMOTE_OFFER -> SignalingState.HAVE_REMOTE_OFFER + PeerConnection.SignalingState.HAVE_LOCAL_PRANSWER -> SignalingState.HAVE_LOCAL_PRANSWER + PeerConnection.SignalingState.HAVE_REMOTE_PRANSWER -> SignalingState.HAVE_REMOTE_PRANSWER + PeerConnection.SignalingState.CLOSED -> SignalingState.CLOSED + null -> null + } + + override fun setRemoteDescription( + type: SdpType, + sdp: String, + ) { + val sdType = + when (type) { + SdpType.OFFER -> SessionDescription.Type.OFFER + SdpType.ANSWER -> SessionDescription.Type.ANSWER + } + webRtcSession.setRemoteDescription(SessionDescription(sdType, sdp)) + } + + override fun addIceCandidate(candidate: IceCandidateData) { + webRtcSession.addIceCandidate( + IceCandidate(candidate.sdpMid, candidate.sdpMLineIndex, candidate.sdp), + ) + } + + override fun createOffer(onSdpCreated: (String) -> Unit) { + webRtcSession.createOffer { sdp -> onSdpCreated(sdp.description) } + } + + override fun createAnswer(onSdpCreated: (String) -> Unit) { + webRtcSession.createAnswer { sdp -> onSdpCreated(sdp.description) } + } + + override fun rollback(onDone: () -> Unit) { + webRtcSession.rollback(onDone) + } + + override fun dispose() { + webRtcSession.dispose() + } +} diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/PeerSessionManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/PeerSessionManager.kt new file mode 100644 index 000000000..7ececb271 --- /dev/null +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/PeerSessionManager.kt @@ -0,0 +1,263 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.commons.call + +import com.vitorpamplona.quartz.nip01Core.core.HexKey + +/** + * Represents a single ICE candidate received from a peer. + * Platform-neutral — does not depend on org.webrtc. + */ +data class IceCandidateData( + val sdp: String, + val sdpMid: String, + val sdpMLineIndex: Int, +) + +/** + * Represents the signaling state of a peer connection. + * Maps 1:1 with WebRTC's PeerConnection.SignalingState. + */ +enum class SignalingState { + STABLE, + HAVE_LOCAL_OFFER, + HAVE_REMOTE_OFFER, + HAVE_LOCAL_PRANSWER, + HAVE_REMOTE_PRANSWER, + CLOSED, +} + +/** + * Abstraction over a single peer connection's signaling operations. + * Implemented by the platform-specific WebRTC wrapper. + */ +interface PeerSession { + fun getSignalingState(): SignalingState? + + fun setRemoteDescription( + type: SdpType, + sdp: String, + ) + + fun addIceCandidate(candidate: IceCandidateData) + + fun createOffer(onSdpCreated: (String) -> Unit) + + fun createAnswer(onSdpCreated: (String) -> Unit) + + fun rollback(onDone: () -> Unit) + + fun dispose() +} + +enum class SdpType { + OFFER, + ANSWER, +} + +/** + * Manages per-peer session state and ICE candidate buffering. + * + * This class encapsulates the NIP-AC spec requirements for: + * - **Two-layer ICE buffering**: global buffer (before session exists) and + * per-session buffer (before remote description is set) + * - **Renegotiation glare handling**: pubkey comparison tiebreaker + * - **Callee-to-callee mesh**: lower pubkey initiates + * + * It is platform-independent and testable without real WebRTC. + */ +class PeerSessionManager( + private val localPubKey: HexKey, +) { + data class SessionEntry( + val session: PeerSession, + var remoteDescriptionSet: Boolean = false, + val pendingIceCandidates: MutableList = mutableListOf(), + ) + + private val sessions = mutableMapOf() + + /** Candidates received before a session exists for the sender. */ + private val globalPendingIce = mutableMapOf>() + + // ---- Session management ---- + + fun registerSession( + peerPubKey: HexKey, + session: PeerSession, + ): SessionEntry { + val globalPending = globalPendingIce.remove(peerPubKey) ?: emptyList() + val entry = SessionEntry(session) + entry.pendingIceCandidates.addAll(globalPending) + sessions[peerPubKey] = entry + return entry + } + + fun getSession(peerPubKey: HexKey): SessionEntry? = sessions[peerPubKey] + + fun hasSession(peerPubKey: HexKey): Boolean = sessions.containsKey(peerPubKey) + + fun removeSession(peerPubKey: HexKey): SessionEntry? { + globalPendingIce.remove(peerPubKey) + return sessions.remove(peerPubKey) + } + + fun allSessionKeys(): Set = sessions.keys.toSet() + + // ---- ICE candidate routing (two-layer buffering) ---- + + /** + * Routes an incoming ICE candidate to the correct destination: + * 1. If session exists AND remote description is set → add directly + * 2. If session exists but remote description NOT set → buffer per-session + * 3. If no session exists → buffer globally (keyed by sender) + * + * Returns the action taken for testability. + */ + fun routeIceCandidate( + senderPubKey: HexKey, + candidate: IceCandidateData, + ): IceRouteAction { + val entry = sessions[senderPubKey] + if (entry != null && entry.remoteDescriptionSet) { + entry.session.addIceCandidate(candidate) + return IceRouteAction.ADDED_DIRECTLY + } else if (entry != null) { + entry.pendingIceCandidates.add(candidate) + return IceRouteAction.BUFFERED_PER_SESSION + } else { + globalPendingIce.getOrPut(senderPubKey) { mutableListOf() }.add(candidate) + return IceRouteAction.BUFFERED_GLOBALLY + } + } + + /** + * Flushes all per-session buffered candidates into the PeerConnection. + * Called after setRemoteDescription succeeds. + */ + fun flushPendingIceCandidates(peerPubKey: HexKey): Int { + val entry = sessions[peerPubKey] ?: return 0 + entry.remoteDescriptionSet = true + val candidates = entry.pendingIceCandidates.toList() + entry.pendingIceCandidates.clear() + candidates.forEach { entry.session.addIceCandidate(it) } + return candidates.size + } + + fun globalPendingCount(peerPubKey: HexKey): Int = globalPendingIce[peerPubKey]?.size ?: 0 + + fun sessionPendingCount(peerPubKey: HexKey): Int = sessions[peerPubKey]?.pendingIceCandidates?.size ?: 0 + + // ---- Renegotiation glare handling ---- + + /** + * Determines how to handle a renegotiation offer when we may have a pending + * local offer (glare condition). + * + * Per NIP-AC spec: "the peer with the higher pubkey wins (their offer takes priority). + * The losing peer MUST roll back their local offer." + */ + fun resolveRenegotiationGlare( + peerPubKey: HexKey, + remoteSdpOffer: String, + onAcceptRemote: (SessionEntry) -> Unit, + ): GlareResolution { + val entry = sessions[peerPubKey] ?: return GlareResolution.NO_SESSION + + val signalingState = entry.session.getSignalingState() + if (signalingState != SignalingState.HAVE_LOCAL_OFFER) { + // No glare — accept the remote offer directly + onAcceptRemote(entry) + return GlareResolution.NO_GLARE + } + + // Glare detected: both sides sent offers simultaneously + return if (localPubKey > peerPubKey) { + // We win — our offer takes priority, ignore remote + GlareResolution.LOCAL_WINS + } else { + // We lose — rollback our local offer, accept remote + entry.session.rollback { + onAcceptRemote(entry) + } + GlareResolution.REMOTE_WINS_ROLLBACK + } + } + + // ---- Callee-to-callee mesh ---- + + /** + * Determines if the local peer should initiate the offer to [peerPubKey]. + * Per NIP-AC spec: "the peer with the lexicographically lower pubkey initiates." + */ + fun shouldInitiateOffer(peerPubKey: HexKey): Boolean = localPubKey < peerPubKey + + // ---- Answer routing ---- + + /** + * Routes an incoming SDP answer to the correct session. + * Returns the action taken. + */ + fun routeAnswer( + peerPubKey: HexKey, + sdpAnswer: String, + ): AnswerRouteAction { + val entry = sessions[peerPubKey] ?: return AnswerRouteAction.NO_SESSION + + val signalingState = entry.session.getSignalingState() + if (signalingState != SignalingState.HAVE_LOCAL_OFFER) { + return AnswerRouteAction.IGNORED_WRONG_STATE + } + + entry.session.setRemoteDescription(SdpType.ANSWER, sdpAnswer) + flushPendingIceCandidates(peerPubKey) + return AnswerRouteAction.APPLIED + } + + // ---- Cleanup ---- + + fun disposeAll() { + for (entry in sessions.values) { + entry.session.dispose() + } + sessions.clear() + globalPendingIce.clear() + } +} + +enum class IceRouteAction { + ADDED_DIRECTLY, + BUFFERED_PER_SESSION, + BUFFERED_GLOBALLY, +} + +enum class GlareResolution { + NO_SESSION, + NO_GLARE, + LOCAL_WINS, + REMOTE_WINS_ROLLBACK, +} + +enum class AnswerRouteAction { + APPLIED, + NO_SESSION, + IGNORED_WRONG_STATE, +} diff --git a/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/PeerSessionManagerTest.kt b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/PeerSessionManagerTest.kt new file mode 100644 index 000000000..f396563bc --- /dev/null +++ b/commons/src/commonTest/kotlin/com/vitorpamplona/amethyst/commons/call/PeerSessionManagerTest.kt @@ -0,0 +1,547 @@ +/* + * Copyright (c) 2025 Vitor Pamplona + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * this software and associated documentation files (the "Software"), to deal in + * the Software without restriction, including without limitation the rights to use, + * copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the + * Software, and to permit persons to whom the Software is furnished to do so, + * subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS + * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR + * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN + * AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION + * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + */ +package com.vitorpamplona.amethyst.commons.call + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Tests for PeerSessionManager — the extracted logic from CallController that + * handles ICE candidate buffering, renegotiation glare, and answer routing. + * + * Uses FakePeerSession to simulate WebRTC PeerConnection behavior without + * native libraries, making these runnable as JVM unit tests. + */ +class PeerSessionManagerTest { + // Identities: alice < bob < carol (lexicographic order of hex chars) + private val alice = "aaaa1111aaaa1111aaaa1111aaaa1111aaaa1111aaaa1111aaaa1111aaaa1111" + private val bob = "bbbb2222bbbb2222bbbb2222bbbb2222bbbb2222bbbb2222bbbb2222bbbb2222" + private val carol = "cccc3333cccc3333cccc3333cccc3333cccc3333cccc3333cccc3333cccc3333" + + private fun makeCandidate(label: String = "candidate:1") = IceCandidateData(sdp = label, sdpMid = "0", sdpMLineIndex = 0) + + // ======================================================================== + // ICE Candidate Buffering — Layer 1: Global Buffer + // ======================================================================== + + @Test + fun iceCandidate_noSession_bufferedGlobally() { + val manager = PeerSessionManager(localPubKey = bob) + val candidate = makeCandidate("early-candidate") + + val action = manager.routeIceCandidate(alice, candidate) + + assertEquals(IceRouteAction.BUFFERED_GLOBALLY, action) + assertEquals(1, manager.globalPendingCount(alice)) + } + + @Test + fun multipleGlobalCandidates_bufferedForSamePeer() { + val manager = PeerSessionManager(localPubKey = bob) + + manager.routeIceCandidate(alice, makeCandidate("c1")) + manager.routeIceCandidate(alice, makeCandidate("c2")) + manager.routeIceCandidate(alice, makeCandidate("c3")) + + assertEquals(3, manager.globalPendingCount(alice)) + } + + @Test + fun globalBuffer_drainedOnSessionCreation() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + + // Buffer 3 candidates before session exists + manager.routeIceCandidate(alice, makeCandidate("c1")) + manager.routeIceCandidate(alice, makeCandidate("c2")) + manager.routeIceCandidate(alice, makeCandidate("c3")) + assertEquals(3, manager.globalPendingCount(alice)) + + // Create session — global candidates should drain into per-session buffer + val entry = manager.registerSession(alice, fake) + + assertEquals(0, manager.globalPendingCount(alice), "Global buffer should be drained") + assertEquals(3, entry.pendingIceCandidates.size, "Candidates should move to per-session buffer") + assertEquals(0, fake.addedCandidates.size, "Candidates should NOT be added directly yet") + } + + @Test + fun globalBuffer_drainedAndFlushedAfterRemoteDescription() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + + // Buffer candidates before session + manager.routeIceCandidate(alice, makeCandidate("c1")) + manager.routeIceCandidate(alice, makeCandidate("c2")) + + // Create session (drains global → per-session) + manager.registerSession(alice, fake) + assertEquals(2, manager.sessionPendingCount(alice)) + + // Flush after remote description set + val flushed = manager.flushPendingIceCandidates(alice) + + assertEquals(2, flushed) + assertEquals(2, fake.addedCandidates.size, "Flushed candidates must reach the PeerConnection") + assertEquals("c1", fake.addedCandidates[0].sdp) + assertEquals("c2", fake.addedCandidates[1].sdp) + assertEquals(0, manager.sessionPendingCount(alice)) + } + + // ======================================================================== + // ICE Candidate Buffering — Layer 2: Per-Session Buffer + // ======================================================================== + + @Test + fun iceCandidate_sessionExists_remoteDescNotSet_bufferedPerSession() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + manager.registerSession(alice, fake) + + val action = manager.routeIceCandidate(alice, makeCandidate("mid-buffer")) + + assertEquals(IceRouteAction.BUFFERED_PER_SESSION, action) + assertEquals(1, manager.sessionPendingCount(alice)) + assertEquals(0, fake.addedCandidates.size, "Should NOT add to PeerConnection yet") + } + + @Test + fun iceCandidate_sessionExists_remoteDescSet_addedDirectly() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + manager.registerSession(alice, fake) + manager.flushPendingIceCandidates(alice) // marks remoteDescriptionSet = true + + val action = manager.routeIceCandidate(alice, makeCandidate("direct")) + + assertEquals(IceRouteAction.ADDED_DIRECTLY, action) + assertEquals(1, fake.addedCandidates.size) + assertEquals("direct", fake.addedCandidates[0].sdp) + } + + @Test + fun perSessionBuffer_notClearedOnSessionCreation_onlyOnFlush() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + + // Buffer globally + manager.routeIceCandidate(alice, makeCandidate("global-c")) + + // Create session → global drains to per-session + manager.registerSession(alice, fake) + + // Buffer more per-session (remote desc not set yet) + manager.routeIceCandidate(alice, makeCandidate("session-c")) + + assertEquals(2, manager.sessionPendingCount(alice)) + assertEquals(0, fake.addedCandidates.size) + + // Now flush + manager.flushPendingIceCandidates(alice) + + assertEquals(2, fake.addedCandidates.size) + assertEquals("global-c", fake.addedCandidates[0].sdp) + assertEquals("session-c", fake.addedCandidates[1].sdp) + } + + // ======================================================================== + // ICE Buffering: Candidates Buffered While Ringing MUST Be Preserved + // ======================================================================== + + @Test + fun candidatesBufferedWhileRinging_notLost_whenAccepting() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + + // Phase 1: Ringing — candidates arrive, no session yet + manager.routeIceCandidate(alice, makeCandidate("ringing-c1")) + manager.routeIceCandidate(alice, makeCandidate("ringing-c2")) + assertEquals(2, manager.globalPendingCount(alice)) + + // Phase 2: Accept call — create session + manager.registerSession(alice, fake) + assertEquals(0, manager.globalPendingCount(alice), "Global buffer drained") + assertEquals(2, manager.sessionPendingCount(alice), "Candidates preserved in session") + + // Phase 3: Set remote description and flush + manager.flushPendingIceCandidates(alice) + + assertEquals(2, fake.addedCandidates.size) + assertEquals("ringing-c1", fake.addedCandidates[0].sdp) + assertEquals("ringing-c2", fake.addedCandidates[1].sdp) + } + + // ======================================================================== + // ICE Buffering: Multi-Peer (Group Call) + // ======================================================================== + + @Test + fun globalBuffers_areSeparatePerPeer() { + val manager = PeerSessionManager(localPubKey = bob) + + manager.routeIceCandidate(alice, makeCandidate("alice-c")) + manager.routeIceCandidate(carol, makeCandidate("carol-c")) + + assertEquals(1, manager.globalPendingCount(alice)) + assertEquals(1, manager.globalPendingCount(carol)) + } + + @Test + fun registeringOneSession_doesNotDrainOtherPeersGlobalBuffer() { + val manager = PeerSessionManager(localPubKey = bob) + + manager.routeIceCandidate(alice, makeCandidate("alice-c")) + manager.routeIceCandidate(carol, makeCandidate("carol-c")) + + manager.registerSession(alice, FakePeerSession()) + + assertEquals(0, manager.globalPendingCount(alice), "Alice's global buffer drained") + assertEquals(1, manager.globalPendingCount(carol), "Carol's global buffer untouched") + } + + // ======================================================================== + // Renegotiation Glare Handling + // ======================================================================== + + @Test + fun renegotiation_noGlare_acceptsRemoteOffer() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession(signalingState = SignalingState.STABLE) + manager.registerSession(alice, fake) + + var accepted = false + val resolution = + manager.resolveRenegotiationGlare(alice, "remote-sdp") { entry -> + accepted = true + } + + assertEquals(GlareResolution.NO_GLARE, resolution) + assertTrue(accepted, "Should accept remote offer when no glare") + } + + @Test + fun renegotiationGlare_localWins_higherPubkey() { + // carol (higher) is local, alice (lower) is remote → carol wins + val manager = PeerSessionManager(localPubKey = carol) + val fake = FakePeerSession(signalingState = SignalingState.HAVE_LOCAL_OFFER) + manager.registerSession(alice, fake) + + var accepted = false + val resolution = + manager.resolveRenegotiationGlare(alice, "remote-sdp") { accepted = true } + + assertEquals(GlareResolution.LOCAL_WINS, resolution) + assertFalse(accepted, "Should NOT accept remote offer when local wins") + } + + @Test + fun renegotiationGlare_remoteWins_rollback() { + // alice (lower) is local, carol (higher) is remote → carol wins, alice rolls back + val manager = PeerSessionManager(localPubKey = alice) + val fake = FakePeerSession(signalingState = SignalingState.HAVE_LOCAL_OFFER) + manager.registerSession(carol, fake) + + var accepted = false + val resolution = + manager.resolveRenegotiationGlare(carol, "remote-sdp") { accepted = true } + + assertEquals(GlareResolution.REMOTE_WINS_ROLLBACK, resolution) + assertTrue(fake.rolledBack, "Must call rollback on the session") + assertTrue(accepted, "Must accept remote offer after rollback") + } + + @Test + fun renegotiationGlare_noSession_returnsNoSession() { + val manager = PeerSessionManager(localPubKey = bob) + + val resolution = + manager.resolveRenegotiationGlare(alice, "sdp") { } + + assertEquals(GlareResolution.NO_SESSION, resolution) + } + + // ======================================================================== + // Callee-to-Callee Mesh: Initiation Tiebreaker + // ======================================================================== + + @Test + fun meshInitiation_lowerPubkey_shouldInitiate() { + val manager = PeerSessionManager(localPubKey = alice) + + assertTrue(manager.shouldInitiateOffer(bob), "Lower pubkey (alice) should initiate to higher (bob)") + assertTrue(manager.shouldInitiateOffer(carol), "Lower pubkey (alice) should initiate to higher (carol)") + } + + @Test + fun meshInitiation_higherPubkey_shouldWait() { + val manager = PeerSessionManager(localPubKey = carol) + + assertFalse(manager.shouldInitiateOffer(alice), "Higher pubkey (carol) should NOT initiate to lower (alice)") + assertFalse(manager.shouldInitiateOffer(bob), "Higher pubkey (carol) should NOT initiate to lower (bob)") + } + + // ======================================================================== + // Answer Routing + // ======================================================================== + + @Test + fun routeAnswer_noSession_returnsNoSession() { + val manager = PeerSessionManager(localPubKey = alice) + + val action = manager.routeAnswer(bob, "answer-sdp") + + assertEquals(AnswerRouteAction.NO_SESSION, action) + } + + @Test + fun routeAnswer_wrongSignalingState_ignored() { + val manager = PeerSessionManager(localPubKey = alice) + val fake = FakePeerSession(signalingState = SignalingState.STABLE) + manager.registerSession(bob, fake) + + val action = manager.routeAnswer(bob, "answer-sdp") + + assertEquals(AnswerRouteAction.IGNORED_WRONG_STATE, action) + assertNull(fake.lastRemoteDescription, "Should NOT set remote description") + } + + @Test + fun routeAnswer_havingLocalOffer_applied() { + val manager = PeerSessionManager(localPubKey = alice) + val fake = FakePeerSession(signalingState = SignalingState.HAVE_LOCAL_OFFER) + manager.registerSession(bob, fake) + + // Buffer some ICE candidates before answer arrives + manager.routeIceCandidate(bob, makeCandidate("pre-answer-c")) + + val action = manager.routeAnswer(bob, "answer-sdp") + + assertEquals(AnswerRouteAction.APPLIED, action) + assertNotNull(fake.lastRemoteDescription) + assertEquals(SdpType.ANSWER, fake.lastRemoteDescription!!.first) + assertEquals("answer-sdp", fake.lastRemoteDescription!!.second) + // Flush should have happened + assertEquals(1, fake.addedCandidates.size, "Buffered candidates should be flushed after answer") + } + + // ======================================================================== + // Session Lifecycle + // ======================================================================== + + @Test + fun removeSession_disposesAndCleansUp() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + manager.registerSession(alice, fake) + manager.routeIceCandidate(alice, makeCandidate("c")) + + val removed = manager.removeSession(alice) + + assertNotNull(removed) + assertFalse(manager.hasSession(alice)) + assertEquals(0, manager.globalPendingCount(alice)) + } + + @Test + fun disposeAll_disposesAllSessions() { + val manager = PeerSessionManager(localPubKey = bob) + val fakeAlice = FakePeerSession() + val fakeCarol = FakePeerSession() + manager.registerSession(alice, fakeAlice) + manager.registerSession(carol, fakeCarol) + + manager.disposeAll() + + assertTrue(fakeAlice.disposed) + assertTrue(fakeCarol.disposed) + assertTrue(manager.allSessionKeys().isEmpty()) + } + + // ======================================================================== + // Full Scenario: P2P Call with ICE Buffering + // ======================================================================== + + @Test + fun fullP2PFlow_iceBufferingThroughAllPhases() { + val manager = PeerSessionManager(localPubKey = bob) + val fake = FakePeerSession() + + // Phase 1: Ringing — candidates arrive before session + manager.routeIceCandidate(alice, makeCandidate("ringing-1")) + manager.routeIceCandidate(alice, makeCandidate("ringing-2")) + assertEquals(2, manager.globalPendingCount(alice)) + + // Phase 2: Accept — create session (drains global → per-session) + manager.registerSession(alice, fake) + assertEquals(2, manager.sessionPendingCount(alice)) + + // Phase 3: More candidates arrive before remote desc set + manager.routeIceCandidate(alice, makeCandidate("pre-desc-1")) + assertEquals(IceRouteAction.BUFFERED_PER_SESSION, manager.routeIceCandidate(alice, makeCandidate("pre-desc-2"))) + assertEquals(4, manager.sessionPendingCount(alice)) + + // Phase 4: Remote description set → flush + val flushed = manager.flushPendingIceCandidates(alice) + assertEquals(4, flushed) + assertEquals(4, fake.addedCandidates.size) + + // Phase 5: Late candidates arrive → added directly + assertEquals(IceRouteAction.ADDED_DIRECTLY, manager.routeIceCandidate(alice, makeCandidate("post-desc"))) + assertEquals(5, fake.addedCandidates.size) + } + + // ======================================================================== + // Full Scenario: Group Call with Mesh ICE Buffering + // ======================================================================== + + @Test + fun groupCall_meshSetup_withIceBuffering() { + // Bob is local. Alice (caller) and Carol (other callee) are peers. + val manager = PeerSessionManager(localPubKey = bob) + + // Phase 1: Ringing — ICE from alice arrives before any session + manager.routeIceCandidate(alice, makeCandidate("alice-ring-c")) + + // Phase 2: Accept — create session for alice + val aliceFake = FakePeerSession() + manager.registerSession(alice, aliceFake) + manager.flushPendingIceCandidates(alice) + assertEquals(1, aliceFake.addedCandidates.size) + + // Phase 3: Carol's ICE arrives before mesh session exists + manager.routeIceCandidate(carol, makeCandidate("carol-early-c")) + assertEquals(1, manager.globalPendingCount(carol)) + + // Phase 4: Mesh setup — should bob initiate to carol? + // bob < carol → bob should initiate + assertTrue(manager.shouldInitiateOffer(carol)) + + // Phase 5: Create session for carol (drains global) + val carolFake = FakePeerSession() + manager.registerSession(carol, carolFake) + assertEquals(1, manager.sessionPendingCount(carol)) + + // Phase 6: Carol's remote description set → flush + manager.flushPendingIceCandidates(carol) + assertEquals(1, carolFake.addedCandidates.size) + assertEquals("carol-early-c", carolFake.addedCandidates[0].sdp) + } + + // ======================================================================== + // Full Scenario: Renegotiation with Glare + // ======================================================================== + + @Test + fun renegotiationGlare_fullFlow_lowerPubkeyRollsBack() { + // Alice (lower) and Bob (higher) both send renegotiation offers simultaneously + // Alice's manager handles Bob's incoming offer + val aliceManager = PeerSessionManager(localPubKey = alice) + val aliceFake = FakePeerSession(signalingState = SignalingState.HAVE_LOCAL_OFFER) + aliceManager.registerSession(bob, aliceFake) + aliceManager.flushPendingIceCandidates(bob) + + var acceptedSdp = "" + val resolution = + aliceManager.resolveRenegotiationGlare(bob, "bob-renego-sdp") { entry -> + acceptedSdp = "bob-renego-sdp" + } + + assertEquals(GlareResolution.REMOTE_WINS_ROLLBACK, resolution, "alice (lower) should lose to bob (higher)") + assertTrue(aliceFake.rolledBack, "alice must rollback her local offer") + assertEquals("bob-renego-sdp", acceptedSdp, "alice must accept bob's offer") + + // Bob's manager handles Alice's incoming offer + val bobManager = PeerSessionManager(localPubKey = bob) + val bobFake = FakePeerSession(signalingState = SignalingState.HAVE_LOCAL_OFFER) + bobManager.registerSession(alice, bobFake) + bobManager.flushPendingIceCandidates(alice) + + var bobAccepted = false + val bobResolution = + bobManager.resolveRenegotiationGlare(alice, "alice-renego-sdp") { bobAccepted = true } + + assertEquals(GlareResolution.LOCAL_WINS, bobResolution, "bob (higher) should win over alice (lower)") + assertFalse(bobFake.rolledBack, "bob should NOT rollback") + assertFalse(bobAccepted, "bob should ignore alice's offer") + } +} + +/** + * Fake PeerSession that records all operations for test assertions. + * No real WebRTC involved. + */ +class FakePeerSession( + private var signalingState: SignalingState = SignalingState.STABLE, +) : PeerSession { + val addedCandidates = mutableListOf() + var lastRemoteDescription: Pair? = null + var rolledBack = false + var disposed = false + var lastCreatedOffer: String? = null + var lastCreatedAnswer: String? = null + + override fun getSignalingState(): SignalingState = signalingState + + override fun setRemoteDescription( + type: SdpType, + sdp: String, + ) { + lastRemoteDescription = type to sdp + if (type == SdpType.ANSWER) { + signalingState = SignalingState.STABLE + } else if (type == SdpType.OFFER) { + signalingState = SignalingState.HAVE_REMOTE_OFFER + } + } + + override fun addIceCandidate(candidate: IceCandidateData) { + addedCandidates.add(candidate) + } + + override fun createOffer(onSdpCreated: (String) -> Unit) { + signalingState = SignalingState.HAVE_LOCAL_OFFER + val sdp = "fake-offer-sdp" + lastCreatedOffer = sdp + onSdpCreated(sdp) + } + + override fun createAnswer(onSdpCreated: (String) -> Unit) { + val sdp = "fake-answer-sdp" + lastCreatedAnswer = sdp + signalingState = SignalingState.STABLE + onSdpCreated(sdp) + } + + override fun rollback(onDone: () -> Unit) { + rolledBack = true + signalingState = SignalingState.STABLE + onDone() + } + + override fun dispose() { + disposed = true + signalingState = SignalingState.CLOSED + } +}