From bac3cbbf60cb0712d468163b13de38853f1681d8 Mon Sep 17 00:00:00 2001 From: lafay <2021211506@stu.hit.edu.cn> Date: Sat, 28 Mar 2026 00:20:56 +0800 Subject: [PATCH] feat(call): enhance call service with reliable message delivery and automatic cleanup - Introduced a new method, PublishToUserOnlineReliable, to ensure critical signaling messages are reliably sent to online users. - Updated CallService to include a cleanup mechanism for expired calls, improving resource management and user experience. - Enhanced call acceptance logic to utilize optimistic locking for state updates, ensuring accurate call status handling across devices. - Refactored wire generation to accommodate the new database dependency in CallService, streamlining service initialization. --- cmd/server/wire_gen.go | 2 +- internal/pkg/ws/hub.go | 38 ++++++++++++ internal/service/call_service.go | 102 +++++++++++++++++++++++++------ internal/wire/service.go | 3 +- 4 files changed, 125 insertions(+), 20 deletions(-) diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index a7f1494..0f8fc7c 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -119,7 +119,7 @@ func InitializeApp() (*App, error) { materialFileRepository := repository.NewMaterialFileRepository(db) callRepository := repository.NewCallRepository(db) materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository) - callService := wire.ProvideCallService(callRepository, hub, config) + callService := wire.ProvideCallService(callRepository, hub, config, db) materialHandler := handler.NewMaterialHandler(materialService) callHandler := handler.NewCallHandler(callService) wsHandler := wire.ProvideWSHandler(hub, chatService, groupService, jwtService, callService) diff --git a/internal/pkg/ws/hub.go b/internal/pkg/ws/hub.go index 2ac0543..e6ffd2a 100644 --- a/internal/pkg/ws/hub.go +++ b/internal/pkg/ws/hub.go @@ -200,6 +200,44 @@ func (h *Hub) PublishToUserOnline(userID string, eventType string, payload inter return true } +// PublishToUserOnlineReliable 可靠地发送事件给在线用户（阻塞发送） +// 用于重要的信令消息（如 SDP, ICE candidate），确保消息送达 +// 如果用户不在线，返回 false +func (h *Hub) PublishToUserOnlineReliable(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: + // 消息已发送 + } + } + 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 54279ee..c60f4b4 100644 --- a/internal/service/call_service.go +++ b/internal/service/call_service.go @@ -11,6 +11,8 @@ import ( "carrot_bbs/internal/model" "carrot_bbs/internal/pkg/ws" "carrot_bbs/internal/repository" + + "gorm.io/gorm" ) var ( @@ -26,7 +28,7 @@ var ( // 通话相关常量 const ( CallLifetimeMs = 60000 // 通话邀请有效期 60秒 (参考 Matrix) - CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒 + CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒 ) // CallService 通话服务接口 @@ -40,12 +42,14 @@ 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() // 新增：启动超时清理 } type callService struct { callRepo repository.CallRepository hub *ws.Hub config *config.Config + db *gorm.DB } // NewCallService 创建通话服务 @@ -53,12 +57,19 @@ func NewCallService( callRepo repository.CallRepository, hub *ws.Hub, cfg *config.Config, + db *gorm.DB, ) CallService { - return &callService{ + svc := &callService{ callRepo: callRepo, hub: hub, config: cfg, + db: db, } + + // 自动启动超时清理 + svc.StartCleanupTicker() + + return svc } func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversationID string) (*model.CallSession, error) { @@ -103,14 +114,13 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati "caller_id": callerID, "call_type": call.CallType, "created_at": now.UnixMilli(), - "lifetime": CallLifetimeMs, // 新增: 有效期 60秒 + "lifetime": CallLifetimeMs, "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) @@ -125,38 +135,54 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*model if err != nil { return nil, fmt.Errorf("%w: %v", ErrCallNotFound, err) } - // === 状态原子检查: 确保单设备接听 === - if call.Status != model.CallStatusCalling { - return nil, ErrCallAlreadyAnswered - } + + // 检查用户是否是参与者 if !isParticipant(call, userID) { return nil, ErrNotParticipant } + // === 修改：使用乐观锁更新状态 === + // 只有 calling 状态的通话才能被接听 now := time.Now() - call.Status = model.CallStatusConnected - call.StartedAt = &now - call.UpdatedAt = 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 err := s.callRepo.UpdateCall(call); err != nil { - return nil, fmt.Errorf("update call connected: %w", err) + if result.Error != nil { + return nil, fmt.Errorf("accept call: %w", result.Error) + } + if result.RowsAffected == 0 { + // 没有更新任何行，说明通话已被其他设备接听 + return nil, 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) } - // === Telegram: 通知拨打方已接听 === + // 重新获取更新后的通话 + call, err = s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return nil, fmt.Errorf("get updated call: %w", err) + } + + // 通知拨打方 s.hub.PublishToUserOnline(call.CallerID, "call_accepted", map[string]interface{}{ "call_id": callID, "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", @@ -190,6 +216,7 @@ func (s *callService) Reject(ctx context.Context, callID, userID string) error { if err == nil { participant.Status = model.ParticipantStatusRejected participant.LeftAt = &now + participant.UpdatedAt = now s.callRepo.UpdateCallParticipant(participant) } @@ -235,7 +262,8 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str if !call.IsActive() { return nil, ErrCallNotActive } - if !isParticipant(call, userID) { + // 允许系统结束通话时 userID 为空 + if userID != "" && !isParticipant(call, userID) { return nil, ErrNotParticipant } @@ -247,7 +275,11 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str endStatus := model.CallStatusEnded if call.Status == model.CallStatusCalling { - endStatus = model.CallStatusCancelled + if reason == "timeout" { + endStatus = model.CallStatusMissed + } else { + endStatus = model.CallStatusCancelled + } } call.Status = endStatus @@ -263,6 +295,7 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str if err == nil { participant.Status = model.ParticipantStatusLeft participant.LeftAt = &now + participant.UpdatedAt = now s.callRepo.UpdateCallParticipant(participant) } @@ -297,7 +330,8 @@ 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.PublishToUserOnline(p.UserID, signalType, map[string]interface{}{ + // 使用可靠发送来转发 SDP 和 ICE 消息 + s.hub.PublishToUserOnlineReliable(p.UserID, signalType, map[string]interface{}{ "call_id": callID, "from_id": fromUserID, "payload": json.RawMessage(payload), @@ -340,6 +374,38 @@ func (s *callService) GetICEServers() []config.ICEServerConfig { return s.config.WebRTC.ICEServers } +// StartCleanupTicker 启动超时清理定时器 +func (s *callService) StartCleanupTicker() { + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for range ticker.C { + s.cleanupExpiredCalls() + } + }() +} + +// cleanupExpiredCalls 清理过期的通话 +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 { diff --git a/internal/wire/service.go b/internal/wire/service.go index ad80254..a4e079a 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -387,6 +387,7 @@ func ProvideCallService( callRepo repository.CallRepository, wsHub *ws.Hub, cfg *config.Config, + db *gorm.DB, ) service.CallService { - return service.NewCallService(callRepo, wsHub, cfg) + return service.NewCallService(callRepo, wsHub, cfg, db) }