From 0488245200988ef8f7423ef10498b313f0e357ea Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Apr 2026 15:36:50 +0000 Subject: [PATCH 1/2] fix: send ICE candidates to all peers and exclude self from peer set Two bugs caused WebRTC group calls to get stuck in "Connecting": 1. CallController.onLocalIceCandidate() used currentPeerPubKey() which returns only the first peer via firstOrNull(). ICE candidates were gift-wrapped to only one peer; the others never received them. Fixed by iterating over all currentPeerPubKeys(). 2. CallManager.acceptCall() set Connecting.peerPubKeys to groupMembers which includes the local user's own pubkey. This caused currentPeerPubKey() to potentially return self, sending ICE candidates to oneself instead of the caller. Fixed by filtering out signer.pubKey from the peer set. https://claude.ai/code/session_01J5fJx9YbSBx1BsBMctiAm8 --- .../vitorpamplona/amethyst/service/call/CallController.kt | 8 +++++--- .../vitorpamplona/amethyst/commons/call/CallManager.kt | 2 +- 2 files changed, 6 insertions(+), 4 deletions(-) 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 25200c4ba0..9b2ebbff73 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 @@ -491,13 +491,15 @@ class CallController( private fun onLocalIceCandidate(candidate: IceCandidate) { Log.d(TAG) { "Local ICE candidate: ${candidate.sdp.take(50)}" } val callId = callManager.currentCallId() ?: return - val peerPubKey = callManager.currentPeerPubKey() ?: return + val peerPubKeys = callManager.currentPeerPubKeys()?.takeIf { it.isNotEmpty() } ?: return val candidateJson = CallIceCandidateEvent.serializeCandidate(candidate.sdp, candidate.sdpMid, candidate.sdpMLineIndex) scope.launch { val signer = signerProvider() - val result = callFactory.createIceCandidate(candidateJson, peerPubKey, callId, signer) - publishWrap(result.wrap) + for (peerPubKey in peerPubKeys) { + val result = callFactory.createIceCandidate(candidateJson, peerPubKey, callId, signer) + publishWrap(result.wrap) + } } } 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 f3672bc6de..07abddbae6 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 @@ -148,7 +148,7 @@ class CallManager( } Log.d("CallManager") { "acceptCall: callId=${current.callId}, transitioning to Connecting, sdpAnswerLength=${sdpAnswer.length}" } - _state.value = CallState.Connecting(current.callId, current.peerPubKeys(), current.callType) + _state.value = CallState.Connecting(current.callId, current.peerPubKeys() - signer.pubKey, current.callType) cancelTimeout() // Include all group members + self so other devices get notified too. From fcecb5241c86e451431b54800bab29cf5aaa62ee Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Apr 2026 16:54:00 +0000 Subject: [PATCH 2/2] feat: implement full-mesh WebRTC group calls with per-peer PeerConnections MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the single-PeerConnection architecture with full mesh topology where each participant maintains one PeerConnection per peer, as specified by NIP-AC. WebRtcCallSession: Refactored to a pure PeerConnection wrapper that accepts a shared PeerConnectionFactory. No longer manages media sources, tracks, or camera — those are now shared across all peer sessions. CallController: Manages per-peer sessions via ConcurrentHashMap. Shared resources (PeerConnectionFactory, EglBase, audio/video sources, camera) are initialized once and reused. Each peer gets its own WebRtcCallSession with per-peer ICE candidate routing. Supports callee-to-callee mesh connections with pubkey-based tie-breaking to avoid ofer glare. CallManager: New methods for per-peer offer/answer publishing (publishOfferToPeer, publishAnswerToPeer, beginOffering). Forwards ALL answers to CallController (not just the first). New callbacks onNewPeerInGroupCall and onMidCallOfferReceived for mesh setup. Handles mid-call offers (same call-id) when already in Connecting/Connected state. AccountViewModel: Updated callback wiring to pass peer pubkey with answer events and register new group call callbacks. https://claude.ai/code/session_01J5fJx9YbSBx1BsBMctiAm8 --- .../amethyst/service/call/CallController.kt | 656 +++++++++++++----- .../service/call/WebRtcCallSession.kt | 142 +--- .../ui/screen/loggedIn/AccountViewModel.kt | 4 +- .../amethyst/commons/call/CallManager.kt | 173 +++-- 4 files changed, 608 insertions(+), 367 deletions(-) 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 9b2ebbff73..638ef64f99 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 @@ -30,6 +30,7 @@ import com.vitorpamplona.amethyst.commons.call.CallManager import com.vitorpamplona.amethyst.commons.call.CallState import com.vitorpamplona.amethyst.model.LocalCache import com.vitorpamplona.amethyst.service.notifications.NotificationUtils +import com.vitorpamplona.quartz.nip01Core.core.HexKey import com.vitorpamplona.quartz.nip01Core.core.hexToByteArray import com.vitorpamplona.quartz.nip19Bech32.toNpub import com.vitorpamplona.quartz.nip59Giftwrap.wraps.GiftWrapEvent @@ -46,17 +47,31 @@ import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow import kotlinx.coroutines.launch import kotlinx.coroutines.withContext +import org.webrtc.AudioSource +import org.webrtc.AudioTrack +import org.webrtc.Camera2Enumerator +import org.webrtc.CameraVideoCapturer +import org.webrtc.DefaultVideoDecoderFactory +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 +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" +private const val VIDEO_MAX_BITRATE_BPS = 1_500_000 class CallController( private val context: Context, @@ -65,25 +80,45 @@ class CallController( private val publishWrap: suspend (GiftWrapEvent) -> Unit, private val signerProvider: suspend () -> com.vitorpamplona.quartz.nip01Core.signers.NostrSigner, ) { - private var webRtcSession: WebRtcCallSession? = null + // ---- Per-peer session state ---- + + private class PeerSessionState( + val session: WebRtcCallSession, + val remoteDescriptionSet: AtomicBoolean = AtomicBoolean(false), + val pendingIceCandidates: CopyOnWriteArrayList = CopyOnWriteArrayList(), + ) + + private val peerSessions = ConcurrentHashMap() + + // ---- Shared WebRTC resources ---- + + private var peerConnectionFactory: PeerConnectionFactory? = null + private var sharedEglBase: EglBase? = null + private var localAudioSource: AudioSource? = null + private var localVideoSource: VideoSource? = null + private var localAudioTrackInternal: AudioTrack? = null + private var localVideoTrackInternal: VideoTrack? = null + private var cameraCapturer: CameraVideoCapturer? = null + private val callFactory = WebRtcCallFactory() - private val remoteDescriptionSet = - java.util.concurrent.atomic - .AtomicBoolean(false) - private val pendingIceCandidates = CopyOnWriteArrayList() val audioManager = CallAudioManager(context) - // Video tracks exposed to UI - private val _remoteVideoTrack = MutableStateFlow(null) - val remoteVideoTrack: StateFlow = _remoteVideoTrack.asStateFlow() + // ---- UI-exposed state ---- + private val _localVideoTrack = MutableStateFlow(null) val localVideoTrack: StateFlow = _localVideoTrack.asStateFlow() - // Error state exposed to UI + // Primary remote track (first connected peer) for backward-compat with P2P UI + private val _remoteVideoTrack = MutableStateFlow(null) + val remoteVideoTrack: StateFlow = _remoteVideoTrack.asStateFlow() + + // All remote tracks keyed by peer pubkey (for group call UI) + private val _remoteVideoTracks = MutableStateFlow>(emptyMap()) + val remoteVideoTracks: StateFlow> = _remoteVideoTracks.asStateFlow() + private val _errorMessage = MutableStateFlow(null) val errorMessage: StateFlow = _errorMessage.asStateFlow() - // Remote video frame monitoring — detects when peer stops sending video private val _isRemoteVideoActive = MutableStateFlow(false) val isRemoteVideoActive: StateFlow = _isRemoteVideoActive.asStateFlow() private val _remoteVideoAspectRatio = MutableStateFlow(null) @@ -100,7 +135,6 @@ class CallController( } } - // Audio/video toggle state (UI concerns, not domain state) private val _isAudioMuted = MutableStateFlow(false) val isAudioMuted: StateFlow = _isAudioMuted.asStateFlow() private val _isVideoEnabled = MutableStateFlow(false) @@ -108,8 +142,10 @@ class CallController( val audioRoute: StateFlow = audioManager.audioRoute val isBluetoothAvailable: StateFlow = audioManager.isBluetoothAvailable - // Tracks whether video is paused because the phone is near the ear private var videoPausedByProximity = false + private var foregroundServiceStarted = false + + // ---- Initialization ---- init { callManager.onRenegotiationOfferReceived = { event -> onRenegotiationOfferReceived(event) } @@ -147,23 +183,24 @@ class CallController( } } - // Pause outgoing video when the phone is held near the ear scope.launch { audioManager.isNearEar.collect { nearEar -> - val session = webRtcSession ?: return@collect + val videoTrack = localVideoTrackInternal ?: return@collect if (nearEar && _isVideoEnabled.value && !videoPausedByProximity) { videoPausedByProximity = true - session.setVideoEnabled(false) - session.stopCamera() + videoTrack.setEnabled(false) + stopCamera() } else if (!nearEar && videoPausedByProximity) { videoPausedByProximity = false - session.setVideoEnabled(true) - session.startCamera() + videoTrack.setEnabled(true) + startCamera() } } } } + // ---- Call initiation (caller side) ---- + fun initiateCall( peerPubKey: String, callType: CallType, @@ -178,6 +215,10 @@ class CallController( initiateCallInternal(peerPubKeys, callType) } + /** + * Creates a separate PeerConnection (and SDP offer) for each callee, + * establishing full-mesh connectivity for group calls. + */ private fun initiateCallInternal( peerPubKeys: Set, callType: CallType, @@ -186,46 +227,40 @@ class CallController( scope.launch { val callId = UUID.randomUUID().toString() _errorMessage.value = null - remoteDescriptionSet.set(false) - pendingIceCandidates.clear() try { - withContext(Dispatchers.IO) { createWebRtcSession() } - Log.d(TAG) { "initiateCall: WebRTC session created" } + withContext(Dispatchers.IO) { initializeSharedResources(callType) } + Log.d(TAG) { "initiateCall: shared resources initialized" } } catch (e: Exception) { - Log.e(TAG, "Failed to create WebRTC session", e) + Log.e(TAG, "Failed to initialize WebRTC", e) _errorMessage.value = "Failed to start call: ${e.message}" return@launch } - val session = - webRtcSession ?: run { - _errorMessage.value = "Failed to create WebRTC session" - return@launch - } + // Set state to Offering before creating peer sessions + callManager.beginOffering(callId, peerPubKeys, callType) - session.addAudioTrack() - if (callType == CallType.VIDEO) { - session.addVideoTrack() - _localVideoTrack.value = session.getLocalVideoTrack() - _isVideoEnabled.value = true - } - - Log.d(TAG) { "initiateCall: creating offer..." } - session.createOffer { sdp -> - Log.d(TAG) { "initiateCall: offer created, sdpLength=${sdp.description.length}, publishing..." } - scope.launch { - if (peerPubKeys.size == 1) { - callManager.initiateCall(peerPubKeys.first(), callType, callId, sdp.description) - } else { - callManager.initiateGroupCall(peerPubKeys, callType, callId, sdp.description) + // Create a PeerConnection + offer for each callee + for (peerPubKey in peerPubKeys) { + try { + val ps = withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } + Log.d(TAG) { "initiateCall: PeerConnection created for ${peerPubKey.take(8)}" } + ps.session.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) + Log.d(TAG) { "initiateCall: offer published for ${peerPubKey.take(8)}" } + } } - Log.d(TAG) { "initiateCall: offer published, callId=$callId, state=${callManager.state.value::class.simpleName}" } + } catch (e: Exception) { + Log.e(TAG, "Failed to create PeerConnection for ${peerPubKey.take(8)}", e) } } } } + // ---- Accept incoming call (callee side) ---- + fun acceptIncomingCall(sdpOffer: String) { val state = callManager.state.value if (state !is CallState.IncomingCall) { @@ -233,45 +268,43 @@ class CallController( return } - Log.d(TAG) { "acceptIncomingCall: callId=${state.callId}, callType=${state.callType}, sdpOfferLength=${sdpOffer.length}, pendingICE=${pendingIceCandidates.size}" } + val callerPubKey = state.callerPubKey + Log.d(TAG) { "acceptIncomingCall: callId=${state.callId}, callType=${state.callType}, sdpOfferLength=${sdpOffer.length}" } scope.launch { _errorMessage.value = null - remoteDescriptionSet.set(false) - // Don't clear pendingIceCandidates here — they were buffered while ringing try { - withContext(Dispatchers.IO) { createWebRtcSession() } - Log.d(TAG) { "acceptIncomingCall: WebRTC session created" } + withContext(Dispatchers.IO) { initializeSharedResources(state.callType) } + Log.d(TAG) { "acceptIncomingCall: shared resources initialized" } } catch (e: Exception) { - Log.e(TAG, "Failed to create WebRTC session", e) + Log.e(TAG, "Failed to initialize WebRTC", e) _errorMessage.value = "Failed to accept call: ${e.message}" return@launch } - val session = - webRtcSession ?: run { - Log.e(TAG, "acceptIncomingCall: webRtcSession is null after creation") - _errorMessage.value = "Failed to create WebRTC session" + // Retrieve any ICE candidates buffered while ringing (before session existed) + val globalPending = getGlobalPendingCandidates(callerPubKey) + + val ps = + try { + withContext(Dispatchers.IO) { createPeerSession(callerPubKey) } + } catch (e: Exception) { + Log.e(TAG, "Failed to create PeerConnection", e) + _errorMessage.value = "Failed to accept call: ${e.message}" return@launch } - session.addAudioTrack() - Log.d(TAG) { "acceptIncomingCall: audio track added" } - if (state.callType == CallType.VIDEO) { - session.addVideoTrack() - _localVideoTrack.value = session.getLocalVideoTrack() - _isVideoEnabled.value = true - Log.d(TAG) { "acceptIncomingCall: video track added" } - } + // Inject any globally-buffered ICE candidates for the caller + globalPending.forEach { ps.pendingIceCandidates.add(it) } Log.d(TAG) { "acceptIncomingCall: setting remote description (OFFER)..." } - session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) - Log.d(TAG) { "acceptIncomingCall: flushing ${pendingIceCandidates.size} pending ICE candidates..." } - flushPendingIceCandidates() + ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) + Log.d(TAG) { "acceptIncomingCall: flushing ${ps.pendingIceCandidates.size} pending ICE candidates..." } + flushPendingIceCandidates(callerPubKey, ps) Log.d(TAG) { "acceptIncomingCall: creating answer..." } - session.createAnswer { sdp -> + ps.session.createAnswer { sdp -> Log.d(TAG) { "acceptIncomingCall: answer created, sdpLength=${sdp.description.length}, publishing..." } scope.launch { callManager.acceptCall(sdp.description) @@ -281,82 +314,246 @@ class CallController( } } - fun onCallAnswerReceived(sdpAnswer: String) { - val session = webRtcSession - if (session == null) { - Log.e(TAG, "Answer received but webRtcSession is NULL — cannot set remote description") - return - } - val signalingState = session.getSignalingState() - Log.d(TAG) { - "Answer received, SDP length=${sdpAnswer.length}, signalingState=$signalingState, " + - "remoteDescriptionSet=${remoteDescriptionSet.get()}, pendingICE=${pendingIceCandidates.size}, " + - "callState=${callManager.state.value::class.simpleName}" - } + // ---- Answer routing ---- - // An answer is only valid when we have a pending local offer. - // Ignore stale answers (e.g. our own renegotiation answer echoed back by the relay). - if (signalingState != PeerConnection.SignalingState.HAVE_LOCAL_OFFER) { - Log.d(TAG) { "Ignoring answer in $signalingState state (no pending local offer)" } - return - } + /** + * Routes an answer to the correct per-peer PeerConnection. + * For the caller: the answer is for a PeerConnection we already created. + * For a callee seeing another callee's answer: we don't have a session + * for them yet, so we trigger callee-to-callee connection. + */ + fun onCallAnswerReceived( + peerPubKey: HexKey, + sdpAnswer: String, + ) { + 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()}" + } - Log.d(TAG) { "Setting remote description (ANSWER)..." } - session.setRemoteDescription(SessionDescription(SessionDescription.Type.ANSWER, sdpAnswer)) - Log.d(TAG) { "Flushing pending ICE candidates after answer..." } - flushPendingIceCandidates() + if (signalingState != PeerConnection.SignalingState.HAVE_LOCAL_OFFER) { + Log.d(TAG) { "Ignoring answer from ${peerPubKey.take(8)} in $signalingState (no pending local offer)" } + return + } + + 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) + } } + // ---- ICE candidate routing ---- + + /** + * Routes an incoming ICE candidate to the correct per-peer session. + * Before a session exists for a peer, candidates are buffered globally. + */ fun onIceCandidateReceived(event: CallIceCandidateEvent) { try { + val senderPubKey = event.pubKey val candidate = IceCandidate(event.sdpMid(), event.sdpMLineIndex(), event.candidateSdp()) - if (webRtcSession != null && remoteDescriptionSet.get()) { - Log.d(TAG) { "Adding ICE candidate directly: ${candidate.sdp.take(50)}" } - webRtcSession?.addIceCandidate(candidate) + 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 { - Log.d(TAG) { "Buffering ICE candidate (session=${webRtcSession != null}, remoteSet=${remoteDescriptionSet.get()})" } - pendingIceCandidates.add(candidate) + // 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) } } catch (e: Exception) { Log.e(TAG, "Failed to parse ICE candidate", e) } } - private fun flushPendingIceCandidates() { - remoteDescriptionSet.set(true) - val session = webRtcSession ?: return - val candidates = pendingIceCandidates.toList() - Log.d(TAG) { "Flushing ${candidates.size} buffered ICE candidates" } - pendingIceCandidates.clear() - candidates.forEach { session.addIceCandidate(it) } + // Candidates received before a PeerSession exists for the sender + private val globalPendingIce = ConcurrentHashMap>() + + 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) } } - // UI toggle controls + private fun onLocalIceCandidate( + peerPubKey: HexKey, + candidate: IceCandidate, + ) { + Log.d(TAG) { "Local ICE candidate for ${peerPubKey.take(8)}: ${candidate.sdp.take(50)}" } + val callId = callManager.currentCallId() ?: return + val candidateJson = CallIceCandidateEvent.serializeCandidate(candidate.sdp, candidate.sdpMid, candidate.sdpMLineIndex) + + scope.launch { + val signer = signerProvider() + val result = callFactory.createIceCandidate(candidateJson, peerPubKey, callId, signer) + publishWrap(result.wrap) + } + } + + // ---- Callee-to-callee mesh connections ---- + + /** + * Another callee joined the group call. Establish a direct PeerConnection + * to them. To avoid glare (both sides sending offers simultaneously), the + * peer with the lexicographically lower pubkey initiates. + */ + fun onNewPeerInGroupCall(peerPubKey: HexKey) { + if (peerSessions.containsKey(peerPubKey)) return // already connected/connecting + + scope.launch { + val myPubKey = signerProvider().pubKey + if (myPubKey < peerPubKey) { + Log.d(TAG) { "Initiating callee-to-callee connection to ${peerPubKey.take(8)} (I have lower pubkey)" } + createAndOfferToPeer(peerPubKey) + } else { + Log.d(TAG) { "Waiting for callee-to-callee offer from ${peerPubKey.take(8)} (they have lower pubkey)" } + } + } + } + + private suspend fun createAndOfferToPeer(peerPubKey: HexKey) { + if (peerConnectionFactory == null) return + val globalPending = getGlobalPendingCandidates(peerPubKey) + + val ps = + try { + withContext(Dispatchers.IO) { createPeerSession(peerPubKey) } + } catch (e: Exception) { + Log.e(TAG, "Failed to create PeerConnection for ${peerPubKey.take(8)}", e) + return + } + + globalPending.forEach { ps.pendingIceCandidates.add(it) } + + ps.session.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) + } + } + } + + /** + * A mid-call offer was received from another callee in the group. + * Create a PeerSession, accept their offer, and send back an answer. + */ + fun onMidCallOfferReceived( + peerPubKey: HexKey, + sdpOffer: String, + ) { + if (peerSessions.containsKey(peerPubKey)) { + Log.d(TAG) { "Mid-call offer from ${peerPubKey.take(8)} but session already exists — ignoring" } + return + } + + Log.d(TAG) { "Mid-call offer from ${peerPubKey.take(8)}, sdpLength=${sdpOffer.length}" } + scope.launch { + if (peerConnectionFactory == null) { + Log.e(TAG, "Mid-call offer but factory not initialized") + return@launch + } + + val globalPending = getGlobalPendingCandidates(peerPubKey) + + val ps = + try { + withContext(Dispatchers.IO) { createPeerSession(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) } + + ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) + flushPendingIceCandidates(peerPubKey, ps) + + ps.session.createAnswer { sdp -> + Log.d(TAG) { "Callee-to-callee answer for ${peerPubKey.take(8)}, sdpLength=${sdp.description.length}" } + scope.launch { + callManager.publishAnswerToPeer(peerPubKey, sdp.description) + } + } + } + } + + // ---- Renegotiation ---- + + private fun onRenegotiationOfferReceived(event: CallRenegotiateEvent) { + val peerPubKey = event.pubKey + val ps = peerSessions[peerPubKey] ?: return + val sdpOffer = event.sdpOffer() + Log.d(TAG) { "Renegotiation offer from ${peerPubKey.take(8)}, sdpLength=${sdpOffer.length}" } + + scope.launch { + ps.session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) + ps.session.createAnswer { sdp -> + scope.launch { + callManager.sendRenegotiationAnswer(sdp.description, peerPubKey) + } + } + } + } + + private fun performRenegotiation(peerPubKey: HexKey) { + val ps = peerSessions[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 -> + scope.launch { + callManager.sendRenegotiation(sdp.description, peerPubKey) + } + } + } + + // ---- UI toggle controls ---- + fun toggleAudioMute() { val muted = !_isAudioMuted.value _isAudioMuted.value = muted - webRtcSession?.setAudioEnabled(!muted) + localAudioTrackInternal?.setEnabled(!muted) } fun toggleVideo() { - val session = webRtcSession ?: return val enabling = !_isVideoEnabled.value if (enabling) { - if (session.getLocalVideoTrack() == null) { - // No video track yet — create one (voice → video upgrade) - session.addVideoTrack() + 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) } + } } else { - // Track exists but camera was stopped — restart it - session.setVideoEnabled(true) - session.startCamera() + localVideoTrackInternal?.setEnabled(true) + startCamera() } - _localVideoTrack.value = session.getLocalVideoTrack() + _localVideoTrack.value = localVideoTrackInternal _isVideoEnabled.value = true } else { - // Disable video: stop camera and disable track - session.setVideoEnabled(false) - session.stopCamera() + localVideoTrackInternal?.setEnabled(false) + stopCamera() _isVideoEnabled.value = false } } @@ -365,23 +562,17 @@ class CallController( audioManager.cycleAudioRoute() } - fun getEglBase() = webRtcSession?.eglBase + fun getEglBase(): EglBase? = sharedEglBase fun invitePeer(peerPubKey: String) { scope.launch { - val session = webRtcSession ?: return@launch - session.createOffer { sdp -> - scope.launch { - callManager.invitePeer(peerPubKey, sdp.description) - } - } + createAndOfferToPeer(peerPubKey) } } fun hangup() { scope.launch { callManager.hangup() - // cleanup is triggered by the state collector when Ended is reached } } @@ -389,54 +580,182 @@ class CallController( _errorMessage.value = null } - private fun onRenegotiationOfferReceived(event: CallRenegotiateEvent) { - val session = webRtcSession ?: return - val sdpOffer = event.sdpOffer() - val peerPubKey = event.pubKey - Log.d(TAG) { "Renegotiation offer received, SDP length=${sdpOffer.length}" } + // ---- Shared resource management ---- - scope.launch { - session.setRemoteDescription(SessionDescription(SessionDescription.Type.OFFER, sdpOffer)) - session.createAnswer { sdp -> - scope.launch { - callManager.sendRenegotiationAnswer(sdp.description, peerPubKey) - } - } + private fun initializeSharedResources(callType: CallType) { + if (peerConnectionFactory != null) return // already initialized + + sharedEglBase = EglBase.create() + + PeerConnectionFactory.initialize( + PeerConnectionFactory + .InitializationOptions + .builder(context) + .createInitializationOptions(), + ) + + peerConnectionFactory = + PeerConnectionFactory + .builder() + .setVideoDecoderFactory(DefaultVideoDecoderFactory(sharedEglBase!!.eglBaseContext)) + .setVideoEncoderFactory(DefaultVideoEncoderFactory(sharedEglBase!!.eglBaseContext, true, true)) + .createPeerConnectionFactory() + + // Audio + localAudioSource = peerConnectionFactory?.createAudioSource(MediaConstraints()) + localAudioTrackInternal = peerConnectionFactory?.createAudioTrack("audio0", localAudioSource) + + // Video (if video call) + if (callType == CallType.VIDEO) { + createVideoResources() } } - private fun performRenegotiation() { - val session = webRtcSession ?: return - val state = callManager.state.value - if (state !is CallState.Connected && state !is CallState.Connecting) return - val peerPubKey = callManager.currentPeerPubKey() ?: return + private fun createVideoResources() { + if (localVideoSource != null) return + val factory = peerConnectionFactory ?: return - Log.d(TAG) { "Starting renegotiation" } - session.createOffer { sdp -> - scope.launch { - callManager.sendRenegotiation(sdp.description, peerPubKey) + localVideoSource = factory.createVideoSource(false) + localVideoTrackInternal = factory.createVideoTrack("video0", localVideoSource) + _localVideoTrack.value = localVideoTrackInternal + _isVideoEnabled.value = true + startCamera() + } + + private fun startCamera() { + if (cameraCapturer != null) return + val source = localVideoSource ?: return + val egl = sharedEglBase ?: return + + val enumerator = Camera2Enumerator(context) + val frontCamera = enumerator.deviceNames.firstOrNull { enumerator.isFrontFacing(it) } + val camera = frontCamera ?: enumerator.deviceNames.firstOrNull() ?: return + + cameraCapturer = + enumerator.createCapturer(camera, null)?.also { + it.initialize( + SurfaceTextureHelper.create("CaptureThread", egl.eglBaseContext), + context, + source.capturerObserver, + ) + it.startCapture(1280, 720, 30) } + } + + private fun stopCamera() { + cameraCapturer?.stopCapture() + cameraCapturer?.dispose() + cameraCapturer = null + } + + // ---- Per-peer PeerConnection creation ---- + + private fun createPeerSession(peerPubKey: HexKey): PeerSessionState { + val factory = peerConnectionFactory ?: throw IllegalStateException("PeerConnectionFactory not initialized") + + val session = + WebRtcCallSession( + peerConnectionFactory = factory, + iceServers = IceServerConfig.buildIceServers(), + onIceCandidate = { candidate -> onLocalIceCandidate(peerPubKey, candidate) }, + onPeerConnected = { + Log.d(TAG) { "Peer ${peerPubKey.take(8)} connected!" } + callManager.onPeerConnected() + if (!foregroundServiceStarted) { + foregroundServiceStarted = true + startForegroundService() + } + }, + onRemoteVideoTrack = { track -> onRemoteVideoTrack(peerPubKey, track) }, + onDisconnected = { onPeerDisconnected(peerPubKey) }, + onError = { error -> _errorMessage.value = error }, + onRenegotiationNeeded = { performRenegotiation(peerPubKey) }, + ) + + try { + session.createPeerConnection() + } catch (e: Exception) { + session.dispose() + throw e + } + + // Add shared local tracks to this PeerConnection + localAudioTrackInternal?.let { session.addTrack(it) } + localVideoTrackInternal?.let { session.addTrack(it, VIDEO_MAX_BITRATE_BPS) } + + val ps = PeerSessionState(session) + peerSessions[peerPubKey] = ps + return ps + } + + private fun onRemoteVideoTrack( + peerPubKey: HexKey, + track: VideoTrack, + ) { + Log.d(TAG) { "Remote video track from ${peerPubKey.take(8)}" } + _remoteVideoTracks.value = _remoteVideoTracks.value + (peerPubKey to track) + // Backward-compat: set the first remote track as the primary + if (_remoteVideoTrack.value == null) { + _remoteVideoTrack.value = track + startRemoteVideoMonitor(track) } } + private fun onPeerDisconnected(peerPubKey: HexKey) { + Log.d(TAG) { "Peer ${peerPubKey.take(8)} disconnected" } + // If all peers disconnected, hang up + val allDisconnected = + peerSessions.keys.all { key -> + key == peerPubKey || peerSessions[key]?.session?.getSignalingState() == PeerConnection.SignalingState.CLOSED + } + if (allDisconnected) { + scope.launch { callManager.hangup() } + } + } + + // ---- Cleanup ---- + fun cleanup() { audioManager.release() stopForegroundService() + foregroundServiceStarted = false NotificationUtils.cancelCallNotification(context) stopRemoteVideoMonitor() + + // Dispose all peer sessions + peerSessions.values.forEach { it.session.dispose() } + peerSessions.clear() + + // Dispose shared resources + stopCamera() + localAudioTrackInternal?.dispose() + localVideoTrackInternal?.dispose() + localAudioSource?.dispose() + localVideoSource?.dispose() + peerConnectionFactory?.dispose() + sharedEglBase?.release() + + localAudioTrackInternal = null + localVideoTrackInternal = null + localAudioSource = null + localVideoSource = null + peerConnectionFactory = null + sharedEglBase = null + + globalPendingIce.clear() + _remoteVideoTrack.value = null + _remoteVideoTracks.value = emptyMap() _localVideoTrack.value = null _isAudioMuted.value = false _isVideoEnabled.value = false _isRemoteVideoActive.value = false _remoteVideoAspectRatio.value = null videoPausedByProximity = false - webRtcSession?.dispose() - webRtcSession = null - remoteDescriptionSet.set(false) - pendingIceCandidates.clear() } + // ---- Remote video monitoring ---- + private fun startRemoteVideoMonitor(track: VideoTrack) { stopRemoteVideoMonitor() lastRemoteFrameTimeMs.set(System.currentTimeMillis()) @@ -460,48 +779,7 @@ class CallController( } } - private fun createWebRtcSession() { - val session = - WebRtcCallSession( - context = context, - iceServers = IceServerConfig.buildIceServers(), - onIceCandidate = { candidate -> onLocalIceCandidate(candidate) }, - onPeerConnected = { - callManager.onPeerConnected() - startForegroundService() - }, - onRemoteVideoTrack = { track -> - _remoteVideoTrack.value = track - startRemoteVideoMonitor(track) - }, - onDisconnected = { scope.launch { callManager.hangup() } }, - onError = { error -> _errorMessage.value = error }, - onRenegotiationNeeded = { performRenegotiation() }, - ) - try { - session.initialize() - session.createPeerConnection() - webRtcSession = session - } catch (e: Exception) { - session.dispose() - throw e - } - } - - private fun onLocalIceCandidate(candidate: IceCandidate) { - Log.d(TAG) { "Local ICE candidate: ${candidate.sdp.take(50)}" } - val callId = callManager.currentCallId() ?: return - val peerPubKeys = callManager.currentPeerPubKeys()?.takeIf { it.isNotEmpty() } ?: return - val candidateJson = CallIceCandidateEvent.serializeCandidate(candidate.sdp, candidate.sdpMid, candidate.sdpMLineIndex) - - scope.launch { - val signer = signerProvider() - for (peerPubKey in peerPubKeys) { - val result = callFactory.createIceCandidate(candidateJson, peerPubKey, callId, signer) - publishWrap(result.wrap) - } - } - } + // ---- Foreground service ---- private fun startForegroundService() { try { @@ -530,6 +808,8 @@ class CallController( } } + // ---- Incoming call notification ---- + private suspend fun showIncomingCallNotification(callerPubKey: String) { val callerUser = LocalCache.getUserIfExists(callerPubKey) val callerName = callerUser?.toBestDisplayName() ?: callerPubKey.take(8) + "..." diff --git a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcCallSession.kt b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcCallSession.kt index c74f41f6e8..44745c0eea 100644 --- a/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcCallSession.kt +++ b/amethyst/src/main/java/com/vitorpamplona/amethyst/service/call/WebRtcCallSession.kt @@ -20,32 +20,32 @@ */ package com.vitorpamplona.amethyst.service.call -import android.content.Context import com.vitorpamplona.quartz.utils.Log -import org.webrtc.AudioSource -import org.webrtc.AudioTrack -import org.webrtc.Camera2Enumerator -import org.webrtc.CameraVideoCapturer import org.webrtc.DataChannel -import org.webrtc.DefaultVideoDecoderFactory -import org.webrtc.DefaultVideoEncoderFactory -import org.webrtc.EglBase import org.webrtc.IceCandidate import org.webrtc.MediaConstraints import org.webrtc.MediaStream +import org.webrtc.MediaStreamTrack import org.webrtc.PeerConnection import org.webrtc.PeerConnectionFactory import org.webrtc.RtpReceiver +import org.webrtc.RtpSender import org.webrtc.SdpObserver import org.webrtc.SessionDescription -import org.webrtc.SurfaceTextureHelper -import org.webrtc.VideoSource import org.webrtc.VideoTrack private const val TAG = "WebRtcCallSession" +/** + * Lightweight wrapper around a single [PeerConnection]. + * + * The caller is responsible for creating and managing the shared + * [PeerConnectionFactory], media sources, tracks, and camera. This class + * only owns the [PeerConnection] itself and delegates lifecycle callbacks + * through the constructor lambdas. + */ class WebRtcCallSession( - private val context: Context, + private val peerConnectionFactory: PeerConnectionFactory, private val iceServers: List, private val onIceCandidate: (IceCandidate) -> Unit, private val onPeerConnected: () -> Unit, @@ -54,31 +54,7 @@ class WebRtcCallSession( private val onError: (String) -> Unit = {}, private val onRenegotiationNeeded: () -> Unit = {}, ) { - private var peerConnectionFactory: PeerConnectionFactory? = null private var peerConnection: PeerConnection? = null - private var localAudioTrack: AudioTrack? = null - private var localVideoTrack: VideoTrack? = null - private var audioSource: AudioSource? = null - private var videoSource: VideoSource? = null - private var cameraCapturer: CameraVideoCapturer? = null - - val eglBase: EglBase = EglBase.create() - - fun initialize() { - PeerConnectionFactory.initialize( - PeerConnectionFactory - .InitializationOptions - .builder(context) - .createInitializationOptions(), - ) - - peerConnectionFactory = - PeerConnectionFactory - .builder() - .setVideoDecoderFactory(DefaultVideoDecoderFactory(eglBase.eglBaseContext)) - .setVideoEncoderFactory(DefaultVideoEncoderFactory(eglBase.eglBaseContext, true, true)) - .createPeerConnectionFactory() - } fun createPeerConnection() { val rtcConfig = @@ -88,7 +64,7 @@ class WebRtcCallSession( } peerConnection = - peerConnectionFactory?.createPeerConnection( + peerConnectionFactory.createPeerConnection( rtcConfig, object : PeerConnection.Observer { override fun onIceCandidate(candidate: IceCandidate?) { @@ -128,9 +104,6 @@ class WebRtcCallSession( } PeerConnection.IceConnectionState.DISCONNECTED -> { - // DISCONNECTED is often transient (network switch, brief - // packet loss). WebRTC will transition to FAILED if - // recovery is impossible, so we only act on FAILED above. Log.d(TAG) { "ICE: DISCONNECTED (transient, waiting for recovery or FAILED)" } } @@ -151,7 +124,6 @@ class WebRtcCallSession( } override fun onAddStream(stream: MediaStream?) { - // Fallback for Plan B SDP — Unified Plan uses onAddTrack stream?.videoTracks?.firstOrNull()?.let { onRemoteVideoTrack(it) } } @@ -168,7 +140,6 @@ class WebRtcCallSession( receiver: RtpReceiver?, streams: Array?, ) { - // Unified Plan: extract video track from receiver val track = receiver?.track() if (track is VideoTrack) { onRemoteVideoTrack(track) @@ -178,60 +149,26 @@ class WebRtcCallSession( ) } - fun addAudioTrack() { - val constraints = MediaConstraints() - audioSource = peerConnectionFactory?.createAudioSource(constraints) - localAudioTrack = - peerConnectionFactory?.createAudioTrack("audio0", audioSource).also { - peerConnection?.addTrack(it) + /** + * Adds an existing track to this PeerConnection. + * + * @param maxBitrateBps optional per-sender bitrate cap (e.g. 1_500_000 for 720p video) + */ + fun addTrack( + track: MediaStreamTrack, + maxBitrateBps: Int? = null, + ): RtpSender? { + val sender = peerConnection?.addTrack(track) ?: return null + if (maxBitrateBps != null) { + val params = sender.parameters + params.encodings.forEach { encoding -> + encoding.maxBitrateBps = maxBitrateBps } + sender.parameters = params + } + return sender } - fun addVideoTrack() { - videoSource = peerConnectionFactory?.createVideoSource(false) - localVideoTrack = - peerConnectionFactory?.createVideoTrack("video0", videoSource).also { - val sender = peerConnection?.addTrack(it) - // Set max bitrate to 1.5 Mbps for good 720p quality - sender?.let { s -> - val params = s.parameters - params.encodings.forEach { encoding -> - encoding.maxBitrateBps = 1_500_000 - } - s.parameters = params - } - } - startCamera() - } - - fun startCamera() { - if (cameraCapturer != null) return // Already running - val source = videoSource ?: return - val enumerator = Camera2Enumerator(context) - val frontCamera = enumerator.deviceNames.firstOrNull { enumerator.isFrontFacing(it) } - val camera = frontCamera ?: enumerator.deviceNames.firstOrNull() ?: return - - cameraCapturer = - enumerator.createCapturer(camera, null)?.also { - it.initialize( - SurfaceTextureHelper.create("CaptureThread", eglBase.eglBaseContext), - context, - source.capturerObserver, - ) - it.startCapture(1280, 720, 30) - } - } - - fun stopCamera() { - cameraCapturer?.stopCapture() - cameraCapturer?.dispose() - cameraCapturer = null - } - - fun getLocalVideoSource(): VideoSource? = videoSource - - fun getLocalVideoTrack(): VideoTrack? = localVideoTrack - fun createOffer(onSdpCreated: (SessionDescription) -> Unit) { val constraints = MediaConstraints().apply { @@ -320,31 +257,10 @@ class WebRtcCallSession( fun getSignalingState(): PeerConnection.SignalingState? = peerConnection?.signalingState() - fun setAudioEnabled(enabled: Boolean) { - localAudioTrack?.setEnabled(enabled) - } - - fun setVideoEnabled(enabled: Boolean) { - localVideoTrack?.setEnabled(enabled) - } - fun dispose() { - stopCamera() - localAudioTrack?.dispose() - localVideoTrack?.dispose() - audioSource?.dispose() - videoSource?.dispose() peerConnection?.close() peerConnection?.dispose() - peerConnectionFactory?.dispose() - eglBase.release() - - localAudioTrack = null - localVideoTrack = null - audioSource = null - videoSource = null peerConnection = null - peerConnectionFactory = null } private fun loggingSdpObserver(label: String) = 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 9c57aae63e..0227d79eee 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 @@ -221,8 +221,10 @@ class AccountViewModel( ) // Set callbacks before exposing controller to avoid timing races - callManager.onAnswerReceived = { event -> controller.onCallAnswerReceived(event.sdpAnswer()) } + callManager.onAnswerReceived = { event -> controller.onCallAnswerReceived(event.pubKey, event.sdpAnswer()) } callManager.onIceCandidateReceived = { event -> controller.onIceCandidateReceived(event) } + callManager.onNewPeerInGroupCall = { peerPubKey -> controller.onNewPeerInGroupCall(peerPubKey) } + callManager.onMidCallOfferReceived = { peerPubKey, sdpOffer -> controller.onMidCallOfferReceived(peerPubKey, sdpOffer) } 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 07abddbae6..faac810c47 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 @@ -53,10 +53,17 @@ class CallManager( private val _state = MutableStateFlow(CallState.Idle) val state: StateFlow = _state.asStateFlow() + /** Called for every answer that should be applied to a PeerConnection (includes peer pubkey). */ var onAnswerReceived: ((CallAnswerEvent) -> Unit)? = null var onIceCandidateReceived: ((CallIceCandidateEvent) -> Unit)? = null var onRenegotiationOfferReceived: ((CallRenegotiateEvent) -> Unit)? = null + /** Called when a new peer joins the group call (callee-to-callee mesh setup). */ + var onNewPeerInGroupCall: ((peerPubKey: HexKey) -> Unit)? = null + + /** Called when a mid-call offer is received from another callee in a group call. */ + var onMidCallOfferReceived: ((peerPubKey: HexKey, sdpOffer: String) -> Unit)? = null + private var timeoutJob: Job? = null private var resetJob: Job? = null private val processedEventIds = mutableSetOf() @@ -69,7 +76,68 @@ class CallManager( private fun isEventTooOld(event: Event): Boolean = TimeUtils.now() - event.createdAt > MAX_EVENT_AGE_SECONDS - // ---- P2P call initiation ---- + // ---- Call initiation (state + publish) ---- + + /** + * Sets state to Offering. Called by CallController before creating + * per-peer offers in group calls. + */ + fun beginOffering( + callId: String, + calleePubKeys: Set, + callType: CallType, + ) { + _state.value = CallState.Offering(callId, calleePubKeys, callType) + startTimeout(callId) + } + + /** + * Publishes a per-peer call offer (group context: p-tags for all members). + * Called once per callee so each gets their own SDP. + */ + suspend fun publishOfferToPeer( + calleePubKey: HexKey, + allCalleePubKeys: Set, + callType: CallType, + callId: String, + sdpOffer: String, + ) { + Log.d("CallManager") { "publishOfferToPeer: to=${calleePubKey.take(8)}, sdpLength=${sdpOffer.length}" } + val result = factory.createCallOffer(sdpOffer, calleePubKey, allCalleePubKeys, callId, callType, signer) + publishEvent(result.wrap) + } + + /** + * Publishes a callee-to-callee offer within an existing call. + * Uses the current call context for call-id, call-type, and group members. + */ + suspend fun publishOfferToPeer( + peerPubKey: HexKey, + sdpOffer: String, + ) { + val callId = currentCallId() ?: return + val callType = currentCallType() ?: return + val allMembers = (currentPeerPubKeys() ?: emptySet()) + signer.pubKey + Log.d("CallManager") { "publishOfferToPeer (mid-call): to=${peerPubKey.take(8)}, sdpLength=${sdpOffer.length}" } + val result = factory.createCallOffer(sdpOffer, peerPubKey, allMembers, callId, callType, signer) + publishEvent(result.wrap) + } + + /** + * Publishes a callee-to-callee answer within an existing call. + */ + suspend fun publishAnswerToPeer( + peerPubKey: HexKey, + sdpAnswer: String, + ) { + val callId = currentCallId() ?: return + val allMembers = (currentPeerPubKeys() ?: emptySet()) + signer.pubKey + Log.d("CallManager") { "publishAnswerToPeer: to=${peerPubKey.take(8)}, sdpLength=${sdpAnswer.length}" } + val result = factory.createCallAnswer(sdpAnswer, peerPubKey, allMembers, callId, signer) + publishEvent(result.wrap) + } + + // ---- P2P call initiation (convenience) ---- suspend fun initiateCall( calleePubKey: HexKey, @@ -85,24 +153,6 @@ class CallManager( Log.d("CallManager") { "initiateCall: offer published, timeout started" } } - // ---- Group call initiation ---- - - /** - * Initiates a group call. A single [CallOfferEvent] is created with `p` - * tags for every callee and then gift-wrapped individually to each one. - */ - suspend fun initiateGroupCall( - calleePubKeys: Set, - callType: CallType, - callId: String, - sdpOffer: String, - ) { - val result = factory.createGroupCallOffer(sdpOffer, calleePubKeys, callId, callType, signer) - _state.value = CallState.Offering(callId, calleePubKeys, callType) - result.wraps.forEach { publishEvent(it) } - startTimeout(callId) - } - // ---- Incoming call handling ---- fun onIncomingCallEvent(event: CallOfferEvent) { @@ -117,9 +167,23 @@ class CallManager( return } - if (_state.value !is CallState.Idle) { - // Already in a call — send a "busy" reject so the caller gets - // immediate feedback instead of waiting for the 60s timeout. + val currentState = _state.value + + // Mid-call offer: same call-id, we're already in the call + if (currentState is CallState.Connecting || currentState is CallState.Connected) { + val currentCallId = + when (currentState) { + is CallState.Connecting -> currentState.callId + is CallState.Connected -> currentState.callId + } + if (callId == currentCallId) { + Log.d("CallManager") { "Mid-call offer from ${callerPubKey.take(8)} for current call — callee-to-callee" } + onMidCallOfferReceived?.invoke(callerPubKey, event.sdpOffer()) + return + } + } + + if (currentState !is CallState.Idle) { scope.launch { val result = factory.createReject(callerPubKey, callId, "busy", signer = signer) publishEvent(result.wrap) @@ -151,7 +215,6 @@ class CallManager( _state.value = CallState.Connecting(current.callId, current.peerPubKeys() - signer.pubKey, current.callType) cancelTimeout() - // Include all group members + self so other devices get notified too. 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) @@ -165,7 +228,6 @@ class CallManager( transitionToEnded(current.callId, current.peerPubKeys(), EndReason.REJECTED) - // Include all group members + self so other devices get notified too. val allRecipients = current.groupMembers + signer.pubKey val result = factory.createGroupReject(allRecipients, current.callId, signer = signer) result.wraps.forEach { publishEvent(it) } @@ -187,7 +249,6 @@ class CallManager( Log.d("CallManager") { "onCallAnswered: callId mismatch (got=$callId, expected=${current.callId})" } return } - // First answer: start the call immediately. Remaining peers stay pending. val pending = current.peerPubKeys - answeringPeer _state.value = CallState.Connecting( @@ -197,12 +258,11 @@ class CallManager( pendingPeerPubKeys = pending, ) cancelTimeout() - Log.d("CallManager") { "onCallAnswered: Offering -> Connecting, invoking onAnswerReceived callback (isNull=${onAnswerReceived == null})" } + Log.d("CallManager") { "onCallAnswered: Offering -> Connecting, forwarding answer to CallController" } onAnswerReceived?.invoke(event) } is CallState.Connecting -> { - // Another peer answered while we're still connecting with the first if (callId != current.callId) return if (answeringPeer in current.pendingPeerPubKeys) { _state.value = @@ -211,30 +271,36 @@ class CallManager( pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer, ) } - // TODO: establish additional WebRTC peer connection for this peer + // Forward to CallController — it routes to the correct PeerSession + Log.d("CallManager") { "onCallAnswered: additional peer ${answeringPeer.take(8)} in Connecting, forwarding" } + 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) { - // Another device of this user answered the call — stop ringing. transitionToEnded(current.callId, current.peerPubKeys(), EndReason.ANSWERED_ELSEWHERE) } - // Otherwise another group member answered — we keep ringing. + // Another group member answered — notify for future mesh connections + // (we'll connect to them after we accept the call ourselves) } is CallState.Connected -> { if (callId != current.callId) return if (answeringPeer in current.pendingPeerPubKeys) { - // A pending peer just joined the group call _state.value = current.copy( peerPubKeys = current.peerPubKeys + answeringPeer, pendingPeerPubKeys = current.pendingPeerPubKeys - answeringPeer, ) - // TODO: establish additional WebRTC peer connection for this peer + // Forward to CallController for routing + onAnswerReceived?.invoke(event) + onNewPeerInGroupCall?.invoke(answeringPeer) } else { - // Renegotiation answer (e.g., peer accepted our video upgrade offer) + // Renegotiation answer onAnswerReceived?.invoke(event) } } @@ -262,14 +328,12 @@ class CallManager( } is CallState.Connecting -> { - // A pending peer rejected while we're already connecting with another if (callId != current.callId) return _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer) } is CallState.Connected -> { - // A pending peer rejected while we're already in the call if (callId != current.callId) return _state.value = current.copy(pendingPeerPubKeys = current.pendingPeerPubKeys - rejectingPeer) @@ -278,10 +342,8 @@ class CallManager( is CallState.IncomingCall -> { if (callId != current.callId) return if (rejectingPeer == signer.pubKey) { - // Another device of this user rejected the call — stop ringing. transitionToEnded(current.callId, current.peerPubKeys(), EndReason.REJECTED) } - // Otherwise another group member rejected — we keep ringing. } else -> { @@ -307,11 +369,6 @@ class CallManager( onRenegotiationOfferReceived?.invoke(event) } - /** - * Sends a renegotiation offer to a specific peer. SDP is per-PeerConnection - * so this is always addressed to a single peer. In group calls the inner - * event includes `p` tags for all members for group context. - */ suspend fun sendRenegotiation( sdpOffer: String, peerPubKey: HexKey, @@ -322,11 +379,6 @@ class CallManager( publishEvent(result.wrap) } - /** - * Sends a renegotiation answer to a specific peer. SDP is per-PeerConnection - * so this is always addressed to a single peer. The inner event includes - * `p` tags for all members for group context. - */ suspend fun sendRenegotiationAnswer( sdpAnswer: String, peerPubKey: HexKey, @@ -355,11 +407,6 @@ class CallManager( ) } - /** - * Invites a new peer into the current call by sending them an offer. - * The inner event includes `p` tags for all existing group members plus - * the new invitee so they can see the full group composition. - */ suspend fun invitePeer( peerPubKey: HexKey, sdpOffer: String, @@ -388,7 +435,6 @@ class CallManager( } } - // All group members: existing peers + the new invitee + ourselves val allMembers = existingMembers + peerPubKey + signer.pubKey val result = factory.createCallOffer(sdpOffer, peerPubKey, allMembers, callId, callType, signer) publishEvent(result.wrap) @@ -517,10 +563,8 @@ class CallManager( else -> null } - /** Returns the first peer pubkey (for P2P calls) or null. */ fun currentPeerPubKey(): HexKey? = currentPeerPubKeys()?.firstOrNull() - /** Returns all peer pubkeys for the current call (connected + pending). */ fun currentPeerPubKeys(): Set? = when (val s = _state.value) { is CallState.Offering -> s.peerPubKeys @@ -530,7 +574,15 @@ class CallManager( else -> null } - /** True when the current call has more than one peer. */ + fun currentCallType(): CallType? = + when (val s = _state.value) { + is CallState.Offering -> s.callType + is CallState.IncomingCall -> s.callType + is CallState.Connecting -> s.callType + is CallState.Connected -> s.callType + else -> null + } + fun isGroupCall(): Boolean = (currentPeerPubKeys()?.size ?: 0) > 1 fun reset() { @@ -588,13 +640,4 @@ class CallManager( } } -/** - * Convenience extension: the peers in an incoming call are all group members - * except the local signer (i.e. ourselves) – but since we don't store the - * local pubkey here we return all members except the caller's own pubkey - * is already the callerPubKey field. In practice the set of "peer" keys - * the UI should track is groupMembers minus self, which the controller - * resolves. Here we simply exclude the caller from the recipients set - * and add back the caller, resulting in the full group minus self. - */ private fun CallState.IncomingCall.peerPubKeys(): Set = groupMembers