From d3579983215dd7bee4a74933cb8009a20d64aa0c Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Sat, 28 Mar 2026 04:34:49 +0800 Subject: [PATCH] feat(call): enhance call handling with user connection events and active call management - Updated CallService to handle user connection and disconnection events, allowing for automatic management of active calls based on user status. - Introduced methods to retrieve active calls for users and send pending call invites when users reconnect. - Modified the Invite method to return the callee's online status, improving user experience during call invitations. - Enhanced WSHandler to support connection and disconnection event handlers, ensuring accurate call state management. --- internal/handler/ws_handler.go | 9 +- internal/pkg/ws/hub.go | 88 ++++++++++++++-- internal/repository/call_repo.go | 20 ++++ internal/service/call_service.go | 173 +++++++++++++++++++++++++++---- 4 files changed, 255 insertions(+), 35 deletions(-) diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 97bf9c5..c20d0ef 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -467,18 +467,13 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay mediaType = "voice" } - call, err := h.callService.Invite(ctx, client.UserID, req.CalleeID, req.ConversationID, mediaType) + call, _, err := h.callService.Invite(ctx, client.UserID, req.CalleeID, req.ConversationID, mediaType) if err != nil { zap.L().Warn("Failed to invite call", zap.String("caller_id", client.UserID), zap.String("callee_id", req.CalleeID), zap.Error(err), ) - // === 区分不同错误类型 === - if errors.Is(err, service.ErrCalleeOffline) { - h.wsHub.SendError(client, "callee_offline", "对方当前离线，无法拨打") - return - } if errors.Is(err, service.ErrCallInProgress) { h.wsHub.SendError(client, "call_in_progress", err.Error()) return @@ -487,6 +482,7 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay return } + // 返回给呼叫方的响应 resp := ws.ResponseMessage{ EventID: h.wsHub.NextID(), Type: "call_invited", @@ -495,6 +491,7 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay "call_id": call.ID, "conversation_id": req.ConversationID, "callee_id": req.CalleeID, + "lifetime": service.CallLifetimeMs, }, } data, _ := json.Marshal(resp) diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go index e6ffd2a..1e8c0ee 100644 --- a/internal/pkg/ws/hub.go +++ b/internal/pkg/ws/hub.go @@ -61,22 +61,36 @@ type subscriber struct { quit chan struct{} } +// DisconnectHandler 用户断开连接时的回调函数 +// userID: 断开的用户ID +// remainingCount: 该用户剩余的连接数（0表示用户完全离线） +type DisconnectHandler func(userID string, remainingCount int) + +// ConnectHandler 用户连接时的回调函数 +// userID: 连接的用户ID +// isFirstConnection: 是否是该用户的第一个连接（之前完全离线） +type ConnectHandler func(userID string, isFirstConnection bool) + // Hub WebSocket连接管理器 type Hub struct { seq uint64 - mu sync.RWMutex - clients map[string]map[uint64]*Client // userID -> clientID -> Client - history map[string][]Event // userID -> []Event (用于断线重连历史回放) - subscribers map[string]map[uint64]*subscriber // userID -> subscriberID -> subscriber + mu sync.RWMutex + clients map[string]map[uint64]*Client // userID -> clientID -> Client + history map[string][]Event // userID -> []Event (用于断线重连历史回放) + subscribers map[string]map[uint64]*subscriber // userID -> subscriberID -> subscriber + disconnectHandlers []DisconnectHandler // 断开连接事件处理器列表 + connectHandlers []ConnectHandler // 连接事件处理器列表 } // NewHub 创建WebSocket Hub func NewHub() *Hub { return &Hub{ - clients: make(map[string]map[uint64]*Client), - history: make(map[string][]Event), - subscribers: make(map[string]map[uint64]*subscriber), + clients: make(map[string]map[uint64]*Client), + history: make(map[string][]Event), + subscribers: make(map[string]map[uint64]*subscriber), + disconnectHandlers: make([]DisconnectHandler, 0), + connectHandlers: make([]ConnectHandler, 0), } } @@ -88,9 +102,12 @@ func (h *Hub) NextID() uint64 { // Register 注册客户端连接 func (h *Hub) Register(client *Client) []Event { h.mu.Lock() - defer h.mu.Unlock() - if _, ok := h.clients[client.UserID]; !ok { + // 检查是否是该用户的第一个连接（之前完全离线） + _, existed := h.clients[client.UserID] + isFirstConnection := !existed + + if !existed { h.clients[client.UserID] = make(map[uint64]*Client) } h.clients[client.UserID][client.ID] = client @@ -105,31 +122,62 @@ func (h *Hub) Register(client *Client) []Event { zap.String("user_id", client.UserID), zap.Uint64("client_id", client.ID), zap.Int("total_clients", len(h.clients[client.UserID])), + zap.Bool("is_first_connection", isFirstConnection), ) + // 复制处理器列表避免锁竞争 + handlers := make([]ConnectHandler, len(h.connectHandlers)) + copy(handlers, h.connectHandlers) + h.mu.Unlock() + + // 异步调用连接处理器，避免阻塞 Register + if len(handlers) > 0 { + go func() { + for _, handler := range handlers { + handler(client.UserID, isFirstConnection) + } + }() + } + return replay } // Unregister 注销客户端连接 func (h *Hub) Unregister(client *Client) { h.mu.Lock() - defer h.mu.Unlock() + var remainingCount int if userClients, ok := h.clients[client.UserID]; ok { if _, exists := userClients[client.ID]; exists { delete(userClients, client.ID) close(client.Quit) + remainingCount = len(userClients) - if len(userClients) == 0 { + if remainingCount == 0 { delete(h.clients, client.UserID) } zap.L().Debug("WebSocket client unregistered", zap.String("user_id", client.UserID), zap.Uint64("client_id", client.ID), + zap.Int("remaining_connections", remainingCount), ) } } + + // 复制处理器列表避免锁竞争 + handlers := make([]DisconnectHandler, len(h.disconnectHandlers)) + copy(handlers, h.disconnectHandlers) + h.mu.Unlock() + + // 异步调用断开处理器，避免阻塞 Unregister + if len(handlers) > 0 { + go func() { + for _, handler := range handlers { + handler(client.UserID, remainingCount) + } + }() + } } // HasClients 检查用户是否有活跃连接 @@ -139,6 +187,24 @@ func (h *Hub) HasClients(userID string) bool { return len(h.clients[userID]) > 0 } +// OnDisconnect 注册断开连接事件处理器 +// 当用户断开 WebSocket 连接时，会调用所有注册的处理器 +// 处理器会收到用户ID和该用户剩余的连接数 +func (h *Hub) OnDisconnect(handler DisconnectHandler) { + h.mu.Lock() + defer h.mu.Unlock() + h.disconnectHandlers = append(h.disconnectHandlers, handler) +} + +// OnConnect 注册连接事件处理器 +// 当用户建立 WebSocket 连接时，会调用所有注册的处理器 +// 处理器会收到用户ID和是否是该用户的第一个连接 +func (h *Hub) OnConnect(handler ConnectHandler) { + h.mu.Lock() + defer h.mu.Unlock() + h.connectHandlers = append(h.connectHandlers, handler) +} + // GetClientCount 获取用户的连接数 func (h *Hub) GetClientCount(userID string) int { h.mu.RLock() diff --git a/internal/repository/call_repo.go b/internal/repository/call_repo.go index 52910b6..48601d2 100644 --- a/internal/repository/call_repo.go +++ b/internal/repository/call_repo.go @@ -16,6 +16,7 @@ type CallRepository interface { GetCallByIDWithParticipants(id string) (*model.CallSession, error) UpdateCall(call *model.CallSession) error GetActiveCallByConversationID(conversationID string) (*model.CallSession, error) + GetActiveCallsByUserID(userID string) ([]model.CallSession, error) // 获取用户参与的活跃通话 // 通话参与者操作 CreateCallParticipant(participant *model.CallParticipant) error @@ -83,6 +84,25 @@ func (r *callRepository) GetActiveCallByConversationID(conversationID string) (* return &call, nil } +// GetActiveCallsByUserID 获取用户参与的所有活跃通话 +func (r *callRepository) GetActiveCallsByUserID(userID string) ([]model.CallSession, error) { + var calls []model.CallSession + + // 子查询获取用户参与的通话ID + subQuery := r.db.Model(&model.CallParticipant{}). + Select("DISTINCT call_id"). + Where("user_id = ?", userID) + + err := r.db.Where("id IN (?) AND status IN ?", subQuery, []model.CallStatus{ + model.CallStatusCalling, + model.CallStatusConnected, + }).Preload("Participants").Find(&calls).Error + if err != nil { + return nil, err + } + return calls, nil +} + // CreateCallParticipant 创建通话参与者 func (r *callRepository) CreateCallParticipant(participant *model.CallParticipant) error { return r.db.Create(participant).Error diff --git a/internal/service/call_service.go b/internal/service/call_service.go index 224a19a..c173aff 100644 --- a/internal/service/call_service.go +++ b/internal/service/call_service.go @@ -12,6 +12,7 @@ import ( "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" + "go.uber.org/zap" "gorm.io/gorm" ) @@ -33,7 +34,7 @@ const ( // CallService 通话服务接口 type CallService interface { - Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*model.CallSession, error) + Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*model.CallSession, bool, error) Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) Reject(ctx context.Context, callID, userID string) error Busy(ctx context.Context, callID, userID string) error @@ -42,7 +43,7 @@ type CallService interface { 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() // 新增：启动超时清理 + StartCleanupTicker() } type callService struct { @@ -66,21 +67,22 @@ func NewCallService( db: db, } + // 订阅 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, error) { - // === Telegram: 检查被叫方是否在线 === - if !s.hub.HasClients(calleeID) { - return nil, ErrCalleeOffline - } - +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 { - return nil, fmt.Errorf("%w: call %s", ErrCallInProgress, active.ID) + return nil, false, fmt.Errorf("%w: call %s", ErrCallInProgress, active.ID) } now := time.Now() @@ -99,14 +101,17 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati } if err := s.callRepo.CreateCallWithParticipants(ctx, call, participants); err != nil { - return nil, fmt.Errorf("create call session: %w", err) + return nil, false, fmt.Errorf("create call session: %w", err) } call, err = s.callRepo.GetCallByIDWithParticipants(call.ID) if err != nil { - return nil, fmt.Errorf("get call with participants: %w", err) + return nil, false, fmt.Errorf("get call with participants: %w", err) } + // 检查被叫方是否在线 + calleeOnline := s.hub.HasClients(calleeID) + // === Element: 包含 lifetime 字段 === payload := map[string]interface{}{ "call_id": call.ID, @@ -119,16 +124,29 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati "ice_servers": s.config.WebRTC.ICEServers, } - // === Telegram: 不存 history === + // 尝试向被叫方发送来电通知 + // 如果对方在线，立即收到来电 + // 如果对方不在线，依然返回成功，让呼叫方正常等待 + // 后端超时清理机制会在超时后自动结束通话并通知呼叫方 online := s.hub.PublishToUserOnline(calleeID, "call_incoming", payload) - if !online { - call.Status = model.CallStatusCancelled - call.EndedAt = &now - s.callRepo.UpdateCall(call) - return nil, ErrCalleeOffline + + 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.String("call_id", call.ID), + zap.String("callee_id", calleeID), + zap.Bool("online", false), + ) } - return call, nil + // 返回通话会话和被叫方在线状态 + // 前端可以根据 calleeOnline 决定是否显示"等待对方上线"的提示 + return call, calleeOnline, nil } func (s *callService) Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) { @@ -416,3 +434,122 @@ func isParticipant(call *model.CallSession, userID string) bool { } return false } + +// handleUserDisconnect 处理用户 WebSocket 断开连接事件 +// 当用户完全离线时（remainingCount == 0），结束该用户参与的所有活跃通话 +func (s *callService) handleUserDisconnect(userID string, remainingCount int) { + // 如果用户还有其他连接，不处理 + if remainingCount > 0 { + return + } + + 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 + } + + 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), + ) + } + } +} + +// 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), + ) + } + } +}