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

@@ -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
 	}

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()

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,