refactor: extract PeerSessionManager for testable ICE buffering and glare handling

Extract the ICE candidate buffering, answer routing, and renegotiation
glare logic from CallController into a platform-independent
PeerSessionManager in commons/. This makes the three most critical
NIP-AC spec requirements testable as JVM unit tests with FakePeerSession:

1. Two-layer ICE candidate buffering (global + per-session)
   - Global buffer: candidates arriving before any PeerConnection exists
   - Per-session buffer: candidates arriving before setRemoteDescription
   - Candidates buffered while ringing are preserved when accepting

2. Renegotiation glare handling (pubkey comparison tiebreaker)
   - Higher pubkey wins, lower pubkey rolls back local offer
   - Full glare scenario test with two PeerSessionManagers

3. Callee-to-callee mesh initiation (lower pubkey initiates)

CallController now delegates to PeerSessionManager via a PeerSession
interface, with WebRtcPeerSessionAdapter bridging to WebRtcCallSession.

https://claude.ai/code/session_01AfRYTRCvtKqqDxeKQujUrx
This commit is contained in:
Claude
2026-04-05 15:17:29 +00:00
parent c2347c6cee
commit e9f1fcbd2a
4 changed files with 1010 additions and 144 deletions
@@ -26,8 +26,14 @@ import coil3.ImageLoader
import coil3.asDrawable import coil3.asDrawable
import coil3.request.ImageRequest import coil3.request.ImageRequest
import coil3.request.allowHardware import coil3.request.allowHardware
import com.vitorpamplona.amethyst.commons.call.AnswerRouteAction
import com.vitorpamplona.amethyst.commons.call.CallManager import com.vitorpamplona.amethyst.commons.call.CallManager
import com.vitorpamplona.amethyst.commons.call.CallState 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.model.LocalCache
import com.vitorpamplona.amethyst.service.notifications.NotificationUtils import com.vitorpamplona.amethyst.service.notifications.NotificationUtils
import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.HexKey
@@ -56,9 +62,7 @@ import org.webrtc.DefaultVideoEncoderFactory
import org.webrtc.EglBase import org.webrtc.EglBase
import org.webrtc.IceCandidate import org.webrtc.IceCandidate
import org.webrtc.MediaConstraints import org.webrtc.MediaConstraints
import org.webrtc.PeerConnection
import org.webrtc.PeerConnectionFactory import org.webrtc.PeerConnectionFactory
import org.webrtc.SessionDescription
import org.webrtc.SurfaceTextureHelper import org.webrtc.SurfaceTextureHelper
import org.webrtc.VideoFrame import org.webrtc.VideoFrame
import org.webrtc.VideoSink import org.webrtc.VideoSink
@@ -66,8 +70,6 @@ import org.webrtc.VideoSource
import org.webrtc.VideoTrack import org.webrtc.VideoTrack
import java.util.UUID import java.util.UUID
import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicLong
private const val TAG = "CallController" private const val TAG = "CallController"
@@ -80,18 +82,20 @@ class CallController(
private val publishWrap: suspend (EphemeralGiftWrapEvent) -> Unit, private val publishWrap: suspend (EphemeralGiftWrapEvent) -> Unit,
private val signerProvider: suspend () -> com.vitorpamplona.quartz.nip01Core.signers.NostrSigner, private val signerProvider: suspend () -> com.vitorpamplona.quartz.nip01Core.signers.NostrSigner,
) { ) {
// ---- Per-peer session state ---- // ---- Per-peer session state (delegated to PeerSessionManager) ----
private class PeerSessionState( // Lazily initialized — needs signerProvider() for localPubKey
val session: WebRtcCallSession, private var peerSessionMgr: PeerSessionManager? = null
val remoteDescriptionSet: AtomicBoolean = AtomicBoolean(false),
val pendingIceCandidates: CopyOnWriteArrayList<IceCandidate> = CopyOnWriteArrayList(),
)
private val peerSessions = ConcurrentHashMap<HexKey, PeerSessionState>() private suspend fun sessionManager(): PeerSessionManager {
if (peerSessionMgr == null) {
peerSessionMgr = PeerSessionManager(signerProvider().pubKey)
}
return peerSessionMgr!!
}
// Candidates received before a PeerSession exists for the sender /** Map from peer pubkey to the WebRtcCallSession for direct access to WebRTC-specific APIs. */
private val globalPendingIce = ConcurrentHashMap<HexKey, CopyOnWriteArrayList<IceCandidate>>() private val webRtcSessions = ConcurrentHashMap<HexKey, WebRtcCallSession>()
// ---- Shared WebRTC resources ---- // ---- Shared WebRTC resources ----
@@ -272,12 +276,16 @@ class CallController(
// Set state to Offering before creating peer sessions // Set state to Offering before creating peer sessions
callManager.beginOffering(callId, peerPubKeys, callType) callManager.beginOffering(callId, peerPubKeys, callType)
val mgr = sessionManager()
// Create a PeerConnection + offer for each callee // Create a PeerConnection + offer for each callee
for (peerPubKey in peerPubKeys) { for (peerPubKey in peerPubKeys) {
try { 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)}" } 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}" } Log.d(TAG) { "initiateCall: offer created for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" }
scope.launch { scope.launch {
callManager.publishOfferToPeer(peerPubKey, peerPubKeys, callType, callId, sdp.description) callManager.publishOfferToPeer(peerPubKey, peerPubKeys, callType, callId, sdp.description)
@@ -315,28 +323,27 @@ class CallController(
return@launch return@launch
} }
// Retrieve any ICE candidates buffered while ringing (before session existed) val mgr = sessionManager()
val globalPending = getGlobalPendingCandidates(callerPubKey)
val ps = val webRtcSession =
try { try {
withContext(Dispatchers.IO) { createPeerSession(callerPubKey) } withContext(Dispatchers.IO) { createWebRtcSession(callerPubKey) }
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to create PeerConnection", e) Log.e(TAG, "Failed to create PeerConnection", e)
_errorMessage.value = "Failed to accept call: ${e.message}" _errorMessage.value = "Failed to accept call: ${e.message}"
return@launch return@launch
} }
// Inject any globally-buffered ICE candidates for the caller val adapter = WebRtcPeerSessionAdapter(webRtcSession)
globalPending.forEach { ps.pendingIceCandidates.add(it) } val entry = mgr.registerSession(callerPubKey, adapter)
Log.d(TAG) { "acceptIncomingCall: setting remote description (OFFER)..." } Log.d(TAG) { "acceptIncomingCall: setting remote description (OFFER)..." }
ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) adapter.setRemoteDescription(SdpType.OFFER, sdpOffer)
Log.d(TAG) { "acceptIncomingCall: flushing ${ps.pendingIceCandidates.size} pending ICE candidates..." } Log.d(TAG) { "acceptIncomingCall: flushing ${entry.pendingIceCandidates.size} pending ICE candidates..." }
flushPendingIceCandidates(callerPubKey, ps) mgr.flushPendingIceCandidates(callerPubKey)
Log.d(TAG) { "acceptIncomingCall: creating answer..." } Log.d(TAG) { "acceptIncomingCall: creating answer..." }
ps.session.createAnswer { sdp -> webRtcSession.createAnswer { sdp ->
Log.d(TAG) { "acceptIncomingCall: answer created, sdpLength=${sdp.description.length}, publishing..." } Log.d(TAG) { "acceptIncomingCall: answer created, sdpLength=${sdp.description.length}, publishing..." }
scope.launch { scope.launch {
callManager.acceptCall(sdp.description) callManager.acceptCall(sdp.description)
@@ -358,29 +365,29 @@ class CallController(
peerPubKey: HexKey, peerPubKey: HexKey,
sdpAnswer: String, sdpAnswer: String,
) { ) {
Log.d(TAG) { "onCallAnswerReceived: from=${peerPubKey.take(8)}, knownSessions=${peerSessions.keys.map { it.take(8) }}" } val mgr = peerSessionMgr
val ps = peerSessions[peerPubKey] if (mgr == null) {
if (ps != null) { Log.d(TAG) { "onCallAnswerReceived: sessionManager not initialized — ignoring" }
// We have a PeerConnection for this peer — set remote description return
val signalingState = ps.session.getSignalingState() }
Log.d(TAG) {
"Answer received from ${peerPubKey.take(8)}, sdpLength=${sdpAnswer.length}, " + Log.d(TAG) { "onCallAnswerReceived: from=${peerPubKey.take(8)}, knownSessions=${mgr.allSessionKeys().map { it.take(8) }}" }
"signalingState=$signalingState, remoteDescSet=${ps.remoteDescriptionSet.get()}"
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) { AnswerRouteAction.NO_SESSION -> {
Log.d(TAG) { "Ignoring answer from ${peerPubKey.take(8)} in $signalingState (no pending local offer)" } Log.d(TAG) { "Answer from unknown peer ${peerPubKey.take(8)} — triggering callee-to-callee connection" }
return onNewPeerInGroupCall(peerPubKey)
} }
Log.d(TAG) { "Setting remote description (ANSWER) for ${peerPubKey.take(8)}..." } AnswerRouteAction.IGNORED_WRONG_STATE -> {
ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.ANSWER, sdpAnswer)) Log.d(TAG) { "Ignoring answer from ${peerPubKey.take(8)} (no pending local offer)" }
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)
} }
} }
@@ -393,37 +400,19 @@ class CallController(
fun onIceCandidateReceived(event: CallIceCandidateEvent) { fun onIceCandidateReceived(event: CallIceCandidateEvent) {
try { try {
val senderPubKey = event.pubKey val senderPubKey = event.pubKey
val candidate = IceCandidate(event.sdpMid(), event.sdpMLineIndex(), event.candidateSdp()) val candidate = IceCandidateData(event.candidateSdp(), event.sdpMid(), event.sdpMLineIndex())
val ps = peerSessions[senderPubKey] val mgr = peerSessionMgr
if (ps != null && ps.remoteDescriptionSet.get()) { if (mgr == null) {
Log.d(TAG) { "Adding ICE candidate from ${senderPubKey.take(8)} directly" } Log.d(TAG) { "Buffering ICE candidate from ${senderPubKey.take(8)} (manager not initialized)" }
ps.session.addIceCandidate(candidate) return
} 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 action = mgr.routeIceCandidate(senderPubKey, candidate)
Log.d(TAG) { "ICE candidate from ${senderPubKey.take(8)}: $action" }
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to parse ICE candidate", e) Log.e(TAG, "Failed to parse ICE candidate", e)
} }
} }
private fun getGlobalPendingCandidates(peerPubKey: HexKey): List<IceCandidate> = 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( private fun onLocalIceCandidate(
peerPubKey: HexKey, peerPubKey: HexKey,
candidate: IceCandidate, candidate: IceCandidate,
@@ -447,15 +436,16 @@ class CallController(
* peer with the lexicographically lower pubkey initiates. * peer with the lexicographically lower pubkey initiates.
*/ */
fun onNewPeerInGroupCall(peerPubKey: HexKey) { 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" } Log.d(TAG) { "onNewPeerInGroupCall: session already exists for ${peerPubKey.take(8)} — skipping" }
return return
} }
scope.launch { scope.launch {
val myPubKey = signerProvider().pubKey val sm = sessionManager()
Log.d(TAG) { "onNewPeerInGroupCall: peer=${peerPubKey.take(8)}, myPubKey=${myPubKey.take(8)}, iAmLower=${myPubKey < peerPubKey}" } Log.d(TAG) { "onNewPeerInGroupCall: peer=${peerPubKey.take(8)}, shouldInitiate=${sm.shouldInitiateOffer(peerPubKey)}" }
if (myPubKey < peerPubKey) { if (sm.shouldInitiateOffer(peerPubKey)) {
Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" } Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" }
createAndOfferToPeer(peerPubKey) createAndOfferToPeer(peerPubKey)
} else { } else {
@@ -466,19 +456,20 @@ class CallController(
private suspend fun createAndOfferToPeer(peerPubKey: HexKey) { private suspend fun createAndOfferToPeer(peerPubKey: HexKey) {
if (peerConnectionFactory == null) return if (peerConnectionFactory == null) return
val globalPending = getGlobalPendingCandidates(peerPubKey) val mgr = sessionManager()
val ps = val webRtcSession =
try { try {
withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } withContext(Dispatchers.IO) { createWebRtcSession(peerPubKey) }
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to create PeerConnection for ${peerPubKey.take(8)}", e) Log.e(TAG, "Failed to create PeerConnection for ${peerPubKey.take(8)}", e)
return 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}" } Log.d(TAG) { "Callee-to-callee offer created for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" }
scope.launch { scope.launch {
callManager.publishOfferToPeer(peerPubKey, sdp.description) callManager.publishOfferToPeer(peerPubKey, sdp.description)
@@ -494,7 +485,8 @@ class CallController(
peerPubKey: HexKey, peerPubKey: HexKey,
sdpOffer: String, 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" } Log.d(TAG) { "Mid-call offer from ${peerPubKey.take(8)} but session already exists — ignoring" }
return return
} }
@@ -506,22 +498,22 @@ class CallController(
return@launch return@launch
} }
val globalPending = getGlobalPendingCandidates(peerPubKey) val sm = sessionManager()
val ps = val webRtcSession =
try { try {
withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } withContext(Dispatchers.IO) { createWebRtcSession(peerPubKey) }
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "Failed to create PeerConnection for mid-call offer from ${peerPubKey.take(8)}", e) Log.e(TAG, "Failed to create PeerConnection for mid-call offer from ${peerPubKey.take(8)}", e)
return@launch 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)) webRtcSession.createAnswer { sdp ->
flushPendingIceCandidates(peerPubKey, ps)
ps.session.createAnswer { sdp ->
Log.d(TAG) { "Callee-to-callee answer for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" } Log.d(TAG) { "Callee-to-callee answer for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" }
scope.launch { scope.launch {
callManager.publishAnswerToPeer(peerPubKey, sdp.description) callManager.publishAnswerToPeer(peerPubKey, sdp.description)
@@ -534,53 +526,39 @@ class CallController(
private fun onRenegotiationOfferReceived(event: CallRenegotiateEvent) { private fun onRenegotiationOfferReceived(event: CallRenegotiateEvent) {
val peerPubKey = event.pubKey val peerPubKey = event.pubKey
val ps = peerSessions[peerPubKey] ?: return val mgr = peerSessionMgr ?: return
val sdpOffer = event.sdpOffer() val sdpOffer = event.sdpOffer()
Log.d(TAG) { "Renegotiation offer from ${peerPubKey.take(8)}, sdpLength=${sdpOffer.length}" } Log.d(TAG) { "Renegotiation offer from ${peerPubKey.take(8)}, sdpLength=${sdpOffer.length}" }
scope.launch { scope.launch {
val signalingState = ps.session.getSignalingState() val resolution =
mgr.resolveRenegotiationGlare(peerPubKey, sdpOffer) { entry ->
if (signalingState == PeerConnection.SignalingState.HAVE_LOCAL_OFFER) { applyRenegotiationOffer(entry.session, peerPubKey, sdpOffer)
// 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
} }
Log.d(TAG) { "Glare with ${peerPubKey.take(8)}: I lose, rolling back my offer" } Log.d(TAG) { "Renegotiation glare resolution with ${peerPubKey.take(8)}: $resolution" }
ps.session.rollback {
scope.launch {
applyRenegotiationOffer(ps, peerPubKey, sdpOffer)
}
}
} else {
applyRenegotiationOffer(ps, peerPubKey, sdpOffer)
}
} }
} }
private fun applyRenegotiationOffer( private fun applyRenegotiationOffer(
ps: PeerSessionState, session: PeerSession,
peerPubKey: HexKey, peerPubKey: HexKey,
sdpOffer: String, sdpOffer: String,
) { ) {
ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) session.setRemoteDescription(SdpType.OFFER, sdpOffer)
ps.session.createAnswer { sdp -> session.createAnswer { sdpAnswer ->
scope.launch { scope.launch {
callManager.sendRenegotiationAnswer(sdp.description, peerPubKey) callManager.sendRenegotiationAnswer(sdpAnswer, peerPubKey)
} }
} }
} }
private fun performRenegotiation(peerPubKey: HexKey) { private fun performRenegotiation(peerPubKey: HexKey) {
val ps = peerSessions[peerPubKey] ?: return val webRtcSession = webRtcSessions[peerPubKey] ?: return
val state = callManager.state.value val state = callManager.state.value
if (state !is CallState.Connected && state !is CallState.Connecting) return if (state !is CallState.Connected && state !is CallState.Connecting) return
Log.d(TAG) { "Starting renegotiation with ${peerPubKey.take(8)}" } Log.d(TAG) { "Starting renegotiation with ${peerPubKey.take(8)}" }
ps.session.createOffer { sdp -> webRtcSession.createOffer { sdp ->
scope.launch { scope.launch {
callManager.sendRenegotiation(sdp.description, peerPubKey) callManager.sendRenegotiation(sdp.description, peerPubKey)
} }
@@ -602,8 +580,8 @@ class CallController(
if (localVideoTrackInternal == null) { if (localVideoTrackInternal == null) {
// Voice → video upgrade: create video source/track and add to all sessions // Voice → video upgrade: create video source/track and add to all sessions
createVideoResources() createVideoResources()
peerSessions.values.forEach { ps -> webRtcSessions.values.forEach { session ->
localVideoTrackInternal?.let { ps.session.addTrack(it, VIDEO_MAX_BITRATE_BPS) } localVideoTrackInternal?.let { session.addTrack(it, VIDEO_MAX_BITRATE_BPS) }
} }
} else { } else {
localVideoTrackInternal?.setEnabled(true) localVideoTrackInternal?.setEnabled(true)
@@ -710,8 +688,8 @@ class CallController(
// ---- Per-peer PeerConnection creation ---- // ---- Per-peer PeerConnection creation ----
private fun createPeerSession(peerPubKey: HexKey): PeerSessionState { private fun createWebRtcSession(peerPubKey: HexKey): WebRtcCallSession {
Log.d(TAG) { "createPeerSession: ${peerPubKey.take(8)}, existing sessions=${peerSessions.keys.map { it.take(8) }}" } Log.d(TAG) { "createWebRtcSession: ${peerPubKey.take(8)}, existing sessions=${webRtcSessions.keys.map { it.take(8) }}" }
val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized") val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized")
val session = val session =
@@ -744,9 +722,8 @@ class CallController(
localAudioTrackInternal?.let { session.addTrack(it) } localAudioTrackInternal?.let { session.addTrack(it) }
localVideoTrackInternal?.let { session.addTrack(it, VIDEO_MAX_BITRATE_BPS) } localVideoTrackInternal?.let { session.addTrack(it, VIDEO_MAX_BITRATE_BPS) }
val ps = PeerSessionState(session) webRtcSessions[peerPubKey] = session
peerSessions[peerPubKey] = ps return session
return ps
} }
private fun onRemoteVideoTrack( private fun onRemoteVideoTrack(
@@ -814,17 +791,17 @@ class CallController(
private fun onPeerDisconnected(peerPubKey: HexKey) { private fun onPeerDisconnected(peerPubKey: HexKey) {
Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected (ICE FAILED)" } Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected (ICE FAILED)" }
// If all peers disconnected, hang up. val mgr = peerSessionMgr
// A session counts as "not active" if it is the one that just failed, if (mgr == null) {
// if it was already closed, or if it never received a remote description scope.launch { callManager.hangup() }
// (e.g. the peer rejected before answering). return
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 allDisconnected = val allDisconnected =
peerSessions.keys.all { key -> mgr.allSessionKeys().all { key ->
key == peerPubKey || key == peerPubKey ||
peerSessions[key]?.session?.getSignalingState() == PeerConnection.SignalingState.CLOSED || mgr.getSession(key)?.session?.getSignalingState() == SignalingState.CLOSED ||
peerSessions[key]?.remoteDescriptionSet?.get() != true mgr.getSession(key)?.remoteDescriptionSet != true
} }
if (allDisconnected) { if (allDisconnected) {
Log.d(TAG) { "onPeerDisconnected: all peers disconnected, hanging up" } Log.d(TAG) { "onPeerDisconnected: all peers disconnected, hanging up" }
@@ -841,11 +818,12 @@ class CallController(
* but the call continues with remaining peers. * but the call continues with remaining peers.
*/ */
fun disposePeerSession(peerPubKey: HexKey) { fun disposePeerSession(peerPubKey: HexKey) {
val ps = peerSessions.remove(peerPubKey) val entry = peerSessionMgr?.removeSession(peerPubKey)
if (ps != null) { val webRtcSession = webRtcSessions.remove(peerPubKey)
Log.d(TAG) { "disposePeerSession: closing session for ${peerPubKey.take(8)}, remaining sessions=${peerSessions.keys.map { it.take(8) }}" } if (entry != null || webRtcSession != null) {
Log.d(TAG) { "disposePeerSession: closing session for ${peerPubKey.take(8)}" }
try { try {
ps.session.dispose() entry?.session?.dispose()
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "disposePeerSession: dispose() failed for ${peerPubKey.take(8)}", e) Log.e(TAG, "disposePeerSession: dispose() failed for ${peerPubKey.take(8)}", e)
} }
@@ -868,14 +846,12 @@ class CallController(
} else { } else {
Log.d(TAG) { "disposePeerSession: no session for ${peerPubKey.take(8)} (already disposed or never created)" } 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 ---- // ---- Cleanup ----
fun 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 // Each block is wrapped individually so that a failure in one
// (e.g. a WebRTC native crash) does not prevent the rest from // (e.g. a WebRTC native crash) does not prevent the rest from
// running. Without this, a single exception could leave the // running. Without this, a single exception could leave the
@@ -900,14 +876,13 @@ class CallController(
} }
// Dispose all peer sessions // Dispose all peer sessions
for (ps in peerSessions.values) { try {
try { peerSessionMgr?.disposeAll()
ps.session.dispose() } catch (e: Exception) {
} catch (e: Exception) { Log.e(TAG, "cleanup: sessionManager.disposeAll() failed", e)
Log.e(TAG, "cleanup: PeerSession.dispose() failed", e)
}
} }
peerSessions.clear() peerSessionMgr = null
webRtcSessions.clear()
// Dispose shared resources — each in its own try-catch so one // Dispose shared resources — each in its own try-catch so one
// failure does not prevent the others from being released. // failure does not prevent the others from being released.
@@ -954,8 +929,6 @@ class CallController(
peerConnectionFactory = null peerConnectionFactory = null
sharedEglBase = null sharedEglBase = null
globalPendingIce.clear()
_remoteVideoTrack.value = null _remoteVideoTrack.value = null
_remoteVideoTracks.value = emptyMap() _remoteVideoTracks.value = emptyMap()
_localVideoTrack.value = null _localVideoTrack.value = null
@@ -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()
}
}
@@ -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<IceCandidateData> = mutableListOf(),
)
private val sessions = mutableMapOf<HexKey, SessionEntry>()
/** Candidates received before a session exists for the sender. */
private val globalPendingIce = mutableMapOf<HexKey, MutableList<IceCandidateData>>()
// ---- 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<HexKey> = 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,
}
@@ -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<IceCandidateData>()
var lastRemoteDescription: Pair<SdpType, String>? = 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
}
}