736344f123
- Remove SetLogService methods from user, post, comment, and admin post services - Add logService as constructor parameter to all dependent services - Centralize call-related and message error definitions in app_errors.go - Add ConversationID field to WSEventResponse for improved message tracking - Simplify wire provider functions by removing manual setter calls
546 lines
15 KiB
Go
546 lines
15 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"time"
|
||
|
||
"carrot_bbs/internal/config"
|
||
apperrors "carrot_bbs/internal/errors"
|
||
"carrot_bbs/internal/model"
|
||
"carrot_bbs/internal/pkg/ws"
|
||
"carrot_bbs/internal/repository"
|
||
|
||
"go.uber.org/zap"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// 通话相关常量
|
||
const (
|
||
CallLifetimeMs = 60000 // 通话邀请有效期 60秒 (参考 Matrix)
|
||
CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒
|
||
)
|
||
|
||
// CallService 通话服务接口
|
||
type CallService interface {
|
||
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
|
||
End(ctx context.Context, callID, userID string, reason string) (*model.CallSession, error)
|
||
RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error
|
||
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 创建通话服务
|
||
func NewCallService(
|
||
callRepo repository.CallRepository,
|
||
hub *ws.Hub,
|
||
cfg *config.Config,
|
||
db *gorm.DB,
|
||
) CallService {
|
||
svc := &callService{
|
||
callRepo: callRepo,
|
||
hub: hub,
|
||
config: cfg,
|
||
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, bool, error) {
|
||
active, err := s.callRepo.GetActiveCallByConversationID(conversationID)
|
||
if err == nil && active != nil {
|
||
return nil, false, apperrors.Wrap(fmt.Errorf("call %s", active.ID), apperrors.ErrCallInProgress)
|
||
}
|
||
|
||
now := time.Now()
|
||
call := &model.CallSession{
|
||
ConversationID: conversationID,
|
||
CallerID: callerID,
|
||
CallType: model.CallTypePrivate,
|
||
Status: model.CallStatusCalling,
|
||
CreatedAt: now,
|
||
UpdatedAt: now,
|
||
}
|
||
|
||
participants := []*model.CallParticipant{
|
||
{UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now, CreatedAt: now, UpdatedAt: now},
|
||
{UserID: calleeID, Status: model.ParticipantStatusInvited, CreatedAt: now, UpdatedAt: now},
|
||
}
|
||
|
||
if err := s.callRepo.CreateCallWithParticipants(ctx, call, participants); err != nil {
|
||
return nil, false, fmt.Errorf("create call session: %w", err)
|
||
}
|
||
|
||
call, err = s.callRepo.GetCallByIDWithParticipants(call.ID)
|
||
if err != nil {
|
||
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,
|
||
"conversation_id": conversationID,
|
||
"caller_id": callerID,
|
||
"call_type": call.CallType,
|
||
"media_type": mediaType,
|
||
"created_at": now.UnixMilli(),
|
||
"lifetime": CallLifetimeMs,
|
||
"ice_servers": s.config.WebRTC.ICEServers,
|
||
}
|
||
|
||
// 尝试向被叫方发送来电通知
|
||
// 如果对方在线，立即收到来电
|
||
// 如果对方不在线，依然返回成功，让呼叫方正常等待
|
||
// 后端超时清理机制会在超时后自动结束通话并通知呼叫方
|
||
online := s.hub.PublishToUserOnline(calleeID, "call_incoming", payload)
|
||
|
||
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),
|
||
)
|
||
}
|
||
|
||
// 返回通话会话和被叫方在线状态
|
||
// 前端可以根据 calleeOnline 决定是否显示"等待对方上线"的提示
|
||
return call, calleeOnline, nil
|
||
}
|
||
|
||
func (s *callService) Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) {
|
||
call, err := s.callRepo.GetCallByIDWithParticipants(callID)
|
||
if err != nil {
|
||
return nil, apperrors.Wrap(err, apperrors.ErrCallNotFound)
|
||
}
|
||
|
||
// 检查用户是否是参与者
|
||
if !isParticipant(call, userID) {
|
||
return nil, apperrors.ErrNotCallParticipant
|
||
}
|
||
|
||
// === 修改：使用乐观锁更新状态 ===
|
||
// 只有 calling 状态的通话才能被接听
|
||
now := time.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 result.Error != nil {
|
||
return nil, fmt.Errorf("accept call: %w", result.Error)
|
||
}
|
||
if result.RowsAffected == 0 {
|
||
// 没有更新任何行，说明通话已被其他设备接听
|
||
return nil, apperrors.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)
|
||
}
|
||
|
||
// 重新获取更新后的通话
|
||
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,
|
||
})
|
||
|
||
// 通知被叫方其他设备
|
||
s.hub.PublishToUserOnline(userID, "call_answered_elsewhere", map[string]interface{}{
|
||
"call_id": callID,
|
||
"reason": "answered_on_another_device",
|
||
})
|
||
|
||
return call, nil
|
||
}
|
||
|
||
func (s *callService) Reject(ctx context.Context, callID, userID string) error {
|
||
call, err := s.callRepo.GetCallByIDWithParticipants(callID)
|
||
if err != nil {
|
||
return apperrors.Wrap(err, apperrors.ErrCallNotFound)
|
||
}
|
||
if call.Status != model.CallStatusCalling {
|
||
return apperrors.ErrInvalidCallState
|
||
}
|
||
if !isParticipant(call, userID) {
|
||
return apperrors.ErrNotCallParticipant
|
||
}
|
||
|
||
now := time.Now()
|
||
call.Status = model.CallStatusRejected
|
||
call.EndedAt = &now
|
||
call.UpdatedAt = now
|
||
|
||
if err := s.callRepo.UpdateCall(call); err != nil {
|
||
return fmt.Errorf("update call rejected: %w", err)
|
||
}
|
||
|
||
participant, err := s.callRepo.GetCallParticipant(callID, userID)
|
||
if err == nil {
|
||
participant.Status = model.ParticipantStatusRejected
|
||
participant.LeftAt = &now
|
||
participant.UpdatedAt = now
|
||
s.callRepo.UpdateCallParticipant(participant)
|
||
}
|
||
|
||
s.hub.PublishToUserOnline(call.CallerID, "call_rejected", map[string]interface{}{
|
||
"call_id": callID,
|
||
"reason": "rejected",
|
||
})
|
||
return nil
|
||
}
|
||
|
||
func (s *callService) Busy(ctx context.Context, callID, userID string) error {
|
||
call, err := s.callRepo.GetCallByIDWithParticipants(callID)
|
||
if err != nil {
|
||
return apperrors.Wrap(err, apperrors.ErrCallNotFound)
|
||
}
|
||
if call.Status != model.CallStatusCalling {
|
||
return apperrors.ErrInvalidCallState
|
||
}
|
||
if !isParticipant(call, userID) {
|
||
return apperrors.ErrNotCallParticipant
|
||
}
|
||
|
||
now := time.Now()
|
||
call.Status = model.CallStatusMissed
|
||
call.EndedAt = &now
|
||
call.UpdatedAt = now
|
||
|
||
if err := s.callRepo.UpdateCall(call); err != nil {
|
||
return fmt.Errorf("update call missed: %w", err)
|
||
}
|
||
|
||
s.hub.PublishToUserOnline(call.CallerID, "call_busy", map[string]interface{}{
|
||
"call_id": callID,
|
||
})
|
||
return nil
|
||
}
|
||
|
||
func (s *callService) End(ctx context.Context, callID, userID string, reason string) (*model.CallSession, error) {
|
||
call, err := s.callRepo.GetCallByIDWithParticipants(callID)
|
||
if err != nil {
|
||
return nil, apperrors.Wrap(err, apperrors.ErrCallNotFound)
|
||
}
|
||
if !call.IsActive() {
|
||
return nil, apperrors.ErrCallNotActive
|
||
}
|
||
// 允许系统结束通话时 userID 为空
|
||
if userID != "" && !isParticipant(call, userID) {
|
||
return nil, apperrors.ErrNotCallParticipant
|
||
}
|
||
|
||
now := time.Now()
|
||
var duration int64
|
||
if call.StartedAt != nil {
|
||
duration = int64(now.Sub(*call.StartedAt).Seconds())
|
||
}
|
||
|
||
endStatus := model.CallStatusEnded
|
||
if call.Status == model.CallStatusCalling {
|
||
if reason == "timeout" {
|
||
endStatus = model.CallStatusMissed
|
||
} else {
|
||
endStatus = model.CallStatusCancelled
|
||
}
|
||
}
|
||
|
||
call.Status = endStatus
|
||
call.EndedAt = &now
|
||
call.Duration = duration
|
||
call.UpdatedAt = now
|
||
|
||
if err := s.callRepo.UpdateCall(call); err != nil {
|
||
return nil, fmt.Errorf("end call: %w", err)
|
||
}
|
||
|
||
participant, err := s.callRepo.GetCallParticipant(callID, userID)
|
||
if err == nil {
|
||
participant.Status = model.ParticipantStatusLeft
|
||
participant.LeftAt = &now
|
||
participant.UpdatedAt = now
|
||
s.callRepo.UpdateCallParticipant(participant)
|
||
}
|
||
|
||
participants, _ := s.callRepo.GetCallParticipants(callID)
|
||
for _, p := range participants {
|
||
if p.UserID != userID {
|
||
s.hub.PublishToUserOnline(p.UserID, "call_ended", map[string]interface{}{
|
||
"call_id": callID,
|
||
"ended_by": userID,
|
||
"reason": reason,
|
||
"duration": duration,
|
||
"ended_at": now.UnixMilli(),
|
||
})
|
||
}
|
||
}
|
||
|
||
return call, nil
|
||
}
|
||
|
||
func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error {
|
||
call, err := s.callRepo.GetCallByIDWithParticipants(callID)
|
||
if err != nil {
|
||
return apperrors.Wrap(err, apperrors.ErrCallNotFound)
|
||
}
|
||
if !call.IsActive() {
|
||
return apperrors.ErrCallNotActive
|
||
}
|
||
if !isParticipant(call, fromUserID) {
|
||
return apperrors.ErrNotCallParticipant
|
||
}
|
||
|
||
participants, _ := s.callRepo.GetCallParticipants(callID)
|
||
for _, p := range participants {
|
||
if p.UserID != fromUserID {
|
||
// 使用可靠发送来转发 SDP 和 ICE 消息
|
||
s.hub.PublishToUserOnlineReliable(p.UserID, signalType, map[string]interface{}{
|
||
"call_id": callID,
|
||
"from_id": fromUserID,
|
||
"payload": json.RawMessage(payload),
|
||
})
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *callService) SetMuted(ctx context.Context, callID, userID string, muted bool) error {
|
||
call, err := s.callRepo.GetCallByIDWithParticipants(callID)
|
||
if err != nil {
|
||
return apperrors.Wrap(err, apperrors.ErrCallNotFound)
|
||
}
|
||
if !call.IsActive() {
|
||
return apperrors.ErrCallNotActive
|
||
}
|
||
if !isParticipant(call, userID) {
|
||
return apperrors.ErrNotCallParticipant
|
||
}
|
||
|
||
participants, _ := s.callRepo.GetCallParticipants(callID)
|
||
for _, p := range participants {
|
||
if p.UserID != userID {
|
||
s.hub.PublishToUserOnline(p.UserID, "call_peer_muted", map[string]interface{}{
|
||
"call_id": callID,
|
||
"user_id": userID,
|
||
"muted": muted,
|
||
})
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *callService) GetCallHistory(ctx context.Context, userID string, page, pageSize int) ([]model.CallSession, int64, error) {
|
||
return s.callRepo.GetCallHistory(userID, page, pageSize)
|
||
}
|
||
|
||
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 {
|
||
if p.UserID == userID {
|
||
return true
|
||
}
|
||
}
|
||
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),
|
||
)
|
||
}
|
||
}
|
||
}
|