package service import ( "context" "encoding/json" "fmt" "strconv" "sync" "time" "with_you/internal/config" apperrors "with_you/internal/errors" "with_you/internal/model" "with_you/internal/pkg/ws" "with_you/internal/repository" "go.uber.org/zap" "gorm.io/gorm" ) // 通话相关常量 const ( CallLifetimeMs = 60000 // 通话邀请有效期 60秒 (参考 Matrix) CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒 ) // ActiveCall 内存中的活跃通话 type ActiveCall struct { ID string ConversationID string CallerID string CalleeID string CallType model.CallType Status model.CallStatus MediaType string // voice 或 video CreatedAt time.Time StartedAt *time.Time Duration int64 // 通话时长(秒) // 参与者状态 Participants map[string]*ActiveParticipant // ICE Servers ICEServers []config.ICEServerConfig } // ActiveParticipant 内存中的活跃参与者 type ActiveParticipant struct { UserID string Status model.ParticipantStatus JoinedAt *time.Time } // CallService 通话服务接口 type CallService interface { Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*ActiveCall, bool, error) Accept(ctx context.Context, callID, userID string) (*ActiveCall, error) Reject(ctx context.Context, callID, userID string) error Busy(ctx context.Context, callID, userID string) error End(ctx context.Context, callID, userID string, reason string) (*ActiveCall, error) RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error SetMuted(ctx context.Context, callID, userID string, muted bool) error GetCallHistory(ctx context.Context, userID string, page, pageSize int) ([]model.CallSession, int64, error) GetICEServers() []config.ICEServerConfig StartCleanupTicker() } type callService struct { callRepo repository.CallRepository hub *ws.Hub config *config.Config db *gorm.DB // 内存中的活跃通话状态 activeCalls map[string]*ActiveCall // callID -> ActiveCall activeCallsByUser map[string]map[string]bool // userID -> set of callIDs activeCallsByConv map[string]string // conversationID -> callID mu sync.RWMutex } // NewCallService 创建通话服务 func NewCallService( callRepo repository.CallRepository, hub *ws.Hub, cfg *config.Config, db *gorm.DB, ) CallService { svc := &callService{ callRepo: callRepo, hub: hub, config: cfg, db: db, activeCalls: make(map[string]*ActiveCall), activeCallsByUser: make(map[string]map[string]bool), activeCallsByConv: make(map[string]string), } // 订阅 WebSocket 断开事件 hub.OnDisconnect(svc.handleUserDisconnect) // 自动启动超时清理 svc.StartCleanupTicker() return svc } // generateCallID 生成通话ID func (s *callService) generateCallID() string { id := time.Now().UnixNano() return strconv.FormatInt(id, 10) } // getActiveCall 从内存获取活跃通话 func (s *callService) getActiveCall(callID string) *ActiveCall { s.mu.RLock() defer s.mu.RUnlock() return s.activeCalls[callID] } // getActiveCallByConversation 从内存获取会话的活跃通话 func (s *callService) getActiveCallByConversation(conversationID string) *ActiveCall { s.mu.RLock() callID, exists := s.activeCallsByConv[conversationID] if !exists { s.mu.RUnlock() return nil } call := s.activeCalls[callID] s.mu.RUnlock() return call } // getActiveCallsByUser 获取用户的所有活跃通话 func (s *callService) getActiveCallsByUser(userID string) []*ActiveCall { s.mu.RLock() callIDs, exists := s.activeCallsByUser[userID] if !exists { s.mu.RUnlock() return nil } var calls []*ActiveCall for callID := range callIDs { if call, ok := s.activeCalls[callID]; ok { calls = append(calls, call) } } s.mu.RUnlock() return calls } // storeActiveCall 存储活跃通话到内存 func (s *callService) storeActiveCall(call *ActiveCall) { s.mu.Lock() defer s.mu.Unlock() s.activeCalls[call.ID] = call // 按会话索引 s.activeCallsByConv[call.ConversationID] = call.ID // 按用户索引 for userID := range call.Participants { if s.activeCallsByUser[userID] == nil { s.activeCallsByUser[userID] = make(map[string]bool) } s.activeCallsByUser[userID][call.ID] = true } } // removeActiveCall 从内存移除活跃通话 func (s *callService) removeActiveCall(callID string) *ActiveCall { s.mu.Lock() defer s.mu.Unlock() call, exists := s.activeCalls[callID] if !exists { return nil } delete(s.activeCalls, callID) delete(s.activeCallsByConv, call.ConversationID) for userID := range call.Participants { if userCalls, ok := s.activeCallsByUser[userID]; ok { delete(userCalls, callID) if len(userCalls) == 0 { delete(s.activeCallsByUser, userID) } } } return call } // isParticipant 检查用户是否是通话参与者 func isParticipant(call *ActiveCall, userID string) bool { _, ok := call.Participants[userID] return ok } func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*ActiveCall, bool, error) { // 检查会话是否有活跃通话 if active := s.getActiveCallByConversation(conversationID); active != nil { return nil, false, apperrors.Wrap(fmt.Errorf("call %s", active.ID), apperrors.ErrCallInProgress) } now := time.Now() callID := s.generateCallID() call := &ActiveCall{ ID: callID, ConversationID: conversationID, CallerID: callerID, CalleeID: calleeID, CallType: model.CallTypePrivate, Status: model.CallStatusCalling, MediaType: mediaType, CreatedAt: now, Participants: map[string]*ActiveParticipant{ callerID: {UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now}, calleeID: {UserID: calleeID, Status: model.ParticipantStatusInvited}, }, ICEServers: s.config.WebRTC.ICEServers, } // 存储到内存 s.storeActiveCall(call) // 检查被叫方是否在线 calleeOnline := s.hub.HasClients(calleeID) // 发送来电通知 payload := map[string]any{ "call_id": call.ID, "conversation_id": conversationID, "caller_id": callerID, "call_type": call.CallType, "media_type": mediaType, "created_at": now.UnixMilli(), "lifetime": CallLifetimeMs, "ice_servers": s.config.WebRTC.ICEServers, } online := s.hub.PublishToUserOnline(calleeID, "call_incoming", payload) if online { zap.L().Debug("Call invite sent to online callee", zap.String("call_id", call.ID), zap.String("callee_id", calleeID), ) } else { zap.L().Debug("Call invite created for offline callee", zap.String("call_id", call.ID), zap.String("callee_id", calleeID), ) } return call, calleeOnline, nil } func (s *callService) Accept(ctx context.Context, callID, userID string) (*ActiveCall, error) { call := s.getActiveCall(callID) if call == nil { return nil, apperrors.ErrCallNotFound } // 检查用户是否是参与者 if !isParticipant(call, userID) { return nil, apperrors.ErrNotCallParticipant } // 检查通话状态 if call.Status != model.CallStatusCalling { return nil, apperrors.ErrCallAlreadyAnswered } // 使用原子操作更新状态 s.mu.Lock() if call.Status != model.CallStatusCalling { s.mu.Unlock() return nil, apperrors.ErrCallAlreadyAnswered } now := time.Now() call.Status = model.CallStatusConnected call.StartedAt = &now call.Participants[userID].Status = model.ParticipantStatusJoined call.Participants[userID].JoinedAt = &now s.mu.Unlock() // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_accepted", map[string]any{ "call_id": callID, "started_at": now.UnixMilli(), "ice_servers": s.config.WebRTC.ICEServers, }) // 通知被叫方其他设备 s.hub.PublishToUserOnline(userID, "call_answered_elsewhere", map[string]any{ "call_id": callID, "reason": "answered_on_another_device", }) return call, nil } func (s *callService) Reject(ctx context.Context, callID, userID string) error { call := s.getActiveCall(callID) if call == nil { return apperrors.ErrCallNotFound } s.mu.Lock() if call.Status != model.CallStatusCalling { s.mu.Unlock() return apperrors.ErrInvalidCallState } if !isParticipant(call, userID) { s.mu.Unlock() return apperrors.ErrNotCallParticipant } call.Status = model.CallStatusRejected call.Participants[userID].Status = model.ParticipantStatusRejected s.mu.Unlock() // 从内存移除 s.removeActiveCall(callID) // 保存到数据库作为历史记录 s.saveCallHistory(call, model.CallStatusRejected, 0) // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_rejected", map[string]any{ "call_id": callID, "reason": "rejected", }) return nil } func (s *callService) Busy(ctx context.Context, callID, userID string) error { call := s.getActiveCall(callID) if call == nil { return apperrors.ErrCallNotFound } s.mu.Lock() if call.Status != model.CallStatusCalling { s.mu.Unlock() return apperrors.ErrInvalidCallState } if !isParticipant(call, userID) { s.mu.Unlock() return apperrors.ErrNotCallParticipant } call.Status = model.CallStatusMissed s.mu.Unlock() // 从内存移除 s.removeActiveCall(callID) // 保存到数据库作为历史记录 s.saveCallHistory(call, model.CallStatusMissed, 0) // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_busy", map[string]any{ "call_id": callID, }) return nil } func (s *callService) End(ctx context.Context, callID, userID string, reason string) (*ActiveCall, error) { call := s.getActiveCall(callID) if call == nil { return nil, apperrors.ErrCallNotFound } s.mu.Lock() if call.Status != model.CallStatusCalling && call.Status != model.CallStatusConnected { s.mu.Unlock() return nil, apperrors.ErrCallNotActive } // 允许系统结束通话时 userID 为空 if userID != "" && !isParticipant(call, userID) { s.mu.Unlock() return nil, apperrors.ErrNotCallParticipant } now := time.Now() var duration int64 if call.StartedAt != nil { duration = int64(now.Sub(*call.StartedAt).Seconds()) } endStatus := model.CallStatusEnded if call.Status == model.CallStatusCalling { switch reason { case "timeout": endStatus = model.CallStatusMissed case "disconnect": endStatus = model.CallStatusMissed default: endStatus = model.CallStatusCancelled } } call.Status = endStatus call.Duration = duration if userID != "" { call.Participants[userID].Status = model.ParticipantStatusLeft } s.mu.Unlock() // 从内存移除 s.removeActiveCall(callID) // 保存到数据库作为历史记录 s.saveCallHistory(call, endStatus, duration) // 通知其他参与者 for pUserID := range call.Participants { if pUserID != userID { s.hub.PublishToUserOnline(pUserID, "call_ended", map[string]any{ "call_id": callID, "ended_by": userID, "reason": reason, "duration": duration, "ended_at": now.UnixMilli(), }) } } return call, nil } func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error { call := s.getActiveCall(callID) if call == nil { return apperrors.ErrCallNotFound } s.mu.RLock() if call.Status != model.CallStatusCalling && call.Status != model.CallStatusConnected { s.mu.RUnlock() return apperrors.ErrCallNotActive } if !isParticipant(call, fromUserID) { s.mu.RUnlock() return apperrors.ErrNotCallParticipant } s.mu.RUnlock() for pUserID := range call.Participants { if pUserID != fromUserID { s.hub.PublishToUserOnlineReliable(pUserID, signalType, map[string]any{ "call_id": callID, "from_id": fromUserID, "payload": json.RawMessage(payload), }) } } return nil } func (s *callService) SetMuted(ctx context.Context, callID, userID string, muted bool) error { call := s.getActiveCall(callID) if call == nil { return apperrors.ErrCallNotFound } s.mu.RLock() if call.Status != model.CallStatusCalling && call.Status != model.CallStatusConnected { s.mu.RUnlock() return apperrors.ErrCallNotActive } if !isParticipant(call, userID) { s.mu.RUnlock() return apperrors.ErrNotCallParticipant } s.mu.RUnlock() for pUserID := range call.Participants { if pUserID != userID { s.hub.PublishToUserOnline(pUserID, "call_peer_muted", map[string]any{ "call_id": callID, "user_id": userID, "muted": muted, }) } } return nil } func (s *callService) GetCallHistory(ctx context.Context, userID string, page, pageSize int) ([]model.CallSession, int64, error) { return s.callRepo.GetCallHistory(userID, page, pageSize) } func (s *callService) GetICEServers() []config.ICEServerConfig { return s.config.WebRTC.ICEServers } // saveCallHistory 保存通话历史到数据库 func (s *callService) saveCallHistory(call *ActiveCall, status model.CallStatus, duration int64) { ctx := context.Background() now := time.Now() dbCall := &model.CallSession{ ID: call.ID, ConversationID: call.ConversationID, CallerID: call.CallerID, CallType: call.CallType, Status: status, StartedAt: call.StartedAt, EndedAt: &now, Duration: duration, CreatedAt: call.CreatedAt, UpdatedAt: now, } var participants []*model.CallParticipant for _, p := range call.Participants { participants = append(participants, &model.CallParticipant{ CallID: call.ID, UserID: p.UserID, Status: p.Status, JoinedAt: p.JoinedAt, CreatedAt: call.CreatedAt, UpdatedAt: now, }) } if err := s.callRepo.CreateCallWithParticipants(ctx, dbCall, participants); err != nil { zap.L().Error("Failed to save call history", zap.String("call_id", call.ID), zap.Error(err), ) } } // StartCleanupTicker 启动超时清理定时器 func (s *callService) StartCleanupTicker() { go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for range ticker.C { s.cleanupExpiredCalls() } }() } // cleanupExpiredCalls 清理过期的通话 func (s *callService) cleanupExpiredCalls() { ctx := context.Background() s.mu.RLock() var expiredCalls []string now := time.Now() for _, call := range s.activeCalls { if call.Status == model.CallStatusCalling { elapsed := now.Sub(call.CreatedAt) if elapsed.Milliseconds() > CallTimeoutMs { expiredCalls = append(expiredCalls, call.ID) } } } s.mu.RUnlock() for _, callID := range expiredCalls { zap.L().Info("Cleaning up expired call", zap.String("call_id", callID), ) _, _ = s.End(ctx, callID, "", "timeout") } } // handleUserDisconnect 处理用户 WebSocket 断开连接事件 func (s *callService) handleUserDisconnect(userID string, remainingCount int) { // 如果用户还有其他连接,不处理 if remainingCount > 0 { return } ctx := context.Background() // 获取用户参与的所有活跃通话 calls := s.getActiveCallsByUser(userID) if len(calls) == 0 { return } zap.L().Info("User disconnected, ending active calls", zap.String("user_id", userID), zap.Int("active_calls", len(calls)), ) // 结束所有活跃通话 for _, call := range calls { _, err := s.End(ctx, call.ID, userID, "disconnect") if err != nil { zap.L().Error("Failed to end call on user disconnect", zap.String("user_id", userID), zap.String("call_id", call.ID), zap.Error(err), ) } } }