63cfbad65c
- Updated the WSHandler to include a media type field in the call invite request, allowing differentiation between voice and video calls. - Modified the CallService interface and implementation to accept the media type parameter during call invitations, enhancing call handling capabilities.
419 lines
12 KiB
Go
419 lines
12 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"carrot_bbs/internal/config"
|
|
"carrot_bbs/internal/model"
|
|
"carrot_bbs/internal/pkg/ws"
|
|
"carrot_bbs/internal/repository"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
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")
|
|
ErrCalleeOffline = errors.New("callee is offline")
|
|
ErrCallAlreadyAnswered = errors.New("call already answered on another device")
|
|
)
|
|
|
|
// 通话相关常量
|
|
const (
|
|
CallLifetimeMs = 60000 // 通话邀请有效期 60秒 (参考 Matrix)
|
|
CallTimeoutMs = 120000 // 通话超时(无人接听) 120秒
|
|
)
|
|
|
|
// CallService 通话服务接口
|
|
type CallService interface {
|
|
Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*model.CallSession, 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,
|
|
}
|
|
|
|
// 自动启动超时清理
|
|
svc.StartCleanupTicker()
|
|
|
|
return svc
|
|
}
|
|
|
|
func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType 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)
|
|
}
|
|
|
|
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, fmt.Errorf("create call session: %w", err)
|
|
}
|
|
|
|
call, err = s.callRepo.GetCallByIDWithParticipants(call.ID)
|
|
if err != nil {
|
|
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,
|
|
"media_type": mediaType,
|
|
"created_at": now.UnixMilli(),
|
|
"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)
|
|
return nil, ErrCalleeOffline
|
|
}
|
|
|
|
return call, 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, fmt.Errorf("%w: %v", ErrCallNotFound, err)
|
|
}
|
|
|
|
// 检查用户是否是参与者
|
|
if !isParticipant(call, userID) {
|
|
return nil, ErrNotParticipant
|
|
}
|
|
|
|
// === 修改：使用乐观锁更新状态 ===
|
|
// 只有 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, 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 fmt.Errorf("%w: %v", ErrCallNotFound, err)
|
|
}
|
|
if call.Status != model.CallStatusCalling {
|
|
return ErrInvalidCallState
|
|
}
|
|
if !isParticipant(call, userID) {
|
|
return ErrNotParticipant
|
|
}
|
|
|
|
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 fmt.Errorf("%w: %v", ErrCallNotFound, err)
|
|
}
|
|
if call.Status != model.CallStatusCalling {
|
|
return ErrInvalidCallState
|
|
}
|
|
if !isParticipant(call, userID) {
|
|
return ErrNotParticipant
|
|
}
|
|
|
|
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, fmt.Errorf("%w: %v", ErrCallNotFound, err)
|
|
}
|
|
if !call.IsActive() {
|
|
return nil, ErrCallNotActive
|
|
}
|
|
// 允许系统结束通话时 userID 为空
|
|
if userID != "" && !isParticipant(call, userID) {
|
|
return nil, ErrNotParticipant
|
|
}
|
|
|
|
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 fmt.Errorf("%w: %v", ErrCallNotFound, err)
|
|
}
|
|
if !call.IsActive() {
|
|
return ErrCallNotActive
|
|
}
|
|
if !isParticipant(call, fromUserID) {
|
|
return ErrNotParticipant
|
|
}
|
|
|
|
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 fmt.Errorf("%w: %v", ErrCallNotFound, err)
|
|
}
|
|
if !call.IsActive() {
|
|
return ErrCallNotActive
|
|
}
|
|
if !isParticipant(call, userID) {
|
|
return ErrNotParticipant
|
|
}
|
|
|
|
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
|
|
}
|