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 fbb96b899..be655c05f 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 @@ -349,6 +349,7 @@ 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 @@ -437,10 +438,14 @@ class CallController( * peer with the lexicographically lower pubkey initiates. */ fun onNewPeerInGroupCall(peerPubKey: HexKey) { - if (peerSessions.containsKey(peerPubKey)) return // already connected/connecting + if (peerSessions.containsKey(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) { Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" } createAndOfferToPeer(peerPubKey) @@ -671,6 +676,7 @@ 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) }}" } val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized") val session = @@ -722,20 +728,62 @@ class CallController( } private fun onPeerDisconnected(peerPubKey: HexKey) { - Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected" } + Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected (ICE FAILED)" } // If all peers disconnected, hang up + val sessionStates = peerSessions.map { (key, ps) -> "${key.take(8)}=${ps.session.getSignalingState()}" } + Log.d(TAG) { "onPeerDisconnected: checking remaining sessions: $sessionStates" } val allDisconnected = peerSessions.keys.all { key -> key == peerPubKey || peerSessions[key]?.session?.getSignalingState() == PeerConnection.SignalingState.CLOSED } if (allDisconnected) { + Log.d(TAG) { "onPeerDisconnected: all peers disconnected, hanging up" } scope.launch { callManager.hangup() } + } else { + Log.d(TAG) { "onPeerDisconnected: other peers still active, continuing call" } } } + // ---- Per-peer cleanup ---- + + /** + * Disposes a single peer's WebRTC session when they leave the call + * 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) }}" } + try { + ps.session.dispose() + } catch (e: Exception) { + Log.e(TAG, "disposePeerSession: dispose() failed for ${peerPubKey.take(8)}", e) + } + // Update remote video tracks + val currentTracks = _remoteVideoTracks.value + if (peerPubKey in currentTracks) { + _remoteVideoTracks.value = currentTracks - peerPubKey + // If the disposed peer was the primary remote track, pick a new one + if (_remoteVideoTrack.value == currentTracks[peerPubKey]) { + stopRemoteVideoMonitor() + val nextTrack = _remoteVideoTracks.value.values.firstOrNull() + _remoteVideoTrack.value = nextTrack + if (nextTrack != null) { + startRemoteVideoMonitor(nextTrack) + } + } + } + } 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}" } // 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 diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt index 73c0f12f8..18f667c64 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/ui/screen/loggedIn/AccountViewModel.kt @@ -225,6 +225,7 @@ class AccountViewModel( callManager.onIceCandidateReceived = { event -> controller.onIceCandidateReceived(event) } callManager.onNewPeerInGroupCall = { peerPubKey -> controller.onNewPeerInGroupCall(peerPubKey) } callManager.onMidCallOfferReceived = { peerPubKey, sdpOffer -> controller.onMidCallOfferReceived(peerPubKey, sdpOffer) } + callManager.onPeerLeft = { peerPubKey -> controller.disposePeerSession(peerPubKey) } callController = controller // Populate ActiveCallHolder so CallActivity can launch even when the app diff --git a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt index a872d5230..369e20150 100644 --- a/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt +++ b/commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt @@ -64,10 +64,17 @@ class CallManager( /** Called when a mid-call offer is received from another callee in a group call. */ var onMidCallOfferReceived: ((peerPubKey: HexKey, sdpOffer: String) -> Unit)? = null + /** Called when a peer leaves the call (hangup) but the call continues with remaining peers. */ + var onPeerLeft: ((peerPubKey: HexKey) -> Unit)? = null + private var timeoutJob: Job? = null private var resetJob: Job? = null private val processedEventIds = mutableSetOf() + /** Peers whose answers we saw while still ringing (IncomingCall). + * After we accept, we trigger callee-to-callee mesh setup with them. */ + private val discoveredCalleePeers = mutableSetOf() + companion object { const val CALL_TIMEOUT_MS = 60_000L // 60 seconds ringing timeout const val ENDED_DISPLAY_MS = 2_000L // show "call ended" briefly before resetting @@ -220,6 +227,17 @@ class CallManager( val result = factory.createGroupCallAnswer(sdpAnswer, allRecipients, current.callId, signer) result.wraps.forEach { publishEvent(it) } Log.d("CallManager") { "acceptCall: answer published, now in Connecting state" } + + // Trigger callee-to-callee mesh connections with peers we discovered + // while still ringing (their answers arrived before we accepted). + val discovered = discoveredCalleePeers.toSet() + discoveredCalleePeers.clear() + if (discovered.isNotEmpty()) { + Log.d("CallManager") { "acceptCall: triggering mesh setup with ${discovered.size} discovered peers: ${discovered.map { it.take(8) }}" } + for (peer in discovered) { + onNewPeerInGroupCall?.invoke(peer) + } + } } suspend fun rejectCall() { @@ -243,6 +261,18 @@ class CallManager( "currentState=${current::class.simpleName}, sdpAnswerLength=${event.sdpAnswer().length}" } + // Self-answer: only meaningful as "answered elsewhere" in IncomingCall. + // In all other states it's our own echo from the relay — ignore it. + if (answeringPeer == signer.pubKey) { + if (current is CallState.IncomingCall && callId == current.callId) { + Log.d("CallManager") { "onCallAnswered: self-answer detected in IncomingCall — answered elsewhere" } + transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE) + } else { + Log.d("CallManager") { "onCallAnswered: ignoring self-answer echo in ${current::class.simpleName}" } + } + return + } + when (current) { is CallState.Offering -> { if (callId != current.callId) { @@ -272,20 +302,17 @@ class CallManager( ) } // Forward to CallController — it routes to the correct PeerSession - Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding" } + // and internally triggers callee-to-callee mesh setup if needed. + Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding to CallController" } onAnswerReceived?.invoke(event) - - // Notify callees about new peer for mesh setup - onNewPeerInGroupCall?.invoke(answeringPeer) } is CallState.IncomingCall -> { if (callId != current.callId) return - if (answeringPeer == signer.pubKey) { - transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE) - } - // Another group member answered — notify for future mesh connections - // (we'll connect to them after we accept the call ourselves) + // Another group member answered — remember them for mesh setup + // after we accept the call ourselves. + Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answered while we're still ringing, storing for later mesh" } + discoveredCalleePeers.add(answeringPeer) } is CallState.Connected -> { @@ -296,13 +323,11 @@ class CallManager( peerPubKeys = current.peerPubKeys + answeringPeer, pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer, ) - // Forward to CallController for routing - onAnswerReceived?.invoke(event) - onNewPeerInGroupCall?.invoke(answeringPeer) - } else { - // Renegotiation answer - onAnswerReceived?.invoke(event) } + // Forward to CallController — it routes to the correct PeerSession + // and internally triggers callee-to-callee mesh setup if needed. + Log.d("CallManager") { "onCallAnswered: peer ${answeringPeer.take(8)} answer in Connected, forwarding to CallController" } + onAnswerReceived?.invoke(event) } else -> { @@ -474,19 +499,24 @@ class CallManager( val callId = event.callId() ?: return val leavingPeer = event.pubKey + Log.d("CallManager") { "onPeerHangup: from=${leavingPeer.take(8)}, callId=$callId, state=${current::class.simpleName}" } + when (current) { is CallState.Connected -> { if (callId != current.callId) return val connectedRemaining = current.peerPubKeys - leavingPeer val pendingRemaining = current.pendingPeerPubKeys - leavingPeer if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) { + Log.d("CallManager") { "onPeerHangup: last peer left, ending call" } transitionToEnded(callId, current.allPeerPubKeys, EndReason.PEER_HANGUP) } else { + Log.d("CallManager") { "onPeerHangup: ${leavingPeer.take(8)} left, remaining=${connectedRemaining.map { it.take(8) }}, pending=${pendingRemaining.map { it.take(8) }}" } _state.value = current.copy( peerPubKeys = connectedRemaining, pendingPeerPubKeys = pendingRemaining, ) + onPeerLeft?.invoke(leavingPeer) } } @@ -495,13 +525,16 @@ class CallManager( val connectedRemaining = current.peerPubKeys - leavingPeer val pendingRemaining = current.pendingPeerPubKeys - leavingPeer if (connectedRemaining.isEmpty() && pendingRemaining.isEmpty()) { + Log.d("CallManager") { "onPeerHangup: last peer left during connecting, ending call" } transitionToEnded(callId, current.peerPubKeys + current.pendingPeerPubKeys, EndReason.PEER_HANGUP) } else { + Log.d("CallManager") { "onPeerHangup: ${leavingPeer.take(8)} left during connecting, remaining=${connectedRemaining.map { it.take(8) }}" } _state.value = current.copy( peerPubKeys = connectedRemaining, pendingPeerPubKeys = pendingRemaining, ) + onPeerLeft?.invoke(leavingPeer) } } @@ -512,6 +545,7 @@ class CallManager( transitionToEnded(callId, current.peerPubKeys, EndReason.PEER_HANGUP) } else { _state.value = current.copy(peerPubKeys = remaining) + onPeerLeft?.invoke(leavingPeer) } } @@ -542,6 +576,16 @@ class CallManager( } if (!processedEventIds.add(event.id)) return + // Filter out our own ICE candidates and hangups echoed back from relays. + // These are never useful: ICE candidates are for the remote peer, and + // hangups are already handled locally by hangup() → transitionToEnded. + // Self-answers and self-rejects are NOT filtered here because they serve + // as "answered/rejected elsewhere" signals when in IncomingCall state. + if (event.pubKey == signer.pubKey && (event is CallIceCandidateEvent || event is CallHangupEvent)) { + Log.d("CallManager") { "Ignoring self-event kind=${event.kind} id=${event.id.take(8)}" } + return + } + Log.d("CallManager") { "Processing signaling event kind=${event.kind} id=${event.id.take(8)} state=${_state.value::class.simpleName}" } when (event) { @@ -591,6 +635,7 @@ class CallManager( resetJob?.cancel() resetJob = null processedEventIds.clear() + discoveredCalleePeers.clear() } private fun transitionToEnded(