import { create } from 'zustand'; import { MediaStream } from 'react-native-webrtc'; import { wsService, WSCallIncomingMessage, WSCallSDPMessage, WSCallICEMessage, WSErrorMessage, } from '@/services/core'; import { webrtcManager, ICEServer } from '@/services/webrtc'; import { getCurrentUserId } from '../auth/sessionStore'; import { useUserStore } from '../userStore'; import { userManager } from '../user'; export type CallStatus = | 'idle' // 空闲状态 | 'calling' // 正在呼出(已发送邀请,等待对方响应) | 'ringing' // 来电响铃中 | 'connecting' // 正在建立连接(WebRTC 协商中) | 'connected' // 已接通 | 'reconnecting' // 网络断开,正在重连 | 'ended' // 已结束 | 'failed'; // 连接失败 export type CallType = 'voice' | 'video'; export interface CallSession { id: string; conversationId: string; peerId: string; peerName?: string; peerAvatar?: string | null; status: CallStatus; callType: CallType; startedAt?: number; duration: number; isMuted: boolean; isSpeakerOn: boolean; isVideoEnabled: boolean; isPeerVideoEnabled: boolean; isInitiator: boolean; } export interface IncomingCallInfo { callId: string; conversationId: string; callerId: string; callerName?: string; callerAvatar?: string | null; callType: string; iceServers: ICEServer[]; receivedAt: number; lifetime?: number; } interface CallState { currentCall: CallSession | null; incomingCall: IncomingCallInfo | null; callDuration: number; peerStream: MediaStream | null; localStream: MediaStream | null; isMinimized: boolean; initCall: () => () => void; startCall: ( conversationId: string, calleeId: string, calleeInfo?: { nickname?: string; username?: string; avatar?: string | null }, callType?: CallType ) => Promise; acceptCall: () => Promise; rejectCall: () => void; endCall: (reason?: string) => Promise; toggleMute: () => void; toggleSpeaker: () => void; toggleMinimize: () => void; toggleVideo: () => Promise; setVideoEnabled: (enabled: boolean) => Promise; reset: () => void; } // Module-level variables for timers let durationTimer: ReturnType | null = null; let callTimeoutTimer: ReturnType | null = null; let initCallUnsub: (() => void) | null = null; let unsubInvited: (() => void) | null = null; let pendingOffer: { callId: string; sdp: string } | null = null; let rtcUnsubscribe: (() => void) | null = null; // WebRTC event subscription // Constants const CALL_LIFETIME_MS = 55000; const CALL_TIMEOUT_MS = 115000; const IGNORE_CALL_ID_TTL = 30000; // Track processed call IDs to prevent duplicates const processedCallIds = new Map(); function cleanupProcessedCallIds() { const now = Date.now(); for (const [callId, timestamp] of processedCallIds) { if (now - timestamp > IGNORE_CALL_ID_TTL) { processedCallIds.delete(callId); } } } /** * Unified WebRTC event handler * This is called from both initiator and receiver paths */ function setupWebRTCEvents(callId: string, myUserId: string): void { // Clean up any existing subscription if (rtcUnsubscribe) { rtcUnsubscribe(); rtcUnsubscribe = null; } rtcUnsubscribe = webrtcManager.onEvent((event) => { switch (event.type) { case 'icecandidate': if (event.candidate) { wsService.sendCallICE(callId, JSON.stringify(event.candidate.toJSON())); } break; case 'remotestream': handleRemoteStream(event.stream); break; case 'negotiationneeded': handleNegotiationNeeded(callId, event.offer); break; case 'connectionstatechange': handleConnectionStateChange(event.state); break; case 'error': console.error('[CallStore] WebRTC error:', event.error); break; } }); } /** * Handle remote stream with video track detection */ function handleRemoteStream(stream: MediaStream): void { // Create a new MediaStream instance to force zustand subscribers to update // because react-native-webrtc may add tracks to the same stream object reference const newStream = new MediaStream(); stream.getTracks().forEach((track) => { newStream.addTrack(track); }); callStore.setState({ peerStream: newStream }); // Detect video tracks in remote stream const videoTracks = newStream.getVideoTracks(); const hasPeerVideo = videoTracks.length > 0 && videoTracks.some((t) => t.enabled); console.log('[CallStore] Remote stream received, hasVideo:', hasPeerVideo, 'videoTracks:', videoTracks.length, 'audioTracks:', newStream.getAudioTracks().length); callStore.setState((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, isPeerVideoEnabled: hasPeerVideo } : null, })); } /** * Handle negotiation needed event - send offer to peer */ function handleNegotiationNeeded(callId: string, offer: RTCSessionDescriptionInit): void { console.log('[CallStore] Negotiation needed, sending offer'); wsService.sendCallSDP(callId, 'offer', offer.sdp || ''); } /** * Handle connection state change with enhanced state machine */ function handleConnectionStateChange(state: string): void { console.log('[CallStore] Connection state changed:', state); const { currentCall } = callStore.getState(); if (!currentCall) return; switch (state) { case 'connected': // 连接成功,开始计时 const now = Date.now(); callStore.setState((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, status: 'connected', startedAt: now } : null, })); if (durationTimer) clearInterval(durationTimer); durationTimer = setInterval(() => { callStore.setState((s) => ({ callDuration: s.callDuration + 1 })); }, 1000); break; case 'disconnected': // 临时断开,进入重连状态 if (currentCall.status === 'connected') { callStore.setState((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, status: 'reconnecting' } : null, })); } break; case 'failed': // 连接失败 if (currentCall.status === 'connected' || currentCall.status === 'reconnecting') { callStore.getState().endCall('connection_failed'); } else if (currentCall.status === 'connecting') { // 初始连接失败 callStore.getState().endCall('connection_failed'); } break; case 'closed': // 连接关闭 if (currentCall.status !== 'ended' && currentCall.status !== 'failed') { callStore.getState().endCall('connection_closed'); } break; } } /** * Handle incoming SDP offer with Glare handling */ async function handleIncomingOffer(msg: WSCallSDPMessage, myUserId: string): Promise { let pc = webrtcManager.getPeerConnection(); if (!pc) { // Cache offer for later processing pendingOffer = { callId: msg.call_id, sdp: msg.payload.sdp }; console.log('[CallStore] Caching pending offer, PeerConnection not ready yet'); return; } let signalingState = pc.signalingState; console.log('[CallStore] Received offer, signalingState:', signalingState); // Glare handling: if we're not in stable state, we have a conflict if (signalingState !== 'stable') { const remoteUserId = msg.from_id; // Compare user IDs to determine who wins // Higher user ID wins the negotiation if (myUserId > remoteUserId) { console.log('[CallStore] Glare: I win (my ID > remote ID), ignoring incoming offer'); return; } console.log('[CallStore] Glare: Remote wins (remote ID > my ID), rolling back'); // Rollback to stable state try { await webrtcManager.rollback(); } catch (err) { console.error('[CallStore] Rollback failed:', err); // If rollback fails because we're already stable, that's fine - just proceed if (pc.signalingState !== 'stable') { return; } } // Re-check state after rollback signalingState = pc.signalingState; if (signalingState !== 'stable') { console.warn('[CallStore] Not stable after rollback, state:', signalingState); return; } } pendingOffer = null; // Set remote description and create answer try { await webrtcManager.setRemoteDescription({ type: msg.payload.sdp_type as 'offer' | 'answer', sdp: msg.payload.sdp, }); const answer = await webrtcManager.createAnswer(); wsService.sendCallSDP(msg.call_id, 'answer', answer.sdp || ''); } catch (err) { console.error('[CallStore] Failed to handle incoming offer:', err); } } /** * Handle incoming SDP answer */ async function handleIncomingAnswer(msg: WSCallSDPMessage): Promise { const pc = webrtcManager.getPeerConnection(); if (!pc) return; const signalingState = pc.signalingState; console.log('[CallStore] Received answer, signalingState:', signalingState); if (signalingState !== 'have-local-offer') { console.warn('[CallStore] Ignoring answer, signaling state is', signalingState); return; } try { await webrtcManager.setRemoteDescription({ type: msg.payload.sdp_type as 'offer' | 'answer', sdp: msg.payload.sdp, }); } catch (err) { console.error('[CallStore] Failed to set remote description from answer:', err); } } /** * Clean up all resources */ function cleanupResources(): void { if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } if (durationTimer) { clearInterval(durationTimer); durationTimer = null; } if (rtcUnsubscribe) { rtcUnsubscribe(); rtcUnsubscribe = null; } pendingOffer = null; } export const callStore = create((set, get) => ({ currentCall: null, incomingCall: null, callDuration: 0, peerStream: null, localStream: null, isMinimized: false, initCall: () => { // Prevent duplicate handler registration if (initCallUnsub) { initCallUnsub(); initCallUnsub = null; } const unsubs: Array<() => void> = []; unsubs.push( wsService.on('call_incoming', async (msg) => { cleanupProcessedCallIds(); if (processedCallIds.has(msg.call_id)) { console.log('[CallStore] Ignoring already processed call:', msg.call_id); return; } const { currentCall, incomingCall } = get(); if (incomingCall) { wsService.sendCallBusy(msg.call_id); processedCallIds.set(msg.call_id, Date.now()); return; } if (currentCall && currentCall.status !== 'idle') { wsService.sendCallBusy(msg.call_id); processedCallIds.set(msg.call_id, Date.now()); return; } // Check lifetime expiry const callAge = Date.now() - msg.created_at; const lifetime = msg.lifetime || 60000; if (callAge > lifetime - 5000) { console.log('[CallStore] Ignoring stale incoming call, age:', callAge); wsService.sendCallReject(msg.call_id); processedCallIds.set(msg.call_id, Date.now()); return; } // Fetch caller info let caller: { nickname?: string; username?: string; avatar?: string | null } | null = useUserStore.getState().userCache[msg.caller_id]; if (!caller) { try { const fetchedCaller = await userManager.getUserById(msg.caller_id); if (fetchedCaller) { caller = { nickname: fetchedCaller.nickname, username: fetchedCaller.username, avatar: fetchedCaller.avatar, }; } } catch (err) { console.log('[CallStore] Failed to fetch caller info:', err); } } const callerName = caller?.nickname || caller?.username || msg.caller_id; set({ incomingCall: { callId: msg.call_id, conversationId: msg.conversation_id, callerId: msg.caller_id, callerName, callerAvatar: caller?.avatar, callType: msg.call_type, iceServers: msg.ice_servers || [], receivedAt: Date.now(), lifetime: msg.lifetime, }, }); processedCallIds.set(msg.call_id, Date.now()); // Set timeout based on lifetime if (callTimeoutTimer) clearTimeout(callTimeoutTimer); const remainingTime = Math.max(lifetime - callAge - 1000, 5000); callTimeoutTimer = setTimeout(() => { const { incomingCall: ic } = get(); if (ic?.callId === msg.call_id) { console.log('[CallStore] Incoming call timeout'); wsService.sendCallReject(msg.call_id); processedCallIds.set(msg.call_id, Date.now()); set({ incomingCall: null }); } }, remainingTime); }) ); unsubs.push( wsService.on('call_accepted', async (msg) => { const { currentCall } = get(); if (!currentCall || currentCall.id !== msg.call_id) return; if (!currentCall.isInitiator) { console.log('[CallStore] Ignoring call_accepted, we are not the initiator'); return; } if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } const myUserId = getCurrentUserId() || ''; try { const isVideoCall = currentCall.callType === 'video'; await webrtcManager.initialize(msg.ice_servers || []); const newStream = await webrtcManager.createLocalStream(!isVideoCall); set({ localStream: newStream }); // Setup unified WebRTC event handler setupWebRTCEvents(currentCall.id, myUserId); // Start call with transceiver-based approach const offer = await webrtcManager.startCall(true, currentCall.callType); if (offer) { wsService.sendCallSDP(currentCall.id, 'offer', offer.sdp || ''); } } catch (err) { console.error('[CallStore] call_accepted error:', err); get().endCall('connection_failed'); } }) ); unsubs.push( wsService.on('call_rejected', (msg) => { const { currentCall } = get(); if (currentCall?.id !== msg.call_id) return; console.log('[CallStore] Call rejected'); get().endCall('rejected'); }) ); unsubs.push( wsService.on('call_busy', (msg) => { const { currentCall } = get(); if (currentCall?.id !== msg.call_id) return; console.log('[CallStore] Call busy'); get().endCall('busy'); }) ); unsubs.push( wsService.on('call_ended', (msg) => { const { currentCall, incomingCall } = get(); processedCallIds.set(msg.call_id, Date.now()); if (currentCall?.id === msg.call_id) { console.log('[CallStore] Call ended, duration:', msg.duration); get().endCall('ended'); return; } if (incomingCall?.callId === msg.call_id) { console.log('[CallStore] Incoming call cancelled by caller'); if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } set({ incomingCall: null }); } }) ); unsubs.push( wsService.on('call_answered_elsewhere', (msg) => { const { incomingCall } = get(); if (incomingCall?.callId === msg.call_id) { console.log('[CallStore] Call answered on another device'); processedCallIds.set(msg.call_id, Date.now()); if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } set({ incomingCall: null }); } }) ); unsubs.push( wsService.on('error', (msg: WSErrorMessage) => { const { currentCall, incomingCall } = get(); console.log('[CallStore] Server error:', msg.code, msg.message); if (msg.code === 'callee_offline') { if (currentCall && currentCall.status === 'ringing') { console.log('[CallStore] Callee is offline'); get().endCall('callee_offline'); } } if (msg.code === 'call_already_answered') { if (incomingCall) { console.log('[CallStore] Call already answered on another device'); processedCallIds.set(incomingCall.callId, Date.now()); if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } set({ incomingCall: null }); } } }) ); unsubs.push( wsService.on('call_sdp', async (msg: WSCallSDPMessage) => { const { currentCall } = get(); if (!currentCall || currentCall.id !== msg.call_id) return; const myUserId = getCurrentUserId() || ''; if (msg.from_id === myUserId) return; if (msg.payload.sdp_type === 'offer') { await handleIncomingOffer(msg, myUserId); } else if (msg.payload.sdp_type === 'answer') { await handleIncomingAnswer(msg); } }) ); unsubs.push( wsService.on('call_ice', (msg: WSCallICEMessage) => { const { currentCall } = get(); if (!currentCall || currentCall.id !== msg.call_id) return; const myUserId = getCurrentUserId() || ''; if (msg.from_id === myUserId) return; try { const candidate = typeof msg.payload.candidate === 'string' ? JSON.parse(msg.payload.candidate) : msg.payload.candidate; webrtcManager.addIceCandidate(candidate); } catch (err) { console.error('[CallStore] call_ice error:', err); } }) ); unsubs.push( wsService.on('call_peer_muted', (msg) => { console.log('[CallStore] Peer muted:', msg.user_id, msg.muted); }) ); const cleanup = () => { cleanupResources(); if (unsubInvited) { unsubInvited(); unsubInvited = null; } unsubs.forEach((unsub) => unsub()); initCallUnsub = null; }; initCallUnsub = cleanup; return cleanup; }, startCall: async ( conversationId: string, calleeId: string, calleeInfo?: { nickname?: string; username?: string; avatar?: string | null }, callType: CallType = 'voice' ) => { const { currentCall } = get(); if (currentCall && currentCall.status !== 'idle') { console.warn('[CallStore] Already in a call'); return; } const myUserId = getCurrentUserId(); if (!myUserId) { console.error('[CallStore] Not logged in'); return; } const cachedCallee = useUserStore.getState().userCache[calleeId]; const callee = calleeInfo || cachedCallee; const calleeName = callee?.nickname || callee?.username || calleeId; set({ currentCall: { id: '', conversationId, peerId: calleeId, peerName: calleeName, peerAvatar: callee?.avatar, callType, status: 'calling', // 改为 'calling' 表示正在呼出 duration: 0, isMuted: false, isSpeakerOn: false, isVideoEnabled: callType === 'video', isPeerVideoEnabled: false, isInitiator: true, }, }); wsService.sendCallInvite(conversationId, calleeId, callType); if (unsubInvited) { unsubInvited(); } unsubInvited = wsService.on('call_invited', (msg) => { set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, id: msg.call_id } : null, })); }); if (callTimeoutTimer) clearTimeout(callTimeoutTimer); callTimeoutTimer = setTimeout(() => { const { currentCall: cc } = get(); // 只有在 'calling' 状态(呼出中)才超时 if (cc && cc.status === 'calling') { console.warn('[CallStore] Call timeout'); get().endCall('timeout'); } }, CALL_TIMEOUT_MS); }, acceptCall: async () => { const { incomingCall } = get(); if (!incomingCall) return; console.log('[CallStore] acceptCall, incomingCall.callType:', incomingCall.callType); if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } const isVideoCall = incomingCall.callType === 'video'; console.log('[CallStore] acceptCall, isVideoCall:', isVideoCall); const myUserId = getCurrentUserId() || ''; set({ currentCall: { id: incomingCall.callId, conversationId: incomingCall.conversationId, peerId: incomingCall.callerId, peerName: incomingCall.callerName, peerAvatar: incomingCall.callerAvatar, callType: incomingCall.callType as CallType, status: 'connecting', duration: 0, isMuted: false, isSpeakerOn: false, isVideoEnabled: isVideoCall, isPeerVideoEnabled: isVideoCall, isInitiator: false, }, incomingCall: null, }); wsService.sendCallAnswer(incomingCall.callId); try { await webrtcManager.initialize(incomingCall.iceServers); const newStream = await webrtcManager.createLocalStream(!isVideoCall); set({ localStream: newStream }); // Setup unified WebRTC event handler setupWebRTCEvents(incomingCall.callId, myUserId); // Start call (non-initiator, will wait for offer) await webrtcManager.startCall(false, incomingCall.callType as CallType); // Note: For non-initiator, we don't create an offer here. // We wait for the initiator's offer via handleIncomingOffer. } catch (err) { console.error('[CallStore] Failed to accept call:', err); get().endCall('connection_failed'); } }, rejectCall: () => { const { incomingCall } = get(); if (!incomingCall) return; if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } wsService.sendCallReject(incomingCall.callId); set({ incomingCall: null }); }, endCall: async (reason = 'ended') => { const { currentCall } = get(); if (!currentCall) return; cleanupResources(); if (unsubInvited) { unsubInvited(); unsubInvited = null; } const callId = currentCall.id; set({ currentCall: null, callDuration: 0, peerStream: null, localStream: null, }); webrtcManager.dispose(); if (callId && reason !== 'ended') { wsService.sendCallEnd(callId, reason); } }, toggleMute: () => { const { currentCall } = get(); if (!currentCall) return; const newMuted = !currentCall.isMuted; webrtcManager.setMuted(newMuted); if (currentCall.id) { wsService.sendCallMute(currentCall.id, newMuted); } set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, isMuted: newMuted } : null, })); }, toggleSpeaker: () => { const { currentCall } = get(); if (!currentCall) return; set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, isSpeakerOn: !s.currentCall.isSpeakerOn } : null, })); }, toggleMinimize: () => { set((s) => ({ isMinimized: !s.isMinimized })); }, toggleVideo: async () => { const { currentCall } = get(); if (!currentCall) return; const newVideoEnabled = !currentCall.isVideoEnabled; await get().setVideoEnabled(newVideoEnabled); }, setVideoEnabled: async (enabled: boolean) => { const { currentCall } = get(); if (!currentCall) return; try { if (enabled) { const newStream = await webrtcManager.enableVideo(); set({ localStream: newStream }); } else { const newStream = await webrtcManager.disableVideo(); set({ localStream: newStream }); } set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, isVideoEnabled: enabled, callType: enabled ? 'video' : 'voice' } : null, })); } catch (err) { console.error('[CallStore] Failed to toggle video:', err); } }, // 重置所有状态,用于登出时清理 reset: () => { // 清理所有定时器和资源 cleanupResources(); // 清理 initCall 的订阅 if (initCallUnsub) { initCallUnsub(); initCallUnsub = null; } // 清理 invited 订阅 if (unsubInvited) { unsubInvited(); unsubInvited = null; } // 清理 WebRTC 资源 webrtcManager.dispose(); // 清理已处理的通话ID缓存 processedCallIds.clear(); // 重置状态 set({ currentCall: null, incomingCall: null, callDuration: 0, peerStream: null, localStream: null, isMinimized: false, }); }, }));