import { create } from 'zustand'; import { MediaStream } from 'react-native-webrtc'; import { wsService, WSCallIncomingMessage, WSCallSDPMessage, WSCallICEMessage, WSErrorMessage, } from '../services/wsService'; import { webrtcManager, ICEServer } from '../services/webrtc'; import { useAuthStore } from './authStore'; import { useUserStore } from './userStore'; import { userManager } from './userManager'; export type CallStatus = 'idle' | 'ringing' | 'connecting' | 'connected' | 'ending'; export interface CallSession { id: string; conversationId: string; peerId: string; peerName?: string; peerAvatar?: string | null; status: CallStatus; startedAt?: number; duration: number; isMuted: boolean; isSpeakerOn: 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; isMinimized: boolean; initCall: () => () => void; startCall: ( conversationId: string, calleeId: string, calleeInfo?: { nickname?: string; username?: string; avatar?: string | null } ) => Promise; acceptCall: () => Promise; rejectCall: () => void; endCall: (reason?: string) => Promise; toggleMute: () => void; toggleSpeaker: () => void; toggleMinimize: () => void; } 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; // === Element + Telegram 结合: 常量 === const CALL_LIFETIME_MS = 55000; // 55秒 (略小于后端60秒) const CALL_TIMEOUT_MS = 115000; // 115秒 (拨打方等待超时,略小于后端120秒) const IGNORE_CALL_ID_TTL = 30000; // 已处理callId保留30秒 // === Element: 已处理的 callId 集合 (防重复) === const processedCallIds = new Map(); // callId -> timestamp // 清理过期的 callId function cleanupProcessedCallIds() { const now = Date.now(); for (const [callId, timestamp] of processedCallIds) { if (now - timestamp > IGNORE_CALL_ID_TTL) { processedCallIds.delete(callId); } } } export const callStore = create((set, get) => ({ currentCall: null, incomingCall: null, callDuration: 0, peerStream: 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) => { // === Element: 清理过期的 callId === cleanupProcessedCallIds(); // === Element: 检查是否已处理过该 callId === 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; } // === Element: 检查 lifetime 过期 === const callAge = Date.now() - msg.created_at; const lifetime = msg.lifetime || 60000; // 默认60秒 if (callAge > lifetime - 5000) { // 留5秒余量 console.log('[CallStore] Ignoring stale incoming call, age:', callAge, 'ms, lifetime:', lifetime); wsService.sendCallReject(msg.call_id); processedCallIds.set(msg.call_id, Date.now()); return; } // Try to get caller info from cache first, then fetch from API 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()); // 设置超时 (使用 lifetime) if (callTimeoutTimer) clearTimeout(callTimeoutTimer); const remainingTime = Math.max(lifetime - callAge - 1000, 5000); // 剩余时间,最少5秒 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; try { await webrtcManager.initialize(msg.ice_servers || []); await webrtcManager.createLocalStream(true); // Listen for WebRTC events const unsubRTC = webrtcManager.onEvent((event) => { if (event.type === 'icecandidate' && event.candidate) { wsService.sendCallICE(currentCall.id, JSON.stringify(event.candidate.toJSON())); } if (event.type === 'remotestream') { set({ peerStream: event.stream }); } if (event.type === 'connectionstatechange') { if (event.state === 'connected') { const now = Date.now(); set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, status: 'connected', startedAt: now } : null, })); if (durationTimer) clearInterval(durationTimer); durationTimer = setInterval(() => { set((s) => ({ callDuration: s.callDuration + 1 })); }, 1000); } if (event.state === 'disconnected' || event.state === 'failed') { get().endCall('connection_lost'); } } }); void unsubRTC; // startCall creates PeerConnection, adds tracks, and creates offer for initiator const offer = await webrtcManager.startCall(true); 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 this is an active call, end it if (currentCall?.id === msg.call_id) { console.log('[CallStore] Call ended, duration:', msg.duration); get().endCall('ended'); return; } // If this is an incoming call that was cancelled by caller, clear it if (incomingCall?.callId === msg.call_id) { console.log('[CallStore] Incoming call cancelled by caller'); if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } set({ incomingCall: null }); } }) ); // === Telegram: 其他设备已接听 === 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 }); } }) ); // === Telegram: 处理服务端错误 === unsubs.push( wsService.on('error', (msg: WSErrorMessage) => { const { currentCall, incomingCall } = get(); console.log('[CallStore] Server error:', msg.code, msg.message); // 处理 callee_offline 错误 if (msg.code === 'callee_offline') { if (currentCall && currentCall.status === 'ringing') { console.log('[CallStore] Callee is offline'); get().endCall('callee_offline'); } } // 处理 call_already_answered 错误 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; // Check that we are not the sender of this SDP message const myUserId = useAuthStore.getState().currentUser?.id; if (myUserId && msg.from_id === myUserId) return; try { const pc = webrtcManager.getPeerConnection(); if (msg.payload.sdp_type === 'offer') { if (!pc) { // Cache offer for later processing when PeerConnection is ready pendingOffer = { callId: msg.call_id, sdp: msg.payload.sdp }; console.log('[CallStore] Caching pending offer, PeerConnection not ready yet'); return; } // We are the callee - only process if in 'stable' state const signalingState = pc.signalingState; if (signalingState !== 'stable') { console.warn('[CallStore] Ignoring offer, signaling state is', signalingState); return; } pendingOffer = null; await webrtcManager.setRemoteDescription({ type: msg.payload.sdp_type, sdp: msg.payload.sdp, }); const answer = await webrtcManager.createAnswer(); wsService.sendCallSDP(msg.call_id, 'answer', answer.sdp || ''); } else if (msg.payload.sdp_type === 'answer') { if (!pc) return; // We are the initiator - only process if in 'have-local-offer' state const signalingState = pc.signalingState; if (signalingState !== 'have-local-offer') { console.warn('[CallStore] Ignoring answer, signaling state is', signalingState); return; } await webrtcManager.setRemoteDescription({ type: msg.payload.sdp_type, sdp: msg.payload.sdp, }); } } catch (err) { console.error('[CallStore] call_sdp error:', err); } }) ); unsubs.push( wsService.on('call_ice', (msg: WSCallICEMessage) => { const { currentCall } = get(); if (!currentCall || currentCall.id !== msg.call_id) return; // Ignore ICE candidates from ourselves const myUserId = useAuthStore.getState().currentUser?.id; if (myUserId && 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 = () => { if (callTimeoutTimer) clearTimeout(callTimeoutTimer); if (durationTimer) clearInterval(durationTimer); unsubs.forEach((unsub) => unsub()); initCallUnsub = null; }; initCallUnsub = cleanup; return cleanup; }, startCall: async ( conversationId: string, calleeId: string, calleeInfo?: { nickname?: string; username?: string; avatar?: string | null } ) => { const { currentCall } = get(); if (currentCall && currentCall.status !== 'idle') { console.warn('[CallStore] Already in a call'); return; } const myUserId = useAuthStore.getState().currentUser?.id; if (!myUserId) { console.error('[CallStore] Not logged in'); return; } // Use provided callee info first, then fall back to userCache const cachedCallee = useUserStore.getState().userCache[calleeId]; const callee = calleeInfo || cachedCallee; const calleeName = callee?.nickname || callee?.username || calleeId; set({ currentCall: { id: '', // Will be filled when call_invited response comes conversationId, peerId: calleeId, peerName: calleeName, peerAvatar: callee?.avatar, status: 'ringing', duration: 0, isMuted: false, isSpeakerOn: false, isInitiator: true, }, }); wsService.sendCallInvite(conversationId, calleeId); // Listen for call_invited to get the call_id if (unsubInvited) { unsubInvited(); } unsubInvited = wsService.on('call_invited', (msg) => { set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, id: msg.call_id } : null, })); }); // Timeout - 使用 CALL_TIMEOUT_MS (115秒,略小于后端 120秒) if (callTimeoutTimer) clearTimeout(callTimeoutTimer); callTimeoutTimer = setTimeout(() => { const { currentCall: cc } = get(); if (cc && cc.status === 'ringing') { console.warn('[CallStore] Call timeout'); get().endCall('timeout'); } }, CALL_TIMEOUT_MS); }, acceptCall: async () => { const { incomingCall } = get(); if (!incomingCall) return; if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } set({ currentCall: { id: incomingCall.callId, conversationId: incomingCall.conversationId, peerId: incomingCall.callerId, peerName: incomingCall.callerName, peerAvatar: incomingCall.callerAvatar, status: 'connecting', duration: 0, isMuted: false, isSpeakerOn: false, isInitiator: false, }, incomingCall: null, }); wsService.sendCallAnswer(incomingCall.callId); try { await webrtcManager.initialize(incomingCall.iceServers); await webrtcManager.createLocalStream(true); // Listen for WebRTC events const unsubRTC = webrtcManager.onEvent((event) => { if (event.type === 'icecandidate' && event.candidate) { wsService.sendCallICE(incomingCall.callId, JSON.stringify(event.candidate.toJSON())); } if (event.type === 'remotestream') { set({ peerStream: event.stream }); } if (event.type === 'connectionstatechange') { if (event.state === 'connected') { const now = Date.now(); set((s) => ({ currentCall: s.currentCall ? { ...s.currentCall, status: 'connected', startedAt: now } : null, })); if (durationTimer) clearInterval(durationTimer); durationTimer = setInterval(() => { set((s) => ({ callDuration: s.callDuration + 1 })); }, 1000); } if (event.state === 'disconnected' || event.state === 'failed') { get().endCall('connection_lost'); } } }); void unsubRTC; // startCall creates PeerConnection and adds tracks (non-initiator, no offer) await webrtcManager.startCall(false); // Process any pending offer that arrived before PeerConnection was ready if (pendingOffer && pendingOffer.callId === incomingCall.callId) { const offerMsg = pendingOffer; pendingOffer = null; console.log('[CallStore] Processing pending offer after PeerConnection ready'); try { await webrtcManager.setRemoteDescription({ type: 'offer', sdp: offerMsg.sdp, }); const answer = await webrtcManager.createAnswer(); wsService.sendCallSDP(offerMsg.callId, 'answer', answer.sdp || ''); } catch (err) { console.error('[CallStore] Failed to process pending offer:', err); } } } 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; if (callTimeoutTimer) { clearTimeout(callTimeoutTimer); callTimeoutTimer = null; } if (durationTimer) { clearInterval(durationTimer); durationTimer = null; } if (unsubInvited) { unsubInvited(); unsubInvited = null; } pendingOffer = null; const callId = currentCall.id; set({ currentCall: null, callDuration: 0, peerStream: 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 })); }, }));