feat(call): implement group calling and enhance call management
All checks were successful
Build Backend / build (push) Successful in 2m14s
Build Backend / build-docker (push) Successful in 1m15s

Introduces group calling capabilities, including group invites and participant
lifecycle management. Enhances the existing call system with support for
media types (voice/video) and improved participant tracking.

- Add `GroupInvite` and `ParticipantJoin/Leave` to `CallService`.
- Implement WebSocket handlers for `call_group_invite` and
  `call_participant_join`.
- Update `CallSession` model to support group IDs, media types, and
  tracking who ended the call.
- Integrate `GroupService` into `CallService` via dependency injection.
- Add error constants for group call limitations and participant capacity.
- Update LiveKit webhook handling to process participant leave events.

Refactor grade synchronization to remove semester-based filtering in favor
of a more flexible user-based approach and improve GPA summary updates.
This commit is contained in:
2026-06-07 00:11:47 +08:00
parent c741a7d61d
commit 60654bc0f4
11 changed files with 394 additions and 69 deletions

View File

@@ -137,7 +137,7 @@ func InitializeApp() (*App, error) {
materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository) materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository)
materialHandler := handler.NewMaterialHandler(materialService) materialHandler := handler.NewMaterialHandler(materialService)
callRepository := repository.NewCallRepository(db) callRepository := repository.NewCallRepository(db)
callService := wire.ProvideCallService(callRepository, messagePublisher, config, db, client) callService := wire.ProvideCallService(callRepository, messagePublisher, config, db, client, groupService)
callHandler := handler.NewCallHandler(callService) callHandler := handler.NewCallHandler(callService)
reportRepository := repository.NewReportRepository(db) reportRepository := repository.NewReportRepository(db)
reportService := wire.ProvideReportService(reportRepository, postRepository, commentRepository, messageRepository, userRepository, systemNotificationRepository, pushService, transactionManager, operationLogService, config) reportService := wire.ProvideReportService(reportRepository, postRepository, commentRepository, messageRepository, userRepository, systemNotificationRepository, pushService, transactionManager, operationLogService, config)

View File

