diff --git a/internal/service/call_service.go b/internal/service/call_service.go index a55abdf..437e652 100644 --- a/internal/service/call_service.go +++ b/internal/service/call_service.go @@ -4,6 +4,8 @@ import ( "context" "encoding/json" "fmt" + "strconv" + "sync" "time" "carrot_bbs/internal/config" @@ -22,13 +24,38 @@ const ( 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) (*model.CallSession, bool, error) - Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) + 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) (*model.CallSession, 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) @@ -41,6 +68,12 @@ type callService struct { 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 创建通话服务 @@ -51,58 +84,150 @@ func NewCallService( db *gorm.DB, ) CallService { svc := &callService{ - callRepo: callRepo, - hub: hub, - config: cfg, - db: db, + 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) - // 订阅 WebSocket 连接事件 - hub.OnConnect(svc.handleUserConnect) - // 自动启动超时清理 svc.StartCleanupTicker() return svc } -func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*model.CallSession, bool, error) { - active, err := s.callRepo.GetActiveCallByConversationID(conversationID) - if err == nil && active != nil { +// 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() - call := &model.CallSession{ + callID := s.generateCallID() + + call := &ActiveCall{ + ID: callID, ConversationID: conversationID, CallerID: callerID, + CalleeID: calleeID, CallType: model.CallTypePrivate, Status: model.CallStatusCalling, + MediaType: mediaType, CreatedAt: now, - UpdatedAt: now, + Participants: map[string]*ActiveParticipant{ + callerID: {UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now}, + calleeID: {UserID: calleeID, Status: model.ParticipantStatusInvited}, + }, + ICEServers: s.config.WebRTC.ICEServers, } - participants := []*model.CallParticipant{ - {UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now, CreatedAt: now, UpdatedAt: now}, - {UserID: calleeID, Status: model.ParticipantStatusInvited, CreatedAt: now, UpdatedAt: now}, - } - - if err := s.callRepo.CreateCallWithParticipants(ctx, call, participants); err != nil { - return nil, false, fmt.Errorf("create call session: %w", err) - } - - call, err = s.callRepo.GetCallByIDWithParticipants(call.ID) - if err != nil { - return nil, false, fmt.Errorf("get call with participants: %w", err) - } + // 存储到内存 + s.storeActiveCall(call) // 检查被叫方是否在线 calleeOnline := s.hub.HasClients(calleeID) - // === Element: 包含 lifetime 字段 === + // 发送来电通知 payload := map[string]interface{}{ "call_id": call.ID, "conversation_id": conversationID, @@ -114,35 +239,27 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati "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), - zap.Bool("online", true), ) } else { - zap.L().Debug("Call invite created for offline callee, waiting for reconnect", + zap.L().Debug("Call invite created for offline callee", zap.String("call_id", call.ID), zap.String("callee_id", calleeID), - zap.Bool("online", false), ) } - // 返回通话会话和被叫方在线状态 - // 前端可以根据 calleeOnline 决定是否显示"等待对方上线"的提示 return call, calleeOnline, nil } -func (s *callService) Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) { - call, err := s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return nil, apperrors.Wrap(err, apperrors.ErrCallNotFound) +func (s *callService) Accept(ctx context.Context, callID, userID string) (*ActiveCall, error) { + call := s.getActiveCall(callID) + if call == nil { + return nil, apperrors.ErrCallNotFound } // 检查用户是否是参与者 @@ -150,39 +267,23 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*model return nil, apperrors.ErrNotCallParticipant } - // === 修改：使用乐观锁更新状态 === - // 只有 calling 状态的通话才能被接听 - now := time.Now() - result := s.db.Model(&model.CallSession{}). - Where("id = ? AND status = ?", callID, model.CallStatusCalling). - Updates(map[string]interface{}{ - "status": model.CallStatusConnected, - "started_at": now, - "updated_at": now, - }) - - if result.Error != nil { - return nil, fmt.Errorf("accept call: %w", result.Error) - } - if result.RowsAffected == 0 { - // 没有更新任何行，说明通话已被其他设备接听 + // 检查通话状态 + if call.Status != model.CallStatusCalling { return nil, apperrors.ErrCallAlreadyAnswered } - // 更新参与者状态 - participant, err := s.callRepo.GetCallParticipant(callID, userID) - if err == nil { - participant.Status = model.ParticipantStatusJoined - participant.JoinedAt = &now - participant.UpdatedAt = now - s.callRepo.UpdateCallParticipant(participant) - } - - // 重新获取更新后的通话 - call, err = s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return nil, fmt.Errorf("get updated call: %w", err) + // 使用原子操作更新状态 + 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]interface{}{ @@ -201,78 +302,87 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*model } func (s *callService) Reject(ctx context.Context, callID, userID string) error { - call, err := s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return apperrors.Wrap(err, apperrors.ErrCallNotFound) + 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 } - now := time.Now() call.Status = model.CallStatusRejected - call.EndedAt = &now - call.UpdatedAt = now + call.Participants[userID].Status = model.ParticipantStatusRejected + s.mu.Unlock() - if err := s.callRepo.UpdateCall(call); err != nil { - return fmt.Errorf("update call rejected: %w", err) - } + // 从内存移除 + s.removeActiveCall(callID) - participant, err := s.callRepo.GetCallParticipant(callID, userID) - if err == nil { - participant.Status = model.ParticipantStatusRejected - participant.LeftAt = &now - participant.UpdatedAt = now - s.callRepo.UpdateCallParticipant(participant) - } + // 保存到数据库作为历史记录 + s.saveCallHistory(call, model.CallStatusRejected, 0) + // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_rejected", map[string]interface{}{ "call_id": callID, "reason": "rejected", }) + return nil } func (s *callService) Busy(ctx context.Context, callID, userID string) error { - call, err := s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return apperrors.Wrap(err, apperrors.ErrCallNotFound) + 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 } - now := time.Now() call.Status = model.CallStatusMissed - call.EndedAt = &now - call.UpdatedAt = now + s.mu.Unlock() - if err := s.callRepo.UpdateCall(call); err != nil { - return fmt.Errorf("update call missed: %w", err) - } + // 从内存移除 + s.removeActiveCall(callID) + // 保存到数据库作为历史记录 + s.saveCallHistory(call, model.CallStatusMissed, 0) + + // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_busy", map[string]interface{}{ "call_id": callID, }) + return nil } -func (s *callService) End(ctx context.Context, callID, userID string, reason string) (*model.CallSession, error) { - call, err := s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return nil, apperrors.Wrap(err, apperrors.ErrCallNotFound) +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 } - if !call.IsActive() { + + 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 } @@ -284,34 +394,33 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str endStatus := model.CallStatusEnded if call.Status == model.CallStatusCalling { - if reason == "timeout" { + switch reason { + case "timeout": endStatus = model.CallStatusMissed - } else { + case "disconnect": + endStatus = model.CallStatusMissed + default: endStatus = model.CallStatusCancelled } } call.Status = endStatus - call.EndedAt = &now call.Duration = duration - call.UpdatedAt = now - - if err := s.callRepo.UpdateCall(call); err != nil { - return nil, fmt.Errorf("end call: %w", err) + if userID != "" { + call.Participants[userID].Status = model.ParticipantStatusLeft } + s.mu.Unlock() - participant, err := s.callRepo.GetCallParticipant(callID, userID) - if err == nil { - participant.Status = model.ParticipantStatusLeft - participant.LeftAt = &now - participant.UpdatedAt = now - s.callRepo.UpdateCallParticipant(participant) - } + // 从内存移除 + s.removeActiveCall(callID) - participants, _ := s.callRepo.GetCallParticipants(callID) - for _, p := range participants { - if p.UserID != userID { - s.hub.PublishToUserOnline(p.UserID, "call_ended", map[string]interface{}{ + // 保存到数据库作为历史记录 + s.saveCallHistory(call, endStatus, duration) + + // 通知其他参与者 + for pUserID := range call.Participants { + if pUserID != userID { + s.hub.PublishToUserOnline(pUserID, "call_ended", map[string]interface{}{ "call_id": callID, "ended_by": userID, "reason": reason, @@ -325,22 +434,25 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str } func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error { - call, err := s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return apperrors.Wrap(err, apperrors.ErrCallNotFound) + call := s.getActiveCall(callID) + if call == nil { + return apperrors.ErrCallNotFound } - if !call.IsActive() { + + 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() - participants, _ := s.callRepo.GetCallParticipants(callID) - for _, p := range participants { - if p.UserID != fromUserID { - // 使用可靠发送来转发 SDP 和 ICE 消息 - s.hub.PublishToUserOnlineReliable(p.UserID, signalType, map[string]interface{}{ + for pUserID := range call.Participants { + if pUserID != fromUserID { + s.hub.PublishToUserOnlineReliable(pUserID, signalType, map[string]interface{}{ "call_id": callID, "from_id": fromUserID, "payload": json.RawMessage(payload), @@ -351,21 +463,25 @@ func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string } func (s *callService) SetMuted(ctx context.Context, callID, userID string, muted bool) error { - call, err := s.callRepo.GetCallByIDWithParticipants(callID) - if err != nil { - return apperrors.Wrap(err, apperrors.ErrCallNotFound) + call := s.getActiveCall(callID) + if call == nil { + return apperrors.ErrCallNotFound } - if !call.IsActive() { + + 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() - participants, _ := s.callRepo.GetCallParticipants(callID) - for _, p := range participants { - if p.UserID != userID { - s.hub.PublishToUserOnline(p.UserID, "call_peer_muted", map[string]interface{}{ + for pUserID := range call.Participants { + if pUserID != userID { + s.hub.PublishToUserOnline(pUserID, "call_peer_muted", map[string]interface{}{ "call_id": callID, "user_id": userID, "muted": muted, @@ -383,6 +499,44 @@ 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() { @@ -399,34 +553,28 @@ func (s *callService) StartCleanupTicker() { func (s *callService) cleanupExpiredCalls() { ctx := context.Background() - // 查找超过 CallTimeoutMs 的 calling 状态通话 - expiredAt := time.Now().Add(-time.Duration(CallTimeoutMs) * time.Millisecond) - var calls []model.CallSession - - if err := s.db. - Where("status = ? AND created_at < ?", model.CallStatusCalling, expiredAt). - Find(&calls).Error; err != nil { - return - } - - for _, call := range calls { - // 标记为 missed - _, _ = s.End(ctx, call.ID, "", "timeout") - } -} - -// isParticipant 检查用户是否是通话参与者 -func isParticipant(call *model.CallSession, userID string) bool { - for _, p := range call.Participants { - if p.UserID == userID { - return true + 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) + } } } - return false + 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 断开连接事件 -// 当用户完全离线时（remainingCount == 0），结束该用户参与的所有活跃通话 func (s *callService) handleUserDisconnect(userID string, remainingCount int) { // 如果用户还有其他连接，不处理 if remainingCount > 0 { @@ -436,15 +584,7 @@ func (s *callService) handleUserDisconnect(userID string, remainingCount int) { ctx := context.Background() // 获取用户参与的所有活跃通话 - calls, err := s.callRepo.GetActiveCallsByUserID(userID) - if err != nil { - zap.L().Error("Failed to get active calls for disconnected user", - zap.String("user_id", userID), - zap.Error(err), - ) - return - } - + calls := s.getActiveCallsByUser(userID) if len(calls) == 0 { return } @@ -466,80 +606,3 @@ func (s *callService) handleUserDisconnect(userID string, remainingCount int) { } } } - -// handleUserConnect 处理用户 WebSocket 连接事件 -// 当用户从离线状态变为在线时，检查是否有待处理的通话邀请并推送 -func (s *callService) handleUserConnect(userID string, isFirstConnection bool) { - // 只有当用户是从完全离线变为在线时才处理 - if !isFirstConnection { - return - } - - // 获取用户作为被叫方的所有活跃通话 - calls, err := s.callRepo.GetActiveCallsByUserID(userID) - if err != nil { - zap.L().Error("Failed to get active calls for connected user", - zap.String("user_id", userID), - zap.Error(err), - ) - return - } - - if len(calls) == 0 { - return - } - - zap.L().Info("User connected, sending pending call invites", - zap.String("user_id", userID), - zap.Int("pending_calls", len(calls)), - ) - - // 推送所有待处理的通话邀请 - for _, call := range calls { - // 检查用户是否是被叫方（status == invited） - isCallee := false - for _, p := range call.Participants { - if p.UserID == userID && p.Status == model.ParticipantStatusInvited { - isCallee = true - break - } - } - - if !isCallee { - continue - } - - // 检查通话是否还在有效期内 - elapsed := time.Since(call.CreatedAt) - if elapsed.Milliseconds() > CallLifetimeMs { - zap.L().Debug("Skipping expired call invite", - zap.String("call_id", call.ID), - zap.Duration("elapsed", elapsed), - ) - continue - } - - remainingLifetime := CallLifetimeMs - elapsed.Milliseconds() - - // 推送来电通知 - payload := map[string]interface{}{ - "call_id": call.ID, - "conversation_id": call.ConversationID, - "caller_id": call.CallerID, - "call_type": call.CallType, - "media_type": "voice", // 默认语音通话 - "created_at": call.CreatedAt.UnixMilli(), - "lifetime": remainingLifetime, - "ice_servers": s.config.WebRTC.ICEServers, - } - - sent := s.hub.PublishToUserOnline(userID, "call_incoming", payload) - if sent { - zap.L().Debug("Pending call invite sent to newly connected user", - zap.String("call_id", call.ID), - zap.String("user_id", userID), - zap.Int64("remaining_lifetime_ms", remainingLifetime), - ) - } - } -}