1344 lines
43 KiB
Go
1344 lines
43 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"log/slog"
|
||
"math/rand"
|
||
"slices"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
roomv1 "hyapp.local/api/proto/room/v1"
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/logx"
|
||
"hyapp/pkg/roomid"
|
||
"hyapp/pkg/xerr"
|
||
)
|
||
|
||
const (
|
||
robotRoomStatusActive = "active"
|
||
robotRoomStatusStopped = "stopped"
|
||
robotRoomMode = "voice"
|
||
robotRoomLuckyPoolID = "robot_lucky_display"
|
||
|
||
maxRobotLuckyComboPerCycle = int64(3)
|
||
|
||
defaultRobotStayMinMS = int64(3 * 60 * 1000)
|
||
defaultRobotStayMaxMS = int64(10 * 60 * 1000)
|
||
defaultRobotReplaceMinMS = int64(0)
|
||
defaultRobotReplaceMaxMS = int64(60 * 1000)
|
||
defaultMaxGiftSenders = int64(1)
|
||
|
||
defaultRobotRoomSeatCount = int32(10)
|
||
)
|
||
|
||
func (s *Service) AdminListRobotRooms(ctx context.Context, req *roomv1.AdminListRobotRoomsRequest) (*roomv1.AdminListRobotRoomsResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
items, total, err := s.repository.ListRobotRooms(ctx, RobotRoomListQuery{
|
||
AppCode: appcode.FromContext(ctx),
|
||
Status: strings.TrimSpace(req.GetStatus()),
|
||
Page: int(req.GetPage()),
|
||
PageSize: int(req.GetPageSize()),
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
resp := &roomv1.AdminListRobotRoomsResponse{Total: total, ServerTimeMs: s.clock.Now().UnixMilli()}
|
||
for _, item := range items {
|
||
resp.Rooms = append(resp.Rooms, robotRoomConfigToProto(item))
|
||
}
|
||
return resp, nil
|
||
}
|
||
|
||
func (s *Service) AdminFilterAvailableRoomRobots(ctx context.Context, req *roomv1.AdminFilterAvailableRoomRobotsRequest) (*roomv1.AdminFilterAvailableRoomRobotsResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
ids := normalizeRobotUserIDs(req.GetUserIds())
|
||
occupied, err := s.repository.OccupiedRobotUserIDs(ctx, ids)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
resp := &roomv1.AdminFilterAvailableRoomRobotsResponse{ServerTimeMs: s.clock.Now().UnixMilli()}
|
||
for _, userID := range ids {
|
||
if occupied[userID] {
|
||
resp.OccupiedUserIds = append(resp.OccupiedUserIds, userID)
|
||
continue
|
||
}
|
||
resp.AvailableUserIds = append(resp.AvailableUserIds, userID)
|
||
}
|
||
slices.Sort(resp.AvailableUserIds)
|
||
slices.Sort(resp.OccupiedUserIds)
|
||
return resp, nil
|
||
}
|
||
|
||
func (s *Service) AdminCreateRobotRoom(ctx context.Context, req *roomv1.AdminCreateRobotRoomRequest) (*roomv1.AdminCreateRobotRoomResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
now := s.clock.Now()
|
||
config, err := s.buildRobotRoomConfig(ctx, req, now)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := s.bootstrapRobotRoom(ctx, config); err != nil {
|
||
logx.Error(ctx, "admin_robot_room_bootstrap_failed", err,
|
||
slog.String("room_id", config.RoomID),
|
||
slog.Int64("owner_robot_user_id", config.OwnerRobotUserID),
|
||
slog.Int("robot_count", len(config.RobotUserIDs)),
|
||
)
|
||
return nil, err
|
||
}
|
||
saved, err := s.repository.CreateRobotRoomConfig(ctx, CreateRobotRoomConfigInput{Config: config, NowMS: now.UnixMilli()})
|
||
if err != nil {
|
||
logx.Error(ctx, "admin_robot_room_config_save_failed", err,
|
||
slog.String("room_id", config.RoomID),
|
||
slog.Int64("owner_robot_user_id", config.OwnerRobotUserID),
|
||
slog.Int("robot_count", len(config.RobotUserIDs)),
|
||
)
|
||
return nil, err
|
||
}
|
||
s.startRobotRoomRuntime(ctx, saved)
|
||
return &roomv1.AdminCreateRobotRoomResponse{Room: robotRoomConfigToProto(saved), ServerTimeMs: now.UnixMilli()}, nil
|
||
}
|
||
|
||
func (s *Service) AdminSetRobotRoomStatus(ctx context.Context, req *roomv1.AdminSetRobotRoomStatusRequest) (*roomv1.AdminSetRobotRoomStatusResponse, error) {
|
||
ctx = contextFromMeta(ctx, req.GetMeta())
|
||
roomID := strings.TrimSpace(req.GetMeta().GetRoomId())
|
||
status := normalizeRobotRoomStatus(req.GetStatus())
|
||
if roomID == "" {
|
||
return nil, xerr.New(xerr.InvalidArgument, "room_id is required")
|
||
}
|
||
if status == "" {
|
||
return nil, xerr.New(xerr.InvalidArgument, "robot room status is invalid")
|
||
}
|
||
config, exists, err := s.repository.UpdateRobotRoomStatus(ctx, roomID, status, s.clock.Now().UnixMilli())
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !exists {
|
||
return nil, xerr.New(xerr.NotFound, "robot room not found")
|
||
}
|
||
if status == robotRoomStatusActive {
|
||
s.startRobotRoomRuntime(ctx, config)
|
||
} else {
|
||
s.stopRobotRoomRuntime(roomID)
|
||
}
|
||
return &roomv1.AdminSetRobotRoomStatusResponse{Room: robotRoomConfigToProto(config), ServerTimeMs: s.clock.Now().UnixMilli()}, nil
|
||
}
|
||
|
||
func (s *Service) buildRobotRoomConfig(ctx context.Context, req *roomv1.AdminCreateRobotRoomRequest, now time.Time) (RobotRoomConfig, error) {
|
||
ownerID := req.GetOwnerRobotUserId()
|
||
if ownerID <= 0 {
|
||
return RobotRoomConfig{}, xerr.New(xerr.InvalidArgument, "owner_robot_user_id is required")
|
||
}
|
||
candidates := normalizeRobotUserIDs(req.GetCandidateRobotUserIds())
|
||
containsOwner := false
|
||
for _, userID := range candidates {
|
||
if userID == ownerID {
|
||
containsOwner = true
|
||
break
|
||
}
|
||
}
|
||
if !containsOwner {
|
||
candidates = append(candidates, ownerID)
|
||
}
|
||
occupied, err := s.repository.OccupiedRobotUserIDs(ctx, candidates)
|
||
if err != nil {
|
||
return RobotRoomConfig{}, err
|
||
}
|
||
if occupied[ownerID] {
|
||
return RobotRoomConfig{}, xerr.New(xerr.Conflict, "owner robot already assigned to active robot room")
|
||
}
|
||
available := make([]int64, 0, len(candidates))
|
||
for _, userID := range candidates {
|
||
if !occupied[userID] {
|
||
available = append(available, userID)
|
||
}
|
||
}
|
||
minCount, maxCount := normalizeRobotCountRange(req.GetMinRobotCount(), req.GetMaxRobotCount())
|
||
if len(available) < minCount {
|
||
return RobotRoomConfig{}, xerr.New(xerr.Conflict, "available robot count is not enough")
|
||
}
|
||
rng := rand.New(rand.NewSource(now.UnixNano()))
|
||
selectedCount := randomIntRange(rng, minCount, maxCount)
|
||
if selectedCount > len(available) {
|
||
selectedCount = len(available)
|
||
}
|
||
robotPool := selectRobotUsers(rng, ownerID, available, len(available))
|
||
rule, err := normalizeRobotRoomGiftRule(req.GetGiftRule())
|
||
if err != nil {
|
||
return RobotRoomConfig{}, err
|
||
}
|
||
seatCount := normalizeRobotRoomSeatCount(req.GetSeatCount())
|
||
if seatCount <= 0 {
|
||
return RobotRoomConfig{}, xerr.New(xerr.InvalidArgument, "robot room seat_count is invalid")
|
||
}
|
||
roomID := fmt.Sprintf("robot_%d_%d", now.UnixMilli(), rng.Intn(900000)+100000)
|
||
if !roomid.ValidStringID(roomID) {
|
||
return RobotRoomConfig{}, xerr.New(xerr.Internal, "generated robot room id is invalid")
|
||
}
|
||
title := strings.TrimSpace(req.GetRoomName())
|
||
if title == "" {
|
||
title = fmt.Sprintf("Robot Room %d", ownerID)
|
||
}
|
||
coverURL := strings.TrimSpace(req.GetRoomAvatar())
|
||
if coverURL == "" {
|
||
coverURL = defaultRoomAvatar
|
||
}
|
||
return RobotRoomConfig{
|
||
AppCode: appcode.FromContext(ctx),
|
||
RoomID: roomID,
|
||
RoomShortID: fmt.Sprintf("%d", ownerID),
|
||
Title: title,
|
||
CoverURL: coverURL,
|
||
VisibleRegionID: normalizeVisibleRegionID(req.GetVisibleRegionId()),
|
||
OwnerCountryCode: normalizeRoomCountryCode(req.GetOwnerCountryCode()),
|
||
Status: robotRoomStatusActive,
|
||
OwnerRobotUserID: ownerID,
|
||
RobotUserIDs: robotPool,
|
||
ActiveRobotCount: int32(selectedCount),
|
||
SeatCount: seatCount,
|
||
GiftRule: rule,
|
||
CreatedByAdminID: req.GetAdminId(),
|
||
CreatedAtMS: now.UnixMilli(),
|
||
UpdatedAtMS: now.UnixMilli(),
|
||
}, nil
|
||
}
|
||
|
||
func (s *Service) bootstrapRobotRoom(ctx context.Context, config RobotRoomConfig) error {
|
||
activeIDs := robotRoomInitialActiveIDs(config)
|
||
_, err := s.CreateRoom(ctx, &roomv1.CreateRoomRequest{
|
||
Meta: &roomv1.RequestMeta{
|
||
RequestId: fmt.Sprintf("admin-robot-room-create-%s", config.RoomID),
|
||
CommandId: fmt.Sprintf("admin-robot-room:%s:create", config.RoomID),
|
||
ActorUserId: config.OwnerRobotUserID,
|
||
RoomId: config.RoomID,
|
||
AppCode: config.AppCode,
|
||
SentAtMs: s.clock.Now().UnixMilli(),
|
||
},
|
||
SeatCount: config.SeatCount,
|
||
Mode: robotRoomMode,
|
||
VisibleRegionId: config.VisibleRegionID,
|
||
OwnerCountryCode: config.OwnerCountryCode,
|
||
RoomName: config.Title,
|
||
RoomAvatar: config.CoverURL,
|
||
RoomShortId: config.RoomShortID,
|
||
RobotRoom: true,
|
||
RobotUserIds: config.RobotUserIDs,
|
||
})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
for _, userID := range activeIDs {
|
||
if userID == config.OwnerRobotUserID {
|
||
continue
|
||
}
|
||
if _, err := s.JoinRoom(ctx, &roomv1.JoinRoomRequest{
|
||
Meta: robotRoomCommandMeta(config, userID, fmt.Sprintf("join:%d", userID)),
|
||
Role: "audience",
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
for index, userID := range activeIDs {
|
||
if _, err := s.RobotVirtualMicUp(ctx, RobotMicUpInput{
|
||
Meta: robotRoomCommandMeta(config, config.OwnerRobotUserID, fmt.Sprintf("mic:%d", userID)),
|
||
TargetUserID: userID,
|
||
SeatNo: int32(index + 1),
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// RunRobotRoomRuntimeManager 恢复 active 机器人房间的进程内送礼循环；配置事实仍以 MySQL 为准。
|
||
func (s *Service) RunRobotRoomRuntimeManager(ctx context.Context, interval time.Duration) {
|
||
if interval <= 0 {
|
||
interval = 10 * time.Second
|
||
}
|
||
logx.Info(ctx, "robot_room_runtime_manager_started", slog.Int64("interval_ms", interval.Milliseconds()))
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
s.startActiveRobotRooms(ctx)
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
s.stopAllRobotRoomRuntimes()
|
||
return
|
||
case <-ticker.C:
|
||
s.startActiveRobotRooms(ctx)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) startActiveRobotRooms(ctx context.Context) {
|
||
configs, err := s.repository.ListActiveRobotRooms(ctx)
|
||
if err != nil {
|
||
logx.Warn(ctx, "robot_room_runtime_scan_failed", slog.String("error", err.Error()))
|
||
return
|
||
}
|
||
logx.Info(ctx, "robot_room_runtime_scan_completed", slog.Int("active_room_count", len(configs)))
|
||
// 在启动/停止决策前先记录已运行集合；扫描期间由管理端并发新建的房间不在本轮 active 列表里，
|
||
// 不能被误判为已停用。
|
||
runningRoomIDs := s.runningRobotRoomIDs()
|
||
activeRoomIDs := make(map[string]bool, len(configs))
|
||
for _, config := range configs {
|
||
activeRoomIDs[config.RoomID] = true
|
||
ownerNodeID, ownedElsewhere, err := s.roomLeaseOwnedElsewhere(ctx, config.AppCode, config.RoomID)
|
||
if err != nil {
|
||
// 目录抖动时保持现状：不认领也不停止，等下一轮扫描重新判定。
|
||
logx.Warn(ctx, "robot_room_runtime_lease_lookup_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
|
||
continue
|
||
}
|
||
if ownedElsewhere {
|
||
// 其他节点持有有效 lease 时，本节点对该房间的任何命令都只会得到 CONFLICT；
|
||
// 不启动新循环，并停掉执行权已迁走后仍残留的本地循环。
|
||
if s.stopRobotRoomRuntime(config.RoomID) {
|
||
logx.Warn(ctx, "robot_room_runtime_ownership_lost", slog.String("room_id", config.RoomID), slog.String("owner_node_id", ownerNodeID))
|
||
}
|
||
continue
|
||
}
|
||
s.startRobotRoomRuntime(ctx, config)
|
||
}
|
||
for _, roomID := range runningRoomIDs {
|
||
if activeRoomIDs[roomID] {
|
||
continue
|
||
}
|
||
// 停用/删除请求可能落在其他节点；配置不再 active 的房间不能在任何节点继续跑循环。
|
||
if s.stopRobotRoomRuntime(roomID) {
|
||
logx.Info(ctx, "robot_room_runtime_stopped_inactive", slog.String("room_id", roomID))
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) startRobotRoomRuntime(parent context.Context, config RobotRoomConfig) {
|
||
if config.Status != robotRoomStatusActive || len(config.RobotUserIDs) < 2 {
|
||
logx.Info(parent, "robot_room_runtime_start_skipped",
|
||
slog.String("room_id", config.RoomID),
|
||
slog.String("status", config.Status),
|
||
slog.Int("robot_count", len(config.RobotUserIDs)),
|
||
)
|
||
return
|
||
}
|
||
if err := parent.Err(); err != nil {
|
||
// 管理端创建/启停请求的 ctx 会随 HTTP/gRPC 请求结束而取消；已取消的 ctx 不能写入
|
||
// runtime map，否则后台恢复扫描会误以为该房间仍在运行。
|
||
logx.Warn(parent, "robot_room_runtime_parent_cancelled", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
|
||
return
|
||
}
|
||
s.robotRuntimeMu.Lock()
|
||
if s.robotRuntimes == nil {
|
||
s.robotRuntimes = make(map[string]robotRoomRuntime)
|
||
}
|
||
if _, exists := s.robotRuntimes[config.RoomID]; exists {
|
||
s.robotRuntimeMu.Unlock()
|
||
logx.Info(parent, "robot_room_runtime_start_skipped",
|
||
slog.String("room_id", config.RoomID),
|
||
slog.String("status", config.Status),
|
||
slog.Int("robot_count", len(config.RobotUserIDs)),
|
||
slog.String("reason", "already_running"),
|
||
)
|
||
return
|
||
}
|
||
s.robotRuntimeMu.Unlock()
|
||
|
||
owned, ownerNodeID, err := s.claimRobotRoomOwnership(parent, config)
|
||
if err != nil {
|
||
logx.Warn(parent, "robot_room_runtime_lease_claim_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
|
||
return
|
||
}
|
||
if !owned {
|
||
// 机器人房循环只能在 lease owner 节点运行；非 owner 节点启动只会对 owner 的房间刷 CONFLICT。
|
||
logx.Info(parent, "robot_room_runtime_start_skipped",
|
||
slog.String("room_id", config.RoomID),
|
||
slog.String("status", config.Status),
|
||
slog.Int("robot_count", len(config.RobotUserIDs)),
|
||
slog.String("reason", "not_owner"),
|
||
slog.String("owner_node_id", ownerNodeID),
|
||
)
|
||
return
|
||
}
|
||
|
||
if err := s.ensureRobotRoomParticipants(parent, config); err != nil {
|
||
logx.Warn(parent, "robot_room_runtime_prepare_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
|
||
return
|
||
}
|
||
|
||
s.robotRuntimeMu.Lock()
|
||
if _, exists := s.robotRuntimes[config.RoomID]; exists {
|
||
s.robotRuntimeMu.Unlock()
|
||
return
|
||
}
|
||
ctx, cancel := context.WithCancel(appcode.WithContext(parent, config.AppCode))
|
||
token := fmt.Sprintf("%s:%d", config.RoomID, time.Now().UTC().UnixNano())
|
||
s.robotRuntimes[config.RoomID] = robotRoomRuntime{cancel: cancel, token: token}
|
||
s.robotRuntimeMu.Unlock()
|
||
logx.Info(ctx, "robot_room_runtime_started",
|
||
slog.String("room_id", config.RoomID),
|
||
slog.Int64("owner_robot_user_id", config.OwnerRobotUserID),
|
||
slog.Int("robot_count", len(config.RobotUserIDs)),
|
||
slog.Int("normal_gift_count", len(config.GiftRule.GiftIDs)),
|
||
slog.Int("lucky_gift_count", len(config.GiftRule.LuckyGiftIDs)),
|
||
)
|
||
go func() {
|
||
defer s.clearRobotRoomRuntime(config.RoomID, token)
|
||
s.runRobotRoomRuntime(ctx, config)
|
||
}()
|
||
go s.runRobotRoomLeaseKeeper(ctx, config, token)
|
||
}
|
||
|
||
// claimRobotRoomOwnership 在无有效 owner 或本节点已是 owner 时认领/续租机器人房 lease；
|
||
// 其他节点持有有效 lease 时返回现任 owner，调用方必须放弃启动循环。
|
||
func (s *Service) claimRobotRoomOwnership(ctx context.Context, config RobotRoomConfig) (bool, string, error) {
|
||
if s.directory == nil {
|
||
// 测试中的极简 Service 可能没有装配 directory；生产 HealthCheck 会拒绝这种配置。
|
||
return true, "", nil
|
||
}
|
||
lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), s.nodeID, s.clock.Now(), s.leaseTTL)
|
||
if err != nil {
|
||
return false, "", err
|
||
}
|
||
if lease.NodeID != s.nodeID {
|
||
return false, lease.NodeID, nil
|
||
}
|
||
return true, "", nil
|
||
}
|
||
|
||
// runRobotRoomLeaseKeeper 以短于 leaseTTL 的周期续租机器人房 lease。
|
||
// 机器人房的写命令间隔可能远大于 leaseTTL，只靠命令续租会让 lease 频繁过期、被其他节点的
|
||
// 扫描抢走；续租发现执行权已被接管时立即停掉本地 runtime，不再对新 owner 的房间空转 CONFLICT。
|
||
func (s *Service) runRobotRoomLeaseKeeper(ctx context.Context, config RobotRoomConfig, token string) {
|
||
if s.directory == nil {
|
||
return
|
||
}
|
||
interval := s.leaseTTL / 3
|
||
if interval <= 0 {
|
||
interval = time.Second
|
||
}
|
||
ticker := time.NewTicker(interval)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
}
|
||
lease, err := s.directory.EnsureOwner(ctx, runtimeRoomKey(config.AppCode, config.RoomID), s.nodeID, s.clock.Now(), s.leaseTTL)
|
||
if err != nil {
|
||
// 目录抖动时保留 runtime 等下一次续租；期间的房间命令也会经 ensureCell 续租。
|
||
logx.Warn(ctx, "robot_room_lease_renew_failed", slog.String("room_id", config.RoomID), slog.String("error", err.Error()))
|
||
continue
|
||
}
|
||
if lease.NodeID != s.nodeID {
|
||
logx.Warn(ctx, "robot_room_runtime_ownership_lost",
|
||
slog.String("room_id", config.RoomID),
|
||
slog.String("owner_node_id", lease.NodeID),
|
||
)
|
||
s.stopRobotRoomRuntimeToken(config.RoomID, token)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// stopRobotRoomRuntime 停止指定房间的本地循环；返回是否确实存在运行中的 runtime。
|
||
func (s *Service) stopRobotRoomRuntime(roomID string) bool {
|
||
s.robotRuntimeMu.Lock()
|
||
runtime, exists := s.robotRuntimes[roomID]
|
||
delete(s.robotRuntimes, roomID)
|
||
s.robotRuntimeMu.Unlock()
|
||
if runtime.cancel != nil {
|
||
runtime.cancel()
|
||
}
|
||
return exists
|
||
}
|
||
|
||
// stopRobotRoomRuntimeToken 只停止仍由该 token 代表的 runtime，避免误停同房间的新一轮启动。
|
||
func (s *Service) stopRobotRoomRuntimeToken(roomID string, token string) {
|
||
s.robotRuntimeMu.Lock()
|
||
runtime, exists := s.robotRuntimes[roomID]
|
||
if !exists || runtime.token != token {
|
||
s.robotRuntimeMu.Unlock()
|
||
return
|
||
}
|
||
delete(s.robotRuntimes, roomID)
|
||
s.robotRuntimeMu.Unlock()
|
||
if runtime.cancel != nil {
|
||
runtime.cancel()
|
||
}
|
||
}
|
||
|
||
func (s *Service) runningRobotRoomIDs() []string {
|
||
s.robotRuntimeMu.Lock()
|
||
defer s.robotRuntimeMu.Unlock()
|
||
roomIDs := make([]string, 0, len(s.robotRuntimes))
|
||
for roomID := range s.robotRuntimes {
|
||
roomIDs = append(roomIDs, roomID)
|
||
}
|
||
return roomIDs
|
||
}
|
||
|
||
func (s *Service) stopAllRobotRoomRuntimes() {
|
||
s.robotRuntimeMu.Lock()
|
||
cancels := make([]context.CancelFunc, 0, len(s.robotRuntimes))
|
||
for roomID, runtime := range s.robotRuntimes {
|
||
if runtime.cancel != nil {
|
||
cancels = append(cancels, runtime.cancel)
|
||
}
|
||
delete(s.robotRuntimes, roomID)
|
||
}
|
||
s.robotRuntimeMu.Unlock()
|
||
for _, cancel := range cancels {
|
||
cancel()
|
||
}
|
||
}
|
||
|
||
func (s *Service) clearRobotRoomRuntime(roomID string, token string) {
|
||
s.robotRuntimeMu.Lock()
|
||
defer s.robotRuntimeMu.Unlock()
|
||
runtime, exists := s.robotRuntimes[roomID]
|
||
if !exists || runtime.token != token {
|
||
return
|
||
}
|
||
delete(s.robotRuntimes, roomID)
|
||
}
|
||
|
||
func (s *Service) ensureRobotRoomParticipants(ctx context.Context, config RobotRoomConfig) error {
|
||
roomCtx := appcode.WithContext(ctx, config.AppCode)
|
||
snapshot, err := s.currentSnapshot(roomCtx, config.RoomID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
online := make(map[int64]bool, len(snapshot.GetOnlineUsers()))
|
||
for _, user := range snapshot.GetOnlineUsers() {
|
||
online[user.GetUserId()] = true
|
||
}
|
||
activeTarget := robotRoomActiveTarget(config)
|
||
activeRobots := onlineRobotIDs(config.RobotUserIDs, online)
|
||
missing := activeTarget - len(activeRobots)
|
||
if missing > 0 {
|
||
// 首次补回缺席机器人也必须从当前 Room Cell 快照选择真实麦位。候选池下标可能
|
||
// 大于 seat_count；同一轮成功补位后立刻占记，避免后续候选复用尚未写回旧快照的麦位。
|
||
occupiedSeats := make(map[int32]bool, len(snapshot.GetMicSeats()))
|
||
for _, seat := range snapshot.GetMicSeats() {
|
||
if seat.GetUserId() > 0 {
|
||
occupiedSeats[seat.GetSeatNo()] = true
|
||
}
|
||
}
|
||
for _, userID := range robotRoomFillCandidates(config, online, missing) {
|
||
// active 机器人房从持久化配置恢复时，旧进程可能已经把机器人 stale 成 left；
|
||
// 这里只补齐后台配置的活跃人数，不再把整个候选池一次性塞回房间。
|
||
seatNo := robotRestoreSeatNo(config, userID, snapshot.GetMicSeats(), occupiedSeats)
|
||
if seatNo <= 0 {
|
||
logx.Warn(roomCtx, "robot_room_runtime_join_restore_skipped", slog.String("room_id", config.RoomID), slog.Int64("robot_user_id", userID), slog.String("reason", "no_free_seat"))
|
||
continue
|
||
}
|
||
if err := s.joinRobotRoomUser(roomCtx, config, userID, seatNo, fmt.Sprintf("runtime-join:%d:%d", userID, time.Now().UTC().UnixNano())); err != nil {
|
||
return err
|
||
}
|
||
occupiedSeats[seatNo] = true
|
||
online[userID] = true
|
||
activeRobots = append(activeRobots, userID)
|
||
}
|
||
}
|
||
|
||
snapshot, err = s.currentSnapshot(roomCtx, config.RoomID)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
seated := make(map[int64]bool, len(config.RobotUserIDs))
|
||
occupiedSeats := make(map[int32]bool, len(snapshot.GetMicSeats()))
|
||
for _, seat := range snapshot.GetMicSeats() {
|
||
userID := seat.GetUserId()
|
||
if userID > 0 {
|
||
seated[userID] = true
|
||
occupiedSeats[seat.GetSeatNo()] = true
|
||
}
|
||
}
|
||
for _, userID := range activeRobots {
|
||
if userID <= 0 || seated[userID] {
|
||
continue
|
||
}
|
||
seatNo := robotRestoreSeatNo(config, userID, snapshot.GetMicSeats(), occupiedSeats)
|
||
if seatNo <= 0 {
|
||
logx.Warn(roomCtx, "robot_room_runtime_mic_restore_skipped", slog.String("room_id", config.RoomID), slog.Int64("robot_user_id", userID), slog.String("reason", "no_free_seat"))
|
||
continue
|
||
}
|
||
// 虚拟麦位只服务机器人房展示；若并发恢复导致机器人已上麦，Conflict 可视为达成目标。
|
||
if _, err := s.RobotVirtualMicUp(roomCtx, RobotMicUpInput{
|
||
Meta: robotRoomCommandMeta(config, config.OwnerRobotUserID, fmt.Sprintf("runtime-mic:%d:%d", userID, time.Now().UTC().UnixNano())),
|
||
TargetUserID: userID,
|
||
SeatNo: seatNo,
|
||
}); err != nil {
|
||
if xerr.CodeOf(err) == xerr.Conflict && strings.Contains(xerr.MessageOf(err), "already on seat") {
|
||
continue
|
||
}
|
||
return err
|
||
}
|
||
occupiedSeats[seatNo] = true
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// robotRestoreSeatNo 只接受当前 Room Cell 快照里真实存在且空闲的麦位。
|
||
// robot_user_ids 是候选池而不是麦位表：历史配置可包含数十个轮换机器人，因此其下标
|
||
// 可能远大于当前 10/15/20 个麦位；这种 preferred 必须回退到快照中的首个空麦，不能
|
||
// 把不存在的 seat_no 发给 RobotVirtualMicUp 后让每 10 秒恢复扫描永久失败。
|
||
func robotRestoreSeatNo(config RobotRoomConfig, userID int64, seats []*roomv1.SeatState, occupied map[int32]bool) int32 {
|
||
preferred := preferredRobotSeatNo(config, userID)
|
||
if preferred > 0 && robotSnapshotSeatFree(seats, preferred) && !occupied[preferred] {
|
||
return preferred
|
||
}
|
||
// snapshot 是本轮恢复开始时的只读副本；occupied 还包含本轮刚成功分配、尚未反映到
|
||
// 该副本的麦位。回退时同时检查二者，避免多个待恢复机器人重复选择同一个空麦。
|
||
var firstFree int32
|
||
for _, seat := range seats {
|
||
seatNo := seat.GetSeatNo()
|
||
if seatNo <= 0 || seat.GetUserId() > 0 || seat.GetLocked() || occupied[seatNo] {
|
||
continue
|
||
}
|
||
if firstFree == 0 || seatNo < firstFree {
|
||
firstFree = seatNo
|
||
}
|
||
}
|
||
return firstFree
|
||
}
|
||
|
||
func robotSnapshotSeatFree(seats []*roomv1.SeatState, seatNo int32) bool {
|
||
for _, seat := range seats {
|
||
if seat.GetSeatNo() == seatNo {
|
||
return seat.GetUserId() == 0 && !seat.GetLocked()
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func (s *Service) runRobotRoomRuntime(ctx context.Context, config RobotRoomConfig) {
|
||
activity := newRobotRoomActivity(config.RobotUserIDs)
|
||
if snapshot, err := s.currentSnapshot(appcode.WithContext(ctx, config.AppCode), config.RoomID); err == nil {
|
||
for _, user := range snapshot.GetOnlineUsers() {
|
||
if robotRoomHasUser(config.RobotUserIDs, user.GetUserId()) {
|
||
activity.markActive(user.GetUserId())
|
||
}
|
||
}
|
||
}
|
||
giftSlots := make(chan struct{}, robotRoomMaxGiftSenders(config))
|
||
for _, senderID := range config.RobotUserIDs {
|
||
senderID := senderID
|
||
go s.runRobotPresenceLoop(ctx, config, activity, senderID)
|
||
go s.runRobotNormalGiftLoop(ctx, config, activity, giftSlots, senderID)
|
||
go s.runRobotLuckyGiftLoop(ctx, config, activity, giftSlots, senderID)
|
||
}
|
||
<-ctx.Done()
|
||
}
|
||
|
||
func (s *Service) runRobotPresenceLoop(ctx context.Context, config RobotRoomConfig, activity *robotRoomActivity, userID int64) {
|
||
if userID <= 0 {
|
||
return
|
||
}
|
||
rng := rand.New(rand.NewSource(time.Now().UnixNano() + userID*31))
|
||
for {
|
||
if !activity.isActive(userID) {
|
||
if !waitRobotRuntimeDelay(ctx, time.Duration(200+rng.Intn(600))*time.Millisecond) {
|
||
return
|
||
}
|
||
continue
|
||
}
|
||
stay := time.Duration(randomInt64Range(rng, config.GiftRule.RobotStayMinMS, config.GiftRule.RobotStayMaxMS)) * time.Millisecond
|
||
if stay <= 0 {
|
||
stay = time.Duration(defaultRobotStayMinMS) * time.Millisecond
|
||
}
|
||
if !waitRobotRuntimeDelay(ctx, stay) {
|
||
return
|
||
}
|
||
if !activity.markInactive(userID) {
|
||
continue
|
||
}
|
||
seatNo := s.robotRoomSeatNo(ctx, config, userID)
|
||
if err := s.leaveRobotRoomUser(ctx, config, userID); err != nil {
|
||
logx.Warn(ctx, "robot_room_leave_failed", slog.String("room_id", config.RoomID), slog.Int64("robot_user_id", userID), slog.String("error", err.Error()))
|
||
activity.markActive(userID)
|
||
continue
|
||
}
|
||
delay := time.Duration(randomInt64Range(rng, config.GiftRule.RobotReplaceMinMS, config.GiftRule.RobotReplaceMaxMS)) * time.Millisecond
|
||
if !waitRobotRuntimeDelay(ctx, delay) {
|
||
return
|
||
}
|
||
replacementID := activity.pickReplacement(rng, userID)
|
||
if replacementID <= 0 {
|
||
replacementID = userID
|
||
}
|
||
if err := s.joinRobotRoomUser(ctx, config, replacementID, seatNo, fmt.Sprintf("replace:%d:%d", replacementID, time.Now().UTC().UnixNano())); err != nil {
|
||
logx.Warn(ctx, "robot_room_replace_join_failed", slog.String("room_id", config.RoomID), slog.Int64("leaving_user_id", userID), slog.Int64("replacement_user_id", replacementID), slog.String("error", err.Error()))
|
||
if restoreErr := s.joinRobotRoomUser(ctx, config, userID, seatNo, fmt.Sprintf("restore:%d:%d", userID, time.Now().UTC().UnixNano())); restoreErr != nil {
|
||
logx.Warn(ctx, "robot_room_replace_restore_failed", slog.String("room_id", config.RoomID), slog.Int64("robot_user_id", userID), slog.String("error", restoreErr.Error()))
|
||
continue
|
||
}
|
||
activity.markActive(userID)
|
||
continue
|
||
}
|
||
activity.markActive(replacementID)
|
||
logx.Info(ctx, "robot_room_replaced_user",
|
||
slog.String("room_id", config.RoomID),
|
||
slog.Int64("left_robot_user_id", userID),
|
||
slog.Int64("joined_robot_user_id", replacementID),
|
||
slog.Int64("stay_ms", stay.Milliseconds()),
|
||
slog.Int64("replace_delay_ms", delay.Milliseconds()),
|
||
)
|
||
}
|
||
}
|
||
|
||
func (s *Service) runRobotNormalGiftLoop(ctx context.Context, config RobotRoomConfig, activity *robotRoomActivity, giftSlots chan struct{}, senderID int64) {
|
||
interval := time.Duration(config.GiftRule.NormalGiftIntervalMS) * time.Millisecond
|
||
if interval <= 0 || len(config.GiftRule.GiftIDs) == 0 {
|
||
return
|
||
}
|
||
rng := rand.New(rand.NewSource(time.Now().UnixNano() + senderID))
|
||
timer := time.NewTimer(randomJitter(interval, rng))
|
||
defer timer.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-timer.C:
|
||
if !activity.isActive(senderID) {
|
||
timer.Reset(interval)
|
||
continue
|
||
}
|
||
targetID := randomRobotTarget(rng, activity.activeIDs(), senderID)
|
||
giftID := randomString(rng, config.GiftRule.GiftIDs)
|
||
if targetID > 0 && giftID != "" {
|
||
s.sendRobotGiftBestEffort(ctx, config, activity, giftSlots, senderID, targetID, giftID, "", "normal")
|
||
}
|
||
timer.Reset(interval)
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) runRobotLuckyGiftLoop(ctx context.Context, config RobotRoomConfig, activity *robotRoomActivity, giftSlots chan struct{}, senderID int64) {
|
||
if len(config.GiftRule.LuckyGiftIDs) == 0 || config.GiftRule.LuckyComboMax <= 0 {
|
||
return
|
||
}
|
||
comboMin := clampRobotLuckyCombo(config.GiftRule.LuckyComboMin)
|
||
comboMax := clampRobotLuckyCombo(config.GiftRule.LuckyComboMax)
|
||
if comboMin <= 0 || comboMax < comboMin {
|
||
return
|
||
}
|
||
rng := rand.New(rand.NewSource(time.Now().UnixNano() + senderID*17))
|
||
for {
|
||
if !activity.isActive(senderID) {
|
||
if !waitRobotRuntimeDelay(ctx, time.Duration(300+rng.Intn(700))*time.Millisecond) {
|
||
return
|
||
}
|
||
continue
|
||
}
|
||
combo := randomInt64Range(rng, comboMin, comboMax)
|
||
for i := int64(0); i < combo; i++ {
|
||
giftID := randomString(rng, config.GiftRule.LuckyGiftIDs)
|
||
targetID := randomRobotTarget(rng, activity.activeIDs(), senderID)
|
||
if targetID > 0 && giftID != "" {
|
||
s.sendRobotGiftBestEffort(ctx, config, activity, giftSlots, senderID, targetID, giftID, robotRoomLuckyPoolID, "lucky")
|
||
}
|
||
if !waitRobotGiftPace(ctx, rng) {
|
||
return
|
||
}
|
||
}
|
||
pause := time.Duration(randomInt64Range(rng, config.GiftRule.LuckyPauseMinMS, config.GiftRule.LuckyPauseMaxMS)) * time.Millisecond
|
||
if pause <= 0 {
|
||
pause = 5 * time.Second
|
||
}
|
||
timer := time.NewTimer(pause)
|
||
select {
|
||
case <-ctx.Done():
|
||
timer.Stop()
|
||
return
|
||
case <-timer.C:
|
||
}
|
||
}
|
||
}
|
||
|
||
func (s *Service) sendRobotGiftBestEffort(ctx context.Context, config RobotRoomConfig, activity *robotRoomActivity, giftSlots chan struct{}, senderID int64, targetID int64, giftID string, poolID string, kind string) {
|
||
if !activity.isActive(senderID) || !activity.isActive(targetID) {
|
||
return
|
||
}
|
||
select {
|
||
case giftSlots <- struct{}{}:
|
||
defer func() { <-giftSlots }()
|
||
default:
|
||
return
|
||
}
|
||
_, err := s.RobotSendGift(ctx, RobotSendGiftInput{
|
||
Meta: robotRoomCommandMeta(config, senderID, fmt.Sprintf("gift:%s:%d:%d", kind, targetID, time.Now().UTC().UnixNano())),
|
||
TargetUserID: targetID,
|
||
GiftID: giftID,
|
||
GiftCount: 1,
|
||
PoolID: poolID,
|
||
RobotUserIDs: config.RobotUserIDs,
|
||
SyntheticRewardCoins: 0,
|
||
SyntheticMultiplierPPM: 0,
|
||
})
|
||
if err != nil {
|
||
logx.Warn(ctx, "robot_room_send_gift_failed", slog.String("room_id", config.RoomID), slog.String("kind", kind), slog.Int64("sender_user_id", senderID), slog.Int64("target_user_id", targetID), slog.String("gift_id", giftID), slog.String("error", err.Error()))
|
||
}
|
||
}
|
||
|
||
func (s *Service) leaveRobotRoomUser(ctx context.Context, config RobotRoomConfig, userID int64) error {
|
||
_, err := s.LeaveRoom(ctx, &roomv1.LeaveRoomRequest{
|
||
Meta: robotRoomCommandMeta(config, userID, fmt.Sprintf("leave:%d:%d", userID, time.Now().UTC().UnixNano())),
|
||
})
|
||
return err
|
||
}
|
||
|
||
func (s *Service) joinRobotRoomUser(ctx context.Context, config RobotRoomConfig, userID int64, seatNo int32, suffix string) error {
|
||
role := "audience"
|
||
if userID == config.OwnerRobotUserID {
|
||
role = "owner"
|
||
}
|
||
if _, err := s.JoinRoom(ctx, &roomv1.JoinRoomRequest{
|
||
Meta: robotRoomCommandMeta(config, userID, suffix),
|
||
Role: role,
|
||
ActorIsRobot: true,
|
||
}); err != nil {
|
||
return err
|
||
}
|
||
if seatNo <= 0 {
|
||
seatNo = preferredRobotSeatNo(config, userID)
|
||
}
|
||
if seatNo <= 0 {
|
||
seatNo = 1
|
||
}
|
||
if _, err := s.RobotVirtualMicUp(ctx, RobotMicUpInput{
|
||
Meta: robotRoomCommandMeta(config, config.OwnerRobotUserID, fmt.Sprintf("mic:%d:%d", userID, time.Now().UTC().UnixNano())),
|
||
TargetUserID: userID,
|
||
SeatNo: seatNo,
|
||
}); err != nil {
|
||
if xerr.CodeOf(err) == xerr.Conflict && strings.Contains(xerr.MessageOf(err), "already on seat") {
|
||
return nil
|
||
}
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *Service) robotRoomSeatNo(ctx context.Context, config RobotRoomConfig, userID int64) int32 {
|
||
snapshot, err := s.currentSnapshot(appcode.WithContext(ctx, config.AppCode), config.RoomID)
|
||
if err != nil {
|
||
return preferredRobotSeatNo(config, userID)
|
||
}
|
||
for _, seat := range snapshot.GetMicSeats() {
|
||
if seat.GetUserId() == userID {
|
||
return seat.GetSeatNo()
|
||
}
|
||
}
|
||
return firstFreeRobotSeatNo(snapshot.GetMicSeats(), robotRoomActiveTarget(config))
|
||
}
|
||
|
||
func robotRoomCommandMeta(config RobotRoomConfig, actorUserID int64, suffix string) *roomv1.RequestMeta {
|
||
commandID := fmt.Sprintf("robot-room:%s:%s", config.RoomID, strings.TrimSpace(suffix))
|
||
if len(commandID) > 128 {
|
||
commandID = commandID[:128]
|
||
}
|
||
return &roomv1.RequestMeta{
|
||
RequestId: commandID,
|
||
CommandId: commandID,
|
||
ActorUserId: actorUserID,
|
||
RoomId: config.RoomID,
|
||
AppCode: config.AppCode,
|
||
SentAtMs: time.Now().UTC().UnixMilli(),
|
||
}
|
||
}
|
||
|
||
func normalizeRobotRoomGiftRule(input *roomv1.AdminRobotRoomGiftRule) (RobotRoomGiftRule, error) {
|
||
if input == nil {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "gift_rule is required")
|
||
}
|
||
rule := RobotRoomGiftRule{
|
||
GiftIDs: normalizeStringSet(input.GetGiftIds()),
|
||
LuckyGiftIDs: normalizeStringSet(input.GetLuckyGiftIds()),
|
||
NormalGiftIntervalMS: input.GetNormalGiftIntervalMs(),
|
||
LuckyComboMin: clampRobotLuckyCombo(input.GetLuckyComboMin()),
|
||
LuckyComboMax: clampRobotLuckyCombo(input.GetLuckyComboMax()),
|
||
LuckyPauseMinMS: input.GetLuckyPauseMinMs(),
|
||
LuckyPauseMaxMS: input.GetLuckyPauseMaxMs(),
|
||
RobotStayMinMS: input.GetRobotStayMinMs(),
|
||
RobotStayMaxMS: input.GetRobotStayMaxMs(),
|
||
RobotReplaceMinMS: input.GetRobotReplaceMinMs(),
|
||
RobotReplaceMaxMS: input.GetRobotReplaceMaxMs(),
|
||
MaxGiftSenders: input.GetMaxGiftSenders(),
|
||
}
|
||
rule = withRobotRoomGiftRuleDefaults(rule)
|
||
if len(rule.GiftIDs) == 0 {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "gift_ids is required")
|
||
}
|
||
if len(rule.LuckyGiftIDs) == 0 {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "lucky_gift_ids is required")
|
||
}
|
||
if rule.NormalGiftIntervalMS <= 0 {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "normal_gift_interval_ms is required")
|
||
}
|
||
if rule.LuckyComboMin <= 0 || rule.LuckyComboMax < rule.LuckyComboMin {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "lucky combo range is invalid")
|
||
}
|
||
if rule.LuckyPauseMinMS <= 0 || rule.LuckyPauseMaxMS < rule.LuckyPauseMinMS {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "lucky pause range is invalid")
|
||
}
|
||
if rule.RobotStayMinMS <= 0 || rule.RobotStayMaxMS < rule.RobotStayMinMS {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "robot stay range is invalid")
|
||
}
|
||
if rule.RobotReplaceMinMS < 0 || rule.RobotReplaceMaxMS < rule.RobotReplaceMinMS {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "robot replace range is invalid")
|
||
}
|
||
if rule.MaxGiftSenders <= 0 {
|
||
return RobotRoomGiftRule{}, xerr.New(xerr.InvalidArgument, "max_gift_senders is required")
|
||
}
|
||
return rule, nil
|
||
}
|
||
|
||
func withRobotRoomGiftRuleDefaults(rule RobotRoomGiftRule) RobotRoomGiftRule {
|
||
if rule.RobotStayMinMS <= 0 && rule.RobotStayMaxMS <= 0 {
|
||
rule.RobotStayMinMS = defaultRobotStayMinMS
|
||
rule.RobotStayMaxMS = defaultRobotStayMaxMS
|
||
}
|
||
if rule.RobotReplaceMinMS == 0 && rule.RobotReplaceMaxMS == 0 {
|
||
rule.RobotReplaceMinMS = defaultRobotReplaceMinMS
|
||
rule.RobotReplaceMaxMS = defaultRobotReplaceMaxMS
|
||
}
|
||
if rule.MaxGiftSenders <= 0 {
|
||
rule.MaxGiftSenders = defaultMaxGiftSenders
|
||
}
|
||
return rule
|
||
}
|
||
|
||
// WithRobotRoomGiftRuleStorageDefaults 让 repository 恢复历史行时复用运行时默认值；
|
||
// 默认值只补空字段，不放宽 normalizeRobotRoomGiftRule 对显式非法范围的校验。
|
||
func WithRobotRoomGiftRuleStorageDefaults(rule RobotRoomGiftRule) RobotRoomGiftRule {
|
||
return withRobotRoomGiftRuleDefaults(rule)
|
||
}
|
||
|
||
func clampRobotLuckyCombo(value int64) int64 {
|
||
if value <= 0 {
|
||
return value
|
||
}
|
||
if value > maxRobotLuckyComboPerCycle {
|
||
return maxRobotLuckyComboPerCycle
|
||
}
|
||
return value
|
||
}
|
||
|
||
func robotRoomConfigToProto(config RobotRoomConfig) *roomv1.AdminRobotRoom {
|
||
return &roomv1.AdminRobotRoom{
|
||
AppCode: config.AppCode,
|
||
RoomId: config.RoomID,
|
||
RoomShortId: config.RoomShortID,
|
||
Title: config.Title,
|
||
CoverUrl: config.CoverURL,
|
||
VisibleRegionId: config.VisibleRegionID,
|
||
Status: config.Status,
|
||
OwnerRobotUserId: config.OwnerRobotUserID,
|
||
RobotUserIds: append([]int64(nil), config.RobotUserIDs...),
|
||
ActiveRobotCount: config.ActiveRobotCount,
|
||
SeatCount: config.SeatCount,
|
||
GiftRule: &roomv1.AdminRobotRoomGiftRule{
|
||
GiftIds: append([]string(nil), config.GiftRule.GiftIDs...),
|
||
LuckyGiftIds: append([]string(nil), config.GiftRule.LuckyGiftIDs...),
|
||
NormalGiftIntervalMs: config.GiftRule.NormalGiftIntervalMS,
|
||
LuckyComboMin: config.GiftRule.LuckyComboMin,
|
||
LuckyComboMax: config.GiftRule.LuckyComboMax,
|
||
LuckyPauseMinMs: config.GiftRule.LuckyPauseMinMS,
|
||
LuckyPauseMaxMs: config.GiftRule.LuckyPauseMaxMS,
|
||
RobotStayMinMs: config.GiftRule.RobotStayMinMS,
|
||
RobotStayMaxMs: config.GiftRule.RobotStayMaxMS,
|
||
RobotReplaceMinMs: config.GiftRule.RobotReplaceMinMS,
|
||
RobotReplaceMaxMs: config.GiftRule.RobotReplaceMaxMS,
|
||
MaxGiftSenders: config.GiftRule.MaxGiftSenders,
|
||
},
|
||
CreatedByAdminId: config.CreatedByAdminID,
|
||
CreatedAtMs: config.CreatedAtMS,
|
||
UpdatedAtMs: config.UpdatedAtMS,
|
||
}
|
||
}
|
||
|
||
func normalizeRobotCountRange(minCount int32, maxCount int32) (int, int) {
|
||
minValue := int(minCount)
|
||
maxValue := int(maxCount)
|
||
if minValue <= 0 {
|
||
minValue = 6
|
||
}
|
||
if maxValue <= 0 {
|
||
maxValue = 8
|
||
}
|
||
if maxValue < minValue {
|
||
maxValue = minValue
|
||
}
|
||
return minValue, maxValue
|
||
}
|
||
|
||
func normalizeRobotRoomSeatCount(seatCount int32) int32 {
|
||
switch seatCount {
|
||
case 0:
|
||
return defaultRobotRoomSeatCount
|
||
case 10, 15, 20:
|
||
return seatCount
|
||
default:
|
||
return 0
|
||
}
|
||
}
|
||
|
||
func selectRobotUsers(rng *rand.Rand, ownerID int64, available []int64, count int) []int64 {
|
||
others := make([]int64, 0, len(available))
|
||
for _, userID := range available {
|
||
if userID != ownerID {
|
||
others = append(others, userID)
|
||
}
|
||
}
|
||
rng.Shuffle(len(others), func(i, j int) { others[i], others[j] = others[j], others[i] })
|
||
selected := []int64{ownerID}
|
||
for _, userID := range others {
|
||
if len(selected) >= count {
|
||
break
|
||
}
|
||
selected = append(selected, userID)
|
||
}
|
||
slices.Sort(selected)
|
||
return selected
|
||
}
|
||
|
||
func robotRoomInitialActiveIDs(config RobotRoomConfig) []int64 {
|
||
target := robotRoomActiveTarget(config)
|
||
if target <= 0 {
|
||
return nil
|
||
}
|
||
if target > len(config.RobotUserIDs) {
|
||
target = len(config.RobotUserIDs)
|
||
}
|
||
active := make([]int64, 0, target)
|
||
if config.OwnerRobotUserID > 0 && robotRoomHasUser(config.RobotUserIDs, config.OwnerRobotUserID) {
|
||
active = append(active, config.OwnerRobotUserID)
|
||
}
|
||
for _, userID := range config.RobotUserIDs {
|
||
if len(active) >= target {
|
||
break
|
||
}
|
||
if userID <= 0 || userID == config.OwnerRobotUserID {
|
||
continue
|
||
}
|
||
active = append(active, userID)
|
||
}
|
||
return active
|
||
}
|
||
|
||
func robotRoomActiveTarget(config RobotRoomConfig) int {
|
||
target := int(config.ActiveRobotCount)
|
||
if target <= 0 {
|
||
target = len(config.RobotUserIDs)
|
||
}
|
||
if target > len(config.RobotUserIDs) {
|
||
target = len(config.RobotUserIDs)
|
||
}
|
||
if target < 0 {
|
||
return 0
|
||
}
|
||
return target
|
||
}
|
||
|
||
func robotRoomMaxGiftSenders(config RobotRoomConfig) int {
|
||
maxSenders := int(config.GiftRule.MaxGiftSenders)
|
||
if maxSenders <= 0 {
|
||
maxSenders = int(defaultMaxGiftSenders)
|
||
}
|
||
if maxSenders > len(config.RobotUserIDs) {
|
||
maxSenders = len(config.RobotUserIDs)
|
||
}
|
||
if maxSenders <= 0 {
|
||
maxSenders = 1
|
||
}
|
||
return maxSenders
|
||
}
|
||
|
||
func robotRoomHasUser(robots []int64, userID int64) bool {
|
||
for _, robotID := range robots {
|
||
if robotID == userID {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func onlineRobotIDs(robots []int64, online map[int64]bool) []int64 {
|
||
out := make([]int64, 0, len(robots))
|
||
for _, userID := range robots {
|
||
if online[userID] {
|
||
out = append(out, userID)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func robotRoomFillCandidates(config RobotRoomConfig, online map[int64]bool, limit int) []int64 {
|
||
if limit <= 0 {
|
||
return nil
|
||
}
|
||
out := make([]int64, 0, limit)
|
||
if config.OwnerRobotUserID > 0 && !online[config.OwnerRobotUserID] && robotRoomHasUser(config.RobotUserIDs, config.OwnerRobotUserID) {
|
||
out = append(out, config.OwnerRobotUserID)
|
||
}
|
||
for _, userID := range config.RobotUserIDs {
|
||
if len(out) >= limit {
|
||
break
|
||
}
|
||
if userID <= 0 || userID == config.OwnerRobotUserID || online[userID] {
|
||
continue
|
||
}
|
||
out = append(out, userID)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func preferredRobotSeatNo(config RobotRoomConfig, userID int64) int32 {
|
||
for index, robotID := range config.RobotUserIDs {
|
||
if robotID == userID {
|
||
return int32(index + 1)
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func firstFreeRobotSeatNo(seats []*roomv1.SeatState, activeTarget int) int32 {
|
||
occupied := make(map[int32]bool, len(seats))
|
||
maxSeatNo := int32(0)
|
||
for _, seat := range seats {
|
||
if seat.GetSeatNo() > maxSeatNo {
|
||
maxSeatNo = seat.GetSeatNo()
|
||
}
|
||
if seat.GetUserId() > 0 {
|
||
occupied[seat.GetSeatNo()] = true
|
||
}
|
||
}
|
||
if maxSeatNo <= 0 {
|
||
maxSeatNo = int32(max(10, activeTarget))
|
||
}
|
||
for seatNo := int32(1); seatNo <= maxSeatNo; seatNo++ {
|
||
if !occupied[seatNo] {
|
||
return seatNo
|
||
}
|
||
}
|
||
return 0
|
||
}
|
||
|
||
func waitRobotRuntimeDelay(ctx context.Context, delay time.Duration) bool {
|
||
if delay <= 0 {
|
||
select {
|
||
case <-ctx.Done():
|
||
return false
|
||
default:
|
||
return true
|
||
}
|
||
}
|
||
timer := time.NewTimer(delay)
|
||
defer timer.Stop()
|
||
select {
|
||
case <-ctx.Done():
|
||
return false
|
||
case <-timer.C:
|
||
return true
|
||
}
|
||
}
|
||
|
||
type robotRoomActivity struct {
|
||
mu sync.Mutex
|
||
active map[int64]bool
|
||
giftSenders map[int64]bool
|
||
pool []int64
|
||
}
|
||
|
||
func newRobotRoomActivity(pool []int64) *robotRoomActivity {
|
||
return &robotRoomActivity{
|
||
active: make(map[int64]bool, len(pool)),
|
||
giftSenders: make(map[int64]bool),
|
||
pool: append([]int64(nil), pool...),
|
||
}
|
||
}
|
||
|
||
func (a *robotRoomActivity) isActive(userID int64) bool {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
return a.active[userID]
|
||
}
|
||
|
||
func (a *robotRoomActivity) markActive(userID int64) {
|
||
if userID <= 0 {
|
||
return
|
||
}
|
||
a.mu.Lock()
|
||
a.active[userID] = true
|
||
a.mu.Unlock()
|
||
}
|
||
|
||
func (a *robotRoomActivity) markInactive(userID int64) bool {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
if !a.active[userID] {
|
||
return false
|
||
}
|
||
delete(a.active, userID)
|
||
delete(a.giftSenders, userID)
|
||
return true
|
||
}
|
||
|
||
func (a *robotRoomActivity) isGiftSender(userID int64) bool {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
return a.giftSenders[userID]
|
||
}
|
||
|
||
func (a *robotRoomActivity) markGiftSender(userID int64) {
|
||
if userID <= 0 {
|
||
return
|
||
}
|
||
a.mu.Lock()
|
||
if a.active[userID] {
|
||
a.giftSenders[userID] = true
|
||
}
|
||
a.mu.Unlock()
|
||
}
|
||
|
||
func (a *robotRoomActivity) markGiftSenderWithinLimit(userID int64, maxSenders int) bool {
|
||
if userID <= 0 || maxSenders <= 0 {
|
||
return false
|
||
}
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
if !a.active[userID] || len(a.giftSenders) >= maxSenders {
|
||
return false
|
||
}
|
||
a.giftSenders[userID] = true
|
||
return true
|
||
}
|
||
|
||
func (a *robotRoomActivity) giftSenderCount() int {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
return len(a.giftSenders)
|
||
}
|
||
|
||
func (a *robotRoomActivity) activeIDs() []int64 {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
out := make([]int64, 0, len(a.active))
|
||
for userID := range a.active {
|
||
out = append(out, userID)
|
||
}
|
||
return out
|
||
}
|
||
|
||
func (a *robotRoomActivity) pickReplacement(rng *rand.Rand, departingUserID int64) int64 {
|
||
a.mu.Lock()
|
||
defer a.mu.Unlock()
|
||
candidates := make([]int64, 0, len(a.pool))
|
||
fallback := make([]int64, 0, 1)
|
||
for _, userID := range a.pool {
|
||
if a.active[userID] {
|
||
continue
|
||
}
|
||
if userID == departingUserID {
|
||
fallback = append(fallback, userID)
|
||
continue
|
||
}
|
||
candidates = append(candidates, userID)
|
||
}
|
||
if len(candidates) == 0 {
|
||
candidates = fallback
|
||
}
|
||
if len(candidates) == 0 {
|
||
return 0
|
||
}
|
||
return candidates[rng.Intn(len(candidates))]
|
||
}
|
||
|
||
func normalizeRobotRoomStatus(status string) string {
|
||
switch strings.TrimSpace(status) {
|
||
case robotRoomStatusActive:
|
||
return robotRoomStatusActive
|
||
case robotRoomStatusStopped:
|
||
return robotRoomStatusStopped
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
func normalizeStringSet(values []string) []string {
|
||
seen := make(map[string]bool, len(values))
|
||
out := make([]string, 0, len(values))
|
||
for _, value := range values {
|
||
value = strings.TrimSpace(value)
|
||
if value == "" || seen[value] {
|
||
continue
|
||
}
|
||
seen[value] = true
|
||
out = append(out, value)
|
||
}
|
||
slices.Sort(out)
|
||
return out
|
||
}
|
||
|
||
func int64SetKeys(values map[int64]bool) []int64 {
|
||
out := make([]int64, 0, len(values))
|
||
for value := range values {
|
||
out = append(out, value)
|
||
}
|
||
slices.Sort(out)
|
||
return out
|
||
}
|
||
|
||
func randomString(rng *rand.Rand, values []string) string {
|
||
if len(values) == 0 {
|
||
return ""
|
||
}
|
||
return values[rng.Intn(len(values))]
|
||
}
|
||
|
||
func randomRobotTarget(rng *rand.Rand, robots []int64, senderID int64) int64 {
|
||
targets := make([]int64, 0, len(robots))
|
||
for _, userID := range robots {
|
||
if userID != senderID {
|
||
targets = append(targets, userID)
|
||
}
|
||
}
|
||
if len(targets) == 0 {
|
||
return 0
|
||
}
|
||
return targets[rng.Intn(len(targets))]
|
||
}
|
||
|
||
func randomIntRange(rng *rand.Rand, minValue int, maxValue int) int {
|
||
if maxValue <= minValue {
|
||
return minValue
|
||
}
|
||
return minValue + rng.Intn(maxValue-minValue+1)
|
||
}
|
||
|
||
func randomInt64Range(rng *rand.Rand, minValue int64, maxValue int64) int64 {
|
||
if maxValue <= minValue {
|
||
return minValue
|
||
}
|
||
return minValue + rng.Int63n(maxValue-minValue+1)
|
||
}
|
||
|
||
func randomJitter(interval time.Duration, rng *rand.Rand) time.Duration {
|
||
if interval <= time.Second {
|
||
return interval
|
||
}
|
||
return time.Duration(rng.Int63n(int64(interval))) + time.Second
|
||
}
|
||
|
||
func waitRobotGiftPace(ctx context.Context, rng *rand.Rand) bool {
|
||
delay := time.Duration(300+rng.Intn(500)) * time.Millisecond
|
||
timer := time.NewTimer(delay)
|
||
defer timer.Stop()
|
||
select {
|
||
case <-ctx.Done():
|
||
return false
|
||
case <-timer.C:
|
||
return true
|
||
}
|
||
}
|