@@ -108,6 +108,8 @@ var (
ErrInvalidCallState = &AppError{Code: "INVALID_CALL_STATE", Message: "通话状态无效"} ErrInvalidCallState = &AppError{Code: "INVALID_CALL_STATE", Message: "通话状态无效"}
ErrCalleeOffline = &AppError{Code: "CALLEE_OFFLINE", Message: "对方不在线"} ErrCalleeOffline = &AppError{Code: "CALLEE_OFFLINE", Message: "对方不在线"}
ErrCallAlreadyAnswered = &AppError{Code: "CALL_ALREADY_ANSWERED", Message: "通话已在其他设备应答"} ErrCallAlreadyAnswered = &AppError{Code: "CALL_ALREADY_ANSWERED", Message: "通话已在其他设备应答"}
ErrGroupCallNotSupported = &AppError{Code: "GROUP_CALL_NOT_SUPPORTED", Message: "群组通话不支持"}
ErrMaxParticipantsReached = &AppError{Code: "MAX_PARTICIPANTS_REACHED", Message: "通话人数已达上限"}
// 消息相关错误 // 消息相关错误
ErrConversationNotFound = &AppError{Code: "CONVERSATION_NOT_FOUND", Message: "会话不存在或无权限"} ErrConversationNotFound = &AppError{Code: "CONVERSATION_NOT_FOUND", Message: "会话不存在或无权限"}

View File

@@ -62,19 +62,14 @@ func (h *GradeHandler) ListGrades(c *gin.Context) {
response.Unauthorized(c, "") response.Unauthorized(c, "")
return return
} }
semester := c.Query("semester")
if semester == "" {
response.BadRequest(c, "semester is required")
return
}
grades, err := h.gradeRepo.ListByUserAndSemester(userID, semester) grades, err := h.gradeRepo.ListByUser(userID)
if err != nil { if err != nil {
response.HandleError(c, err, "failed to list grades") response.HandleError(c, err, "failed to list grades")
return return
} }
gpaSummary, _ := h.gradeRepo.GetGpaSummary(userID, semester) gpaSummary, _ := h.gradeRepo.GetLatestGpaSummary(userID)
response.Success(c, gin.H{ response.Success(c, gin.H{
"grades": grades, "grades": grades,

View File

@@ -127,10 +127,21 @@ func (h *LiveKitHandler) HandleWebhook(c *gin.Context) {
case "room_finished": case "room_finished":
h.handleRoomFinished(c.Request.Context(), event) h.handleRoomFinished(c.Request.Context(), event)
case "participant_left": case "participant_left":
identity := event.Participant.GetIdentity()
roomName := event.Room.GetName()
h.logger.Info("participant left room", h.logger.Info("participant left room",
zap.String("room", event.Room.GetName()), zap.String("room", roomName),
zap.String("identity", event.Participant.GetIdentity()), zap.String("identity", identity),
) )
if identity != "" && roomName != "" {
if err := h.callService.ParticipantLeave(c.Request.Context(), roomName, identity, "left_livekit"); err != nil {
h.logger.Warn("failed to handle participant_left",
zap.String("room", roomName),
zap.String("identity", identity),
zap.Error(err),
)
}
}
} }
c.JSON(http.StatusOK, gin.H{"status": "ok"}) c.JSON(http.StatusOK, gin.H{"status": "ok"})

View File

@@ -377,6 +377,8 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) {
// 通话信令 // 通话信令
case "call_invite": case "call_invite":
h.handleCallInvite(ctx, client, msg.Payload) h.handleCallInvite(ctx, client, msg.Payload)
case "call_group_invite":
h.handleCallGroupInvite(ctx, client, msg.Payload)
case "call_answer": case "call_answer":
h.handleCallAnswer(ctx, client, msg.Payload) h.handleCallAnswer(ctx, client, msg.Payload)
case "call_reject": case "call_reject":
@@ -389,6 +391,8 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) {
h.handleCallEnd(ctx, client, msg.Payload) h.handleCallEnd(ctx, client, msg.Payload)
case "call_mute": case "call_mute":
h.handleCallMute(ctx, client, msg.Payload) h.handleCallMute(ctx, client, msg.Payload)
case "call_participant_join":
h.handleCallParticipantJoin(ctx, client, msg.Payload)
default: default:
h.publisher.SendError(client, "unknown_type", fmt.Sprintf("unknown message type: %s", msg.Type)) h.publisher.SendError(client, "unknown_type", fmt.Sprintf("unknown message type: %s", msg.Type))
} }
@@ -648,6 +652,61 @@ func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, pay
} }
} }
// handleCallGroupInvite 处理群组通话邀请
func (h *WSHandler) handleCallGroupInvite(ctx context.Context, client *ws.Client, payload json.RawMessage) {
if !h.isVerified(ctx, client) {
return
}
var req struct {
GroupID string `json:"group_id"`
ConversationID string `json:"conversation_id"`
MediaType string `json:"call_type"`
}
if err := json.Unmarshal(payload, &req); err != nil {
h.publisher.SendError(client, "parse_error", "invalid call_group_invite format")
return
}
if req.GroupID == "" || req.ConversationID == "" {
h.publisher.SendError(client, "invalid_params", "group_id and conversation_id are required")
return
}
mediaType := req.MediaType
if mediaType != "video" {
mediaType = "voice"
}
call, err := h.callService.GroupInvite(ctx, client.UserID, req.GroupID, req.ConversationID, mediaType)
if err != nil {
zap.L().Warn("Failed to invite group call",
zap.String("caller_id", client.UserID),
zap.String("group_id", req.GroupID),
zap.Error(err),
)
h.publisher.SendError(client, "call_group_invite_failed", err.Error())
return
}
resp := ws.ResponseMessage{
EventID: h.publisher.NextID(),
Type: "call_invited",
TS: time.Now().UnixMilli(),
Payload: map[string]any{
"call_id": call.ID,
"conversation_id": req.ConversationID,
"group_id": req.GroupID,
"call_type": "group",
"lifetime": service.CallLifetimeMs,
},
}
data, _ := json.Marshal(resp)
select {
case <-client.Quit:
case client.Send <- data:
}
}
// handleCallAnswer 处理接听 // handleCallAnswer 处理接听
func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, payload json.RawMessage) { func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, payload json.RawMessage) {
var req struct { var req struct {
@@ -778,3 +837,18 @@ func (h *WSHandler) handleCallMute(ctx context.Context, client *ws.Client, paylo
h.publisher.SendError(client, "call_mute_failed", err.Error()) h.publisher.SendError(client, "call_mute_failed", err.Error())
} }
} }
// handleCallParticipantJoin 处理群组通话参与者加入
func (h *WSHandler) handleCallParticipantJoin(ctx context.Context, client *ws.Client, payload json.RawMessage) {
var req struct {
CallID string `json:"call_id"`
}
if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" {
h.publisher.SendError(client, "invalid_params", "call_id is required")
return
}
if err := h.callService.ParticipantJoin(ctx, req.CallID, client.UserID); err != nil {
h.publisher.SendError(client, "call_participant_join_failed", err.Error())
}
}

View File

