feat(call): implement active call management and memory storage

- Introduced ActiveCall and ActiveParticipant types to manage call state in memory.
- Updated CallService interface to return ActiveCall instances for Invite, Accept, and End methods.
- Added methods for storing, retrieving, and removing active calls from memory, enhancing call handling efficiency.
- Implemented concurrency control with sync.RWMutex to ensure thread-safe access to active call data.

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),
-			)
-		}
-	}
-}