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

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

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

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 {

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