@@ -43,7 +43,9 @@ type CallSession struct {
GroupID *string `gorm:"index" json:"group_id,omitempty"` // 群组ID群聊时使用 GroupID *string `gorm:"index" json:"group_id,omitempty"` // 群组ID群聊时使用
CallerID string `gorm:"column:caller_id;type:varchar(50);index;not null" json:"caller_id"` // 发起者ID CallerID string `gorm:"column:caller_id;type:varchar(50);index;not null" json:"caller_id"` // 发起者ID
CallType CallType `gorm:"type:varchar(20);not null" json:"call_type"` // 通话类型 CallType CallType `gorm:"type:varchar(20);not null" json:"call_type"` // 通话类型
MediaType string `gorm:"type:varchar(20);default:'voice'" json:"media_type"` // 媒体类型: voice/video
Status CallStatus `gorm:"type:varchar(20);default:'calling'" json:"status"` // 通话状态 Status CallStatus `gorm:"type:varchar(20);default:'calling'" json:"status"` // 通话状态
EndedBy *string `gorm:"type:varchar(50)" json:"ended_by,omitempty"` // 结束通话的用户ID
StartedAt *time.Time `json:"started_at,omitempty"` // 接通时间 StartedAt *time.Time `json:"started_at,omitempty"` // 接通时间
EndedAt *time.Time `json:"ended_at,omitempty"` // 结束时间 EndedAt *time.Time `json:"ended_at,omitempty"` // 结束时间
Duration int64 `gorm:"default:0" json:"duration"` // 通话时长(秒) Duration int64 `gorm:"default:0" json:"duration"` // 通话时长(秒)

View File

@@ -9,11 +9,12 @@ import (
) )
type GradeRepository interface { type GradeRepository interface {
ListByUserAndSemester(userID, semester string) ([]*model.Grade, error) ListByUser(userID string) ([]*model.Grade, error)
DeleteByUserAndSemester(ctx context.Context, userID, semester string) error DeleteByUser(ctx context.Context, userID string) error
BatchCreate(ctx context.Context, grades []*model.Grade) error BatchCreate(ctx context.Context, grades []*model.Grade) error
GetGpaSummary(userID, semester string) (*model.GpaSummary, error) GetLatestGpaSummary(userID string) (*model.GpaSummary, error)
SaveGpaSummary(ctx context.Context, summary *model.GpaSummary) error SaveGpaSummary(ctx context.Context, summary *model.GpaSummary) error
DeleteGpaSummaryByUser(ctx context.Context, userID string) error
} }
type gradeRepository struct { type gradeRepository struct {
@@ -24,17 +25,17 @@ func NewGradeRepository(db *gorm.DB) GradeRepository {
return &gradeRepository{db: db} return &gradeRepository{db: db}
} }
func (r *gradeRepository) ListByUserAndSemester(userID, semester string) ([]*model.Grade, error) { func (r *gradeRepository) ListByUser(userID string) ([]*model.Grade, error) {
var grades []*model.Grade var grades []*model.Grade
err := r.db.Where("user_id = ? AND term = ?", userID, semester). err := r.db.Where("user_id = ?", userID).
Order("created_at ASC"). Order("term DESC, created_at ASC").
Find(&grades).Error Find(&grades).Error
return grades, err return grades, err
} }
func (r *gradeRepository) DeleteByUserAndSemester(ctx context.Context, userID, semester string) error { func (r *gradeRepository) DeleteByUser(ctx context.Context, userID string) error {
return r.db.WithContext(ctx). return r.db.WithContext(ctx).
Where("user_id = ? AND term = ?", userID, semester). Where("user_id = ?", userID).
Delete(&model.Grade{}).Error Delete(&model.Grade{}).Error
} }
@@ -42,9 +43,11 @@ func (r *gradeRepository) BatchCreate(ctx context.Context, grades []*model.Grade
return r.db.WithContext(ctx).CreateInBatches(grades, 100).Error return r.db.WithContext(ctx).CreateInBatches(grades, 100).Error
} }
func (r *gradeRepository) GetGpaSummary(userID, semester string) (*model.GpaSummary, error) { func (r *gradeRepository) GetLatestGpaSummary(userID string) (*model.GpaSummary, error) {
var summary model.GpaSummary var summary model.GpaSummary
err := r.db.Where("user_id = ? AND semester = ?", userID, semester).First(&summary).Error err := r.db.Where("user_id = ?", userID).
Order("updated_at DESC").
First(&summary).Error
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -54,3 +57,9 @@ func (r *gradeRepository) GetGpaSummary(userID, semester string) (*model.GpaSumm
func (r *gradeRepository) SaveGpaSummary(ctx context.Context, summary *model.GpaSummary) error { func (r *gradeRepository) SaveGpaSummary(ctx context.Context, summary *model.GpaSummary) error {
return r.db.WithContext(ctx).Save(summary).Error return r.db.WithContext(ctx).Save(summary).Error
} }
func (r *gradeRepository) DeleteGpaSummaryByUser(ctx context.Context, userID string) error {
return r.db.WithContext(ctx).
Where("user_id = ?", userID).
Delete(&model.GpaSummary{}).Error
}

View File

@@ -9,6 +9,7 @@ import (
"time" "time"
redispkg "with_you/internal/pkg/redis" redispkg "with_you/internal/pkg/redis"
"with_you/internal/pkg/utils"
"with_you/internal/config" "with_you/internal/config"
apperrors "with_you/internal/errors" apperrors "with_you/internal/errors"
@@ -37,12 +38,14 @@ type activeCallRedisData struct {
ID string `json:"id"` ID string `json:"id"`
ConversationID string `json:"conversation_id"` ConversationID string `json:"conversation_id"`
CallerID string `json:"caller_id"` CallerID string `json:"caller_id"`
CalleeID string `json:"callee_id"` CalleeID string `json:"callee_id,omitempty"` // 兼容旧数据1v1时仍有值
CallType string `json:"call_type"` CallType string `json:"call_type"`
Status string `json:"status"` Status string `json:"status"`
MediaType string `json:"media_type"` MediaType string `json:"media_type"`
GroupID string `json:"group_id,omitempty"`
CreatedAt int64 `json:"created_at"` CreatedAt int64 `json:"created_at"`
StartedAt int64 `json:"started_at"` // 0 means nil StartedAt int64 `json:"started_at"` // 0 means nil
ParticipantIDs []string `json:"participant_ids"`
} }
// ActiveCall 内存中的活跃通话 // ActiveCall 内存中的活跃通话
@@ -50,7 +53,8 @@ type ActiveCall struct {
ID string ID string
ConversationID string ConversationID string
CallerID string CallerID string
CalleeID string CalleeID string // 1v1通话的被叫方ID兼容
GroupID string // 群组通话的群组ID
CallType model.CallType CallType model.CallType
Status model.CallStatus Status model.CallStatus
MediaType string // voice 或 video MediaType string // voice 或 video
@@ -66,15 +70,19 @@ type ActiveParticipant struct {
UserID string UserID string
Status model.ParticipantStatus Status model.ParticipantStatus
JoinedAt *time.Time JoinedAt *time.Time
LeftAt *time.Time
} }
// CallService 通话服务接口 // CallService 通话服务接口
type CallService interface { type CallService interface {
Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*ActiveCall, bool, error) Invite(ctx context.Context, callerID, calleeID, conversationID, mediaType string) (*ActiveCall, bool, error)
GroupInvite(ctx context.Context, callerID, groupID, conversationID, mediaType string) (*ActiveCall, error)
Accept(ctx context.Context, callID, userID string) (*ActiveCall, error) Accept(ctx context.Context, callID, userID string) (*ActiveCall, error)
Reject(ctx context.Context, callID, userID string) error Reject(ctx context.Context, callID, userID string) error
Busy(ctx context.Context, callID, userID string) error Busy(ctx context.Context, callID, userID string) error
End(ctx context.Context, callID, userID string, reason string) (*ActiveCall, error) End(ctx context.Context, callID, userID string, reason string) (*ActiveCall, error)
ParticipantJoin(ctx context.Context, callID, userID string) error
ParticipantLeave(ctx context.Context, callID, userID string, reason string) error
SetMuted(ctx context.Context, callID, userID string, muted bool) error SetMuted(ctx context.Context, callID, userID string, muted bool) error
Ready(ctx context.Context, callID, userID string) error Ready(ctx context.Context, callID, userID string) error
GetActiveCall(ctx context.Context, callID string) *ActiveCall GetActiveCall(ctx context.Context, callID string) *ActiveCall
@@ -88,6 +96,7 @@ type callService struct {
config *config.Config config *config.Config
db *gorm.DB db *gorm.DB
redis *redispkg.Client redis *redispkg.Client
groupService GroupService
// 内存中的活跃通话状态 // 内存中的活跃通话状态
activeCalls map[string]*ActiveCall // callID -> ActiveCall activeCalls map[string]*ActiveCall // callID -> ActiveCall
@@ -103,6 +112,7 @@ func NewCallService(
cfg *config.Config, cfg *config.Config,
db *gorm.DB, db *gorm.DB,
redisClient *redispkg.Client, redisClient *redispkg.Client,
groupService GroupService,
) CallService { ) CallService {
svc := &callService{ svc := &callService{
callRepo: callRepo, callRepo: callRepo,
@@ -110,6 +120,7 @@ func NewCallService(
config: cfg, config: cfg,
db: db, db: db,
redis: redisClient, redis: redisClient,
groupService: groupService,
activeCalls: make(map[string]*ActiveCall), activeCalls: make(map[string]*ActiveCall),
activeCallsByUser: make(map[string]map[string]bool), activeCallsByUser: make(map[string]map[string]bool),
activeCallsByConv: make(map[string]string), activeCallsByConv: make(map[string]string),
@@ -124,9 +135,13 @@ func NewCallService(
return svc return svc
} }
// generateCallID 生成通话ID // generateCallID 生成通话ID使用雪花算法与数据库ID一致
func (s *callService) generateCallID() string { func (s *callService) generateCallID() string {
id := time.Now().UnixNano() id, err := utils.GetSnowflake().GenerateID()
if err != nil {
zap.L().Error("Failed to generate snowflake call ID", zap.Error(err))
return strconv.FormatInt(time.Now().UnixNano(), 10)
}
return strconv.FormatInt(id, 10) return strconv.FormatInt(id, 10)
} }
@@ -141,6 +156,12 @@ func (s *callService) redisStoreCall(call *ActiveCall) {
if call.StartedAt != nil { if call.StartedAt != nil {
startedAt = call.StartedAt.UnixMilli() startedAt = call.StartedAt.UnixMilli()
} }
participantIDs := make([]string, 0, len(call.Participants))
for id := range call.Participants {
participantIDs = append(participantIDs, id)
}
data := activeCallRedisData{ data := activeCallRedisData{
ID: call.ID, ID: call.ID,
ConversationID: call.ConversationID, ConversationID: call.ConversationID,
@@ -149,8 +170,10 @@ func (s *callService) redisStoreCall(call *ActiveCall) {
CallType: string(call.CallType), CallType: string(call.CallType),
Status: string(call.Status), Status: string(call.Status),
MediaType: call.MediaType, MediaType: call.MediaType,
GroupID: call.GroupID,
CreatedAt: call.CreatedAt.UnixMilli(), CreatedAt: call.CreatedAt.UnixMilli(),
StartedAt: startedAt, StartedAt: startedAt,
ParticipantIDs: participantIDs,
} }
jsonData, err := json.Marshal(data) jsonData, err := json.Marshal(data)
if err != nil { if err != nil {
@@ -159,13 +182,13 @@ func (s *callService) redisStoreCall(call *ActiveCall) {
} }
callKey := redisCallKeyPrefix + call.ID callKey := redisCallKeyPrefix + call.ID
callerKey := redisCallByUserPrefix + call.CallerID
calleeKey := redisCallByUserPrefix + call.CalleeID
pipe := s.redis.Pipeline() pipe := s.redis.Pipeline()
pipe.Set(ctx, callKey, jsonData, redisCallTTL) pipe.Set(ctx, callKey, jsonData, redisCallTTL)
pipe.Set(ctx, callerKey, call.ID, redisCallTTL) // 为每个参与者建立索引
pipe.Set(ctx, calleeKey, call.ID, redisCallTTL) for userID := range call.Participants {
pipe.Set(ctx, redisCallByUserPrefix+userID, call.ID, redisCallTTL)
}
if _, err := pipe.Exec(ctx); err != nil { if _, err := pipe.Exec(ctx); err != nil {
zap.L().Error("Failed to store call in Redis", zap.String("call_id", call.ID), zap.Error(err)) zap.L().Error("Failed to store call in Redis", zap.String("call_id", call.ID), zap.Error(err))
} }
@@ -180,10 +203,11 @@ func (s *callService) redisRemoveCall(call *ActiveCall) {
keys := []string{ keys := []string{
redisCallKeyPrefix + call.ID, redisCallKeyPrefix + call.ID,
redisCallByUserPrefix + call.CallerID,
redisCallByUserPrefix + call.CalleeID,
redisCallAcceptPrefix + call.ID, redisCallAcceptPrefix + call.ID,
} }
for userID := range call.Participants {
keys = append(keys, redisCallByUserPrefix+userID)
}
if err := s.redis.Del(ctx, keys...); err != nil { if err := s.redis.Del(ctx, keys...); err != nil {
zap.L().Error("Failed to remove call from Redis", zap.String("call_id", call.ID), zap.Error(err)) zap.L().Error("Failed to remove call from Redis", zap.String("call_id", call.ID), zap.Error(err))
} }
@@ -207,18 +231,38 @@ func (s *callService) redisGetCall(callID string) *ActiveCall {
return nil return nil
} }
// 构建参与者列表
participants := make(map[string]*ActiveParticipant)
for _, pid := range data.ParticipantIDs {
status := model.ParticipantStatusInvited
if pid == data.CallerID {
status = model.ParticipantStatusJoined
}
participants[pid] = &ActiveParticipant{
UserID: pid,
Status: status,
}
}
// 兼容旧数据:无 ParticipantIDs 时使用 CallerID/CalleeID
if len(participants) == 0 && data.CalleeID != "" {
participants[data.CallerID] = &ActiveParticipant{
UserID: data.CallerID, Status: model.ParticipantStatusJoined,
}
participants[data.CalleeID] = &ActiveParticipant{
UserID: data.CalleeID, Status: model.ParticipantStatusInvited,
}
}
call := &ActiveCall{ call := &ActiveCall{
ID: data.ID, ID: data.ID,
ConversationID: data.ConversationID, ConversationID: data.ConversationID,
CallerID: data.CallerID, CallerID: data.CallerID,
CalleeID: data.CalleeID, CalleeID: data.CalleeID,
GroupID: data.GroupID,
CallType: model.CallType(data.CallType), CallType: model.CallType(data.CallType),
Status: model.CallStatus(data.Status), Status: model.CallStatus(data.Status),
MediaType: data.MediaType, MediaType: data.MediaType,
Participants: map[string]*ActiveParticipant{ Participants: participants,
data.CallerID: {UserID: data.CallerID, Status: model.ParticipantStatusJoined},
data.CalleeID: {UserID: data.CalleeID, Status: model.ParticipantStatusInvited},
},
} }
call.CreatedAt = time.UnixMilli(data.CreatedAt) call.CreatedAt = time.UnixMilli(data.CreatedAt)
@@ -226,10 +270,11 @@ func (s *callService) redisGetCall(callID string) *ActiveCall {
t := time.UnixMilli(data.StartedAt) t := time.UnixMilli(data.StartedAt)
call.StartedAt = &t call.StartedAt = &t
// 如果已经开始,被叫方也已加入 // 如果已经开始,被叫方也已加入
call.Participants[data.CalleeID] = &ActiveParticipant{ if data.CalleeID != "" {
UserID: data.CalleeID, if p, ok := call.Participants[data.CalleeID]; ok {
Status: model.ParticipantStatusJoined, p.Status = model.ParticipantStatusJoined
JoinedAt: call.StartedAt, p.JoinedAt = call.StartedAt
}
} }
} }
@@ -243,10 +288,9 @@ func (s *callService) redisRefreshTTL(call *ActiveCall) {
} }
ctx := context.Background() ctx := context.Background()
keys := []string{ keys := []string{redisCallKeyPrefix + call.ID}
redisCallKeyPrefix + call.ID, for userID := range call.Participants {
redisCallByUserPrefix + call.CallerID, keys = append(keys, redisCallByUserPrefix+userID)
redisCallByUserPrefix + call.CalleeID,
} }
for _, key := range keys { for _, key := range keys {
_, _ = s.redis.Expire(ctx, key, redisCallTTL) _, _ = s.redis.Expire(ctx, key, redisCallTTL)
@@ -506,7 +550,7 @@ func (s *callService) Reject(ctx context.Context, callID, userID string) error {
s.redisRemoveCall(call) s.redisRemoveCall(call)
// 保存到数据库作为历史记录 // 保存到数据库作为历史记录
s.saveCallHistory(call, model.CallStatusRejected, 0) s.saveCallHistory(call, model.CallStatusRejected, 0, userID)
// 通知拨打方 // 通知拨打方
s.hub.PublishToUserOnline(call.CallerID, "call_rejected", map[string]any{ s.hub.PublishToUserOnline(call.CallerID, "call_rejected", map[string]any{
@@ -543,7 +587,7 @@ func (s *callService) Busy(ctx context.Context, callID, userID string) error {
s.redisRemoveCall(call) s.redisRemoveCall(call)
// 保存到数据库作为历史记录 // 保存到数据库作为历史记录
s.saveCallHistory(call, model.CallStatusMissed, 0) s.saveCallHistory(call, model.CallStatusMissed, 0, "")
// 通知拨打方 // 通知拨打方
s.hub.PublishToUserOnline(call.CallerID, "call_busy", map[string]any{ s.hub.PublishToUserOnline(call.CallerID, "call_busy", map[string]any{
@@ -602,7 +646,7 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str
s.redisRemoveCall(call) s.redisRemoveCall(call)
// 保存到数据库作为历史记录 // 保存到数据库作为历史记录
s.saveCallHistory(call, endStatus, duration) s.saveCallHistory(call, endStatus, duration, userID)
// 通知其他参与者 // 通知其他参与者
for pUserID := range call.Participants { for pUserID := range call.Participants {
@@ -654,6 +698,174 @@ func (s *callService) GetActiveCall(ctx context.Context, callID string) *ActiveC
return s.getActiveCall(callID) return s.getActiveCall(callID)
} }
// GroupInvite 发起群组通话
func (s *callService) GroupInvite(ctx context.Context, callerID, groupID, conversationID, mediaType string) (*ActiveCall, error) {
if s.groupService == nil {
return nil, apperrors.ErrGroupCallNotSupported
}
// 验证调用者是群组成员
member, err := s.groupService.GetMember(groupID, callerID)
if err != nil {
return nil, apperrors.ErrNotGroupMember
}
if member == nil {
return nil, apperrors.ErrNotGroupMember
}
// 检查会话是否有活跃通话
if active := s.getActiveCallByConversation(conversationID); active != nil {
return nil, apperrors.Wrap(fmt.Errorf("call %s", active.ID), apperrors.ErrCallInProgress)
}
// 获取群组成员
members, _, err := s.groupService.GetMembers(groupID, 1, 50)
if err != nil {
return nil, apperrors.Wrap(err, apperrors.ErrInternal)
}
now := time.Now()
callID := s.generateCallID()
// 构建参与者列表
participants := map[string]*ActiveParticipant{
callerID: {UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now},
}
for _, m := range members {
if m.UserID != callerID {
participants[m.UserID] = &ActiveParticipant{
UserID: m.UserID,
Status: model.ParticipantStatusInvited,
}
}
}
call := &ActiveCall{
ID: callID,
ConversationID: conversationID,
CallerID: callerID,
GroupID: groupID,
CallType: model.CallTypeGroup,
Status: model.CallStatusCalling,
MediaType: mediaType,
CreatedAt: now,
Participants: participants,
}
s.storeActiveCall(call)
s.redisStoreCall(call)
// 通知所有在线成员
payload := map[string]any{
"call_id": call.ID,
"conversation_id": conversationID,
"caller_id": callerID,
"call_type": "group",
"media_type": mediaType,
"group_id": groupID,
"created_at": now.UnixMilli(),
"lifetime": CallLifetimeMs,
}
for pID := range participants {
if pID != callerID {
s.hub.PublishToUserOnline(pID, "call_incoming", payload)
}
}
return call, nil
}
// ParticipantJoin 参与者加入通话
func (s *callService) ParticipantJoin(ctx context.Context, callID, userID string) error {
call := s.getActiveCall(callID)
if call == nil {
return apperrors.ErrCallNotFound
}
s.mu.Lock()
p, ok := call.Participants[userID]
if !ok {
s.mu.Unlock()
return apperrors.ErrNotCallParticipant
}
now := time.Now()
p.Status = model.ParticipantStatusJoined
p.JoinedAt = &now
// 如果是第一个被叫方加入,标记通话已接通
if call.Status == model.CallStatusCalling {
call.Status = model.CallStatusConnected
call.StartedAt = &now
}
s.mu.Unlock()
s.redisStoreCall(call)
// 通知其他参与者
for pID := range call.Participants {
if pID != userID {
s.hub.PublishToUserOnline(pID, "call_participant_joined", map[string]any{
"call_id": callID,
"user_id": userID,
})
}
}
return nil
}
// ParticipantLeave 参与者离开通话
func (s *callService) ParticipantLeave(ctx context.Context, callID, userID string, reason string) error {
call := s.getActiveCall(callID)
if call == nil {
return apperrors.ErrCallNotFound
}
s.mu.Lock()
p, ok := call.Participants[userID]
if !ok {
s.mu.Unlock()
return apperrors.ErrNotCallParticipant
}
now := time.Now()
p.Status = model.ParticipantStatusLeft
p.LeftAt = &now
// 计算剩余活跃参与者数量
activeCount := 0
for _, pp := range call.Participants {
if pp.Status == model.ParticipantStatusJoined {
activeCount++
}
}
shouldEnd := false
if call.CallType == model.CallTypeGroup && activeCount <= 1 {
shouldEnd = true
} else if call.CallType == model.CallTypePrivate {
shouldEnd = true
}
s.mu.Unlock()
// 通知其他参与者
for pID := range call.Participants {
if pID != userID {
s.hub.PublishToUserOnline(pID, "call_participant_left", map[string]any{
"call_id": callID,
"user_id": userID,
"reason": reason,
})
}
}
if shouldEnd {
_, _ = s.End(ctx, callID, userID, reason)
}
return nil
}
// Ready 标记参与者已加入 LiveKit 房间 // Ready 标记参与者已加入 LiveKit 房间
func (s *callService) Ready(ctx context.Context, callID, userID string) error { func (s *callService) Ready(ctx context.Context, callID, userID string) error {
call := s.getActiveCall(callID) call := s.getActiveCall(callID)
@@ -686,16 +898,23 @@ func (s *callService) GetCallHistory(ctx context.Context, userID string, page, p
// saveCallHistory 保存通话历史到数据库 // saveCallHistory 保存通话历史到数据库
func (s *callService) saveCallHistory(call *ActiveCall, status model.CallStatus, duration int64) { func (s *callService) saveCallHistory(call *ActiveCall, status model.CallStatus, duration int64, endedBy string) {
ctx := context.Background() ctx := context.Background()
now := time.Now() now := time.Now()
var endedByPtr *string
if endedBy != "" {
endedByPtr = &endedBy
}
dbCall := &model.CallSession{ dbCall := &model.CallSession{
ID: call.ID, ID: call.ID,
ConversationID: call.ConversationID, ConversationID: call.ConversationID,
CallerID: call.CallerID, CallerID: call.CallerID,
CallType: call.CallType, CallType: call.CallType,
MediaType: call.MediaType,
Status: status, Status: status,
EndedBy: endedByPtr,
StartedAt: call.StartedAt, StartedAt: call.StartedAt,
EndedAt: &now, EndedAt: &now,
Duration: duration, Duration: duration,

View File

@@ -81,7 +81,7 @@ func (s *gradeSyncService) SyncUserGrades(ctx context.Context, userID string, re
} }
grades := s.convertGrades(userID, &resultData) grades := s.convertGrades(userID, &resultData)
if err := s.saveGrades(ctx, userID, resultData.Semester, grades); err != nil { if err := s.saveGrades(ctx, userID, grades); err != nil {
s.logger.Error("failed to save grades", s.logger.Error("failed to save grades",
zap.String("user_id", userID), zap.String("user_id", userID),
zap.Error(err)) zap.Error(err))
@@ -89,7 +89,12 @@ func (s *gradeSyncService) SyncUserGrades(ctx context.Context, userID string, re
} }
if resultData.GpaSummary != nil { if resultData.GpaSummary != nil {
summary := s.convertGpaSummary(userID, resultData.Semester, resultData.GpaSummary) if err := s.gradeRepo.DeleteGpaSummaryByUser(ctx, userID); err != nil {
s.logger.Warn("failed to delete old gpa summary",
zap.String("user_id", userID),
zap.Error(err))
}
summary := s.convertGpaSummary(userID, resultData.GpaSummary)
if err := s.gradeRepo.SaveGpaSummary(ctx, summary); err != nil { if err := s.gradeRepo.SaveGpaSummary(ctx, summary); err != nil {
s.logger.Warn("failed to save gpa summary", s.logger.Warn("failed to save gpa summary",
zap.String("user_id", userID), zap.String("user_id", userID),
@@ -128,11 +133,11 @@ func (s *gradeSyncService) convertGrades(userID string, data *pb.GradesResultDat
return grades return grades
} }
func (s *gradeSyncService) convertGpaSummary(userID, semester string, gpa *pb.GpaSummary) *model.GpaSummary { func (s *gradeSyncService) convertGpaSummary(userID string, gpa *pb.GpaSummary) *model.GpaSummary {
return &model.GpaSummary{ return &model.GpaSummary{
ID: uuid.New().String(), ID: uuid.New().String(),
UserID: userID, UserID: userID,
Semester: semester, Semester: gpa.Semesters,
AverageGpa: gpa.AverageGpa, AverageGpa: gpa.AverageGpa,
Rank: gpa.Rank, Rank: gpa.Rank,
ExamTotalGpa: gpa.ExamTotalGpa, ExamTotalGpa: gpa.ExamTotalGpa,
@@ -142,8 +147,8 @@ func (s *gradeSyncService) convertGpaSummary(userID, semester string, gpa *pb.Gp
} }
} }
func (s *gradeSyncService) saveGrades(ctx context.Context, userID, semester string, grades []*model.Grade) error { func (s *gradeSyncService) saveGrades(ctx context.Context, userID string, grades []*model.Grade) error {
if err := s.gradeRepo.DeleteByUserAndSemester(ctx, userID, semester); err != nil { if err := s.gradeRepo.DeleteByUser(ctx, userID); err != nil {
return fmt.Errorf("failed to delete old grades: %w", err) return fmt.Errorf("failed to delete old grades: %w", err)
} }
if len(grades) == 0 { if len(grades) == 0 {

View File

@@ -17,9 +17,15 @@ import (
"google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/encoding/protojson"
) )
// RoomConfig 房间配置选项
type RoomConfig struct {
MaxParticipants uint32
EmptyTimeout uint32 // 秒
}
// LiveKitService LiveKit 令牌生成和 Webhook 处理服务 // LiveKitService LiveKit 令牌生成和 Webhook 处理服务
type LiveKitService interface { type LiveKitService interface {
GenerateToken(roomName, userID string, canPublish, canSubscribe bool) (string, error) GenerateToken(roomName, userID string, canPublish, canSubscribe bool, roomCfg ...RoomConfig) (string, error)
ProcessWebhook(ctx context.Context, body []byte, authHeader string) (*livekit.WebhookEvent, error) ProcessWebhook(ctx context.Context, body []byte, authHeader string) (*livekit.WebhookEvent, error)
} }
@@ -37,7 +43,7 @@ func NewLiveKitService(cfg *config.Config, logger *zap.Logger) LiveKitService {
} }
// GenerateToken 为指定 room 和 user 生成 LiveKit 访问令牌 // GenerateToken 为指定 room 和 user 生成 LiveKit 访问令牌
func (s *liveKitService) GenerateToken(roomName, userID string, canPublish, canSubscribe bool) (string, error) { func (s *liveKitService) GenerateToken(roomName, userID string, canPublish, canSubscribe bool, roomCfg ...RoomConfig) (string, error) {
if !s.config.Enabled { if !s.config.Enabled {
return "", fmt.Errorf("livekit is not enabled") return "", fmt.Errorf("livekit is not enabled")
} }
@@ -57,8 +63,9 @@ func (s *liveKitService) GenerateToken(roomName, userID string, canPublish, canS
Room: roomName, Room: roomName,
CanPublish: &canPublish, CanPublish: &canPublish,
CanSubscribe: &canSubscribe, CanSubscribe: &canSubscribe,
CanPublishData: &canSubscribe, CanPublishData: &canPublish,
} }
at.SetVideoGrant(grant) at.SetVideoGrant(grant)
token, err := at.ToJWT() token, err := at.ToJWT()

View File

@@ -460,8 +460,9 @@ func ProvideCallService(
cfg *config.Config, cfg *config.Config,
db *gorm.DB, db *gorm.DB,
redisClient *redis.Client, redisClient *redis.Client,
groupService service.GroupService,
) service.CallService { ) service.CallService {
return service.NewCallService(callRepo, publisher, cfg, db, redisClient) return service.NewCallService(callRepo, publisher, cfg, db, redisClient, groupService)
} }
// ProvideLiveKitService 提供 LiveKit 服务 // ProvideLiveKitService 提供 LiveKit 服务