Merge pull request #2209 from vitorpamplona/claude/review-webrtc-calls-Vc6v0

Improve thread safety and error handling in call management

amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallController.kt

@@ -53,6 +53,8 @@ import org.webrtc.EglBase
 import org.webrtc.IceCandidate
 import org.webrtc.VideoTrack
 import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicBoolean
 
 private const val TAG = "CallController"
 private const val VIDEO_MAX_BITRATE_BPS_DEFAULT = 1_500_000
@@ -95,9 +97,11 @@ class CallController(
     val audioRoute: StateFlow<AudioRoute> = audioManager.audioRoute
     val isBluetoothAvailable: StateFlow<Boolean> = audioManager.isBluetoothAvailable
 
-    private var videoPausedByProximity = false
-    private var foregroundServiceStarted = false
-    private val videoSenders = mutableMapOf<HexKey, org.webrtc.RtpSender>()
+    @Volatile private var videoPausedByProximity = false
+
+    @Volatile private var foregroundServiceStarted = false
+    private val cleanedUp = AtomicBoolean(false)
+    private val videoSenders = ConcurrentHashMap<HexKey, org.webrtc.RtpSender>()
 
     private val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
     private var networkCallbackRegistered = false
@@ -126,6 +130,7 @@ class CallController(
             callManager.state.collect { state ->
                 when (state) {
                     is CallState.IncomingCall -> {
+                        ensureForegroundService()
                         withContext(Dispatchers.IO) { audioManager.startRinging() }
                         scope.launch {
                             NotificationUtils.showIncomingCallNotification(state.callerPubKey, context)
@@ -209,6 +214,7 @@ class CallController(
             } catch (e: Exception) {
                 Log.e(TAG, "Failed to initialize WebRTC", e)
                 _errorMessage.value = "Failed to start call: ${e.message}"
+                callManager.hangup()
                 return@launch
             }
 
@@ -247,6 +253,7 @@ class CallController(
             } catch (e: Exception) {
                 Log.e(TAG, "Failed to initialize WebRTC", e)
                 _errorMessage.value = "Failed to accept call: ${e.message}"
+                callManager.rejectCall()
                 return@launch
             }
 
@@ -256,6 +263,7 @@ class CallController(
                 } catch (e: Exception) {
                     Log.e(TAG, "Failed to create PeerConnection", e)
                     _errorMessage.value = "Failed to accept call: ${e.message}"
+                    callManager.rejectCall()
                     return@launch
                 }
 
@@ -582,6 +590,7 @@ class CallController(
     // ---- Cleanup ----
 
     fun cleanup() {
+        if (!cleanedUp.compareAndSet(false, true)) return
         unregisterNetworkCallback()
         try {
             audioManager.release()
@@ -610,6 +619,7 @@ class CallController(
         _isAudioMuted.value = false
         videoPausedByProximity = false
         videoSenders.clear()
+        cleanedUp.set(false)
     }
 
     // ---- Foreground service ----

amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/CallNotificationReceiver.kt

@@ -24,6 +24,9 @@ import android.content.BroadcastReceiver
 import android.content.Context
 import android.content.Intent
 import com.vitorpamplona.amethyst.service.notifications.NotificationUtils
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
 
 /**
@@ -40,7 +43,8 @@ class CallNotificationReceiver : BroadcastReceiver() {
     ) {
         val callManager = CallSessionBridge.callManager ?: return
         val pendingResult = goAsync()
-        kotlinx.coroutines.MainScope().launch {
+        val scope = CoroutineScope(SupervisorJob() + kotlinx.coroutines.Dispatchers.Main.immediate)
+        scope.launch {
             try {
                 when (intent.action) {
                     ACTION_REJECT_CALL -> {
@@ -54,6 +58,7 @@ class CallNotificationReceiver : BroadcastReceiver() {
                 }
             } finally {
                 pendingResult.finish()
+                scope.cancel()
             }
         }
     }

amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcCallSession.kt

@@ -55,8 +55,9 @@ class WebRtcCallSession(
     private val onRenegotiationNeeded: () -> Unit = {},
     private val onIceRestartOffer: (SessionDescription) -> Unit = {},
 ) {
-    private var peerConnection: PeerConnection? = null
-    private var iceRestartAttempted = false
+    @Volatile private var peerConnection: PeerConnection? = null
+
+    @Volatile private var iceRestartAttempted = false
 
     fun createPeerConnection() {
         val rtcConfig =
@@ -65,7 +66,7 @@ class WebRtcCallSession(
                 continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY
             }
 
-        peerConnection =
+        val pc =
             peerConnectionFactory.createPeerConnection(
                 rtcConfig,
                 object : PeerConnection.Observer {
@@ -155,7 +156,8 @@ class WebRtcCallSession(
                         }
                     }
                 },
-            )
+            ) ?: throw IllegalStateException("PeerConnectionFactory.createPeerConnection returned null")
+        peerConnection = pc
     }
 
     /**
@@ -345,9 +347,10 @@ class WebRtcCallSession(
     }
 
     fun dispose() {
-        peerConnection?.close()
-        peerConnection?.dispose()
+        val pc = peerConnection ?: return
         peerConnection = null
+        pc.close()
+        pc.dispose()
     }
 
     private fun loggingSdpObserver(label: String) =

commons/src/commonMain/kotlin/com/vitorpamplona/amethyst/commons/call/CallManager.kt

@@ -201,9 +201,9 @@ class CallManager(
         val result = factory.createCallOffer(sdpOffer, calleePubKey, callId, callType, signer)
         stateMutex.withLock {
             _state.value = CallState.Offering(callId, setOf(calleePubKey), callType)
+            startTimeout(callId)
         }
         publishEvent(result.wrap)
-        startTimeout(callId)
         Log.d("CallManager") { "initiateCall: offer published, timeout started" }
     }
 
@@ -281,9 +281,10 @@ class CallManager(
             discoveredCalleePeers.clear()
         }
 
-        val allRecipients = current.groupMembers + signer.pubKey
-        Log.d("CallManager") { "acceptCall: publishing answer to ${allRecipients.size} recipients" }
-        val result = factory.createGroupCallAnswer(sdpAnswer, allRecipients, current.callId, signer)
+        val allMembers = current.groupMembers + signer.pubKey
+        val otherMembers = allMembers - signer.pubKey
+        Log.d("CallManager") { "acceptCall: publishing answer to ${otherMembers.size} recipients" }
+        val result = factory.createGroupCallAnswer(sdpAnswer, otherMembers, current.callId, signer)
         result.wraps.forEach { publishEvent(it) }
         Log.d("CallManager") { "acceptCall: answer published, now in Connecting state" }
 
@@ -306,8 +307,8 @@ class CallManager(
             transitionToEnded(current.callId, current.peerPubKeys(), EndReason.REJECTED)
         }
 
-        val allRecipients = current.groupMembers + signer.pubKey
-        val result = factory.createGroupReject(allRecipients, current.callId, signer = signer)
+        val otherMembers = current.groupMembers - signer.pubKey
+        val result = factory.createGroupReject(otherMembers, current.callId, signer = signer)
         result.wraps.forEach { publishEvent(it) }
     }