930 lines
37 KiB
Go
930 lines
37 KiB
Go
// Package service 提供 meeting 模块的业务逻辑
|
||
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/echochat/backend/app/constants"
|
||
"github.com/echochat/backend/app/meeting/dao"
|
||
"github.com/echochat/backend/app/meeting/model"
|
||
"github.com/echochat/backend/pkg/logs"
|
||
"github.com/redis/go-redis/v9"
|
||
"go.uber.org/zap"
|
||
)
|
||
|
||
// MeetingSignalService WS 信令事件业务处理
|
||
// Task 6 落地：处理设计 §6.3 的 11 个 meeting.* 事件，分 3 组：
|
||
// - 房间组（meeting.room.join/leave）：绑定 WS 连接 ↔ roomCode，辅助断连清理
|
||
// - 成员组（meeting.member.state.changed）：麦/视频状态变更 + host 静音他人
|
||
// - 媒体组（meeting.transport.*/produce.*/consume.*/producer.close）：
|
||
// 对接 MediaOrchestrator，实现 mediasoup signaling 桥接
|
||
//
|
||
// 所有方法统一返回 (ackData, error)：error 非 nil 表示业务失败，由 Handler 映射为 ACK code=-1
|
||
// 广播副作用（如 meeting.member.producer.new）在方法内部通过 broadcaster 发出，不进 ACK
|
||
type MeetingSignalService struct {
|
||
roomDAO *dao.MeetingRoomDAO
|
||
participantDAO *dao.MeetingParticipantDAO
|
||
redis *redis.Client
|
||
|
||
broadcaster *MeetingBroadcaster
|
||
mediaOrchestrator MediaOrchestrator
|
||
lifecycleSvc *MeetingLifecycleService
|
||
}
|
||
|
||
// NewMeetingSignalService 构造 WS 信令服务
|
||
// Task 8 注入 lifecycleSvc：host 掉线 / 重连的生命周期联动
|
||
func NewMeetingSignalService(
|
||
roomDAO *dao.MeetingRoomDAO,
|
||
participantDAO *dao.MeetingParticipantDAO,
|
||
redis *redis.Client,
|
||
broadcaster *MeetingBroadcaster,
|
||
mediaOrchestrator MediaOrchestrator,
|
||
lifecycleSvc *MeetingLifecycleService,
|
||
) *MeetingSignalService {
|
||
return &MeetingSignalService{
|
||
roomDAO: roomDAO,
|
||
participantDAO: participantDAO,
|
||
redis: redis,
|
||
broadcaster: broadcaster,
|
||
mediaOrchestrator: mediaOrchestrator,
|
||
lifecycleSvc: lifecycleSvc,
|
||
}
|
||
}
|
||
|
||
// Redis 资源追踪 key（设计 §九 - 断线清理）
|
||
// Set 内元素格式："transport:{id}" / "producer:{id}" / "consumer:{id}"
|
||
func resourceTrackKey(roomCode string, userID int64) string {
|
||
return fmt.Sprintf("echo:meeting:resource:%s:%d", roomCode, userID)
|
||
}
|
||
|
||
// memberStateKey Redis Hash，保存用户当前的音视频开关状态
|
||
// fields：audio_enabled / video_enabled，值为 "true" / "false"
|
||
// 用途：后入者加入房间时，后端从这里读取每个已有成员的最新状态定向补推 state.changed，
|
||
// 解决"后入者的成员列表里其他人图标一直灰色"的问题
|
||
func memberStateKey(roomCode string, userID int64) string {
|
||
return fmt.Sprintf("echo:meeting:member_state:%s:%d", roomCode, userID)
|
||
}
|
||
|
||
// screenOwnerKey Redis String，记录当前会议正在共享屏幕的人 + Producer ID
|
||
// 值格式："{userId}:{producerId}"；为空 / Nil 表示当前无人共享
|
||
// 单字符串而非 Hash 是因为"会议同时只允许 1 份屏幕共享"，无需多字段
|
||
//
|
||
// 生命周期：
|
||
// - OnProduceStart 收到 appData.screen=true 时 SET（同时广播 meeting.screen.started）
|
||
// - OnProducerClose / cleanupUserResources 命中该 Producer 时 DEL（同时广播 meeting.screen.stopped）
|
||
// - 房间销毁路径（cleanupRoomRedisResidual）一并 Del
|
||
func screenOwnerKey(roomCode string) string {
|
||
return fmt.Sprintf("echo:meeting:screen_owner:%s", roomCode)
|
||
}
|
||
|
||
// formatScreenOwnerValue / parseScreenOwnerValue 统一 screenOwnerKey 的取值格式
|
||
// 复用于 set / parse 双向，避免格式漂移
|
||
func formatScreenOwnerValue(userID int64, producerID string) string {
|
||
return fmt.Sprintf("%d:%s", userID, producerID)
|
||
}
|
||
|
||
func parseScreenOwnerValue(raw string) (int64, string, bool) {
|
||
if raw == "" {
|
||
return 0, "", false
|
||
}
|
||
parts := strings.SplitN(raw, ":", 2)
|
||
if len(parts) != 2 || parts[1] == "" {
|
||
return 0, "", false
|
||
}
|
||
var uid int64
|
||
if _, err := fmt.Sscanf(parts[0], "%d", &uid); err != nil || uid <= 0 {
|
||
return 0, "", false
|
||
}
|
||
return uid, parts[1], true
|
||
}
|
||
|
||
// resourceTTL 单个用户资源追踪集合 TTL
|
||
// 设计：会议期间维持可达即可；若用户长期不活跃由断线清理接管
|
||
// Task 16 Nit：常量已迁出至 constants.MeetingResourceTrackTTLSeconds，此处保留计算式 wrapper 方便调用侧零改动
|
||
var resourceTTL = time.Duration(constants.MeetingResourceTrackTTLSeconds) * time.Second
|
||
|
||
// trackResource 记录用户在会议中持有的媒体资源 ID
|
||
func (s *MeetingSignalService) trackResource(ctx context.Context, roomCode string, userID int64, kind, id string) {
|
||
key := resourceTrackKey(roomCode, userID)
|
||
member := kind + ":" + id
|
||
if err := s.redis.SAdd(ctx, key, member).Err(); err != nil {
|
||
logs.Warn(ctx, "service.meeting_signal_service.trackResource", "追踪媒体资源失败",
|
||
zap.String("key", key), zap.String("member", member), zap.Error(err))
|
||
return
|
||
}
|
||
_ = s.redis.Expire(ctx, key, resourceTTL).Err()
|
||
}
|
||
|
||
// untrackResource 从集合中移除资源 ID（关闭 producer/consumer 时）
|
||
func (s *MeetingSignalService) untrackResource(ctx context.Context, roomCode string, userID int64, kind, id string) {
|
||
key := resourceTrackKey(roomCode, userID)
|
||
member := kind + ":" + id
|
||
_ = s.redis.SRem(ctx, key, member).Err()
|
||
}
|
||
|
||
// assertOwnsResource 校验指定资源（kind: transport/producer/consumer）是否由 userID 在该 roomCode 中创建
|
||
// 背景：WS 信令入口原本只校验"用户是否在会议中"，但客户端上报的 transport_id / producer_id / consumer_id 完全可伪造，
|
||
// 造成任意成员可关闭他人 producer、把 consumer 挂到他人 recv transport 等横向越权（P0-1 / P0-2 审计）
|
||
// 判据：trackResource 记录的 Redis Set 是最权威的归属来源（kind:id 写入 / untrack 时移除）
|
||
// 返回：
|
||
// - nil：归属合法
|
||
// - ErrResourceNotOwned：Set 里不存在该元素（越权尝试）
|
||
// - Redis 错误时同样返回 ErrResourceNotOwned 并记录 Warn，按"fail-closed"保守拒绝，避免服务抖动打开权限口子
|
||
func (s *MeetingSignalService) assertOwnsResource(ctx context.Context, roomCode string, userID int64, kind, id string) error {
|
||
if id == "" {
|
||
return fmt.Errorf("%s_id 不能为空", kind)
|
||
}
|
||
key := resourceTrackKey(roomCode, userID)
|
||
member := kind + ":" + id
|
||
ok, err := s.redis.SIsMember(ctx, key, member).Result()
|
||
if err != nil {
|
||
logs.Warn(ctx, "service.meeting_signal_service.assertOwnsResource", "查询资源归属失败",
|
||
zap.String("key", key), zap.String("member", member), zap.Error(err))
|
||
return ErrResourceNotOwned
|
||
}
|
||
if !ok {
|
||
logs.Warn(ctx, "service.meeting_signal_service.assertOwnsResource", "资源归属不匹配，拒绝越权操作",
|
||
zap.String("room_code", roomCode), zap.Int64("user_id", userID),
|
||
zap.String("kind", kind), zap.String("id", id))
|
||
return ErrResourceNotOwned
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// updateMemberState 将某用户的音视频开关状态持久化到 Redis Hash
|
||
// 非 nil 的字段才写入；audio/video 任意一个 nil 都不碰它，避免误覆盖另一维状态
|
||
// 容错：任何一步失败仅 Warn 日志，不阻断业务流程（前端已发 state.changed 作为权威广播）
|
||
func (s *MeetingSignalService) updateMemberState(ctx context.Context, roomCode string, userID int64, audio, video *bool) {
|
||
if audio == nil && video == nil {
|
||
return
|
||
}
|
||
key := memberStateKey(roomCode, userID)
|
||
values := make([]interface{}, 0, 4)
|
||
if audio != nil {
|
||
values = append(values, "audio_enabled", boolToStr(*audio))
|
||
}
|
||
if video != nil {
|
||
values = append(values, "video_enabled", boolToStr(*video))
|
||
}
|
||
if err := s.redis.HSet(ctx, key, values...).Err(); err != nil {
|
||
logs.Warn(ctx, "service.meeting_signal_service.updateMemberState", "写入成员音视频状态失败",
|
||
zap.String("key", key), zap.Error(err))
|
||
return
|
||
}
|
||
_ = s.redis.Expire(ctx, key, resourceTTL).Err()
|
||
}
|
||
|
||
// loadMemberState 读取某用户的音视频开关状态；Hash 不存在时两个字段都返回 (nil, nil)
|
||
func (s *MeetingSignalService) loadMemberState(ctx context.Context, roomCode string, userID int64) (audio *bool, video *bool) {
|
||
key := memberStateKey(roomCode, userID)
|
||
fields, err := s.redis.HGetAll(ctx, key).Result()
|
||
if err != nil || len(fields) == 0 {
|
||
return nil, nil
|
||
}
|
||
if v, ok := fields["audio_enabled"]; ok {
|
||
b := v == "true"
|
||
audio = &b
|
||
}
|
||
if v, ok := fields["video_enabled"]; ok {
|
||
b := v == "true"
|
||
video = &b
|
||
}
|
||
return audio, video
|
||
}
|
||
|
||
func boolToStr(b bool) string {
|
||
if b {
|
||
return "true"
|
||
}
|
||
return "false"
|
||
}
|
||
|
||
// loadRoomAndParticipant 通用前置校验：拉取房间 + 确认用户是活跃参会者
|
||
// 所有信令事件在进入业务前都要过这一关；返回的 *MeetingRoom 供后续广播使用 roomID
|
||
func (s *MeetingSignalService) loadRoomAndParticipant(ctx context.Context, roomCode string, userID int64) (*model.MeetingRoom, error) {
|
||
room, err := s.roomDAO.GetByCode(ctx, roomCode)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if room == nil {
|
||
return nil, ErrMeetingNotFound
|
||
}
|
||
if room.Status == constants.MeetingStatusEnded {
|
||
return nil, ErrMeetingEnded
|
||
}
|
||
p, err := s.participantDAO.GetByRoomAndUser(ctx, room.ID, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if p == nil || !p.IsActive() {
|
||
return nil, ErrNotInMeeting
|
||
}
|
||
return room, nil
|
||
}
|
||
|
||
// ========== 房间组（2 个 C→S）==========
|
||
|
||
// OnRoomJoin 处理 meeting.room.join 事件
|
||
// 语义：客户端 REST 加入会议成功后，通过 WS 宣告在线；服务端记录 userID ↔ roomCode 映射
|
||
// 副作用（Task 15 增强）：补推"房间内其他用户已有的 producer 列表"给刚 join 的用户，
|
||
// 解决 Mediasoup SFU "后入者错过历史 producer.new"的经典问题
|
||
func (s *MeetingSignalService) OnRoomJoin(ctx context.Context, userID int64, roomCode string) error {
|
||
room, err := s.loadRoomAndParticipant(ctx, roomCode, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 触发资源追踪 key 续期（空集合 TTL 续期无副作用）
|
||
key := resourceTrackKey(roomCode, userID)
|
||
_ = s.redis.Expire(ctx, key, resourceTTL).Err()
|
||
|
||
// Task 8：若本次 WS 在线的用户是会议 host，尝试撤销 host 宽限期
|
||
// 本方法幂等：若此前无 host_grace key（正常场景）DEL 返回 0，不产生副作用
|
||
if s.lifecycleSvc != nil && room.HostID == userID {
|
||
s.lifecycleSvc.OnHostReconnect(ctx, roomCode, userID)
|
||
}
|
||
|
||
logs.Info(ctx, "service.meeting_signal_service.OnRoomJoin", "用户宣告 WS 在线",
|
||
zap.String("room_code", roomCode),
|
||
zap.Int64("user_id", userID),
|
||
zap.Int64("room_id", room.ID))
|
||
|
||
// P1-8：pushExistingRoomState 同步化（Task 16）
|
||
// 旧版异步 go 导致 REST `meeting.member.joined` 与 WS `pushExistingRoomState` 并行，
|
||
// 极端场景下新人先收到未知 user_id 的 state.changed 再收到 REST participant（名称头像短暂缺失）。
|
||
// 前端本就在等 room.join 的 ACK，这里同步在 ACK 返回前完成补推，可消除并行窗口。
|
||
// 单次调用仅 O(当前房间 producer 数量) Redis 读 + 若干 WS 发送，P99 <30ms，不影响 ACK 体验。
|
||
s.pushExistingRoomState(ctx, room.ID, room.RoomCode, userID)
|
||
|
||
return nil
|
||
}
|
||
|
||
// pushExistingRoomState 向刚加入者定向推送房间的历史媒体状态，分两件事：
|
||
// 1. producer 列表：复用 meeting.member.producer.new 事件语义，前端自动发起 consume
|
||
// 2. 成员 audio/video 开关：复用 meeting.member.state.changed 事件语义，前端更新成员面板图标
|
||
//
|
||
// 容错：任何单步失败仅 Warn 日志，不中断流程
|
||
func (s *MeetingSignalService) pushExistingRoomState(ctx context.Context, roomID int64, roomCode string, userID int64) {
|
||
s.pushExistingProducers(ctx, roomID, roomCode, userID)
|
||
s.pushExistingMemberStates(ctx, roomID, roomCode, userID)
|
||
s.pushExistingScreenShare(ctx, roomCode, userID)
|
||
}
|
||
|
||
// pushExistingScreenShare 后入者定向补推当前屏幕共享状态（Phase 3）
|
||
// 复用 meeting.screen.started 事件语义，前端共用同一 handler，无需新事件类型
|
||
// 没有人在共享 / Redis 异常 / 解析失败 都静默 no-op
|
||
func (s *MeetingSignalService) pushExistingScreenShare(ctx context.Context, roomCode string, userID int64) {
|
||
funcName := "service.meeting_signal_service.pushExistingScreenShare"
|
||
|
||
raw, err := s.redis.Get(ctx, screenOwnerKey(roomCode)).Result()
|
||
if err != nil {
|
||
if err != redis.Nil {
|
||
logs.Warn(ctx, funcName, "读取屏幕共享拥有者失败（忽略）",
|
||
zap.String("room_code", roomCode), zap.Error(err))
|
||
}
|
||
return
|
||
}
|
||
ownerUID, producerID, ok := parseScreenOwnerValue(raw)
|
||
if !ok {
|
||
return
|
||
}
|
||
if ownerUID == userID {
|
||
// 自己就是共享者（理论上不会发生：刚 join 不应已是 owner），跳过避免回响
|
||
return
|
||
}
|
||
if err := s.broadcaster.PublishToUser(ctx, userID, constants.MeetingWSEventScreenStarted, map[string]interface{}{
|
||
"room_code": roomCode,
|
||
"user_id": ownerUID,
|
||
"producer_id": producerID,
|
||
"existing": true,
|
||
}); err != nil {
|
||
logs.Warn(ctx, funcName, "定向推送 existing screen.started 失败",
|
||
zap.Int64("to_user", userID),
|
||
zap.Int64("owner_user", ownerUID),
|
||
zap.String("producer_id", producerID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
|
||
// pushExistingProducers 向刚加入者定向推送房间里其他用户已产生的 producer 列表
|
||
// 复用 meeting.member.producer.new 事件语义，前端 _onProducerNew handler 无需改动
|
||
// 容错：任何单步失败仅 Warn 日志，不中断流程
|
||
func (s *MeetingSignalService) pushExistingProducers(ctx context.Context, roomID int64, roomCode string, userID int64) {
|
||
funcName := "service.meeting_signal_service.pushExistingProducers"
|
||
|
||
actives, err := s.participantDAO.ListActiveByRoom(ctx, roomID)
|
||
if err != nil {
|
||
logs.Warn(ctx, funcName, "列出房间活跃参会者失败",
|
||
zap.Int64("room_id", roomID), zap.Error(err))
|
||
return
|
||
}
|
||
|
||
pushCount := 0
|
||
for i := range actives {
|
||
other := &actives[i]
|
||
if other.UserID == userID {
|
||
continue
|
||
}
|
||
otherKey := resourceTrackKey(roomCode, other.UserID)
|
||
members, err := s.redis.SMembers(ctx, otherKey).Result()
|
||
if err != nil {
|
||
logs.Warn(ctx, funcName, "读取他人资源追踪集合失败",
|
||
zap.String("key", otherKey), zap.Error(err))
|
||
continue
|
||
}
|
||
for _, m := range members {
|
||
// 格式："kind:id"；仅关心 producer
|
||
// Nit（代码审查 2026-04-23）：改用 strings.SplitN，避免 rune 解码开销
|
||
parts := strings.SplitN(m, ":", 2)
|
||
if len(parts) != 2 {
|
||
continue
|
||
}
|
||
kind, id := parts[0], parts[1]
|
||
if kind != "producer" || id == "" {
|
||
continue
|
||
}
|
||
if err := s.broadcaster.PublishToUser(ctx, userID, constants.MeetingWSEventMemberProducerNew, map[string]interface{}{
|
||
"room_code": roomCode,
|
||
"user_id": other.UserID,
|
||
"producer_id": id,
|
||
"existing": true,
|
||
}); err != nil {
|
||
logs.Warn(ctx, funcName, "定向推送 existing producer 失败",
|
||
zap.Int64("to_user", userID),
|
||
zap.Int64("owner_user", other.UserID),
|
||
zap.String("producer_id", id),
|
||
zap.Error(err))
|
||
continue
|
||
}
|
||
pushCount++
|
||
}
|
||
}
|
||
|
||
if pushCount > 0 {
|
||
logs.Info(ctx, funcName, "补推已有 producer 完成",
|
||
zap.String("room_code", roomCode),
|
||
zap.Int64("to_user", userID),
|
||
zap.Int("push_count", pushCount))
|
||
}
|
||
}
|
||
|
||
// pushExistingMemberStates 向刚加入者定向推送房间里其他活跃成员当前的音视频开关状态
|
||
// 数据源：updateMemberState 在每次 OnMemberStateChanged 成功后写入的 Redis Hash
|
||
// 前端 _onMemberStateChanged 处理逻辑已存在，收到本事件后会刷新 MemberPanel 图标色彩
|
||
// 容错：读取 Hash 失败、该用户暂无状态记录都直接跳过
|
||
func (s *MeetingSignalService) pushExistingMemberStates(ctx context.Context, roomID int64, roomCode string, userID int64) {
|
||
funcName := "service.meeting_signal_service.pushExistingMemberStates"
|
||
|
||
actives, err := s.participantDAO.ListActiveByRoom(ctx, roomID)
|
||
if err != nil {
|
||
logs.Warn(ctx, funcName, "列出房间活跃参会者失败",
|
||
zap.Int64("room_id", roomID), zap.Error(err))
|
||
return
|
||
}
|
||
|
||
pushCount := 0
|
||
for i := range actives {
|
||
other := &actives[i]
|
||
if other.UserID == userID {
|
||
continue
|
||
}
|
||
audio, video := s.loadMemberState(ctx, roomCode, other.UserID)
|
||
if audio == nil && video == nil {
|
||
continue
|
||
}
|
||
data := map[string]interface{}{
|
||
"room_code": roomCode,
|
||
"user_id": other.UserID,
|
||
"changed_by": other.UserID,
|
||
"existing": true,
|
||
}
|
||
if audio != nil {
|
||
data["audio_enabled"] = *audio
|
||
}
|
||
if video != nil {
|
||
data["video_enabled"] = *video
|
||
}
|
||
if err := s.broadcaster.PublishToUser(ctx, userID, constants.MeetingWSEventMemberStateChange, data); err != nil {
|
||
logs.Warn(ctx, funcName, "定向推送 existing state.changed 失败",
|
||
zap.Int64("to_user", userID),
|
||
zap.Int64("owner_user", other.UserID),
|
||
zap.Error(err))
|
||
continue
|
||
}
|
||
pushCount++
|
||
}
|
||
|
||
if pushCount > 0 {
|
||
logs.Info(ctx, funcName, "补推已有成员状态完成",
|
||
zap.String("room_code", roomCode),
|
||
zap.Int64("to_user", userID),
|
||
zap.Int("push_count", pushCount))
|
||
}
|
||
}
|
||
|
||
// OnWSDisconnect WS 断线钩子，实现 ws.MeetingDisconnectHook 接口（Task 8）
|
||
// 由 ws.handler 的 SetOnDisconnect 回调触发；场景：用户的最后一条 WS 连接被移除
|
||
// 职责：
|
||
// 1. 若该用户当前存在活跃 meeting_participants 记录：清理其媒体资源 + 若是 host 启动 host 宽限期
|
||
// 2. 普通成员：仅清媒体资源（设计 Q1=A：不动 meeting_participants，长期不活跃由 4h 兜底清理）
|
||
//
|
||
// 容错：所有异常仅 Warn 日志，不返回 error；保证 ws.handler 主断连流程不受影响
|
||
func (s *MeetingSignalService) OnWSDisconnect(ctx context.Context, userID int64) {
|
||
funcName := "service.meeting_signal_service.OnWSDisconnect"
|
||
|
||
active, err := s.participantDAO.FindActiveByUser(ctx, userID)
|
||
if err != nil {
|
||
logs.Warn(ctx, funcName, "查询活跃会议失败", zap.Int64("user_id", userID), zap.Error(err))
|
||
return
|
||
}
|
||
if active == nil {
|
||
// 用户当前不在任何会议
|
||
return
|
||
}
|
||
room, err := s.roomDAO.GetByID(ctx, active.RoomID)
|
||
if err != nil || room == nil {
|
||
logs.Warn(ctx, funcName, "加载 room 失败或已不存在",
|
||
zap.Int64("room_id", active.RoomID), zap.Error(err))
|
||
return
|
||
}
|
||
if room.Status == constants.MeetingStatusEnded {
|
||
return
|
||
}
|
||
|
||
s.cleanupUserResources(ctx, room.ID, room.RoomCode, userID)
|
||
|
||
if s.lifecycleSvc != nil && room.HostID == userID {
|
||
s.lifecycleSvc.OnHostDisconnect(ctx, room.RoomCode, userID)
|
||
}
|
||
|
||
logs.Info(ctx, funcName, "WS 断线已处理",
|
||
zap.String("room_code", room.RoomCode),
|
||
zap.Int64("user_id", userID),
|
||
zap.Bool("is_host", room.HostID == userID))
|
||
}
|
||
|
||
// OnRoomLeave 处理 meeting.room.leave 事件
|
||
// 语义：WS 层面的主动离会（等价 REST leave 但不强制要求落库事务；
|
||
// 当前实现：仅清理该用户在本会议的所有媒体资源（batch close producer/consumer）+ 广播 meeting.member.left
|
||
// 参会者表的 LeaveRoom 逻辑仍由 REST API 负责（避免 WS 并发引起 left_at 重复写入）
|
||
func (s *MeetingSignalService) OnRoomLeave(ctx context.Context, userID int64, roomCode string) error {
|
||
room, err := s.loadRoomAndParticipant(ctx, roomCode, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
s.cleanupUserResources(ctx, room.ID, roomCode, userID)
|
||
|
||
// P2-8 修复：使用常量 MeetingLeftReasonDisconnect，避免 "ws_disconnect" 等硬编码
|
||
// 与前端 MEETING_LEFT_REASON_LABEL 字面值不一致
|
||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventMemberLeft, map[string]interface{}{
|
||
"room_code": roomCode,
|
||
"user_id": userID,
|
||
"reason": constants.MeetingLeftReasonDisconnect,
|
||
}, userID)
|
||
return nil
|
||
}
|
||
|
||
// ========== 成员组（1 个双向）==========
|
||
|
||
// MemberStateChangePayload meeting.member.state.changed 请求载荷
|
||
// host 可通过 target_user_id 静音他人 / 关其摄像头；非 host 传该字段将被拒绝
|
||
type MemberStateChangePayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
TargetUserID int64 `json:"target_user_id,omitempty"` // 可选：host 强制他人状态
|
||
AudioEnabled *bool `json:"audio_enabled,omitempty"` // nil 表示不改
|
||
VideoEnabled *bool `json:"video_enabled,omitempty"`
|
||
}
|
||
|
||
// OnMemberStateChanged 处理 meeting.member.state.changed 事件
|
||
// 权限：操作自己无限制；操作他人必须是 host
|
||
// 行为：广播 meeting.member.state.changed 给房间其他成员（发起者自己不收到回显）
|
||
func (s *MeetingSignalService) OnMemberStateChanged(ctx context.Context, fromUserID int64, payload *MemberStateChangePayload) error {
|
||
room, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, fromUserID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
targetID := fromUserID
|
||
if payload.TargetUserID != 0 && payload.TargetUserID != fromUserID {
|
||
if room.HostID != fromUserID {
|
||
return ErrNotMeetingHost
|
||
}
|
||
targetP, err := s.participantDAO.GetByRoomAndUser(ctx, room.ID, payload.TargetUserID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if targetP == nil || !targetP.IsActive() {
|
||
return ErrTransferTargetInvalid
|
||
}
|
||
targetID = payload.TargetUserID
|
||
}
|
||
|
||
data := map[string]interface{}{
|
||
"room_code": payload.RoomCode,
|
||
"user_id": targetID,
|
||
"changed_by": fromUserID,
|
||
}
|
||
if payload.AudioEnabled != nil {
|
||
data["audio_enabled"] = *payload.AudioEnabled
|
||
}
|
||
if payload.VideoEnabled != nil {
|
||
data["video_enabled"] = *payload.VideoEnabled
|
||
}
|
||
|
||
// 持久化到 Redis Hash；作为后入者补推 state.changed 的数据源
|
||
// 注意：写 Hash 之所以同步执行而非放进 goroutine，是因为同一用户并发开关操作需要顺序一致
|
||
s.updateMemberState(ctx, payload.RoomCode, targetID, payload.AudioEnabled, payload.VideoEnabled)
|
||
|
||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventMemberStateChange, data, fromUserID)
|
||
return nil
|
||
}
|
||
|
||
// ========== 媒体组（5 个，mediasoup signaling）==========
|
||
|
||
// TransportCreatePayload meeting.transport.create 请求载荷
|
||
type TransportCreatePayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
Direction string `json:"direction"` // "send" | "recv"
|
||
}
|
||
|
||
// OnTransportCreate 处理 meeting.transport.create 事件
|
||
func (s *MeetingSignalService) OnTransportCreate(ctx context.Context, userID int64, payload *TransportCreatePayload) (*TransportInfo, error) {
|
||
if payload.Direction != "send" && payload.Direction != "recv" {
|
||
return nil, fmt.Errorf("direction 非法，必须是 send 或 recv")
|
||
}
|
||
if _, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, userID); err != nil {
|
||
return nil, err
|
||
}
|
||
info, err := s.mediaOrchestrator.CreateTransport(ctx, &CreateTransportReq{
|
||
RoomCode: payload.RoomCode,
|
||
UserID: userID,
|
||
Direction: payload.Direction,
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s.trackResource(ctx, payload.RoomCode, userID, "transport", info.ID)
|
||
return info, nil
|
||
}
|
||
|
||
// TransportConnectPayload meeting.transport.connect 请求载荷
|
||
type TransportConnectPayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
TransportID string `json:"transport_id"`
|
||
DtlsParameters json.RawMessage `json:"dtls_parameters"`
|
||
}
|
||
|
||
// OnTransportConnect 处理 meeting.transport.connect 事件
|
||
func (s *MeetingSignalService) OnTransportConnect(ctx context.Context, userID int64, payload *TransportConnectPayload) error {
|
||
if _, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, userID); err != nil {
|
||
return err
|
||
}
|
||
if err := s.assertOwnsResource(ctx, payload.RoomCode, userID, "transport", payload.TransportID); err != nil {
|
||
return err
|
||
}
|
||
return s.mediaOrchestrator.ConnectTransport(ctx, payload.TransportID, payload.DtlsParameters)
|
||
}
|
||
|
||
// ProduceStartPayload meeting.produce.start 请求载荷
|
||
type ProduceStartPayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
TransportID string `json:"transport_id"`
|
||
Kind string `json:"kind"` // "audio" | "video"
|
||
RtpParameters json.RawMessage `json:"rtp_parameters"`
|
||
// AppData 客户端透传给 mediasoup Producer 的自定义元信息
|
||
// Phase 3 屏幕共享引入：约定 {"screen": true} 表示该 video Producer 是屏幕分享流；
|
||
// 服务端识别后会写入 screenOwnerKey + 广播 meeting.screen.started，前端将该流提升大画面。
|
||
// 服务端会强制把 userId / roomCode 注入回 appData，覆盖客户端伪造尝试（见 HTTPMediaOrchestrator.CreateProducer）。
|
||
AppData json.RawMessage `json:"app_data,omitempty"`
|
||
}
|
||
|
||
// isScreenAppData 判断客户端 app_data 是否声明了屏幕共享标识
|
||
// 容错：raw 为空 / 非对象 / 字段缺失均返回 false
|
||
func isScreenAppData(raw json.RawMessage) bool {
|
||
if len(raw) == 0 {
|
||
return false
|
||
}
|
||
var m map[string]any
|
||
if err := json.Unmarshal(raw, &m); err != nil {
|
||
return false
|
||
}
|
||
v, ok := m["screen"]
|
||
if !ok {
|
||
return false
|
||
}
|
||
switch x := v.(type) {
|
||
case bool:
|
||
return x
|
||
case string:
|
||
return x == "true" || x == "1"
|
||
case float64:
|
||
return x != 0
|
||
}
|
||
return false
|
||
}
|
||
|
||
// ProduceStartResult 返回给客户端的 producerID
|
||
type ProduceStartResult struct {
|
||
ProducerID string `json:"producer_id"`
|
||
}
|
||
|
||
// OnProduceStart 处理 meeting.produce.start 事件
|
||
// 成功后广播 meeting.member.producer.new 给房间内其他成员，驱动对端自动创建 Consumer
|
||
func (s *MeetingSignalService) OnProduceStart(ctx context.Context, userID int64, payload *ProduceStartPayload) (*ProduceStartResult, error) {
|
||
if payload.Kind != "audio" && payload.Kind != "video" {
|
||
return nil, fmt.Errorf("kind 非法，必须是 audio 或 video")
|
||
}
|
||
room, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, userID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.assertOwnsResource(ctx, payload.RoomCode, userID, "transport", payload.TransportID); err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// Phase 3：识别屏幕共享意图，强制 video kind 才允许（音频流不参与屏幕共享语义）
|
||
isScreen := payload.Kind == "video" && isScreenAppData(payload.AppData)
|
||
|
||
// Phase 3：单会议同时仅允许 1 份屏幕共享。若已有他人在共享，直接拒绝；
|
||
// 同一用户重复请求（例如客户端重发）则容忍，由后续 SET 覆盖
|
||
if isScreen {
|
||
if existingRaw, err := s.redis.Get(ctx, screenOwnerKey(payload.RoomCode)).Result(); err == nil {
|
||
if ownerUID, _, ok := parseScreenOwnerValue(existingRaw); ok && ownerUID != userID {
|
||
return nil, fmt.Errorf("当前会议已有成员正在共享屏幕")
|
||
}
|
||
}
|
||
}
|
||
|
||
producerID, err := s.mediaOrchestrator.CreateProducer(ctx, &CreateProducerReq{
|
||
RoomCode: payload.RoomCode,
|
||
UserID: userID,
|
||
TransportID: payload.TransportID,
|
||
Kind: payload.Kind,
|
||
RtpParameters: payload.RtpParameters,
|
||
AppData: payload.AppData,
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s.trackResource(ctx, payload.RoomCode, userID, "producer", producerID)
|
||
|
||
// Phase 3：屏幕共享额外维护 owner key + 触发 meeting.screen.started 广播
|
||
// 注意：先广播 producer.new（保证消费者侧 Consumer 创建），再广播 screen.started（标记大画面提升）
|
||
if isScreen {
|
||
if err := s.redis.Set(ctx, screenOwnerKey(payload.RoomCode), formatScreenOwnerValue(userID, producerID), resourceTTL).Err(); err != nil {
|
||
logs.Warn(ctx, "service.meeting_signal_service.OnProduceStart", "写入屏幕共享拥有者失败（不阻断业务）",
|
||
zap.String("room_code", payload.RoomCode),
|
||
zap.Int64("user_id", userID),
|
||
zap.String("producer_id", producerID),
|
||
zap.Error(err))
|
||
}
|
||
}
|
||
|
||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventMemberProducerNew, map[string]interface{}{
|
||
"room_code": payload.RoomCode,
|
||
"user_id": userID,
|
||
"producer_id": producerID,
|
||
"kind": payload.Kind,
|
||
"screen": isScreen,
|
||
}, userID)
|
||
|
||
if isScreen {
|
||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventScreenStarted, map[string]interface{}{
|
||
"room_code": payload.RoomCode,
|
||
"user_id": userID,
|
||
"producer_id": producerID,
|
||
}) // 不排除任何人：共享者本人也接收，便于前端统一更新大画面 UI
|
||
}
|
||
|
||
return &ProduceStartResult{ProducerID: producerID}, nil
|
||
}
|
||
|
||
// ConsumeStartPayload meeting.consume.start 请求载荷
|
||
type ConsumeStartPayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
TransportID string `json:"transport_id"` // 客户端 recv Transport
|
||
ProducerID string `json:"producer_id"` // 要订阅的远端 Producer
|
||
RtpCapabilities json.RawMessage `json:"rtp_capabilities"`
|
||
}
|
||
|
||
// OnConsumeStart 处理 meeting.consume.start 事件
|
||
// P0-2 修复：新增 transport_id 归属校验，拒绝把 consumer 挂到他人的 recv transport
|
||
// （producer_id 归属天然不需要校验：消费他人 producer 正是订阅逻辑本身，Node 侧会验证 producer 是否存在）
|
||
func (s *MeetingSignalService) OnConsumeStart(ctx context.Context, userID int64, payload *ConsumeStartPayload) (*ConsumerInfo, error) {
|
||
if payload.ProducerID == "" {
|
||
return nil, fmt.Errorf("producer_id 不能为空")
|
||
}
|
||
if _, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, userID); err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.assertOwnsResource(ctx, payload.RoomCode, userID, "transport", payload.TransportID); err != nil {
|
||
return nil, err
|
||
}
|
||
info, err := s.mediaOrchestrator.CreateConsumer(ctx, &CreateConsumerReq{
|
||
RoomCode: payload.RoomCode,
|
||
UserID: userID,
|
||
TransportID: payload.TransportID,
|
||
ProducerID: payload.ProducerID,
|
||
RtpCapabilities: payload.RtpCapabilities,
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s.trackResource(ctx, payload.RoomCode, userID, "consumer", info.ID)
|
||
return info, nil
|
||
}
|
||
|
||
// ConsumeResumePayload meeting.consume.resume 请求载荷（Task 9）
|
||
// 前端 recv Transport 与 track 就绪后发送，用于把 Node 侧 paused Consumer 切到 active
|
||
type ConsumeResumePayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
ConsumerID string `json:"consumer_id"`
|
||
}
|
||
|
||
// OnConsumeResume 处理 meeting.consume.resume 事件（Task 9）
|
||
// 语义：告知 media-server 把指定 Consumer 从 paused 切到 active
|
||
// 权限：仅当 userID 是会议活跃成员且 consumerID 归属该用户时允许
|
||
// 幂等：Node 对已 active Consumer 再次 resume 不报错；Consumer 不存在则 ACK 返回友好错误
|
||
func (s *MeetingSignalService) OnConsumeResume(ctx context.Context, userID int64, payload *ConsumeResumePayload) error {
|
||
if _, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, userID); err != nil {
|
||
return err
|
||
}
|
||
if err := s.assertOwnsResource(ctx, payload.RoomCode, userID, "consumer", payload.ConsumerID); err != nil {
|
||
return err
|
||
}
|
||
if err := s.mediaOrchestrator.ResumeConsumer(ctx, payload.ConsumerID); err != nil {
|
||
logs.Warn(ctx, "service.meeting_signal_service.OnConsumeResume", "恢复 Consumer 失败",
|
||
zap.String("room_code", payload.RoomCode),
|
||
zap.Int64("user_id", userID),
|
||
zap.String("consumer_id", payload.ConsumerID),
|
||
zap.Error(err))
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// ProducerClosePayload meeting.producer.close 请求载荷
|
||
type ProducerClosePayload struct {
|
||
RoomCode string `json:"room_code"`
|
||
ProducerID string `json:"producer_id"`
|
||
}
|
||
|
||
// OnProducerClose 处理 meeting.producer.close 事件
|
||
// 成功后广播给房间内其他成员（与 mediasoup 的 producerclose 级联动作平级）
|
||
// P0-1 修复：新增 producer 归属校验，拒绝一名参会人关闭他人的 producer（横向越权，审计 P0-1）
|
||
func (s *MeetingSignalService) OnProducerClose(ctx context.Context, userID int64, payload *ProducerClosePayload) error {
|
||
room, err := s.loadRoomAndParticipant(ctx, payload.RoomCode, userID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := s.assertOwnsResource(ctx, payload.RoomCode, userID, "producer", payload.ProducerID); err != nil {
|
||
return err
|
||
}
|
||
if err := s.mediaOrchestrator.CloseProducer(ctx, payload.ProducerID); err != nil {
|
||
logs.Warn(ctx, "service.meeting_signal_service.OnProducerClose", "关闭 Producer 失败",
|
||
zap.String("producer_id", payload.ProducerID), zap.Error(err))
|
||
}
|
||
s.untrackResource(ctx, payload.RoomCode, userID, "producer", payload.ProducerID)
|
||
|
||
// Phase 3：若关闭的恰好是当前屏幕共享 Producer，则一并清 screenOwnerKey + 广播 screen.stopped
|
||
s.maybeReleaseScreenOwner(ctx, room.ID, payload.RoomCode, userID, payload.ProducerID)
|
||
|
||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventMemberProducerNew, map[string]interface{}{
|
||
"room_code": payload.RoomCode,
|
||
"user_id": userID,
|
||
"producer_id": payload.ProducerID,
|
||
"closed": true,
|
||
}, userID)
|
||
return nil
|
||
}
|
||
|
||
// maybeReleaseScreenOwner 检查并释放屏幕共享拥有者状态
|
||
// 当 producerID 恰好等于 screenOwnerKey 当前值的 producer 段时，认为这次关闭是屏幕分享停止：
|
||
// - DEL screenOwnerKey
|
||
// - 广播 meeting.screen.stopped 给所有活跃成员（含共享者本人）
|
||
//
|
||
// 非屏幕 Producer 或 key 已被他人覆盖时静默 no-op，保证幂等
|
||
func (s *MeetingSignalService) maybeReleaseScreenOwner(ctx context.Context, roomID int64, roomCode string, userID int64, producerID string) {
|
||
funcName := "service.meeting_signal_service.maybeReleaseScreenOwner"
|
||
|
||
key := screenOwnerKey(roomCode)
|
||
raw, err := s.redis.Get(ctx, key).Result()
|
||
if err != nil {
|
||
// redis.Nil（无人共享）属正常路径，其他错误仅 Warn 不阻塞
|
||
if err != redis.Nil {
|
||
logs.Warn(ctx, funcName, "读取屏幕共享拥有者失败（忽略）",
|
||
zap.String("room_code", roomCode), zap.Error(err))
|
||
}
|
||
return
|
||
}
|
||
ownerUID, ownerProducerID, ok := parseScreenOwnerValue(raw)
|
||
if !ok || ownerProducerID != producerID {
|
||
return
|
||
}
|
||
// 防御性日志：理论上 ownerUID 必然等于 userID（trackResource 已校验归属）
|
||
if ownerUID != userID {
|
||
logs.Warn(ctx, funcName, "屏幕共享拥有者与关闭者不一致（仍按停止处理）",
|
||
zap.Int64("owner_uid", ownerUID),
|
||
zap.Int64("closer_uid", userID),
|
||
zap.String("producer_id", producerID))
|
||
}
|
||
|
||
if err := s.redis.Del(ctx, key).Err(); err != nil {
|
||
logs.Warn(ctx, funcName, "删除屏幕共享拥有者失败（忽略）",
|
||
zap.String("room_code", roomCode), zap.Error(err))
|
||
}
|
||
|
||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), roomID, constants.MeetingWSEventScreenStopped, map[string]interface{}{
|
||
"room_code": roomCode,
|
||
"user_id": ownerUID,
|
||
"producer_id": producerID,
|
||
})
|
||
}
|
||
|
||
// ========== 资源清理 ==========
|
||
|
||
// cleanupUserResources 批量关闭指定用户在某会议的所有媒体资源
|
||
// WS 断开、主动离会、被踢时使用；依赖 Redis 集合中追踪的资源 ID
|
||
//
|
||
// Phase 3 屏幕共享：调用方需提供 roomID，便于在该用户恰好是屏幕共享者时触发
|
||
// meeting.screen.stopped 广播。roomID==0 时跳过屏幕广播（仅清 Redis），保持向后兼容。
|
||
func (s *MeetingSignalService) cleanupUserResources(ctx context.Context, roomID int64, roomCode string, userID int64) {
|
||
funcName := "service.meeting_signal_service.cleanupUserResources"
|
||
|
||
key := resourceTrackKey(roomCode, userID)
|
||
members, err := s.redis.SMembers(ctx, key).Result()
|
||
if err != nil {
|
||
logs.Warn(ctx, funcName, "读取资源追踪集合失败", zap.String("key", key), zap.Error(err))
|
||
return
|
||
}
|
||
for _, m := range members {
|
||
// 格式："kind:id"
|
||
// Nit（代码审查 2026-04-23）：原先使用 for-range + rune 匹配 ':'，对 ASCII 过度包装；
|
||
// 改用 strings.SplitN 限定 2 段更清晰，且避免 rune 解码开销
|
||
parts := strings.SplitN(m, ":", 2)
|
||
if len(parts) != 2 {
|
||
continue
|
||
}
|
||
kind, id := parts[0], parts[1]
|
||
switch kind {
|
||
case "producer":
|
||
_ = s.mediaOrchestrator.CloseProducer(ctx, id)
|
||
// Phase 3：若该 Producer 是当前屏幕共享流，释放 owner key 并广播 screen.stopped
|
||
// 复用 OnProducerClose 同款幂等逻辑；roomID==0 跳过避免无效广播
|
||
if roomID > 0 {
|
||
s.maybeReleaseScreenOwner(ctx, roomID, roomCode, userID, id)
|
||
}
|
||
case "consumer":
|
||
_ = s.mediaOrchestrator.CloseConsumer(ctx, id)
|
||
case "transport":
|
||
// Task 16 P2-1：补全 transport 精确清理，短暂抖动重连场景下 Router 不会级联关闭自己的 transport
|
||
_ = s.mediaOrchestrator.CloseTransport(ctx, id)
|
||
}
|
||
}
|
||
_ = s.redis.Del(ctx, key).Err()
|
||
// 一并清理成员音视频状态，避免对方下次入会时读到旧 host 的僵尸 AV 状态
|
||
_ = s.redis.Del(ctx, memberStateKey(roomCode, userID)).Err()
|
||
|
||
if len(members) > 0 {
|
||
logs.Info(ctx, funcName, "清理用户媒体资源",
|
||
zap.String("room_code", roomCode),
|
||
zap.Int64("user_id", userID),
|
||
zap.Int("resource_count", len(members)))
|
||
}
|
||
}
|
||
|
||
// cleanupRoomRedisResidual 批量清理指定 roomCode 下一批用户的 Redis 资源追踪集合 + 音视频状态 Hash
|
||
// Task 16 资源清理专项：
|
||
// - EndRoom / HandleEmptyRoomExpired 等会议销毁路径调用，避免正常 WS 断开钩子尚未触发就结束会议导致的残留
|
||
// - userIDs 为销毁前活跃成员快照；无活跃成员可传 nil/空切片，此时仅走上层 lifecycle 清理
|
||
// - 使用 Pipeline 减少 Redis 往返；失败仅 Warn，不阻断上层销毁流程
|
||
// - package 私有：允许 meeting_service / meeting_lifecycle_service 等同包文件直接调用
|
||
func cleanupRoomRedisResidual(ctx context.Context, rdb *redis.Client, roomCode string, userIDs []int64) {
|
||
funcName := "service.meeting_signal_service.cleanupRoomRedisResidual"
|
||
if rdb == nil {
|
||
return
|
||
}
|
||
// Phase 3：会议销毁路径必删 screenOwnerKey（即使无活跃用户也清理）
|
||
// 单独 Del 不依赖 pipeline，因为会议销毁是低频路径，多一次 RTT 可接受
|
||
_ = rdb.Del(ctx, screenOwnerKey(roomCode)).Err()
|
||
|
||
if len(userIDs) == 0 {
|
||
return
|
||
}
|
||
pipe := rdb.Pipeline()
|
||
for _, uid := range userIDs {
|
||
pipe.Del(ctx, resourceTrackKey(roomCode, uid))
|
||
pipe.Del(ctx, memberStateKey(roomCode, uid))
|
||
}
|
||
if _, err := pipe.Exec(ctx); err != nil {
|
||
logs.Warn(ctx, funcName, "批量清理用户 Redis 残留失败（忽略，继续）",
|
||
zap.String("room_code", roomCode), zap.Int("user_count", len(userIDs)), zap.Error(err))
|
||
return
|
||
}
|
||
logs.Debug(ctx, funcName, "会议销毁，用户 Redis 残留已清理",
|
||
zap.String("room_code", roomCode), zap.Int("user_count", len(userIDs)))
|
||
}
|