840 lines
32 KiB
Go
840 lines
32 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"log/slog"
|
|
"math/rand"
|
|
"strings"
|
|
"time"
|
|
|
|
roomv1 "hyapp.local/api/proto/room/v1"
|
|
"hyapp/pkg/appcode"
|
|
"hyapp/pkg/logx"
|
|
"hyapp/pkg/xerr"
|
|
)
|
|
|
|
const (
|
|
humanRobotLuckyPoolID = "human_robot_lucky_display"
|
|
humanRobotSuperLuckyPoolID = "human_robot_super_lucky_display"
|
|
|
|
defaultHumanRobotCandidateRoomMaxOnline = int32(5)
|
|
defaultHumanRobotRoomFullStopOnline = int32(10)
|
|
defaultHumanRobotRoomTargetMinOnline = int32(8)
|
|
defaultHumanRobotRoomTargetMaxOnline = int32(10)
|
|
defaultHumanRobotStayMinMS = int64(60 * 1000)
|
|
defaultHumanRobotStayMaxMS = int64(10 * 60 * 1000)
|
|
defaultHumanRobotReplaceMinMS = int64(60 * 1000)
|
|
defaultHumanRobotReplaceMaxMS = int64(3 * 60 * 1000)
|
|
defaultHumanRobotNormalGiftIntervalMS = int64(10 * 1000)
|
|
defaultHumanRobotNormalGiftIntervalMinMS = int64(1 * 1000)
|
|
defaultHumanRobotNormalGiftIntervalMaxMS = int64(20 * 1000)
|
|
defaultHumanRobotLuckyPauseMinMS = int64(5 * 1000)
|
|
defaultHumanRobotLuckyPauseMaxMS = int64(20 * 1000)
|
|
defaultHumanRobotLuckyComboMin = int64(1)
|
|
defaultHumanRobotLuckyComboMax = int64(3)
|
|
defaultHumanRobotMaxGiftSenders = int64(1)
|
|
)
|
|
|
|
func (s *Service) AdminGetHumanRoomRobotConfig(ctx context.Context, req *roomv1.AdminGetHumanRoomRobotConfigRequest) (*roomv1.AdminGetHumanRoomRobotConfigResponse, error) {
|
|
ctx = contextFromMeta(ctx, req.GetMeta())
|
|
config, exists, err := s.repository.GetHumanRoomRobotConfig(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !exists {
|
|
config = defaultHumanRoomRobotConfig(appcode.FromContext(ctx), s.clock.Now().UnixMilli())
|
|
}
|
|
return &roomv1.AdminGetHumanRoomRobotConfigResponse{Config: humanRoomRobotConfigToProto(config), ServerTimeMs: s.clock.Now().UnixMilli()}, nil
|
|
}
|
|
|
|
func (s *Service) AdminUpdateHumanRoomRobotConfig(ctx context.Context, req *roomv1.AdminUpdateHumanRoomRobotConfigRequest) (*roomv1.AdminUpdateHumanRoomRobotConfigResponse, error) {
|
|
ctx = contextFromMeta(ctx, req.GetMeta())
|
|
config, err := normalizeHumanRoomRobotConfig(appcode.FromContext(ctx), req.GetConfig(), req.GetAdminId(), s.clock.Now().UnixMilli())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
saved, err := s.repository.UpsertHumanRoomRobotConfig(ctx, config)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if !saved.Enabled {
|
|
s.stopAllHumanRoomRobotRuntimes()
|
|
}
|
|
return &roomv1.AdminUpdateHumanRoomRobotConfigResponse{Config: humanRoomRobotConfigToProto(saved), ServerTimeMs: s.clock.Now().UnixMilli()}, nil
|
|
}
|
|
|
|
// RunHumanRoomRobotRuntimeManager 定期把各国家空闲机器人补进低人数真人房间。
|
|
func (s *Service) RunHumanRoomRobotRuntimeManager(ctx context.Context, interval time.Duration) {
|
|
if interval <= 0 {
|
|
interval = 15 * time.Second
|
|
}
|
|
logx.Info(ctx, "human_room_robot_runtime_manager_started", slog.Int64("interval_ms", interval.Milliseconds()))
|
|
ticker := time.NewTicker(interval)
|
|
defer ticker.Stop()
|
|
s.scanHumanRoomRobotRuntimes(ctx)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
s.stopAllHumanRoomRobotRuntimes()
|
|
return
|
|
case <-ticker.C:
|
|
s.scanHumanRoomRobotRuntimes(ctx)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) scanHumanRoomRobotRuntimes(ctx context.Context) {
|
|
config, exists, err := s.repository.GetHumanRoomRobotConfig(ctx)
|
|
if err != nil {
|
|
logx.Warn(ctx, "human_room_robot_config_load_failed", slog.String("error", err.Error()))
|
|
return
|
|
}
|
|
if !exists {
|
|
return
|
|
}
|
|
config = withHumanRoomRobotDefaults(config)
|
|
if !config.Enabled {
|
|
s.stopAllHumanRoomRobotRuntimes()
|
|
return
|
|
}
|
|
pools, err := s.loadHumanRoomRobotPools(ctx, config.AppCode)
|
|
if err != nil {
|
|
logx.Warn(ctx, "human_room_robot_pool_load_failed", slog.String("error", err.Error()))
|
|
return
|
|
}
|
|
config.CountryPools = pools
|
|
for _, pool := range config.CountryPools {
|
|
countryCode := normalizeRoomCountryCode(pool.CountryCode)
|
|
if countryCode == "" || len(pool.RobotUserIDs) == 0 {
|
|
continue
|
|
}
|
|
rooms, err := s.repository.ListHumanRobotCandidateRooms(ctx, config.AppCode, countryCode, config.CandidateRoomMaxOnline, config.RoomTargetMaxOnline, 50)
|
|
if err != nil {
|
|
logx.Warn(ctx, "human_room_robot_candidate_rooms_failed", slog.String("country_code", countryCode), slog.String("error", err.Error()))
|
|
continue
|
|
}
|
|
for _, room := range rooms {
|
|
if s.humanRoomRobotRuntimeConfigChanged(room.RoomID, config.UpdatedAtMS) {
|
|
s.stopHumanRoomRobotRuntime(room.RoomID)
|
|
}
|
|
if s.humanRoomRobotRuntimeExists(room.RoomID) {
|
|
continue
|
|
}
|
|
s.startHumanRoomRobotRuntime(ctx, config, countryCode, room.RoomID, pool.RobotUserIDs)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) loadHumanRoomRobotPools(ctx context.Context, appCode string) ([]HumanRoomRobotCountryPool, error) {
|
|
if s.humanRobotPoolProvider == nil {
|
|
return nil, xerr.New(xerr.Unavailable, "human room robot pool provider is not configured")
|
|
}
|
|
items, err := s.humanRobotPoolProvider.ListHumanRoomRobotPools(ctx, appCode)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
pools := make([]HumanRoomRobotCountryPool, 0, len(items))
|
|
for _, item := range items {
|
|
countryCode := normalizeRoomCountryCode(item.CountryCode)
|
|
ids := normalizeRobotUserIDs(item.RobotUserIDs)
|
|
if countryCode == "" || len(ids) == 0 {
|
|
continue
|
|
}
|
|
pools = append(pools, HumanRoomRobotCountryPool{CountryCode: countryCode, RobotUserIDs: ids})
|
|
}
|
|
return pools, nil
|
|
}
|
|
|
|
func (s *Service) startHumanRoomRobotRuntime(parent context.Context, config HumanRoomRobotConfig, countryCode string, roomID string, pool []int64) {
|
|
if parent.Err() != nil || roomID == "" || len(pool) == 0 {
|
|
return
|
|
}
|
|
s.humanRobotRuntimeMu.Lock()
|
|
if s.humanRobotRuntimes == nil {
|
|
s.humanRobotRuntimes = make(map[string]robotRoomRuntime)
|
|
}
|
|
if _, exists := s.humanRobotRuntimes[roomID]; exists {
|
|
s.humanRobotRuntimeMu.Unlock()
|
|
return
|
|
}
|
|
s.humanRobotRuntimeMu.Unlock()
|
|
|
|
roomCtx := appcode.WithContext(parent, config.AppCode)
|
|
targetOnline := randomHumanRoomTargetOnline(config)
|
|
selected, err := s.prepareHumanRoomRobots(roomCtx, config, roomID, pool, targetOnline)
|
|
if err != nil {
|
|
logx.Warn(roomCtx, "human_room_robot_prepare_failed", slog.String("room_id", roomID), slog.String("country_code", countryCode), slog.String("error", err.Error()))
|
|
return
|
|
}
|
|
if len(selected) == 0 {
|
|
return
|
|
}
|
|
|
|
s.humanRobotRuntimeMu.Lock()
|
|
if _, exists := s.humanRobotRuntimes[roomID]; exists {
|
|
s.humanRobotRuntimeMu.Unlock()
|
|
return
|
|
}
|
|
ctx, cancel := context.WithCancel(roomCtx)
|
|
token := fmt.Sprintf("%s:%d", roomID, time.Now().UTC().UnixNano())
|
|
s.humanRobotRuntimes[roomID] = robotRoomRuntime{cancel: cancel, token: token, configUpdatedAtMS: config.UpdatedAtMS}
|
|
s.humanRobotRuntimeMu.Unlock()
|
|
|
|
logx.Info(ctx, "human_room_robot_runtime_started",
|
|
slog.String("room_id", roomID),
|
|
slog.String("country_code", countryCode),
|
|
slog.Int("target_online", targetOnline),
|
|
slog.Int("active_robot_count", len(selected)),
|
|
)
|
|
go func() {
|
|
defer s.clearHumanRoomRobotRuntime(roomID, token)
|
|
s.runHumanRoomRobotRuntime(ctx, config, roomID, pool, selected, targetOnline)
|
|
}()
|
|
}
|
|
|
|
func (s *Service) prepareHumanRoomRobots(ctx context.Context, config HumanRoomRobotConfig, roomID string, pool []int64, targetOnline int) ([]int64, error) {
|
|
snapshot, err := s.currentSnapshot(ctx, roomID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if snapshot.GetStatus() != "active" {
|
|
return nil, nil
|
|
}
|
|
online := make(map[int64]bool, len(snapshot.GetOnlineUsers()))
|
|
for _, user := range snapshot.GetOnlineUsers() {
|
|
online[user.GetUserId()] = true
|
|
}
|
|
humanOnline := humanRoomRealOnlineCount(snapshot.GetOnlineUsers(), allHumanRoomRobotIDs(config))
|
|
if humanOnline >= int(config.CandidateRoomMaxOnline) || len(online) >= targetOnline {
|
|
return nil, nil
|
|
}
|
|
limit := targetOnline - len(online)
|
|
if limit <= 0 {
|
|
return nil, nil
|
|
}
|
|
freeSeats := freeSeatNumbers(snapshot.GetMicSeats())
|
|
if len(freeSeats) < limit {
|
|
limit = len(freeSeats)
|
|
}
|
|
if limit <= 0 {
|
|
return nil, nil
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
candidates := append([]int64(nil), pool...)
|
|
rng.Shuffle(len(candidates), func(i, j int) { candidates[i], candidates[j] = candidates[j], candidates[i] })
|
|
selected := make([]int64, 0, limit)
|
|
for _, userID := range candidates {
|
|
if len(selected) >= limit {
|
|
break
|
|
}
|
|
if userID <= 0 || online[userID] || s.robotHasActiveRoomPresence(ctx, userID, roomID) {
|
|
continue
|
|
}
|
|
seatNo := freeSeats[len(selected)]
|
|
if err := s.joinHumanRoomRobot(ctx, roomID, userID, seatNo, fmt.Sprintf("initial:%d:%d", userID, time.Now().UTC().UnixNano())); err != nil {
|
|
logx.Warn(ctx, "human_room_robot_initial_join_failed", slog.String("room_id", roomID), slog.Int64("robot_user_id", userID), slog.String("error", err.Error()))
|
|
continue
|
|
}
|
|
selected = append(selected, userID)
|
|
}
|
|
return selected, nil
|
|
}
|
|
|
|
func (s *Service) runHumanRoomRobotRuntime(ctx context.Context, config HumanRoomRobotConfig, roomID string, pool []int64, initialActive []int64, targetOnline int) {
|
|
activity := newRobotRoomActivity(pool)
|
|
for _, userID := range initialActive {
|
|
activity.markActive(userID)
|
|
}
|
|
markInitialHumanRoomGiftSenders(activity, initialActive, int(config.MaxGiftSenders))
|
|
giftSlots := make(chan struct{}, int(max(1, int(config.MaxGiftSenders))))
|
|
for _, userID := range pool {
|
|
userID := userID
|
|
go s.runHumanRobotPresenceLoop(ctx, config, roomID, pool, activity, userID, targetOnline)
|
|
go s.runHumanRobotNormalGiftLoop(ctx, config, roomID, activity, giftSlots, userID)
|
|
go s.runHumanRobotLuckyGiftLoop(ctx, config, roomID, activity, giftSlots, userID, config.LuckyGiftIDs, humanRobotLuckyPoolID, "lucky", 5_000_000)
|
|
go s.runHumanRobotLuckyGiftLoop(ctx, config, roomID, activity, giftSlots, userID, config.SuperLuckyGiftIDs, humanRobotSuperLuckyPoolID, "super_lucky", 20_000_000)
|
|
}
|
|
go s.runHumanRobotFillLoop(ctx, config, roomID, pool, activity, targetOnline)
|
|
<-ctx.Done()
|
|
}
|
|
|
|
func (s *Service) runHumanRobotFillLoop(ctx context.Context, config HumanRoomRobotConfig, roomID string, pool []int64, activity *robotRoomActivity, targetOnline int) {
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
timer := time.NewTimer(time.Duration(1+rng.Intn(3)) * time.Second)
|
|
defer timer.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timer.C:
|
|
s.fillHumanRoomRobotsOnce(ctx, config, roomID, pool, activity, rng, targetOnline)
|
|
timer.Reset(time.Duration(5+rng.Intn(5)) * time.Second)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) fillHumanRoomRobotsOnce(ctx context.Context, config HumanRoomRobotConfig, roomID string, pool []int64, activity *robotRoomActivity, rng *rand.Rand, targetOnline int) {
|
|
snapshot, err := s.currentSnapshot(ctx, roomID)
|
|
if err != nil || snapshot.GetStatus() != "active" {
|
|
return
|
|
}
|
|
humanOnline := humanRoomRealOnlineCount(snapshot.GetOnlineUsers(), allHumanRoomRobotIDs(config))
|
|
if humanOnline >= int(config.CandidateRoomMaxOnline) || len(snapshot.GetOnlineUsers()) >= targetOnline {
|
|
return
|
|
}
|
|
limit := targetOnline - len(snapshot.GetOnlineUsers())
|
|
freeSeats := freeSeatNumbers(snapshot.GetMicSeats())
|
|
if len(freeSeats) < limit {
|
|
limit = len(freeSeats)
|
|
}
|
|
if limit <= 0 {
|
|
return
|
|
}
|
|
online := make(map[int64]bool, len(snapshot.GetOnlineUsers()))
|
|
for _, user := range snapshot.GetOnlineUsers() {
|
|
online[user.GetUserId()] = true
|
|
}
|
|
candidates := append([]int64(nil), pool...)
|
|
rng.Shuffle(len(candidates), func(i, j int) { candidates[i], candidates[j] = candidates[j], candidates[i] })
|
|
joined := 0
|
|
for _, userID := range candidates {
|
|
if joined >= limit {
|
|
return
|
|
}
|
|
if userID <= 0 || online[userID] || activity.isActive(userID) || s.robotHasActiveRoomPresence(ctx, userID, roomID) {
|
|
continue
|
|
}
|
|
seatNo := freeSeats[joined]
|
|
if err := s.joinHumanRoomRobot(ctx, roomID, userID, seatNo, fmt.Sprintf("fill:%d:%d", userID, time.Now().UTC().UnixNano())); err != nil {
|
|
logx.Warn(ctx, "human_room_robot_fill_join_failed", slog.String("room_id", roomID), slog.Int64("robot_user_id", userID), slog.String("error", err.Error()))
|
|
continue
|
|
}
|
|
activity.markActive(userID)
|
|
activity.markGiftSenderWithinLimit(userID, int(config.MaxGiftSenders))
|
|
online[userID] = true
|
|
joined++
|
|
}
|
|
}
|
|
|
|
func (s *Service) runHumanRobotPresenceLoop(ctx context.Context, config HumanRoomRobotConfig, roomID string, pool []int64, activity *robotRoomActivity, userID int64, targetOnline int) {
|
|
if userID <= 0 {
|
|
return
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano() + userID*43))
|
|
for {
|
|
if !activity.isActive(userID) {
|
|
if !waitRobotRuntimeDelay(ctx, time.Duration(300+rng.Intn(700))*time.Millisecond) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
stay := time.Duration(randomInt64Range(rng, config.RobotStayMinMS, config.RobotStayMaxMS)) * time.Millisecond
|
|
if !waitRobotRuntimeDelay(ctx, stay) {
|
|
return
|
|
}
|
|
// 送礼名额绑定在房间内的少数机器人身上；送礼机器人离开时，补位机器人接替这个名额，保证房间内送礼者数量稳定且不会扩散到所有机器人。
|
|
wasGiftSender := activity.isGiftSender(userID)
|
|
if !activity.markInactive(userID) {
|
|
continue
|
|
}
|
|
seatNo := s.humanRoomRobotSeatNo(ctx, roomID, userID)
|
|
if err := s.leaveHumanRoomRobot(ctx, roomID, userID); err != nil {
|
|
logx.Warn(ctx, "human_room_robot_leave_failed", slog.String("room_id", roomID), slog.Int64("robot_user_id", userID), slog.String("error", err.Error()))
|
|
activity.markActive(userID)
|
|
if wasGiftSender {
|
|
activity.markGiftSender(userID)
|
|
}
|
|
continue
|
|
}
|
|
delay := time.Duration(randomInt64Range(rng, config.RobotReplaceMinMS, config.RobotReplaceMaxMS)) * time.Millisecond
|
|
if !waitRobotRuntimeDelay(ctx, delay) {
|
|
return
|
|
}
|
|
replacementID := s.pickHumanRoomReplacement(ctx, config, roomID, pool, activity, userID)
|
|
if replacementID <= 0 {
|
|
continue
|
|
}
|
|
if seatNo <= 0 {
|
|
seatNo = s.firstHumanRoomFreeSeatNo(ctx, roomID)
|
|
}
|
|
if seatNo <= 0 || !s.humanRoomNeedsRobot(ctx, config, roomID, targetOnline) {
|
|
continue
|
|
}
|
|
if err := s.joinHumanRoomRobot(ctx, roomID, replacementID, seatNo, fmt.Sprintf("replace:%d:%d", replacementID, time.Now().UTC().UnixNano())); err != nil {
|
|
logx.Warn(ctx, "human_room_robot_replace_join_failed", slog.String("room_id", roomID), slog.Int64("replacement_user_id", replacementID), slog.String("error", err.Error()))
|
|
continue
|
|
}
|
|
activity.markActive(replacementID)
|
|
if wasGiftSender || activity.giftSenderCount() < int(config.MaxGiftSenders) {
|
|
activity.markGiftSenderWithinLimit(replacementID, int(config.MaxGiftSenders))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) runHumanRobotNormalGiftLoop(ctx context.Context, config HumanRoomRobotConfig, roomID string, activity *robotRoomActivity, giftSlots chan struct{}, senderID int64) {
|
|
if config.NormalGiftIntervalMinMS <= 0 || config.NormalGiftIntervalMaxMS <= 0 || len(config.GiftIDs) == 0 {
|
|
return
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano() + senderID*59))
|
|
timer := time.NewTimer(humanRoomRobotNormalGiftDelay(config, rng))
|
|
defer timer.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timer.C:
|
|
if activity.isActive(senderID) && activity.isGiftSender(senderID) {
|
|
targetID := randomRobotTarget(rng, activity.activeIDs(), senderID)
|
|
giftID := randomString(rng, config.GiftIDs)
|
|
s.sendHumanRoomRobotGiftBestEffort(ctx, config, roomID, activity, giftSlots, senderID, targetID, giftID, "", "normal", 0)
|
|
}
|
|
timer.Reset(humanRoomRobotNormalGiftDelay(config, rng))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) runHumanRobotLuckyGiftLoop(ctx context.Context, config HumanRoomRobotConfig, roomID string, activity *robotRoomActivity, giftSlots chan struct{}, senderID int64, giftIDs []string, poolID string, kind string, multiplierPPM int64) {
|
|
if len(giftIDs) == 0 || config.LuckyComboMax <= 0 {
|
|
return
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano() + senderID*71))
|
|
for {
|
|
if !activity.isActive(senderID) || !activity.isGiftSender(senderID) {
|
|
if !waitRobotRuntimeDelay(ctx, time.Duration(300+rng.Intn(700))*time.Millisecond) {
|
|
return
|
|
}
|
|
continue
|
|
}
|
|
combo := randomInt64Range(rng, config.LuckyComboMin, config.LuckyComboMax)
|
|
for i := int64(0); i < combo; i++ {
|
|
if !activity.isActive(senderID) || !activity.isGiftSender(senderID) {
|
|
break
|
|
}
|
|
targetID := randomRobotTarget(rng, activity.activeIDs(), senderID)
|
|
giftID := randomString(rng, giftIDs)
|
|
s.sendHumanRoomRobotGiftBestEffort(ctx, config, roomID, activity, giftSlots, senderID, targetID, giftID, poolID, kind, multiplierPPM)
|
|
if !waitRobotGiftPace(ctx, rng) {
|
|
return
|
|
}
|
|
}
|
|
pause := time.Duration(randomInt64Range(rng, config.LuckyPauseMinMS, config.LuckyPauseMaxMS)) * time.Millisecond
|
|
if !waitRobotRuntimeDelay(ctx, pause) {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *Service) sendHumanRoomRobotGiftBestEffort(ctx context.Context, config HumanRoomRobotConfig, roomID string, activity *robotRoomActivity, giftSlots chan struct{}, senderID int64, targetID int64, giftID string, poolID string, kind string, multiplierPPM int64) {
|
|
if !activity.isActive(senderID) || !activity.isGiftSender(senderID) || !activity.isActive(targetID) || giftID == "" {
|
|
return
|
|
}
|
|
select {
|
|
case giftSlots <- struct{}{}:
|
|
defer func() { <-giftSlots }()
|
|
default:
|
|
return
|
|
}
|
|
_, err := s.RobotSendGift(ctx, RobotSendGiftInput{
|
|
Meta: humanRoomRobotCommandMeta(config, roomID, senderID, fmt.Sprintf("gift:%s:%d:%d", kind, targetID, time.Now().UTC().UnixNano())),
|
|
TargetUserID: targetID,
|
|
GiftID: giftID,
|
|
GiftCount: 1,
|
|
PoolID: poolID,
|
|
RobotUserIDs: activity.pool,
|
|
SyntheticMultiplierPPM: multiplierPPM,
|
|
RealRoomHeat: true,
|
|
})
|
|
if err != nil {
|
|
logx.Warn(ctx, "human_room_robot_send_gift_failed", slog.String("room_id", 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 markInitialHumanRoomGiftSenders(activity *robotRoomActivity, initialActive []int64, maxSenders int) {
|
|
if activity == nil || maxSenders <= 0 || len(initialActive) == 0 {
|
|
return
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
candidates := append([]int64(nil), initialActive...)
|
|
rng.Shuffle(len(candidates), func(i int, j int) {
|
|
candidates[i], candidates[j] = candidates[j], candidates[i]
|
|
})
|
|
for _, userID := range candidates {
|
|
if activity.giftSenderCount() >= maxSenders {
|
|
return
|
|
}
|
|
activity.markGiftSenderWithinLimit(userID, maxSenders)
|
|
}
|
|
}
|
|
|
|
func humanRoomRobotNormalGiftDelay(config HumanRoomRobotConfig, rng *rand.Rand) time.Duration {
|
|
minMS := config.NormalGiftIntervalMinMS
|
|
maxMS := config.NormalGiftIntervalMaxMS
|
|
if minMS <= 0 && maxMS <= 0 && config.NormalGiftIntervalMS > 0 {
|
|
minMS = config.NormalGiftIntervalMS
|
|
maxMS = config.NormalGiftIntervalMS
|
|
}
|
|
if minMS <= 0 {
|
|
minMS = defaultHumanRobotNormalGiftIntervalMinMS
|
|
}
|
|
if maxMS < minMS {
|
|
maxMS = minMS
|
|
}
|
|
return time.Duration(randomInt64Range(rng, minMS, maxMS)) * time.Millisecond
|
|
}
|
|
|
|
func randomHumanRoomTargetOnline(config HumanRoomRobotConfig) int {
|
|
minOnline := config.RoomTargetMinOnline
|
|
maxOnline := config.RoomTargetMaxOnline
|
|
if minOnline <= 0 && maxOnline <= 0 && config.RoomFullStopOnline > 0 {
|
|
minOnline = config.RoomFullStopOnline
|
|
maxOnline = config.RoomFullStopOnline
|
|
}
|
|
if minOnline <= 0 {
|
|
minOnline = defaultHumanRobotRoomTargetMinOnline
|
|
}
|
|
if maxOnline < minOnline {
|
|
maxOnline = minOnline
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
return int(randomInt64Range(rng, int64(minOnline), int64(maxOnline)))
|
|
}
|
|
|
|
func (s *Service) joinHumanRoomRobot(ctx context.Context, roomID string, userID int64, seatNo int32, suffix string) error {
|
|
if _, err := s.JoinRoom(ctx, &roomv1.JoinRoomRequest{
|
|
Meta: humanRoomRobotCommandMeta(HumanRoomRobotConfig{AppCode: appcode.FromContext(ctx)}, roomID, userID, "join:"+suffix),
|
|
Role: "audience",
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
if _, err := s.RobotVirtualMicUp(ctx, RobotMicUpInput{
|
|
Meta: humanRoomRobotCommandMeta(HumanRoomRobotConfig{AppCode: appcode.FromContext(ctx)}, roomID, userID, "mic:"+suffix),
|
|
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) leaveHumanRoomRobot(ctx context.Context, roomID string, userID int64) error {
|
|
_, err := s.LeaveRoom(ctx, &roomv1.LeaveRoomRequest{
|
|
Meta: humanRoomRobotCommandMeta(HumanRoomRobotConfig{AppCode: appcode.FromContext(ctx)}, roomID, userID, fmt.Sprintf("leave:%d:%d", userID, time.Now().UTC().UnixNano())),
|
|
})
|
|
return err
|
|
}
|
|
|
|
func (s *Service) pickHumanRoomReplacement(ctx context.Context, config HumanRoomRobotConfig, roomID string, pool []int64, activity *robotRoomActivity, departingUserID int64) int64 {
|
|
active := make(map[int64]bool, len(activity.activeIDs()))
|
|
for _, id := range activity.activeIDs() {
|
|
active[id] = true
|
|
}
|
|
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
candidates := append([]int64(nil), pool...)
|
|
rng.Shuffle(len(candidates), func(i, j int) { candidates[i], candidates[j] = candidates[j], candidates[i] })
|
|
for _, userID := range candidates {
|
|
if userID <= 0 || active[userID] || userID == departingUserID || s.robotHasActiveRoomPresence(ctx, userID, roomID) {
|
|
continue
|
|
}
|
|
return userID
|
|
}
|
|
if !s.robotHasActiveRoomPresence(ctx, departingUserID, roomID) {
|
|
return departingUserID
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func (s *Service) humanRoomNeedsRobot(ctx context.Context, config HumanRoomRobotConfig, roomID string, targetOnline int) bool {
|
|
snapshot, err := s.currentSnapshot(ctx, roomID)
|
|
if err != nil || snapshot.GetStatus() != "active" {
|
|
return false
|
|
}
|
|
humanOnline := humanRoomRealOnlineCount(snapshot.GetOnlineUsers(), allHumanRoomRobotIDs(config))
|
|
return humanOnline < int(config.CandidateRoomMaxOnline) && len(snapshot.GetOnlineUsers()) < targetOnline
|
|
}
|
|
|
|
func (s *Service) robotHasActiveRoomPresence(ctx context.Context, userID int64, allowedRoomID string) bool {
|
|
presence, exists, err := s.repository.GetCurrentRoomPresence(ctx, userID)
|
|
if err != nil || !exists {
|
|
return false
|
|
}
|
|
return presence.RoomID != "" && presence.RoomID != allowedRoomID
|
|
}
|
|
|
|
func (s *Service) humanRoomRobotSeatNo(ctx context.Context, roomID string, userID int64) int32 {
|
|
snapshot, err := s.currentSnapshot(ctx, roomID)
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
for _, seat := range snapshot.GetMicSeats() {
|
|
if seat.GetUserId() == userID {
|
|
return seat.GetSeatNo()
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
func (s *Service) firstHumanRoomFreeSeatNo(ctx context.Context, roomID string) int32 {
|
|
snapshot, err := s.currentSnapshot(ctx, roomID)
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
seats := freeSeatNumbers(snapshot.GetMicSeats())
|
|
if len(seats) == 0 {
|
|
return 0
|
|
}
|
|
return seats[0]
|
|
}
|
|
|
|
func (s *Service) humanRoomRobotRuntimeExists(roomID string) bool {
|
|
s.humanRobotRuntimeMu.Lock()
|
|
defer s.humanRobotRuntimeMu.Unlock()
|
|
_, exists := s.humanRobotRuntimes[roomID]
|
|
return exists
|
|
}
|
|
|
|
func (s *Service) humanRoomRobotRuntimeConfigChanged(roomID string, updatedAtMS int64) bool {
|
|
if updatedAtMS <= 0 {
|
|
return false
|
|
}
|
|
s.humanRobotRuntimeMu.Lock()
|
|
defer s.humanRobotRuntimeMu.Unlock()
|
|
runtime, exists := s.humanRobotRuntimes[roomID]
|
|
return exists && runtime.configUpdatedAtMS > 0 && runtime.configUpdatedAtMS != updatedAtMS
|
|
}
|
|
|
|
func (s *Service) stopHumanRoomRobotRuntime(roomID string) {
|
|
s.humanRobotRuntimeMu.Lock()
|
|
runtime := s.humanRobotRuntimes[roomID]
|
|
delete(s.humanRobotRuntimes, roomID)
|
|
s.humanRobotRuntimeMu.Unlock()
|
|
if runtime.cancel != nil {
|
|
runtime.cancel()
|
|
}
|
|
}
|
|
|
|
func (s *Service) clearHumanRoomRobotRuntime(roomID string, token string) {
|
|
s.humanRobotRuntimeMu.Lock()
|
|
defer s.humanRobotRuntimeMu.Unlock()
|
|
runtime, exists := s.humanRobotRuntimes[roomID]
|
|
if !exists || runtime.token != token {
|
|
return
|
|
}
|
|
delete(s.humanRobotRuntimes, roomID)
|
|
}
|
|
|
|
func (s *Service) stopAllHumanRoomRobotRuntimes() {
|
|
s.humanRobotRuntimeMu.Lock()
|
|
cancels := make([]context.CancelFunc, 0, len(s.humanRobotRuntimes))
|
|
for roomID, runtime := range s.humanRobotRuntimes {
|
|
if runtime.cancel != nil {
|
|
cancels = append(cancels, runtime.cancel)
|
|
}
|
|
delete(s.humanRobotRuntimes, roomID)
|
|
}
|
|
s.humanRobotRuntimeMu.Unlock()
|
|
for _, cancel := range cancels {
|
|
cancel()
|
|
}
|
|
}
|
|
|
|
func normalizeHumanRoomRobotConfig(appCode string, input *roomv1.AdminHumanRoomRobotConfig, adminID uint64, nowMS int64) (HumanRoomRobotConfig, error) {
|
|
if input == nil {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "human room robot config is required")
|
|
}
|
|
config := HumanRoomRobotConfig{
|
|
AppCode: appcode.Normalize(appCode),
|
|
Enabled: input.GetEnabled(),
|
|
CandidateRoomMaxOnline: input.GetCandidateRoomMaxOnline(),
|
|
RoomFullStopOnline: input.GetRoomFullStopOnline(),
|
|
RoomTargetMinOnline: input.GetRoomTargetMinOnline(),
|
|
RoomTargetMaxOnline: input.GetRoomTargetMaxOnline(),
|
|
RobotStayMinMS: input.GetRobotStayMinMs(),
|
|
RobotStayMaxMS: input.GetRobotStayMaxMs(),
|
|
RobotReplaceMinMS: input.GetRobotReplaceMinMs(),
|
|
RobotReplaceMaxMS: input.GetRobotReplaceMaxMs(),
|
|
NormalGiftIntervalMS: input.GetNormalGiftIntervalMs(),
|
|
NormalGiftIntervalMinMS: input.GetNormalGiftIntervalMinMs(),
|
|
NormalGiftIntervalMaxMS: input.GetNormalGiftIntervalMaxMs(),
|
|
GiftIDs: normalizeStringSet(input.GetGiftIds()),
|
|
LuckyGiftIDs: normalizeStringSet(input.GetLuckyGiftIds()),
|
|
SuperLuckyGiftIDs: normalizeStringSet(input.GetSuperLuckyGiftIds()),
|
|
LuckyComboMin: input.GetLuckyComboMin(),
|
|
LuckyComboMax: input.GetLuckyComboMax(),
|
|
LuckyPauseMinMS: input.GetLuckyPauseMinMs(),
|
|
LuckyPauseMaxMS: input.GetLuckyPauseMaxMs(),
|
|
MaxGiftSenders: input.GetMaxGiftSenders(),
|
|
UpdatedByAdminID: adminID,
|
|
CreatedAtMS: input.GetCreatedAtMs(),
|
|
UpdatedAtMS: nowMS,
|
|
}
|
|
config = withHumanRoomRobotDefaults(config)
|
|
if config.CandidateRoomMaxOnline <= 0 || config.RoomTargetMinOnline <= 0 || config.RoomTargetMaxOnline < config.RoomTargetMinOnline {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "room online thresholds are invalid")
|
|
}
|
|
if config.RobotStayMinMS <= 0 || config.RobotStayMaxMS < config.RobotStayMinMS {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "robot stay range is invalid")
|
|
}
|
|
if config.RobotReplaceMinMS < 0 || config.RobotReplaceMaxMS < config.RobotReplaceMinMS {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "robot replace range is invalid")
|
|
}
|
|
if config.NormalGiftIntervalMinMS <= 0 || config.NormalGiftIntervalMaxMS < config.NormalGiftIntervalMinMS {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "normal gift interval range is invalid")
|
|
}
|
|
if config.Enabled && len(config.GiftIDs) == 0 {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "gift_ids is required")
|
|
}
|
|
if config.Enabled && len(config.LuckyGiftIDs) == 0 && len(config.SuperLuckyGiftIDs) == 0 {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "lucky gift ids are required")
|
|
}
|
|
if config.LuckyComboMin <= 0 || config.LuckyComboMax < config.LuckyComboMin {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "lucky combo range is invalid")
|
|
}
|
|
if config.LuckyPauseMinMS <= 0 || config.LuckyPauseMaxMS < config.LuckyPauseMinMS {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "lucky pause range is invalid")
|
|
}
|
|
if config.MaxGiftSenders <= 0 {
|
|
return HumanRoomRobotConfig{}, xerr.New(xerr.InvalidArgument, "max_gift_senders is required")
|
|
}
|
|
return config, nil
|
|
}
|
|
|
|
func withHumanRoomRobotDefaults(config HumanRoomRobotConfig) HumanRoomRobotConfig {
|
|
if config.AppCode == "" {
|
|
config.AppCode = appcode.Default
|
|
}
|
|
if config.CandidateRoomMaxOnline <= 0 {
|
|
config.CandidateRoomMaxOnline = defaultHumanRobotCandidateRoomMaxOnline
|
|
}
|
|
if config.RoomTargetMinOnline <= 0 && config.RoomTargetMaxOnline <= 0 {
|
|
if config.RoomFullStopOnline > 0 {
|
|
config.RoomTargetMinOnline = config.RoomFullStopOnline
|
|
config.RoomTargetMaxOnline = config.RoomFullStopOnline
|
|
} else {
|
|
config.RoomTargetMinOnline = defaultHumanRobotRoomTargetMinOnline
|
|
config.RoomTargetMaxOnline = defaultHumanRobotRoomTargetMaxOnline
|
|
}
|
|
}
|
|
if config.RoomFullStopOnline <= 0 {
|
|
config.RoomFullStopOnline = config.RoomTargetMaxOnline
|
|
}
|
|
if config.RobotStayMinMS <= 0 && config.RobotStayMaxMS <= 0 {
|
|
config.RobotStayMinMS = defaultHumanRobotStayMinMS
|
|
config.RobotStayMaxMS = defaultHumanRobotStayMaxMS
|
|
}
|
|
if config.RobotReplaceMinMS == 0 && config.RobotReplaceMaxMS == 0 {
|
|
config.RobotReplaceMinMS = defaultHumanRobotReplaceMinMS
|
|
config.RobotReplaceMaxMS = defaultHumanRobotReplaceMaxMS
|
|
}
|
|
if config.NormalGiftIntervalMinMS <= 0 && config.NormalGiftIntervalMaxMS <= 0 {
|
|
if config.NormalGiftIntervalMS > 0 {
|
|
config.NormalGiftIntervalMinMS = config.NormalGiftIntervalMS
|
|
config.NormalGiftIntervalMaxMS = config.NormalGiftIntervalMS
|
|
} else {
|
|
config.NormalGiftIntervalMinMS = defaultHumanRobotNormalGiftIntervalMinMS
|
|
config.NormalGiftIntervalMaxMS = defaultHumanRobotNormalGiftIntervalMaxMS
|
|
}
|
|
}
|
|
if config.NormalGiftIntervalMS <= 0 {
|
|
config.NormalGiftIntervalMS = config.NormalGiftIntervalMaxMS
|
|
}
|
|
if config.LuckyComboMin <= 0 && config.LuckyComboMax <= 0 {
|
|
config.LuckyComboMin = defaultHumanRobotLuckyComboMin
|
|
config.LuckyComboMax = defaultHumanRobotLuckyComboMax
|
|
}
|
|
if config.LuckyPauseMinMS <= 0 && config.LuckyPauseMaxMS <= 0 {
|
|
config.LuckyPauseMinMS = defaultHumanRobotLuckyPauseMinMS
|
|
config.LuckyPauseMaxMS = defaultHumanRobotLuckyPauseMaxMS
|
|
}
|
|
if config.MaxGiftSenders <= 0 {
|
|
config.MaxGiftSenders = defaultHumanRobotMaxGiftSenders
|
|
}
|
|
return config
|
|
}
|
|
|
|
func defaultHumanRoomRobotConfig(appCode string, nowMS int64) HumanRoomRobotConfig {
|
|
return withHumanRoomRobotDefaults(HumanRoomRobotConfig{AppCode: appcode.Normalize(appCode), CreatedAtMS: nowMS, UpdatedAtMS: nowMS})
|
|
}
|
|
|
|
func humanRoomRobotConfigToProto(config HumanRoomRobotConfig) *roomv1.AdminHumanRoomRobotConfig {
|
|
config = withHumanRoomRobotDefaults(config)
|
|
out := &roomv1.AdminHumanRoomRobotConfig{
|
|
AppCode: config.AppCode,
|
|
Enabled: config.Enabled,
|
|
CandidateRoomMaxOnline: config.CandidateRoomMaxOnline,
|
|
RoomFullStopOnline: config.RoomFullStopOnline,
|
|
RoomTargetMinOnline: config.RoomTargetMinOnline,
|
|
RoomTargetMaxOnline: config.RoomTargetMaxOnline,
|
|
RobotStayMinMs: config.RobotStayMinMS,
|
|
RobotStayMaxMs: config.RobotStayMaxMS,
|
|
RobotReplaceMinMs: config.RobotReplaceMinMS,
|
|
RobotReplaceMaxMs: config.RobotReplaceMaxMS,
|
|
NormalGiftIntervalMs: config.NormalGiftIntervalMS,
|
|
NormalGiftIntervalMinMs: config.NormalGiftIntervalMinMS,
|
|
NormalGiftIntervalMaxMs: config.NormalGiftIntervalMaxMS,
|
|
GiftIds: append([]string(nil), config.GiftIDs...),
|
|
LuckyGiftIds: append([]string(nil), config.LuckyGiftIDs...),
|
|
SuperLuckyGiftIds: append([]string(nil), config.SuperLuckyGiftIDs...),
|
|
LuckyComboMin: config.LuckyComboMin,
|
|
LuckyComboMax: config.LuckyComboMax,
|
|
LuckyPauseMinMs: config.LuckyPauseMinMS,
|
|
LuckyPauseMaxMs: config.LuckyPauseMaxMS,
|
|
MaxGiftSenders: config.MaxGiftSenders,
|
|
UpdatedByAdminId: config.UpdatedByAdminID,
|
|
CreatedAtMs: config.CreatedAtMS,
|
|
UpdatedAtMs: config.UpdatedAtMS,
|
|
}
|
|
return out
|
|
}
|
|
|
|
func humanRoomRobotCommandMeta(config HumanRoomRobotConfig, roomID string, actorUserID int64, suffix string) *roomv1.RequestMeta {
|
|
commandID := fmt.Sprintf("human-room-robot:%s:%s", roomID, strings.TrimSpace(suffix))
|
|
if len(commandID) > 128 {
|
|
commandID = commandID[:128]
|
|
}
|
|
return &roomv1.RequestMeta{
|
|
RequestId: commandID,
|
|
CommandId: commandID,
|
|
ActorUserId: actorUserID,
|
|
RoomId: roomID,
|
|
AppCode: config.AppCode,
|
|
SentAtMs: time.Now().UTC().UnixMilli(),
|
|
}
|
|
}
|
|
|
|
func humanRoomRealOnlineCount(users []*roomv1.RoomUser, robotIDs map[int64]bool) int {
|
|
count := 0
|
|
for _, user := range users {
|
|
if user.GetUserId() <= 0 || robotIDs[user.GetUserId()] {
|
|
continue
|
|
}
|
|
count++
|
|
}
|
|
return count
|
|
}
|
|
|
|
func allHumanRoomRobotIDs(config HumanRoomRobotConfig) map[int64]bool {
|
|
out := make(map[int64]bool)
|
|
for _, pool := range config.CountryPools {
|
|
for _, id := range pool.RobotUserIDs {
|
|
if id > 0 {
|
|
out[id] = true
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func freeSeatNumbers(seats []*roomv1.SeatState) []int32 {
|
|
out := make([]int32, 0, len(seats))
|
|
for _, seat := range seats {
|
|
if seat.GetSeatNo() > 0 && seat.GetUserId() == 0 && !seat.GetLocked() {
|
|
out = append(out, seat.GetSeatNo())
|
|
}
|
|
}
|
|
return out
|
|
}
|