From fcecb5241c86e451431b54800bab29cf5aaa62ee Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Apr 2026 16:54:00 +0000 Subject: [PATCH] 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 9b2ebbff7..638ef64f9 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 c74f41f6e..44745c0ee 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 9c57aae63..0227d79ee 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 07abddbae..faac810c4 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