// Package service 承载 room-service 的 Room Cell 管理、命令流水线、恢复、快照和 outbox 补偿。 package service import ( "context" "errors" "fmt" "sync" "time" "google.golang.org/protobuf/proto" roomeventsv1 "hyapp/api/proto/events/room/v1" imv1 "hyapp/api/proto/im/v1" roomv1 "hyapp/api/proto/room/v1" walletv1 "hyapp/api/proto/wallet/v1" "hyapp/pkg/clock" "hyapp/pkg/idgen" "hyapp/pkg/xerr" "hyapp/services/room-service/internal/integration" "hyapp/services/room-service/internal/room/cell" "hyapp/services/room-service/internal/room/command" "hyapp/services/room-service/internal/room/outbox" "hyapp/services/room-service/internal/room/rank" "hyapp/services/room-service/internal/room/state" "hyapp/services/room-service/internal/router" ) // Config 收敛 room-service 领域服务的运行参数。 type Config struct { // NodeID 是当前 room-service 实例标识,Redis lease 以它作为房间 owner。 NodeID string // LeaseTTL 是房间 owner 租约有效期,过期后其他节点可以接管并恢复房间。 LeaseTTL time.Duration // RankLimit 是房间本地礼物榜保留长度。 RankLimit int // SnapshotEveryN 表示每 N 个房间版本保存一次快照。 SnapshotEveryN int64 // Clock 允许测试注入稳定时间,生产默认使用系统时钟。 Clock clock.Clock } // Service 承载 room-service 的核心房间领域逻辑。 type Service struct { // nodeID 是当前实例 ID,必须和 Redis lease 中的 owner 对齐。 nodeID string // leaseTTL 控制当前节点持有房间执行权的时间窗口。 leaseTTL time.Duration // rankLimit 控制每个 Room Cell 的 LocalRank top N。 rankLimit int // snapshotEveryN 控制快照频率,值越小恢复越快但写放大越高。 snapshotEveryN int64 // clock 是房间命令、事件和快照的统一时间来源。 clock clock.Clock // directory 负责 room_id -> node_id lease,生产路径是 Redis。 directory router.Directory // repository 是 MySQL 持久化边界,保存 meta、snapshot、command log 和 outbox。 repository Repository // wallet 是 SendGift 的同步扣费依赖,扣费失败不能进入 Room Cell 状态变更。 wallet integration.WalletClient // syncPublisher 是 room-service 到 im-service 的低时延房间系统事件桥。 syncPublisher integration.RoomEventPublisher // outboxPublisher 是补偿 worker 对外部消费者的异步投递抽象。 outboxPublisher integration.OutboxPublisher // mu 保护本进程内 room_id -> RoomCell 注册表。 mu sync.Mutex // cells 保存当前节点已经装载的单房间执行单元。 cells map[string]*cell.RoomCell } // mutationResult 是命令在 Room Cell 内执行后的统一结果包。 type mutationResult struct { // applied 表示本次命令确实改变并提交了房间状态。 applied bool // snapshot 是命令执行后对外返回的房间快照。 snapshot *roomv1.RoomSnapshot // user 是 JoinRoom 等命令需要返回的用户投影。 user *roomv1.RoomUser // seatNo 是麦位命令返回的目标或原麦位编号。 seatNo int32 // billingReceiptID 是 SendGift 从 wallet-service 得到的账务回执。 billingReceiptID string // roomHeat 是 SendGift 后的房间热度。 roomHeat int64 // giftRank 是 SendGift 后的本地礼物榜投影。 giftRank []*roomv1.RankItem // syncEvent 是需要同步推给 im-service 的低时延房间系统事件。 syncEvent *imv1.RoomEvent } // New 初始化 room-service 领域服务。 func New(cfg Config, directory router.Directory, repository Repository, wallet integration.WalletClient, syncPublisher integration.RoomEventPublisher, outboxPublisher integration.OutboxPublisher) *Service { usedClock := cfg.Clock if usedClock == nil { // 生产默认系统时钟,测试可注入固定时钟保证断言稳定。 usedClock = clock.System{} } snapshotEveryN := cfg.SnapshotEveryN if snapshotEveryN <= 0 { // 默认每次状态变更都保存快照,优先保证首版恢复简单。 snapshotEveryN = 1 } rankLimit := cfg.RankLimit if rankLimit <= 0 { // 默认只保留房间内前 20 名礼物榜,避免快照和广播载荷过大。 rankLimit = 20 } return &Service{ nodeID: cfg.NodeID, leaseTTL: cfg.LeaseTTL, rankLimit: rankLimit, snapshotEveryN: snapshotEveryN, clock: usedClock, directory: directory, repository: repository, wallet: wallet, syncPublisher: syncPublisher, outboxPublisher: outboxPublisher, cells: make(map[string]*cell.RoomCell), } } // HealthCheck 校验 room-service 核心状态 owner 和持久化依赖已经完成装配。 func (s *Service) HealthCheck() error { if s == nil { // nil service 说明 App 装配失败,live/ready 都不能放行。 return errors.New("room service is not initialized") } if s.directory == nil { // 没有 directory 就无法保证单房间单活 owner。 return errors.New("room directory is not initialized") } if s.repository == nil { // 没有 repository 就无法保证 command log、snapshot 和 outbox 持久化。 return errors.New("room repository is not initialized") } if s.cells == nil { // cells 注册表缺失会导致已恢复房间无法复用同一个 Room Cell。 return errors.New("room cell registry is not initialized") } return nil } // CreateRoom 创建一个新的房间执行单元,并把首个快照和事件落盘。 func (s *Service) CreateRoom(ctx context.Context, req *roomv1.CreateRoomRequest) (*roomv1.CreateRoomResponse, error) { if req.GetMeta() == nil || req.GetMeta().GetRoomId() == "" { // 创建房间必须指定 room_id,后续 lease、meta、snapshot 都依赖它。 return nil, xerr.New(xerr.InvalidArgument, "room_id is required") } // protobuf 请求先收敛为 command,后续幂等、持久化和恢复都只认识 command 模型。 cmd := command.CreateRoom{ Base: baseFromMeta(req.GetMeta()), OwnerUserID: req.GetOwnerUserId(), HostUserID: req.GetHostUserId(), SeatCount: req.GetSeatCount(), Mode: req.GetMode(), } if cmd.SeatCount <= 0 { // 没有麦位的房间不符合语音房 v1 的状态模型。 return nil, xerr.New(xerr.InvalidArgument, "seat_count must be positive") } if cmd.OwnerUserID == 0 || cmd.HostUserID == 0 { // owner/host 是默认管理员集合来源,不能缺失。 return nil, xerr.New(xerr.InvalidArgument, "owner and host are required") } if seen, err := s.repository.HasCommand(ctx, cmd.RoomID(), cmd.ID()); err != nil { return nil, err } else if seen { // 重复 CreateRoom 不再写入状态,直接返回当前快照保证命令幂等。 snapshot, err := s.currentSnapshot(ctx, cmd.RoomID()) if err != nil { return nil, err } return &roomv1.CreateRoomResponse{ Result: commandResult(false, snapshot.GetVersion(), s.clock.Now()), Room: snapshot, }, nil } // 创建房间前先获得 owner lease,避免两个节点同时创建同一个 room_id。 lease, err := s.directory.EnsureOwner(ctx, cmd.RoomID(), s.nodeID, s.clock.Now(), s.leaseTTL) if err != nil { return nil, err } if lease.NodeID != s.nodeID { // 有其他有效 owner 时当前节点不能创建或接管该房间。 return nil, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID)) } if _, exists, err := s.repository.GetRoomMeta(ctx, cmd.RoomID()); err != nil { return nil, err } else if exists { // meta 已存在但 command 未命中,说明 room_id 已被其他创建命令占用。 return nil, xerr.New(xerr.Conflict, "room already exists") } // 创建态直接初始化 RoomState,owner 作为首个在线用户进入房间。 now := s.clock.Now() roomState := state.NewRoomState(cmd.RoomID(), cmd.OwnerUserID, cmd.HostUserID, cmd.SeatCount, cmd.Mode) roomState.Version = 1 roomState.OnlineUsers[cmd.ActorUserID()] = &state.RoomUserState{ UserID: cmd.ActorUserID(), Role: "owner", JoinedAtMS: now.UnixMilli(), LastSeenAtMS: now.UnixMilli(), } payload, err := command.Serialize(cmd) if err != nil { return nil, err } // RoomCreated 进入 outbox,供 activity/audit 等房间外系统异步消费。 createdEvent, err := outbox.Build(cmd.RoomID(), "RoomCreated", roomState.Version, now, &roomeventsv1.RoomCreated{ OwnerUserId: cmd.OwnerUserID, HostUserId: cmd.HostUserID, SeatCount: cmd.SeatCount, Mode: cmd.Mode, }) if err != nil { return nil, err } // 首次创建按 meta -> command log -> outbox -> snapshot 的顺序落盘,保证恢复有基础元数据和首版本命令。 if err := s.repository.SaveRoomMeta(ctx, RoomMeta{ RoomID: cmd.RoomID(), OwnerUserID: cmd.OwnerUserID, HostUserID: cmd.HostUserID, SeatCount: cmd.SeatCount, Mode: cmd.Mode, Status: "active", }); err != nil { return nil, err } if err := s.repository.SaveCommand(ctx, CommandRecord{ RoomID: cmd.RoomID(), RoomVersion: roomState.Version, CommandID: cmd.ID(), CommandType: cmd.Type(), Payload: payload, Replayable: true, CreatedAt: now, }); err != nil { return nil, err } if err := s.repository.SaveOutbox(ctx, []outbox.Record{createdEvent}); err != nil { return nil, err } snapshot := roomState.ToProto() if err := s.persistSnapshot(ctx, snapshot); err != nil { return nil, err } // 持久化成功后再安装内存 Cell,避免内存有房间但恢复来源不完整。 s.installCell(cmd.RoomID(), cell.New(roomState.Clone(), rank.FromState(roomState.GiftRank, s.rankLimit))) return &roomv1.CreateRoomResponse{ Result: commandResult(true, snapshot.GetVersion(), now), Room: snapshot, }, nil } // JoinRoom 把用户业务态接入房间。 func (s *Service) JoinRoom(ctx context.Context, req *roomv1.JoinRoomRequest) (*roomv1.JoinRoomResponse, error) { // JoinRoom 只维护 room-service presence,不创建 im-service WebSocket 订阅。 cmd := command.JoinRoom{ Base: baseFromMeta(req.GetMeta()), Role: req.GetRole(), } if cmd.Role == "" { // 客户端不传角色时按普通观众进入房间。 cmd.Role = "audience" } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { if current.BanUsers[cmd.ActorUserID()] { // 被踢或封禁用户不能重新进入业务 presence。 return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "user is banned") } if _, exists := current.OnlineUsers[cmd.ActorUserID()]; exists { // 用户已经在房间时不递增版本,也不重复写 command log/outbox。 snapshot := current.ToProto() return mutationResult{ snapshot: snapshot, user: findProtoUser(snapshot, cmd.ActorUserID()), }, nil, nil } current.OnlineUsers[cmd.ActorUserID()] = &state.RoomUserState{ UserID: cmd.ActorUserID(), Role: cmd.Role, JoinedAtMS: now.UnixMilli(), LastSeenAtMS: now.UnixMilli(), } current.Version++ // JoinRoom 成功需要 outbox 事件,同时同步推给 im-service 做房间系统消息。 joinedEvent, err := outbox.Build(current.RoomID, "RoomUserJoined", current.Version, now, &roomeventsv1.RoomUserJoined{ UserId: cmd.ActorUserID(), Role: cmd.Role, }) if err != nil { return mutationResult{}, nil, err } snapshot := current.ToProto() return mutationResult{ snapshot: snapshot, user: findProtoUser(snapshot, cmd.ActorUserID()), syncEvent: &imv1.RoomEvent{ EventId: joinedEvent.EventID, RoomId: current.RoomID, EventType: "room_user_joined", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.ActorUserID(), RoomVersion: current.Version, }, }, []outbox.Record{joinedEvent}, nil }) if err != nil { return nil, err } return &roomv1.JoinRoomResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), User: result.user, Room: result.snapshot, }, nil } // LeaveRoom 把用户从房间业务态移出。 func (s *Service) LeaveRoom(ctx context.Context, req *roomv1.LeaveRoomRequest) (*roomv1.LeaveRoomResponse, error) { // LeaveRoom 只移除 room-service 业务 presence,真实 WS 订阅由 im-service unsubscribe 或断线清理。 cmd := command.LeaveRoom{ Base: baseFromMeta(req.GetMeta()), } result, err := s.mutateRoom(ctx, cmd, false, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { if _, exists := current.OnlineUsers[cmd.ActorUserID()]; !exists { // 离房幂等:用户不在房间时返回当前快照,不写命令日志。 return mutationResult{ snapshot: current.ToProto(), }, nil, nil } delete(current.OnlineUsers, cmd.ActorUserID()) current.Version++ // 离房事件进入 outbox,并尝试同步通知 im-service。 leftEvent, err := outbox.Build(current.RoomID, "RoomUserLeft", current.Version, now, &roomeventsv1.RoomUserLeft{ UserId: cmd.ActorUserID(), }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), syncEvent: &imv1.RoomEvent{ EventId: leftEvent.EventID, RoomId: current.RoomID, EventType: "room_user_left", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.ActorUserID(), RoomVersion: current.Version, }, }, []outbox.Record{leftEvent}, nil }) if err != nil { return nil, err } return &roomv1.LeaveRoomResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), Room: result.snapshot, }, nil } // MicUp 把目标用户放到指定麦位。 func (s *Service) MicUp(ctx context.Context, req *roomv1.MicUpRequest) (*roomv1.MicUpResponse, error) { // MicUp 以 actor 作为上麦目标,管理员代操作走 MicDown/ChangeMicSeat 等显式 target 命令。 cmd := command.MicUp{ Base: baseFromMeta(req.GetMeta()), SeatNo: req.GetSeatNo(), } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { if _, exists := current.OnlineUsers[cmd.ActorUserID()]; !exists { // 不在 room-service presence 的用户不能上麦。 return mutationResult{}, nil, xerr.New(xerr.NotFound, "user not in room") } if _, exists := current.SeatByUser(cmd.ActorUserID()); exists { // 一个用户同一时间只能占一个麦位。 return mutationResult{}, nil, xerr.New(xerr.Conflict, "user already on seat") } index := current.SeatIndex(cmd.SeatNo) if index < 0 { // 麦位编号必须来自房间初始化的 MicSeats。 return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "seat does not exist") } if current.MicSeats[index].Locked { // 锁定麦位不能被普通上麦命令占用。 return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "seat is locked") } if current.MicSeats[index].UserID != 0 { // 非空麦位不能被覆盖,避免两个用户同时占同一麦位。 return mutationResult{}, nil, xerr.New(xerr.Conflict, "seat is occupied") } // 麦位占用和房间版本必须在同一个 Room Cell 任务里更新。 current.MicSeats[index].UserID = cmd.ActorUserID() current.Version++ // RoomMicChanged 是房间外消费者和 im-service 共同使用的麦位事件源。 micEvent, err := outbox.Build(current.RoomID, "RoomMicChanged", current.Version, now, &roomeventsv1.RoomMicChanged{ ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.ActorUserID(), FromSeat: 0, ToSeat: cmd.SeatNo, Action: "up", }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), seatNo: cmd.SeatNo, syncEvent: &imv1.RoomEvent{ EventId: micEvent.EventID, RoomId: current.RoomID, EventType: "room_mic_up", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.ActorUserID(), SeatNo: cmd.SeatNo, RoomVersion: current.Version, }, }, []outbox.Record{micEvent}, nil }) if err != nil { return nil, err } return &roomv1.MicUpResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), SeatNo: result.seatNo, Room: result.snapshot, }, nil } // MicDown 把目标用户从当前麦位移下。 func (s *Service) MicDown(ctx context.Context, req *roomv1.MicDownRequest) (*roomv1.MicDownResponse, error) { // MicDown 支持操作者和目标用户不同,权限策略后续可在命令入口扩展。 cmd := command.MicDown{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { seat, exists := current.SeatByUser(cmd.TargetUserID) if !exists { // 目标用户不在麦上时不能生成下麦事件。 return mutationResult{}, nil, xerr.New(xerr.NotFound, "target not on seat") } // SeatByUser 返回的是值拷贝,真正修改需要再定位切片下标。 index := current.SeatIndex(seat.SeatNo) current.MicSeats[index].UserID = 0 current.Version++ micEvent, err := outbox.Build(current.RoomID, "RoomMicChanged", current.Version, now, &roomeventsv1.RoomMicChanged{ ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, FromSeat: seat.SeatNo, ToSeat: 0, Action: "down", }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), seatNo: seat.SeatNo, syncEvent: &imv1.RoomEvent{ EventId: micEvent.EventID, RoomId: current.RoomID, EventType: "room_mic_down", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, SeatNo: seat.SeatNo, RoomVersion: current.Version, }, }, []outbox.Record{micEvent}, nil }) if err != nil { return nil, err } return &roomv1.MicDownResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), SeatNo: result.seatNo, Room: result.snapshot, }, nil } // ChangeMicSeat 调整指定用户所在麦位。 func (s *Service) ChangeMicSeat(ctx context.Context, req *roomv1.ChangeMicSeatRequest) (*roomv1.ChangeMicSeatResponse, error) { // ChangeMicSeat 是原子换位:从原麦位清空和写入新麦位必须同版本完成。 cmd := command.ChangeMicSeat{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), SeatNo: req.GetSeatNo(), } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { fromSeat, exists := current.SeatByUser(cmd.TargetUserID) if !exists { // 未上麦用户没有可换出的源麦位。 return mutationResult{}, nil, xerr.New(xerr.NotFound, "target not on seat") } toIndex := current.SeatIndex(cmd.SeatNo) if toIndex < 0 { // 目标麦位必须存在。 return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "seat does not exist") } if current.MicSeats[toIndex].Locked { // 锁定麦位不能换入。 return mutationResult{}, nil, xerr.New(xerr.PermissionDenied, "seat is locked") } if current.MicSeats[toIndex].UserID != 0 { // 目标麦位被占用时拒绝换位,避免覆盖另一个用户。 return mutationResult{}, nil, xerr.New(xerr.Conflict, "seat is occupied") } // 清空源麦位和占用目标麦位在同一 Cell 任务内完成,客户端不会看到中间态。 fromIndex := current.SeatIndex(fromSeat.SeatNo) current.MicSeats[fromIndex].UserID = 0 current.MicSeats[toIndex].UserID = cmd.TargetUserID current.Version++ micEvent, err := outbox.Build(current.RoomID, "RoomMicChanged", current.Version, now, &roomeventsv1.RoomMicChanged{ ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, FromSeat: fromSeat.SeatNo, ToSeat: cmd.SeatNo, Action: "change", }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), syncEvent: &imv1.RoomEvent{ EventId: micEvent.EventID, RoomId: current.RoomID, EventType: "room_mic_change", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, SeatNo: cmd.SeatNo, RoomVersion: current.Version, }, }, []outbox.Record{micEvent}, nil }) if err != nil { return nil, err } return &roomv1.ChangeMicSeatResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), Room: result.snapshot, }, nil } // MuteUser 修改用户禁言状态。 func (s *Service) MuteUser(ctx context.Context, req *roomv1.MuteUserRequest) (*roomv1.MuteUserResponse, error) { // 禁言状态由 room-service 持有,im-service 发公屏前只同步查询 CheckSpeakPermission。 cmd := command.MuteUser{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), Muted: req.GetMuted(), } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { if cmd.Muted { // map set 表达禁言集合,true 才会被快照导出。 current.MuteUsers[cmd.TargetUserID] = true } else { // 解除禁言直接删除集合成员。 delete(current.MuteUsers, cmd.TargetUserID) } current.Version++ // 禁言变更进入 outbox,外部系统可以做审核和操作审计。 muteEvent, err := outbox.Build(current.RoomID, "RoomUserMuted", current.Version, now, &roomeventsv1.RoomUserMuted{ ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, Muted: cmd.Muted, }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), syncEvent: &imv1.RoomEvent{ EventId: muteEvent.EventID, RoomId: current.RoomID, EventType: "room_user_muted", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, RoomVersion: current.Version, }, }, []outbox.Record{muteEvent}, nil }) if err != nil { return nil, err } return &roomv1.MuteUserResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), Room: result.snapshot, }, nil } // KickUser 把目标用户踢出房间并移除其麦位与 presence。 func (s *Service) KickUser(ctx context.Context, req *roomv1.KickUserRequest) (*roomv1.KickUserResponse, error) { // KickUser 同时修改 presence、麦位和 ban 集合,是典型 Room Cell 强关联命令。 cmd := command.KickUser{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) { if seat, exists := current.SeatByUser(cmd.TargetUserID); exists { // 被踢用户如果在麦上,必须先释放麦位,避免恢复后出现幽灵占位。 index := current.SeatIndex(seat.SeatNo) current.MicSeats[index].UserID = 0 } // ban 集合会让后续 JoinRoom 和 VerifyRoomPresence 都失败。 delete(current.OnlineUsers, cmd.TargetUserID) current.BanUsers[cmd.TargetUserID] = true current.Version++ kickEvent, err := outbox.Build(current.RoomID, "RoomUserKicked", current.Version, now, &roomeventsv1.RoomUserKicked{ ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), syncEvent: &imv1.RoomEvent{ EventId: kickEvent.EventID, RoomId: current.RoomID, EventType: "room_user_kicked", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, RoomVersion: current.Version, }, }, []outbox.Record{kickEvent}, nil }) if err != nil { return nil, err } return &roomv1.KickUserResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), Room: result.snapshot, }, nil } // SendGift 先同步调用 wallet,再在 Room Cell 内完成热度、榜单和房间事件落地。 func (s *Service) SendGift(ctx context.Context, req *roomv1.SendGiftRequest) (*roomv1.SendGiftResponse, error) { // SendGift 的账务不在 room-service 内结算,但房间表现、热度和榜单由 Room Cell 同步更新。 cmd := command.SendGift{ Base: baseFromMeta(req.GetMeta()), TargetUserID: req.GetTargetUserId(), GiftID: req.GetGiftId(), GiftCount: req.GetGiftCount(), GiftUnitValue: req.GetGiftUnitValue(), } result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, localRank *rank.LocalRank) (mutationResult, []outbox.Record, error) { if _, exists := current.OnlineUsers[cmd.ActorUserID()]; !exists { // 送礼者必须在房间业务 presence 内。 return mutationResult{}, nil, xerr.New(xerr.NotFound, "sender not in room") } if _, exists := current.OnlineUsers[cmd.TargetUserID]; !exists { // v1 要求礼物目标在房间内,避免给不存在的房间表现目标计榜。 return mutationResult{}, nil, xerr.New(xerr.NotFound, "target not in room") } if cmd.GiftCount <= 0 || cmd.GiftUnitValue <= 0 { // 非正数礼物会破坏账务和榜单累计语义。 return mutationResult{}, nil, xerr.New(xerr.InvalidArgument, "gift_count and gift_unit_value must be positive") } // 钱包扣费在房间状态变更前完成;失败时不写 command log、不进入 Room Cell 提交态。 billing, err := s.wallet.DebitGift(ctx, &walletv1.DebitGiftRequest{ CommandId: cmd.ID(), RoomId: cmd.RoomID(), SenderUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, GiftId: cmd.GiftID, GiftCount: cmd.GiftCount, GiftUnitValue: cmd.GiftUnitValue, }) if err != nil { return mutationResult{}, nil, err } // 扣费成功后,Room Cell 同步更新热度和本地礼物榜。 current.Heat += billing.GetTotalGiftValue() current.GiftRank = localRank.ApplyGift(cmd.ActorUserID(), billing.GetTotalGiftValue(), now) current.Version++ // 送礼事件、热度事件和榜单事件使用同一个房间版本,方便消费者关联同一次命令。 giftEvent, err := outbox.Build(current.RoomID, "RoomGiftSent", current.Version, now, &roomeventsv1.RoomGiftSent{ SenderUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, GiftId: cmd.GiftID, GiftCount: cmd.GiftCount, GiftValue: billing.GetTotalGiftValue(), BillingReceiptId: billing.GetBillingReceiptId(), }) if err != nil { return mutationResult{}, nil, err } heatEvent, err := outbox.Build(current.RoomID, "RoomHeatChanged", current.Version, now, &roomeventsv1.RoomHeatChanged{ Delta: billing.GetTotalGiftValue(), CurrentHeat: current.Heat, }) if err != nil { return mutationResult{}, nil, err } rankItem := findRankItem(current.GiftRank, cmd.ActorUserID()) rankEvent, err := outbox.Build(current.RoomID, "RoomRankChanged", current.Version, now, &roomeventsv1.RoomRankChanged{ UserId: rankItem.UserID, Score: rankItem.Score, GiftValue: rankItem.GiftValue, }) if err != nil { return mutationResult{}, nil, err } return mutationResult{ snapshot: current.ToProto(), billingReceiptID: billing.GetBillingReceiptId(), roomHeat: current.Heat, giftRank: cloneProtoRank(current.GiftRank), syncEvent: &imv1.RoomEvent{ // 同步广播只选 GiftSent 主事件作为客户端房间系统消息,Heat/Rank 通过字段携带。 EventId: giftEvent.EventID, RoomId: current.RoomID, EventType: "room_gift_sent", ActorUserId: cmd.ActorUserID(), TargetUserId: cmd.TargetUserID, GiftValue: billing.GetTotalGiftValue(), RoomHeat: current.Heat, RoomVersion: current.Version, Attributes: map[string]string{ "gift_id": cmd.GiftID, "gift_count": fmt.Sprintf("%d", cmd.GiftCount), "billing_receipt_id": billing.GetBillingReceiptId(), }, }, }, []outbox.Record{giftEvent, heatEvent, rankEvent}, nil }) if err != nil { return nil, err } return &roomv1.SendGiftResponse{ Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()), BillingReceiptId: result.billingReceiptID, RoomHeat: result.roomHeat, GiftRank: result.giftRank, Room: result.snapshot, }, nil } // CheckSpeakPermission 给 im-service 提供同步发言守卫。 func (s *Service) CheckSpeakPermission(ctx context.Context, req *roomv1.CheckSpeakPermissionRequest) (*roomv1.CheckSpeakPermissionResponse, error) { // 守卫查询优先读内存 Cell;没有 Cell 时从快照读取,避免 im-service 越权自判。 snapshot, err := s.currentSnapshot(ctx, req.GetRoomId()) if err != nil { return nil, err } if snapshot == nil { // 没有快照代表 room-service 不认识该房间。 return &roomv1.CheckSpeakPermissionResponse{ Allowed: false, Reason: "room_not_found", }, nil } // 允许发言必须同时满足未 ban、未 mute、房间开启聊天、用户在业务 presence 内。 if !containsID(snapshot.GetBanUserIds(), req.GetUserId()) && !containsID(snapshot.GetMuteUserIds(), req.GetUserId()) && snapshot.GetChatEnabled() && findProtoUser(snapshot, req.GetUserId()) != nil { return &roomv1.CheckSpeakPermissionResponse{ Allowed: true, RoomVersion: snapshot.GetVersion(), }, nil } reason := "not_in_room" switch { case containsID(snapshot.GetBanUserIds(), req.GetUserId()): // ban 优先级最高,被踢用户即使仍有旧 WS 连接也不能发言。 reason = "user_banned" case containsID(snapshot.GetMuteUserIds(), req.GetUserId()): reason = "user_muted" case !snapshot.GetChatEnabled(): reason = "chat_disabled" } return &roomv1.CheckSpeakPermissionResponse{ Allowed: false, Reason: reason, RoomVersion: snapshot.GetVersion(), }, nil } // VerifyRoomPresence 给 im-service 提供订阅房间前的同步 presence 守卫。 func (s *Service) VerifyRoomPresence(ctx context.Context, req *roomv1.VerifyRoomPresenceRequest) (*roomv1.VerifyRoomPresenceResponse, error) { // subscribe_room 必须通过该守卫,不能只依赖客户端先调用过 JoinRoom。 snapshot, err := s.currentSnapshot(ctx, req.GetRoomId()) if err != nil { return nil, err } if snapshot == nil { // 房间不存在时 im-service 不应把连接加入房间频道。 return &roomv1.VerifyRoomPresenceResponse{ Present: false, Reason: "room_not_found", }, nil } if containsID(snapshot.GetBanUserIds(), req.GetUserId()) { // ban 用户不能订阅,即使快照里还有旧 presence 也以 ban 为准。 return &roomv1.VerifyRoomPresenceResponse{ Present: false, Reason: "user_banned", RoomVersion: snapshot.GetVersion(), }, nil } if findProtoUser(snapshot, req.GetUserId()) == nil { // 没有 room-service presence 的用户不能偷收房间消息。 return &roomv1.VerifyRoomPresenceResponse{ Present: false, Reason: "not_in_room", RoomVersion: snapshot.GetVersion(), }, nil } return &roomv1.VerifyRoomPresenceResponse{ Present: true, RoomVersion: snapshot.GetVersion(), }, nil } // ProcessPendingOutbox 把待投递事件补偿性地推送给异步消费者。 func (s *Service) ProcessPendingOutbox(ctx context.Context, limit int) error { // 补偿 worker 从 repository 扫描 pending 事件;同步 room->im 失败不会回滚房间状态。 records, err := s.repository.ListPendingOutbox(ctx, limit) if err != nil { return err } for _, record := range records { if err := s.outboxPublisher.PublishOutboxEvent(ctx, record.Envelope); err != nil { // 投递失败保持 pending,并记录错误和重试次数,等待下一轮 worker 再试。 if markErr := s.repository.MarkOutboxFailed(ctx, record.EventID, err.Error()); markErr != nil { return markErr } continue } // 只有外部发布成功后才能标记 published。 if err := s.repository.MarkOutboxPublished(ctx, record.EventID); err != nil { return err } } return nil } func (s *Service) mutateRoom(ctx context.Context, cmd command.Command, replayable bool, mutate func(now time.Time, current *state.RoomState, localRank *rank.LocalRank) (mutationResult, []outbox.Record, error)) (mutationResult, error) { if cmd.RoomID() == "" || cmd.ID() == "" || cmd.ActorUserID() == 0 { // command meta 是幂等、路由、审计和权限判断的共同基础,缺一不可。 return mutationResult{}, xerr.New(xerr.InvalidArgument, "command meta is incomplete") } if seen, err := s.repository.HasCommand(ctx, cmd.RoomID(), cmd.ID()); err != nil { return mutationResult{}, err } else if seen { // 幂等命令不重新执行 mutate,也不会再次调用 wallet 或写 outbox。 snapshot, err := s.currentSnapshot(ctx, cmd.RoomID()) if err != nil { return mutationResult{}, err } return mutationResult{snapshot: snapshot}, nil } // 确保当前节点持有房间 lease,并在没有内存 Cell 时完成恢复。 roomCell, err := s.ensureCell(ctx, cmd.RoomID()) if err != nil { return mutationResult{}, err } value, err := roomCell.Do(ctx, func(current *state.RoomState, currentRank *rank.LocalRank) (any, error) { // 在副本上执行命令,只有 command log/outbox 保存成功后才整体替换当前状态。 nextState := current.Clone() nextRank := currentRank.Clone() now := s.clock.Now() result, outboxRecords, err := mutate(now, nextState, nextRank) if err != nil { // 校验、钱包或业务失败时不提交状态,不写 command log。 return nil, err } if result.snapshot == nil { // 无特殊返回需求的命令默认返回变更后的快照。 result.snapshot = nextState.ToProto() } if result.snapshot.GetVersion() != current.Version { // 版本变化代表本命令产生实际状态变更,需要持久化命令和事件。 payload, err := command.Serialize(cmd) if err != nil { return nil, err } if err := s.repository.SaveCommand(ctx, CommandRecord{ RoomID: cmd.RoomID(), RoomVersion: nextState.Version, CommandID: cmd.ID(), CommandType: cmd.Type(), Payload: payload, Replayable: replayable, CreatedAt: now, }); err != nil { return nil, err } if len(outboxRecords) > 0 { // outbox 与 command log 共同构成恢复和房间外投递来源。 if err := s.repository.SaveOutbox(ctx, outboxRecords); err != nil { return nil, err } } // 持久化成功后再替换内存状态,避免内存提交领先于恢复来源。 current.ReplaceWith(nextState) currentRank.ReplaceWith(nextRank) result.snapshot = current.ToProto() result.applied = true } return result, nil }) if err != nil { return mutationResult{}, err } result := value.(mutationResult) if result.applied && shouldSnapshot(result.snapshot, s.snapshotEveryN) { // 快照失败时返回错误,因为没有快照会增加恢复成本并可能影响 guard 查询。 if err := s.persistSnapshot(ctx, result.snapshot); err != nil { return mutationResult{}, err } } if result.applied && result.syncEvent != nil { // 同步 im-service 失败不回滚房间状态;outbox 后续负责补偿投递。 _ = s.syncPublisher.PublishRoomEvent(ctx, result.syncEvent) } return result, nil } func (s *Service) ensureCell(ctx context.Context, roomID string) (*cell.RoomCell, error) { // 每次写命令先刷新或接管 lease,确保同一房间同一时刻只有一个执行节点。 lease, err := s.directory.EnsureOwner(ctx, roomID, s.nodeID, s.clock.Now(), s.leaseTTL) if err != nil { return nil, err } if lease.NodeID != s.nodeID { // 仍有其他有效 owner 时当前节点不能执行该房间命令。 return nil, xerr.New(xerr.Conflict, fmt.Sprintf("room is owned by %s", lease.NodeID)) } if roomCell := s.loadCell(roomID); roomCell != nil { // 本节点已经装载该房间,直接复用已有 Room Cell。 return roomCell, nil } // 没有内存 Cell 时从持久化恢复,支持 lease 过期后的新节点接管。 roomMeta, exists, err := s.repository.GetRoomMeta(ctx, roomID) if err != nil { return nil, err } if !exists { // 没有 meta 说明房间从未创建,不能凭请求临时造状态。 return nil, xerr.New(xerr.NotFound, "room not found") } restoredState, err := s.recoverRoom(ctx, roomMeta) if err != nil { // 恢复失败时返回 unavailable,调用方应重试或等待节点修复,而不是返回 room not found。 return nil, xerr.New(xerr.Unavailable, fmt.Sprintf("recover room failed: %v", err)) } // 恢复完成后创建新的 Room Cell,并从恢复后的 GiftRank 重建 LocalRank 索引。 roomCell := cell.New(restoredState, rank.FromState(restoredState.GiftRank, s.rankLimit)) s.installCell(roomID, roomCell) return roomCell, nil } func (s *Service) recoverRoom(ctx context.Context, roomMeta RoomMeta) (*state.RoomState, error) { // 恢复优先从最新快照开始,再回放快照版本之后的 replayable command log。 snapshotRecord, hasSnapshot, err := s.repository.GetLatestSnapshot(ctx, roomMeta.RoomID) if err != nil { return nil, err } var recovered *state.RoomState var fromVersion int64 if hasSnapshot { // 快照 payload 是 roomv1.RoomSnapshot 的 protobuf 序列化结果。 var snapshot roomv1.RoomSnapshot if err := proto.Unmarshal(snapshotRecord.Payload, &snapshot); err != nil { return nil, err } recovered = state.FromProto(&snapshot) fromVersion = snapshot.GetVersion() } else { // 没有快照时从 rooms meta 重建初始状态,再从版本 0 回放命令。 recovered = state.NewRoomState(roomMeta.RoomID, roomMeta.OwnerUserID, roomMeta.HostUserID, roomMeta.SeatCount, roomMeta.Mode) recovered.Status = roomMeta.Status fromVersion = 0 } records, err := s.repository.ListCommandsAfter(ctx, roomMeta.RoomID, fromVersion) if err != nil { return nil, err } for _, record := range records { if !record.Replayable { // 不可回放命令保留审计价值,但不参与状态重建。 continue } cmd, err := command.Deserialize(record.CommandType, record.Payload) if err != nil { return nil, err } if cmd == nil { // 未知命令类型跳过,保留兼容未来命令的恢复容错。 continue } if err := replay(recovered, cmd); err != nil { // replay 失败说明命令日志和当前状态不兼容,不能继续装载错误 Cell。 return nil, err } } return recovered, nil } func replay(current *state.RoomState, cmd command.Command) error { // replay 只重建房间内状态,不重复写 outbox、不重复调用 wallet、不重复同步 im-service。 switch typed := cmd.(type) { case *command.CreateRoom: // 创建命令的 meta 已由 RoomMeta 或快照恢复,这里只保证版本进入初始提交态。 current.Version = 1 case *command.JoinRoom: // presence 使用命令发送时间恢复,真实连接需要客户端重新订阅 im-service。 current.OnlineUsers[typed.ActorUserID()] = &state.RoomUserState{ UserID: typed.ActorUserID(), Role: typed.Role, JoinedAtMS: typed.SentAtMS, LastSeenAtMS: typed.SentAtMS, } current.Version++ case *command.LeaveRoom: // 离房回放只删除业务 presence,不涉及 im-service session。 delete(current.OnlineUsers, typed.ActorUserID()) current.Version++ case *command.MicUp: index := current.SeatIndex(typed.SeatNo) if index >= 0 { // 回放对不存在麦位保持容错,避免旧命令阻塞新结构恢复。 current.MicSeats[index].UserID = typed.ActorUserID() current.Version++ } case *command.MicDown: if seat, exists := current.SeatByUser(typed.TargetUserID); exists { // 只有目标仍在麦上时才清空,重复或乱序异常不会制造负状态。 index := current.SeatIndex(seat.SeatNo) current.MicSeats[index].UserID = 0 current.Version++ } case *command.ChangeMicSeat: if from, exists := current.SeatByUser(typed.TargetUserID); exists { // 换位回放必须同时找到源麦位和目标麦位。 fromIndex := current.SeatIndex(from.SeatNo) toIndex := current.SeatIndex(typed.SeatNo) if fromIndex >= 0 && toIndex >= 0 { current.MicSeats[fromIndex].UserID = 0 current.MicSeats[toIndex].UserID = typed.TargetUserID current.Version++ } } case *command.MuteUser: if typed.Muted { // 回放禁言集合,供 CheckSpeakPermission 使用。 current.MuteUsers[typed.TargetUserID] = true } else { delete(current.MuteUsers, typed.TargetUserID) } current.Version++ case *command.KickUser: if seat, exists := current.SeatByUser(typed.TargetUserID); exists { // 被踢用户恢复后不能继续占麦位。 index := current.SeatIndex(seat.SeatNo) current.MicSeats[index].UserID = 0 } delete(current.OnlineUsers, typed.TargetUserID) current.BanUsers[typed.TargetUserID] = true current.Version++ case *command.SendGift: // 送礼回放不能再次调用 wallet,只使用命令中记录的礼物价值重建展示态。 total := typed.GiftUnitValue * int64(typed.GiftCount) current.Heat += total applied := false for index := range current.GiftRank { if current.GiftRank[index].UserID == typed.ActorUserID() { current.GiftRank[index].Score += total current.GiftRank[index].GiftValue += total current.GiftRank[index].UpdatedAtMS = typed.SentAtMS applied = true break } } if !applied { // 用户首次出现在榜单时追加榜单项,排序由 LocalRank/快照展示阶段处理。 current.GiftRank = append(current.GiftRank, state.RankItem{ UserID: typed.ActorUserID(), Score: total, GiftValue: total, UpdatedAtMS: typed.SentAtMS, }) } current.Version++ } return nil } func (s *Service) currentSnapshot(ctx context.Context, roomID string) (*roomv1.RoomSnapshot, error) { if roomCell := s.loadCell(roomID); roomCell != nil { // 内存 Cell 是最新状态来源,guard 查询优先读它。 currentState, _, err := roomCell.Snapshot(ctx) if err != nil { return nil, err } return currentState.ToProto(), nil } // 没有内存 Cell 时从持久化快照读取,支持 im-service guard 在恢复前查询。 record, exists, err := s.repository.GetLatestSnapshot(ctx, roomID) if err != nil || !exists { return nil, err } var snapshot roomv1.RoomSnapshot if err := proto.Unmarshal(record.Payload, &snapshot); err != nil { // 快照反序列化失败说明持久化数据不可用,不能返回半状态。 return nil, err } return &snapshot, nil } func (s *Service) persistSnapshot(ctx context.Context, snapshot *roomv1.RoomSnapshot) error { // 快照统一使用 protobuf,保持和跨服务契约一致。 payload, err := proto.Marshal(snapshot) if err != nil { return err } return s.repository.SaveSnapshot(ctx, SnapshotRecord{ RoomID: snapshot.GetRoomId(), RoomVersion: snapshot.GetVersion(), Payload: payload, CreatedAt: s.clock.Now(), }) } func (s *Service) installCell(roomID string, roomCell *cell.RoomCell) { s.mu.Lock() defer s.mu.Unlock() // 注册表只保存当前进程内已装载 Cell,不代表 Redis lease 的全局 owner。 s.cells[roomID] = roomCell } func (s *Service) loadCell(roomID string) *cell.RoomCell { s.mu.Lock() defer s.mu.Unlock() // 返回指针后由 RoomCell 自己的 mailbox 保证状态串行化。 return s.cells[roomID] } func shouldSnapshot(snapshot *roomv1.RoomSnapshot, every int64) bool { if snapshot == nil || every <= 1 { // every<=1 表示每次有快照结果都保存,nil 快照永远不保存。 return snapshot != nil } // 按房间版本取模做简单采样,避免引入额外计数状态。 return snapshot.GetVersion()%every == 0 } func commandResult(applied bool, roomVersion int64, now time.Time) *roomv1.CommandResult { // 所有命令响应统一带 applied、room_version 和服务端时间,便于客户端处理幂等结果。 return &roomv1.CommandResult{ Applied: applied, RoomVersion: roomVersion, ServerTimeMs: now.UnixMilli(), } } func baseFromMeta(meta *roomv1.RequestMeta) command.Base { if meta == nil { // 上层 mutateRoom 会校验关键字段,nil meta 在这里先收敛为空 Base。 return command.Base{} } // RequestMeta 是跨服务 protobuf 契约,领域层统一转换成 command.Base。 return command.Base{ RequestID: meta.GetRequestId(), CommandID: meta.GetCommandId(), ActorID: meta.GetActorUserId(), Room: meta.GetRoomId(), GatewayNodeID: meta.GetGatewayNodeId(), SessionID: meta.GetSessionId(), SentAtMS: meta.GetSentAtMs(), } } func findProtoUser(snapshot *roomv1.RoomSnapshot, userID int64) *roomv1.RoomUser { if snapshot == nil { return nil } // protobuf 快照用户列表规模受房间在线人数限制,线性扫描足够简单。 for _, user := range snapshot.GetOnlineUsers() { if user.GetUserId() == userID { return user } } return nil } func containsID(values []int64, target int64) bool { // 快照里的集合以有序数组表达,守卫查询用线性查找即可。 for _, value := range values { if value == target { return true } } return false } func cloneProtoRank(items []state.RankItem) []*roomv1.RankItem { // SendGift 响应需要 protobuf 榜单项,不能直接暴露内部 state.RankItem。 result := make([]*roomv1.RankItem, 0, len(items)) for _, item := range items { result = append(result, &roomv1.RankItem{ UserId: item.UserID, Score: item.Score, GiftValue: item.GiftValue, UpdatedAtMs: item.UpdatedAtMS, }) } return result } func findRankItem(items []state.RankItem, userID int64) state.RankItem { // 送礼后通常能找到发送方榜单项;找不到时返回 user_id 占位避免空事件。 for _, item := range items { if item.UserID == userID { return item } } return state.RankItem{UserID: userID} } // NewRequestMeta 是测试和 stub 组装命令元信息时的便捷函数。 func NewRequestMeta(roomID string, actorUserID int64) *roomv1.RequestMeta { // 该函数只用于测试和 stub,真实入口应由 gateway-service 生成稳定 RequestMeta。 return &roomv1.RequestMeta{ RequestId: idgen.New("req"), CommandId: idgen.New("cmd"), ActorUserId: actorUserID, RoomId: roomID, GatewayNodeId: "gateway-local", SessionId: idgen.New("sess"), SentAtMs: time.Now().UnixMilli(), } }