From 20b1e0476471b2564599900daeb33ed3c7b68e4e Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Fri, 27 Mar 2026 17:10:30 +0800 Subject: [PATCH] feat(call): enhance call handling with error differentiation and online notifications - Added error handling for call invite and answer processes to differentiate between callee offline and call already answered scenarios. - Introduced a new method to publish messages only to online users, improving real-time communication during call signaling. - Updated call service to check callee's online status before sending call invites, ensuring better user experience. - Enhanced call acceptance logic to notify both caller and callee about call status changes, including when a call is answered on another device. --- internal/handler/ws_handler.go | 15 ++++++++ internal/pkg/ws/hub.go | 35 ++++++++++++++++++ internal/service/call_service.go | 62 ++++++++++++++++++++++++-------- 3 files changed, 97 insertions(+), 15 deletions(-) diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 0b7a036..3b81e87 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -3,6 +3,7 @@ package handler import ( "context" "encoding/json" + "errors" "fmt" "net/http" "strconv" @@ -466,6 +467,15 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay 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 + } h.wsHub.SendError(client, "call_invite_failed", err.Error()) return } @@ -499,6 +509,11 @@ func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, pay call, err := h.callService.Accept(ctx, req.CallID, client.UserID) if err != nil { + // === 区分已接听错误 === + if errors.Is(err, service.ErrCallAlreadyAnswered) { + h.wsHub.SendError(client, "call_already_answered", "通话已被其他设备接听") + return + } h.wsHub.SendError(client, "call_accept_failed", err.Error()) return } diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go index 3627f9d..2ac0543 100644 --- a/internal/pkg/ws/hub.go +++ b/internal/pkg/ws/hub.go @@ -165,6 +165,41 @@ func (h *Hub) PublishToUsers(userIDs []string, eventType string, payload interfa } } +// PublishToUserOnline 仅在用户在线时发送事件，不存入历史回放 +// 适用于通话信令等实时性消息，避免断线重连时重放过期消息 +func (h *Hub) PublishToUserOnline(userID string, eventType string, payload interface{}) bool { + h.mu.RLock() + targets := make([]*Client, 0, len(h.clients[userID])) + for _, c := range h.clients[userID] { + targets = append(targets, c) + } + h.mu.RUnlock() + + if len(targets) == 0 { + return false + } + + msg := ResponseMessage{ + EventID: h.NextID(), + Type: eventType, + TS: time.Now().UnixMilli(), + Payload: payload, + } + data, err := json.Marshal(msg) + if err != nil { + return false + } + + for _, c := range targets { + select { + case <-c.Quit: + case c.Send <- data: + default: + } + } + return true +} + // publish 内部发布方法 func (h *Hub) publish(userID string, ev Event) { h.mu.Lock() diff --git a/internal/service/call_service.go b/internal/service/call_service.go index 2e76e74..54279ee 100644 --- a/internal/service/call_service.go +++ b/internal/service/call_service.go @@ -14,11 +14,19 @@ import ( ) var ( - ErrCallNotFound = errors.New("call not found") - ErrCallNotActive = errors.New("call is not active") - ErrCallInProgress = errors.New("call already in progress between users") - ErrNotParticipant = errors.New("user is not a participant of this call") - ErrInvalidCallState = errors.New("invalid call state for this operation") + ErrCallNotFound = errors.New("call not found") + ErrCallNotActive = errors.New("call is not active") + ErrCallInProgress = errors.New("call already in progress between users") + ErrNotParticipant = errors.New("user is not a participant of this call") + ErrInvalidCallState = errors.New("invalid call state for this operation") + ErrCalleeOffline = errors.New("callee is offline") + ErrCallAlreadyAnswered = errors.New("call already answered on another device") +) + +// 通话相关常量 +const ( + CallLifetimeMs = 60000 // 通话邀请有效期 60秒 (参考 Matrix) + CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒 ) // CallService 通话服务接口 @@ -54,6 +62,11 @@ func NewCallService( } func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversationID string) (*model.CallSession, error) { + // === Telegram: 检查被叫方是否在线 === + if !s.hub.HasClients(calleeID) { + return nil, ErrCalleeOffline + } + active, err := s.callRepo.GetActiveCallByConversationID(conversationID) if err == nil && active != nil { return nil, fmt.Errorf("%w: call %s", ErrCallInProgress, active.ID) @@ -83,15 +96,26 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati return nil, fmt.Errorf("get call with participants: %w", err) } + // === Element: 包含 lifetime 字段 === payload := map[string]interface{}{ "call_id": call.ID, "conversation_id": conversationID, "caller_id": callerID, "call_type": call.CallType, "created_at": now.UnixMilli(), + "lifetime": CallLifetimeMs, // 新增: 有效期 60秒 "ice_servers": s.config.WebRTC.ICEServers, } - s.hub.PublishToUser(calleeID, "call_incoming", payload) + + // === 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 + } return call, nil } @@ -101,8 +125,9 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*model if err != nil { return nil, fmt.Errorf("%w: %v", ErrCallNotFound, err) } - if !call.IsActive() { - return nil, ErrCallNotActive + // === 状态原子检查: 确保单设备接听 === + if call.Status != model.CallStatusCalling { + return nil, ErrCallAlreadyAnswered } if !isParticipant(call, userID) { return nil, ErrNotParticipant @@ -124,12 +149,19 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*model s.callRepo.UpdateCallParticipant(participant) } - s.hub.PublishToUser(call.CallerID, "call_accepted", map[string]interface{}{ + // === Telegram: 通知拨打方已接听 === + s.hub.PublishToUserOnline(call.CallerID, "call_accepted", map[string]interface{}{ "call_id": callID, - "started_at": now.UnixMilli(), + "started_at": now.UnixMilli(), "ice_servers": s.config.WebRTC.ICEServers, }) + // === Telegram: 通知被叫方其他设备 call_answered_elsewhere === + s.hub.PublishToUserOnline(userID, "call_answered_elsewhere", map[string]interface{}{ + "call_id": callID, + "reason": "answered_on_another_device", + }) + return call, nil } @@ -161,7 +193,7 @@ func (s *callService) Reject(ctx context.Context, callID, userID string) error { s.callRepo.UpdateCallParticipant(participant) } - s.hub.PublishToUser(call.CallerID, "call_rejected", map[string]interface{}{ + s.hub.PublishToUserOnline(call.CallerID, "call_rejected", map[string]interface{}{ "call_id": callID, "reason": "rejected", }) @@ -189,7 +221,7 @@ func (s *callService) Busy(ctx context.Context, callID, userID string) error { return fmt.Errorf("update call missed: %w", err) } - s.hub.PublishToUser(call.CallerID, "call_busy", map[string]interface{}{ + s.hub.PublishToUserOnline(call.CallerID, "call_busy", map[string]interface{}{ "call_id": callID, }) return nil @@ -237,7 +269,7 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str participants, _ := s.callRepo.GetCallParticipants(callID) for _, p := range participants { if p.UserID != userID { - s.hub.PublishToUser(p.UserID, "call_ended", map[string]interface{}{ + s.hub.PublishToUserOnline(p.UserID, "call_ended", map[string]interface{}{ "call_id": callID, "ended_by": userID, "reason": reason, @@ -265,7 +297,7 @@ func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string participants, _ := s.callRepo.GetCallParticipants(callID) for _, p := range participants { if p.UserID != fromUserID { - s.hub.PublishToUser(p.UserID, signalType, map[string]interface{}{ + s.hub.PublishToUserOnline(p.UserID, signalType, map[string]interface{}{ "call_id": callID, "from_id": fromUserID, "payload": json.RawMessage(payload), @@ -290,7 +322,7 @@ func (s *callService) SetMuted(ctx context.Context, callID, userID string, muted participants, _ := s.callRepo.GetCallParticipants(callID) for _, p := range participants { if p.UserID != userID { - s.hub.PublishToUser(p.UserID, "call_peer_muted", map[string]interface{}{ + s.hub.PublishToUserOnline(p.UserID, "call_peer_muted", map[string]interface{}{ "call_id": callID, "user_id": userID, "muted": muted,