diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index 14198b8..1e582fd 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -137,7 +137,7 @@ func InitializeApp() (*App, error) { materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository) materialHandler := handler.NewMaterialHandler(materialService) 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) reportRepository := repository.NewReportRepository(db) reportService := wire.ProvideReportService(reportRepository, postRepository, commentRepository, messageRepository, userRepository, systemNotificationRepository, pushService, transactionManager, operationLogService, config) diff --git a/internal/errors/app_errors.go b/internal/errors/app_errors.go index 5390ee5..bea87e8 100644 --- a/internal/errors/app_errors.go +++ b/internal/errors/app_errors.go @@ -108,6 +108,8 @@ var ( ErrInvalidCallState = &AppError{Code: "INVALID_CALL_STATE", Message: "通话状态无效"} ErrCalleeOffline = &AppError{Code: "CALLEE_OFFLINE", 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: "会话不存在或无权限"} diff --git a/internal/handler/grade_handler.go b/internal/handler/grade_handler.go index badaee1..8e5e442 100644 --- a/internal/handler/grade_handler.go +++ b/internal/handler/grade_handler.go @@ -62,19 +62,14 @@ func (h *GradeHandler) ListGrades(c *gin.Context) { response.Unauthorized(c, "") 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 { response.HandleError(c, err, "failed to list grades") return } - gpaSummary, _ := h.gradeRepo.GetGpaSummary(userID, semester) + gpaSummary, _ := h.gradeRepo.GetLatestGpaSummary(userID) response.Success(c, gin.H{ "grades": grades, diff --git a/internal/handler/livekit_handler.go b/internal/handler/livekit_handler.go index 1db8e32..fd73976 100644 --- a/internal/handler/livekit_handler.go +++ b/internal/handler/livekit_handler.go @@ -127,10 +127,21 @@ func (h *LiveKitHandler) HandleWebhook(c *gin.Context) { case "room_finished": h.handleRoomFinished(c.Request.Context(), event) case "participant_left": + identity := event.Participant.GetIdentity() + roomName := event.Room.GetName() h.logger.Info("participant left room", - zap.String("room", event.Room.GetName()), - zap.String("identity", event.Participant.GetIdentity()), + zap.String("room", roomName), + 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"}) diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 98be153..d4919a8 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -377,6 +377,8 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) { // 通话信令 case "call_invite": h.handleCallInvite(ctx, client, msg.Payload) + case "call_group_invite": + h.handleCallGroupInvite(ctx, client, msg.Payload) case "call_answer": h.handleCallAnswer(ctx, client, msg.Payload) case "call_reject": @@ -389,6 +391,8 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) { h.handleCallEnd(ctx, client, msg.Payload) case "call_mute": h.handleCallMute(ctx, client, msg.Payload) + case "call_participant_join": + h.handleCallParticipantJoin(ctx, client, msg.Payload) default: 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 处理接听 func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, payload json.RawMessage) { 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()) } } + +// 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()) + } +} diff --git a/internal/model/call.go b/internal/model/call.go index edbe055..860d42d 100644 --- a/internal/model/call.go +++ b/internal/model/call.go @@ -43,7 +43,9 @@ type CallSession struct { 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 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"` // 通话状态 + EndedBy *string `gorm:"type:varchar(50)" json:"ended_by,omitempty"` // 结束通话的用户ID StartedAt *time.Time `json:"started_at,omitempty"` // 接通时间 EndedAt *time.Time `json:"ended_at,omitempty"` // 结束时间 Duration int64 `gorm:"default:0" json:"duration"` // 通话时长（秒） diff --git a/internal/repository/grade_repo.go b/internal/repository/grade_repo.go index 9484f7b..ac3eac8 100644 --- a/internal/repository/grade_repo.go +++ b/internal/repository/grade_repo.go @@ -9,11 +9,12 @@ import ( ) type GradeRepository interface { - ListByUserAndSemester(userID, semester string) ([]*model.Grade, error) - DeleteByUserAndSemester(ctx context.Context, userID, semester string) error + ListByUser(userID string) ([]*model.Grade, error) + DeleteByUser(ctx context.Context, userID string) 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 + DeleteGpaSummaryByUser(ctx context.Context, userID string) error } type gradeRepository struct { @@ -24,17 +25,17 @@ func NewGradeRepository(db *gorm.DB) GradeRepository { 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 - err := r.db.Where("user_id = ? AND term = ?", userID, semester). - Order("created_at ASC"). + err := r.db.Where("user_id = ?", userID). + Order("term DESC, created_at ASC"). Find(&grades).Error 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). - Where("user_id = ? AND term = ?", userID, semester). + Where("user_id = ?", userID). 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 } -func (r *gradeRepository) GetGpaSummary(userID, semester string) (*model.GpaSummary, error) { +func (r *gradeRepository) GetLatestGpaSummary(userID string) (*model.GpaSummary, error) { 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 { 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 { 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 +} diff --git a/internal/service/call_service.go b/internal/service/call_service.go index a4c9145..7788d74 100644 --- a/internal/service/call_service.go +++ b/internal/service/call_service.go @@ -9,6 +9,7 @@ import ( "time" redispkg "with_you/internal/pkg/redis" + "with_you/internal/pkg/utils" "with_you/internal/config" apperrors "with_you/internal/errors" @@ -34,15 +35,17 @@ const ( // activeCallRedisData Redis 中存储的通话数据（精简字段） type activeCallRedisData struct { - ID string `json:"id"` - ConversationID string `json:"conversation_id"` - CallerID string `json:"caller_id"` - CalleeID string `json:"callee_id"` - CallType string `json:"call_type"` - Status string `json:"status"` - MediaType string `json:"media_type"` - CreatedAt int64 `json:"created_at"` - StartedAt int64 `json:"started_at"` // 0 means nil + ID string `json:"id"` + ConversationID string `json:"conversation_id"` + CallerID string `json:"caller_id"` + CalleeID string `json:"callee_id,omitempty"` // 兼容旧数据，1v1时仍有值 + CallType string `json:"call_type"` + Status string `json:"status"` + MediaType string `json:"media_type"` + GroupID string `json:"group_id,omitempty"` + CreatedAt int64 `json:"created_at"` + StartedAt int64 `json:"started_at"` // 0 means nil + ParticipantIDs []string `json:"participant_ids"` } // ActiveCall 内存中的活跃通话 @@ -50,7 +53,8 @@ type ActiveCall struct { ID string ConversationID string CallerID string - CalleeID string + CalleeID string // 1v1通话的被叫方ID（兼容） + GroupID string // 群组通话的群组ID CallType model.CallType Status model.CallStatus MediaType string // voice 或 video @@ -66,15 +70,19 @@ type ActiveParticipant struct { UserID string Status model.ParticipantStatus JoinedAt *time.Time + LeftAt *time.Time } // CallService 通话服务接口 type CallService interface { 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) 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) (*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 Ready(ctx context.Context, callID, userID string) error GetActiveCall(ctx context.Context, callID string) *ActiveCall @@ -83,11 +91,12 @@ type CallService interface { } type callService struct { - callRepo repository.CallRepository - hub ws.MessagePublisher - config *config.Config - db *gorm.DB - redis *redispkg.Client + callRepo repository.CallRepository + hub ws.MessagePublisher + config *config.Config + db *gorm.DB + redis *redispkg.Client + groupService GroupService // 内存中的活跃通话状态 activeCalls map[string]*ActiveCall // callID -> ActiveCall @@ -103,6 +112,7 @@ func NewCallService( cfg *config.Config, db *gorm.DB, redisClient *redispkg.Client, + groupService GroupService, ) CallService { svc := &callService{ callRepo: callRepo, @@ -110,6 +120,7 @@ func NewCallService( config: cfg, db: db, redis: redisClient, + groupService: groupService, activeCalls: make(map[string]*ActiveCall), activeCallsByUser: make(map[string]map[string]bool), activeCallsByConv: make(map[string]string), @@ -124,9 +135,13 @@ func NewCallService( return svc } -// generateCallID 生成通话ID +// generateCallID 生成通话ID（使用雪花算法，与数据库ID一致） 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) } @@ -141,6 +156,12 @@ func (s *callService) redisStoreCall(call *ActiveCall) { if call.StartedAt != nil { startedAt = call.StartedAt.UnixMilli() } + + participantIDs := make([]string, 0, len(call.Participants)) + for id := range call.Participants { + participantIDs = append(participantIDs, id) + } + data := activeCallRedisData{ ID: call.ID, ConversationID: call.ConversationID, @@ -149,8 +170,10 @@ func (s *callService) redisStoreCall(call *ActiveCall) { CallType: string(call.CallType), Status: string(call.Status), MediaType: call.MediaType, + GroupID: call.GroupID, CreatedAt: call.CreatedAt.UnixMilli(), StartedAt: startedAt, + ParticipantIDs: participantIDs, } jsonData, err := json.Marshal(data) if err != nil { @@ -159,13 +182,13 @@ func (s *callService) redisStoreCall(call *ActiveCall) { } callKey := redisCallKeyPrefix + call.ID - callerKey := redisCallByUserPrefix + call.CallerID - calleeKey := redisCallByUserPrefix + call.CalleeID pipe := s.redis.Pipeline() 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 { 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{ redisCallKeyPrefix + call.ID, - redisCallByUserPrefix + call.CallerID, - redisCallByUserPrefix + call.CalleeID, redisCallAcceptPrefix + call.ID, } + for userID := range call.Participants { + keys = append(keys, redisCallByUserPrefix+userID) + } 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)) } @@ -207,18 +231,38 @@ func (s *callService) redisGetCall(callID string) *ActiveCall { 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{ ID: data.ID, ConversationID: data.ConversationID, CallerID: data.CallerID, CalleeID: data.CalleeID, + GroupID: data.GroupID, CallType: model.CallType(data.CallType), Status: model.CallStatus(data.Status), MediaType: data.MediaType, - Participants: map[string]*ActiveParticipant{ - data.CallerID: {UserID: data.CallerID, Status: model.ParticipantStatusJoined}, - data.CalleeID: {UserID: data.CalleeID, Status: model.ParticipantStatusInvited}, - }, + Participants: participants, } call.CreatedAt = time.UnixMilli(data.CreatedAt) @@ -226,10 +270,11 @@ func (s *callService) redisGetCall(callID string) *ActiveCall { t := time.UnixMilli(data.StartedAt) call.StartedAt = &t // 如果已经开始，被叫方也已加入 - call.Participants[data.CalleeID] = &ActiveParticipant{ - UserID: data.CalleeID, - Status: model.ParticipantStatusJoined, - JoinedAt: call.StartedAt, + if data.CalleeID != "" { + if p, ok := call.Participants[data.CalleeID]; ok { + p.Status = model.ParticipantStatusJoined + p.JoinedAt = call.StartedAt + } } } @@ -243,10 +288,9 @@ func (s *callService) redisRefreshTTL(call *ActiveCall) { } ctx := context.Background() - keys := []string{ - redisCallKeyPrefix + call.ID, - redisCallByUserPrefix + call.CallerID, - redisCallByUserPrefix + call.CalleeID, + keys := []string{redisCallKeyPrefix + call.ID} + for userID := range call.Participants { + keys = append(keys, redisCallByUserPrefix+userID) } for _, key := range keys { _, _ = 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.saveCallHistory(call, model.CallStatusRejected, 0) + s.saveCallHistory(call, model.CallStatusRejected, 0, userID) // 通知拨打方 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.saveCallHistory(call, model.CallStatusMissed, 0) + s.saveCallHistory(call, model.CallStatusMissed, 0, "") // 通知拨打方 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.saveCallHistory(call, endStatus, duration) + s.saveCallHistory(call, endStatus, duration, userID) // 通知其他参与者 for pUserID := range call.Participants { @@ -654,6 +698,174 @@ func (s *callService) GetActiveCall(ctx context.Context, callID string) *ActiveC 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 房间 func (s *callService) Ready(ctx context.Context, callID, userID string) error { call := s.getActiveCall(callID) @@ -686,16 +898,23 @@ func (s *callService) GetCallHistory(ctx context.Context, userID string, page, p // 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() now := time.Now() + var endedByPtr *string + if endedBy != "" { + endedByPtr = &endedBy + } + dbCall := &model.CallSession{ ID: call.ID, ConversationID: call.ConversationID, CallerID: call.CallerID, CallType: call.CallType, + MediaType: call.MediaType, Status: status, + EndedBy: endedByPtr, StartedAt: call.StartedAt, EndedAt: &now, Duration: duration, diff --git a/internal/service/grade_sync_service.go b/internal/service/grade_sync_service.go index bcff11c..a3082cb 100644 --- a/internal/service/grade_sync_service.go +++ b/internal/service/grade_sync_service.go @@ -81,7 +81,7 @@ func (s *gradeSyncService) SyncUserGrades(ctx context.Context, userID string, re } 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", zap.String("user_id", userID), zap.Error(err)) @@ -89,7 +89,12 @@ func (s *gradeSyncService) SyncUserGrades(ctx context.Context, userID string, re } 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 { s.logger.Warn("failed to save gpa summary", zap.String("user_id", userID), @@ -128,11 +133,11 @@ func (s *gradeSyncService) convertGrades(userID string, data *pb.GradesResultDat 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{ ID: uuid.New().String(), UserID: userID, - Semester: semester, + Semester: gpa.Semesters, AverageGpa: gpa.AverageGpa, Rank: gpa.Rank, 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 { - if err := s.gradeRepo.DeleteByUserAndSemester(ctx, userID, semester); err != nil { +func (s *gradeSyncService) saveGrades(ctx context.Context, userID string, grades []*model.Grade) error { + if err := s.gradeRepo.DeleteByUser(ctx, userID); err != nil { return fmt.Errorf("failed to delete old grades: %w", err) } if len(grades) == 0 { diff --git a/internal/service/livekit_service.go b/internal/service/livekit_service.go index 711963e..a86bf9d 100644 --- a/internal/service/livekit_service.go +++ b/internal/service/livekit_service.go @@ -17,9 +17,15 @@ import ( "google.golang.org/protobuf/encoding/protojson" ) +// RoomConfig 房间配置选项 +type RoomConfig struct { + MaxParticipants uint32 + EmptyTimeout uint32 // 秒 +} + // LiveKitService LiveKit 令牌生成和 Webhook 处理服务 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) } @@ -37,7 +43,7 @@ func NewLiveKitService(cfg *config.Config, logger *zap.Logger) LiveKitService { } // 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 { return "", fmt.Errorf("livekit is not enabled") } @@ -57,8 +63,9 @@ func (s *liveKitService) GenerateToken(roomName, userID string, canPublish, canS Room: roomName, CanPublish: &canPublish, CanSubscribe: &canSubscribe, - CanPublishData: &canSubscribe, + CanPublishData: &canPublish, } + at.SetVideoGrant(grant) token, err := at.ToJWT() diff --git a/internal/wire/service.go b/internal/wire/service.go index e4c57d2..df80b20 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -460,8 +460,9 @@ func ProvideCallService( cfg *config.Config, db *gorm.DB, redisClient *redis.Client, + groupService service.GroupService, ) service.CallService { - return service.NewCallService(callRepo, publisher, cfg, db, redisClient) + return service.NewCallService(callRepo, publisher, cfg, db, redisClient, groupService) } // ProvideLiveKitService 提供 LiveKit 服务