2026-07-23 16:47:40 +08:00

494 lines
19 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package service
import (
"context"
"strings"
"time"
"unicode/utf8"
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
roomv1 "hyapp.local/api/proto/room/v1"
"hyapp/pkg/appcode"
"hyapp/pkg/tencentim"
"hyapp/pkg/xerr"
"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"
)
func (s *Service) AdminListRooms(ctx context.Context, req *roomv1.AdminListRoomsRequest) (*roomv1.AdminListRoomsResponse, error) {
ctx = appcode.WithContext(ctx, req.GetMeta().GetAppCode())
rooms, total, err := s.repository.AdminListRooms(ctx, AdminRoomListQuery{
AppCode: appcode.FromContext(ctx),
Page: int(req.GetPage()),
PageSize: int(req.GetPageSize()),
Keyword: strings.TrimSpace(req.GetKeyword()),
Status: strings.TrimSpace(req.GetStatus()),
RegionID: req.GetVisibleRegionId(),
OwnerUserID: req.GetOwnerUserId(),
OwnerUserIDs: positiveUniqueRoomOwnerIDs(req.GetOwnerUserIds()),
SortBy: strings.TrimSpace(req.GetSortBy()),
SortDirection: strings.TrimSpace(req.GetSortDirection()),
NowMS: s.clock.Now().UnixMilli(),
})
if err != nil {
return nil, err
}
items := make([]*roomv1.AdminRoomListItem, 0, len(rooms))
for _, room := range rooms {
items = append(items, adminRoomListItemProto(room))
}
return &roomv1.AdminListRoomsResponse{Rooms: items, Total: total, ServerTimeMs: s.clock.Now().UnixMilli()}, nil
}
func (s *Service) AdminGetRoom(ctx context.Context, req *roomv1.AdminGetRoomRequest) (*roomv1.AdminGetRoomResponse, error) {
ctx = appcode.WithContext(ctx, req.GetMeta().GetAppCode())
room, exists, err := s.repository.AdminGetRoom(ctx, req.GetRoomId(), s.clock.Now().UnixMilli())
if err != nil {
return nil, err
}
if !exists {
return nil, xerr.New(xerr.NotFound, "room not found")
}
return &roomv1.AdminGetRoomResponse{Room: adminRoomListItemProto(room), ServerTimeMs: s.clock.Now().UnixMilli()}, nil
}
func (s *Service) AdminUpdateRoom(ctx context.Context, req *roomv1.AdminUpdateRoomRequest) (*roomv1.AdminUpdateRoomResponse, error) {
ctx = contextFromMeta(ctx, req.GetMeta())
cmd, err := s.adminUpdateCommand(req)
if err != nil {
return nil, err
}
result, err := s.applyAdminRoomMutation(ctx, cmd, cmd)
if err != nil {
return nil, err
}
return &roomv1.AdminUpdateRoomResponse{
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
Room: result.snapshot,
}, nil
}
// AdminBanRoom 通过 Room Cell 原子持久化 banned 状态、清场事实和命令日志。
func (s *Service) AdminBanRoom(ctx context.Context, req *roomv1.AdminBanRoomRequest) (*roomv1.AdminBanRoomResponse, error) {
ctx = contextFromMeta(ctx, req.GetMeta())
if req == nil || req.GetMeta() == nil || strings.TrimSpace(req.GetMeta().GetRoomId()) == "" {
return nil, xerr.New(xerr.InvalidArgument, "room_id is required")
}
reason := strings.TrimSpace(req.GetReason())
if reason == "" {
return nil, xerr.New(xerr.InvalidArgument, "ban reason is required")
}
cmd := command.AdminBanRoom{
Base: baseFromMeta(req.GetMeta()),
Reason: reason,
AdminID: req.GetAdminId(),
AdminName: strings.TrimSpace(req.GetAdminName()),
}
rtcKickTargets := make([]int64, 0)
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
if current == nil {
return mutationResult{}, nil, xerr.New(xerr.NotFound, "room not found")
}
if current.Status == state.RoomStatusBanned {
// command_id 之外再提供状态幂等,重复人工操作不能重复发送踢人事实。
return mutationResult{snapshot: current.ToProto()}, nil, nil
}
if current.Status == state.RoomStatusDeleted {
return mutationResult{}, nil, xerr.New(xerr.NotFound, "room not found")
}
if err := s.ensureNoRecoverableGiftOperations(ctx, current.RoomID); err != nil {
return mutationResult{}, nil, err
}
kickTargets := closeRoomKickTargets(current)
rtcKickTargets = append(rtcKickTargets[:0], kickTargets...)
type occupiedMicSeat struct {
userID int64
seatNo int32
micSessionID string
}
occupiedSeats := make([]occupiedMicSeat, 0, len(current.MicSeats))
for _, seat := range current.MicSeats {
if seat.UserID != 0 && seat.MicSessionID != "" {
occupiedSeats = append(occupiedSeats, occupiedMicSeat{userID: seat.UserID, seatNo: seat.SeatNo, micSessionID: seat.MicSessionID})
}
}
current.Status = state.RoomStatusBanned
current.OnlineUsers = map[int64]*state.RoomUserState{}
for index := range current.MicSeats {
current.ClearMicSession(index)
}
current.Version++
records := make([]outbox.Record, 0, len(kickTargets)+len(occupiedSeats)+1)
bannedEvent, err := outbox.Build(current.RoomID, "RoomBanned", current.Version, now, &roomeventsv1.RoomBanned{
ActorUserId: cmd.ActorUserID(),
Reason: cmd.Reason,
})
if err != nil {
return mutationResult{}, nil, err
}
records = append(records, bannedEvent)
for _, targetUserID := range kickTargets {
kickEvent, buildErr := outbox.Build(current.RoomID, "RoomUserKicked", current.Version, now, &roomeventsv1.RoomUserKicked{
ActorUserId: cmd.ActorUserID(),
TargetUserId: targetUserID,
})
if buildErr != nil {
return mutationResult{}, nil, buildErr
}
records = append(records, kickEvent)
}
for _, seat := range occupiedSeats {
micEvent, buildErr := buildMicDownEventForSeat(current.RoomID, current.Version, now, cmd.ActorUserID(), seat.userID, seat.seatNo, seat.micSessionID, "room_banned")
if buildErr != nil {
return mutationResult{}, nil, buildErr
}
records = append(records, micEvent)
}
return mutationResult{
snapshot: current.ToProto(),
roomStatus: state.RoomStatusBanned,
closeInfo: &RoomCloseInfo{Reason: cmd.Reason, AdminID: cmd.AdminID, AdminName: cmd.AdminName, ClosedAtMS: now.UnixMilli()},
syncEvent: &tencentim.RoomEvent{
EventID: bannedEvent.EventID,
RoomID: current.RoomID,
EventType: "room_banned",
ActorUserID: tencentim.FormatOptionalUserID(cmd.ActorUserID()),
RoomVersion: current.Version,
Attributes: map[string]string{"reason": cmd.Reason},
},
}, records, nil
})
if err != nil {
return nil, err
}
if result.applied && s.rtcUserRemover != nil {
for _, targetUserID := range rtcKickTargets {
// RTC 属于连接态;房间封禁事实提交成功后才执行外部移除,失败不回滚 Room Cell。
_ = s.rtcUserRemover.RemoveUserByStrRoomID(ctx, cmd.RoomID(), targetUserID)
}
}
return &roomv1.AdminBanRoomResponse{
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
Room: result.snapshot,
}, nil
}
// AdminUnbanRoom 只执行 banned -> closed房主后续按普通进房路径恢复 active。
func (s *Service) AdminUnbanRoom(ctx context.Context, req *roomv1.AdminUnbanRoomRequest) (*roomv1.AdminUnbanRoomResponse, error) {
ctx = contextFromMeta(ctx, req.GetMeta())
if req == nil || req.GetMeta() == nil || strings.TrimSpace(req.GetMeta().GetRoomId()) == "" {
return nil, xerr.New(xerr.InvalidArgument, "room_id is required")
}
reason := strings.TrimSpace(req.GetReason())
if reason == "" {
return nil, xerr.New(xerr.InvalidArgument, "unban reason is required")
}
cmd := command.AdminUnbanRoom{
Base: baseFromMeta(req.GetMeta()),
Reason: reason,
AdminID: req.GetAdminId(),
AdminName: strings.TrimSpace(req.GetAdminName()),
}
result, err := s.mutateRoom(ctx, cmd, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
if current == nil {
return mutationResult{}, nil, xerr.New(xerr.NotFound, "room not found")
}
if current.Status != state.RoomStatusBanned {
return mutationResult{}, nil, xerr.New(xerr.Conflict, "room is not banned")
}
current.Status = state.RoomStatusClosed
current.Version++
unbannedEvent, buildErr := outbox.Build(current.RoomID, "RoomUnbanned", current.Version, now, &roomeventsv1.RoomUnbanned{
ActorUserId: cmd.ActorUserID(),
Reason: cmd.Reason,
})
if buildErr != nil {
return mutationResult{}, nil, buildErr
}
return mutationResult{
snapshot: current.ToProto(),
roomStatus: state.RoomStatusClosed,
closeInfo: &RoomCloseInfo{ClearOnOpen: true},
}, []outbox.Record{unbannedEvent}, nil
})
if err != nil {
return nil, err
}
return &roomv1.AdminUnbanRoomResponse{
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
Room: result.snapshot,
}, nil
}
func (s *Service) AdminDeleteRoom(ctx context.Context, req *roomv1.AdminDeleteRoomRequest) (*roomv1.AdminDeleteRoomResponse, error) {
ctx = contextFromMeta(ctx, req.GetMeta())
if req.GetMeta() == nil || strings.TrimSpace(req.GetMeta().GetRoomId()) == "" {
return nil, xerr.New(xerr.InvalidArgument, "room_id is required")
}
cmd := command.AdminDeleteRoom{
Base: baseFromMeta(req.GetMeta()),
AdminID: req.GetAdminId(),
AdminName: strings.TrimSpace(req.GetAdminName()),
}
result, err := s.mutateRoom(ctx, cmd, false, func(_ time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
if current == nil {
return mutationResult{}, nil, xerr.New(xerr.NotFound, "room not found")
}
if err := s.ensureNoRecoverableGiftOperations(ctx, current.RoomID); err != nil {
return mutationResult{}, nil, err
}
// 删除不是“关闭”的另一种展示状态,而是后台确认移除房间恢复来源;状态只服务本次响应和内存 Cell 收尾。
current.Status = state.RoomStatusDeleted
current.OnlineUsers = map[int64]*state.RoomUserState{}
for index := range current.MicSeats {
current.ClearMicSession(index)
}
current.Version++
return mutationResult{
snapshot: current.ToProto(),
deleteRoom: true,
}, nil, nil
})
if err != nil {
return nil, err
}
if result.applied {
s.deleteRoomListCacheBestEffort(ctx, cmd.RoomID())
s.forgetLoadedRoom(ctx, cmd.RoomID())
}
return &roomv1.AdminDeleteRoomResponse{
Result: commandResult(result.applied, result.snapshot.GetVersion(), s.clock.Now()),
Room: result.snapshot,
}, nil
}
func positiveUniqueRoomOwnerIDs(values []int64) []int64 {
seen := make(map[int64]struct{}, len(values))
result := make([]int64, 0, len(values))
for _, value := range values {
if value <= 0 {
continue
}
if _, exists := seen[value]; exists {
continue
}
seen[value] = struct{}{}
result = append(result, value)
}
return result
}
func (s *Service) adminUpdateCommand(req *roomv1.AdminUpdateRoomRequest) (command.AdminUpdateRoom, error) {
if req == nil || req.GetMeta() == nil || strings.TrimSpace(req.GetMeta().GetRoomId()) == "" {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room_id is required")
}
cmd := command.AdminUpdateRoom{
Base: baseFromMeta(req.GetMeta()),
CloseReason: strings.TrimSpace(req.GetCloseReason()),
AdminID: req.GetAdminId(),
AdminName: strings.TrimSpace(req.GetAdminName()),
}
if req.Title != nil {
value := strings.TrimSpace(req.GetTitle())
if value == "" {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room title is required")
}
if utf8.RuneCountInString(value) > maxRoomNameRunes {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room title is too long")
}
cmd.Title = &value
}
if req.CoverUrl != nil {
value := strings.TrimSpace(req.GetCoverUrl())
if utf8.RuneCountInString(value) > maxRoomAvatarRunes {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room cover is too long")
}
if value == "" {
value = defaultRoomAvatar
}
cmd.CoverURL = &value
}
if req.Description != nil {
value := strings.TrimSpace(req.GetDescription())
if utf8.RuneCountInString(value) > maxRoomDescriptionRunes {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room description is too long")
}
cmd.Description = &value
}
if req.Mode != nil {
value := strings.TrimSpace(req.GetMode())
if value == "" || utf8.RuneCountInString(value) > 64 {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room mode is invalid")
}
cmd.Mode = &value
}
if req.Status != nil {
value := strings.TrimSpace(req.GetStatus())
switch value {
case state.RoomStatusActive, state.RoomStatusClosed:
cmd.Status = &value
default:
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "room status is invalid")
}
if value != state.RoomStatusActive && strings.TrimSpace(cmd.CloseReason) == "" {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "close reason is required")
}
}
if req.OwnerCountryCode != nil {
value := normalizeRoomCountryCode(req.GetOwnerCountryCode())
if value == "" {
// 国家码只服务内部同步和运维 backfill入口仍按列表筛选同一套规则拒绝非法值。
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "owner country code is invalid")
}
cmd.OwnerCountryCode = &value
}
if req.VisibleRegionId != nil {
value := req.GetVisibleRegionId()
if value < 0 {
return command.AdminUpdateRoom{}, xerr.New(xerr.InvalidArgument, "visible_region_id is invalid")
}
cmd.VisibleRegionID = &value
}
return cmd, nil
}
func (s *Service) applyAdminRoomMutation(ctx context.Context, logCommand command.Command, cmd command.AdminUpdateRoom) (mutationResult, error) {
return s.mutateRoom(ctx, logCommand, true, func(now time.Time, current *state.RoomState, _ *rank.LocalRank) (mutationResult, []outbox.Record, error) {
if current == nil {
return mutationResult{}, nil, xerr.New(xerr.NotFound, "room not found")
}
if current.Status == state.RoomStatusBanned && cmd.Status != nil {
// banned 只能通过 AdminUnbanRoom 解除,通用资料修改不得伪造 active/closed 跳过审计。
return mutationResult{}, nil, xerr.New(xerr.Conflict, "banned room status requires dedicated unban command")
}
if cmd.Status != nil && (*cmd.Status == state.RoomStatusClosed || *cmd.Status == state.RoomStatusDeleted) {
if err := s.ensureNoRecoverableGiftOperations(ctx, current.RoomID); err != nil {
return mutationResult{}, nil, err
}
}
changed := applyAdminRoomPatch(current, cmd)
records := make([]outbox.Record, 0, 2)
result := mutationResult{
// 房主国家是列表国家筛选和恢复态的一致性字段,内部 backfill 必须立刻把 RoomExt 写进 snapshot。
forceSnapshot: cmd.OwnerCountryCode != nil,
}
if cmd.Status != nil {
switch *cmd.Status {
case state.RoomStatusClosed, state.RoomStatusDeleted:
kickTargets := closeRoomKickTargets(current)
current.OnlineUsers = map[int64]*state.RoomUserState{}
for index := range current.MicSeats {
current.ClearMicSession(index)
}
closedEvent, err := outbox.Build(current.RoomID, "RoomClosed", current.Version+1, now, &roomeventsv1.RoomClosed{
ActorUserId: cmd.ActorUserID(),
Reason: cmd.CloseReason,
})
if err != nil {
return mutationResult{}, nil, err
}
records = append(records, closedEvent)
for _, targetUserID := range kickTargets {
kickEvent, err := outbox.Build(current.RoomID, "RoomUserKicked", current.Version+1, now, &roomeventsv1.RoomUserKicked{
ActorUserId: cmd.ActorUserID(),
TargetUserId: targetUserID,
})
if err != nil {
return mutationResult{}, nil, err
}
records = append(records, kickEvent)
}
result.roomStatus = *cmd.Status
result.closeInfo = &RoomCloseInfo{Reason: cmd.CloseReason, AdminID: cmd.AdminID, AdminName: cmd.AdminName, ClosedAtMS: now.UnixMilli()}
result.syncEvent = &tencentim.RoomEvent{EventID: closedEvent.EventID, RoomID: current.RoomID, EventType: "room_closed", ActorUserID: tencentim.FormatOptionalUserID(cmd.ActorUserID()), RoomVersion: current.Version + 1}
case state.RoomStatusActive:
result.roomStatus = state.RoomStatusActive
result.closeInfo = &RoomCloseInfo{ClearOnOpen: true}
}
}
if cmd.Mode != nil {
result.roomMode = cmd.Mode
}
if cmd.VisibleRegionID != nil {
result.visibleRegionID = cmd.VisibleRegionID
}
if !changed && len(records) == 0 && result.roomMode == nil && result.visibleRegionID == nil && result.roomStatus == "" {
result.snapshot = current.ToProto()
return result, nil, nil
}
current.Version++
result.snapshot = current.ToProto()
return result, records, nil
})
}
func applyAdminRoomPatch(current *state.RoomState, cmd command.AdminUpdateRoom) bool {
changed := false
if current.RoomExt == nil {
current.RoomExt = map[string]string{}
}
if cmd.Title != nil && current.RoomExt[roomExtTitleKey] != *cmd.Title {
current.RoomExt[roomExtTitleKey] = *cmd.Title
changed = true
}
if cmd.CoverURL != nil && current.RoomExt[roomExtCoverURLKey] != *cmd.CoverURL {
current.RoomExt[roomExtCoverURLKey] = *cmd.CoverURL
changed = true
}
if cmd.Description != nil && current.RoomExt[roomExtDescriptionKey] != *cmd.Description {
current.RoomExt[roomExtDescriptionKey] = *cmd.Description
changed = true
}
if cmd.Mode != nil && current.Mode != *cmd.Mode {
current.Mode = *cmd.Mode
changed = true
}
if cmd.VisibleRegionID != nil && current.VisibleRegionID != *cmd.VisibleRegionID {
current.VisibleRegionID = *cmd.VisibleRegionID
changed = true
}
if cmd.OwnerCountryCode != nil {
// 房主国家只作为发现页分类投影字段保存在 RoomExt后台 HTTP 不开放直改入口,只有 user-service 国家事实同步会写入。
ownerCountryCode := normalizeRoomCountryCode(*cmd.OwnerCountryCode)
if current.RoomExt[roomExtOwnerCountryKey] != ownerCountryCode {
current.RoomExt[roomExtOwnerCountryKey] = ownerCountryCode
changed = true
}
}
if cmd.Status != nil && current.Status != *cmd.Status {
current.Status = *cmd.Status
changed = true
}
return changed
}
func adminRoomListItemProto(item AdminRoomListEntry) *roomv1.AdminRoomListItem {
return &roomv1.AdminRoomListItem{
RoomId: item.RoomID,
RoomShortId: item.RoomShortID,
VisibleRegionId: item.VisibleRegionID,
OwnerUserId: item.OwnerUserID,
Title: item.Title,
CoverUrl: item.CoverURL,
Mode: item.Mode,
Status: item.Status,
CloseReason: item.CloseReason,
ClosedByAdminId: item.ClosedByAdminID,
ClosedByAdminName: item.ClosedByAdminName,
ClosedAtMs: item.ClosedAtMS,
Heat: item.Heat,
OnlineCount: item.OnlineCount,
SeatCount: item.SeatCount,
OccupiedSeatCount: item.OccupiedSeatCount,
SortScore: item.SortScore,
CreatedAtMs: item.CreatedAtMS,
UpdatedAtMs: item.UpdatedAtMS,
Locked: item.Locked,
}
}