From de9c1e9eef183ed14da13e18bd941110f829d915 Mon Sep 17 00:00:00 2001 From: zhx Date: Fri, 17 Jul 2026 16:46:04 +0800 Subject: [PATCH] fix(room): make robot gifts display-only --- docs/数据大屏统计口径.md | 3 + docs/机器人房间技术架构文档.md | 50 +-- services/room-service/internal/app/app.go | 7 +- .../integration/robot_gift_catalog.go | 18 + .../internal/integration/tencent_im.go | 17 +- .../internal/integration/tencent_im_test.go | 38 +- .../internal/room/service/gift.go | 6 + .../internal/room/service/gift_flow.go | 21 +- .../internal/room/service/gift_lucky.go | 23 +- .../internal/room/service/gift_robot.go | 338 +++++++++++++++++- .../lucky_gift_display_internal_test.go | 13 + .../service/outbox_worker_internal_test.go | 49 +++ .../room/service/robot_display_publisher.go | 25 +- .../room/service/robot_display_sampler.go | 2 +- .../internal/room/service/room_rocket_test.go | 33 +- .../internal/room/service/service.go | 14 +- .../room/service/user_gift_value_test.go | 260 +++++++++++--- 17 files changed, 796 insertions(+), 121 deletions(-) create mode 100644 services/room-service/internal/integration/robot_gift_catalog.go diff --git a/docs/数据大屏统计口径.md b/docs/数据大屏统计口径.md index a7118324..6f192799 100644 --- a/docs/数据大屏统计口径.md +++ b/docs/数据大屏统计口径.md @@ -49,6 +49,9 @@ | `real_room_robot_total_gift_coin` | 真人房机器人总礼物 | 上面三类机器人礼物金币合计。 | | `real_room_robot_gift_room_count` | 真人房机器人投放房间数 | 有真人房机器人礼物的房间去重数。 | | `real_room_robot_avg_room_gift_coin` | 真人房机器人房均投放 | `real_room_robot_total_gift_coin / real_room_robot_gift_room_count`。 | + +> 2026-07-17 起 robot 送礼改为纯腾讯云 IM 展示，不再写 room/wallet 数据库或发布 MQ。`real_room_robot_*` 字段只保留历史兼容；新事件不会再增长，不能继续作为当前机器人投放口径。 + | `lucky_gift_turnover` | lucky 礼物流水 | `RoomGiftSent.pool_id` 非空的送礼金币。 | | `lucky_gift_payout` | lucky 礼物产出 | `WalletLuckyGiftRewardCredited.amount`，UTC 查询会优先用 activity 的 lucky pool 日表覆盖旧统计行。 | | `lucky_gift_payers` | lucky 礼物付费用户 | 发送 lucky 礼物的用户按日去重。 | diff --git a/docs/机器人房间技术架构文档.md b/docs/机器人房间技术架构文档.md index f4264258..ca4301d3 100644 --- a/docs/机器人房间技术架构文档.md +++ b/docs/机器人房间技术架构文档.md @@ -37,11 +37,9 @@ ### wallet-service -- 机器人礼物扣费入口：`DebitRobotGift`。 -- 使用专用资产 `ROBOT_COIN`。 -- 自动补足机器人专用金币后扣减。 -- 不发出真实 `WalletGiftDebited` 事件。 -- 不写主播周期钻石和真实礼物墙。 +- 只通过 `ListGiftConfigs` 提供 active、区域可用的礼物展示快照。 +- `RobotSendGift` 不调用 `DebitRobotGift`，不创建机器人钱包账户、交易、分录或 wallet outbox。 +- 礼物目录在 room-service 按 App/区域短缓存；缓存只保存配置，不保存任何送礼事实。 ### activity/statistics @@ -101,7 +99,7 @@ room-service gRPC： wallet-service gRPC： -- `DebitRobotGift` +- `ListGiftConfigs`（只读） ## 创建流程 @@ -155,24 +153,22 @@ room-service 启动时会扫描所有 active 机器人房间并恢复运行时 room-service 在机器人送礼时强校验： -- 当前房间必须是机器人房间。 +- 专属机器人房必须带 `robot_room=true`；真人房机器人使用运行时机器人池校验，不把真人加入允许集合。 - sender 必须在该机器人房间 `robot_user_ids_json` 内。 - target 必须在同一机器人房间 `robot_user_ids_json` 内。 - sender 不能给真人送礼。 - 当前机器人礼物只允许单目标调用。 -## 钱包隔离 +## 纯展示隔离 -机器人礼物使用 `DebitRobotGift`： +`RobotSendGift` 不进入通用 `sendGift/mutateRoom`，只执行： -- 资产类型固定为 `ROBOT_COIN`。 -- 如果机器人 `ROBOT_COIN` 不足，同一事务内自动补足。 -- 立即扣减本次礼物价格。 -- 交易类型为 `robot_gift_debit`。 -- 不写真实 `gift_debit`。 -- 不发 `WalletGiftDebited`。 -- 不写 `user_gift_wall`。 -- 不写 `host_period_diamond_accounts`。 +1. 获得或续租 Redis owner lease，并只读当前 Room Cell（缺失时从 MySQL 只读恢复），校验 sender、target 和机器人池。 +2. 从 wallet 礼物目录读取并短缓存名称、图标、动画、类型和展示价值。 +3. 在内存构造 `RoomGiftSent`；幸运礼物额外构造 `RoomRobotLuckyGiftDrawn`。 +4. 经过 2 秒展示采样后，best-effort 直发腾讯云 IM。 + +这条链路不写 `room_gift_operations`、`room_command_log`、`room_snapshots`、`room_outbox`、房间贡献统计、钱包账本或任何 outbox，也不调用 RocketMQ producer。内存里的 `outbox.Record` 只复用 protobuf 信封结构，不是持久化 outbox。 ## 幸运礼物隔离 @@ -189,25 +185,14 @@ room-service 只生成展示事件： ## 排行榜与统计隔离 -`RoomGiftSent` 增加： +机器人展示用 `RoomGiftSent` 保留： - `is_robot_gift` - `synthetic_lucky_gift` -下游服务看到 `is_robot_gift=true` 时跳过： +该事件不会进入 room outbox/MQ，所以下游活动、CP、任务、成长和 statistics 根本不会收到机器人送礼事实；`is_robot_gift=true` 只供 IM 展示兼容。 -- 活动播报统计 -- 周星 -- 房间流水奖励 -- 每日任务 -- 成长值 -- statistics 礼物消费与幸运礼物统计 - -room-service 本地仍允许机器人房内： - -- 房间热度增长 -- 麦位热度增长 -- 房内展示事件发送 +机器人送礼同时不修改房间热度、礼物榜、麦位 `gift_value`、火箭燃料、周贡献或 Redis 跨房贡献榜。真人房机器人也遵守同一边界；历史 `RealRoomHeat` 参数仅用于区分参与者校验范围，不能授权真实贡献。 ## 本地接口测试 @@ -232,7 +217,7 @@ go run ./tools/robot-room-api-test 5. 创建机器人房间。 6. 查询 active 列表。 7. 停止、启动、再次停止机器人房间。 -8. 校验数据库中机器人房间配置和真实 `gift_debit` 隔离。 +8. 校验 robot 展示前后 `room_gift_operations`、command log、snapshot、room outbox、房间统计行数不变，且 `DebitRobotGift` 调用次数为 0。 ## 运维注意事项 @@ -241,3 +226,4 @@ go run ./tools/robot-room-api-test - 同一机器人账号不能被多个 active 机器人房间占用。 - 停止机器人房间只停止 runtime，不删除房间。 - 如果需要彻底释放机器人账号，应增加删除/归档能力后再开放。 +- 腾讯云 IM 是唯一实时展示通道，失败不补写数据库或 MQ，也不重试成业务事实；丢失的动画/公屏不恢复，房间快照只保证真实热度、榜单和麦位值不被机器人展示污染。 diff --git a/services/room-service/internal/app/app.go b/services/room-service/internal/app/app.go index dc4505ac..32708d6f 100644 --- a/services/room-service/internal/app/app.go +++ b/services/room-service/internal/app/app.go @@ -164,7 +164,7 @@ func New(cfg config.Config) (*App, error) { roomPublisher := integration.NewNoopRoomEventPublisher() outboxPublishers := make([]integration.OutboxPublisher, 0, 3) var rtcUserRemover integration.RTCUserRemover - var tencentPublisher *integration.TencentIMPublisher + var tencentPublisher integration.OutboxPublisher if cfg.TencentIM.Enabled { // 腾讯云 IM 替代自研 IM；room-service 只负责服务端 REST 群消息桥接。 tencentClient, err := tencentim.NewRESTClient(cfg.TencentIM.RESTConfig()) @@ -175,8 +175,9 @@ func New(cfg config.Config) (*App, error) { _ = redisClient.Close() return nil, err } - tencentPublisher = integration.NewTencentIMPublisher(tencentClient) - roomPublisher = tencentPublisher + concreteTencentPublisher := integration.NewTencentIMPublisher(tencentClient) + tencentPublisher = concreteTencentPublisher + roomPublisher = concreteTencentPublisher } if cfg.TencentRTC.Enabled { rtcClient, err := tencentrtc.NewRESTClient(cfg.TencentRTC.RESTConfig()) diff --git a/services/room-service/internal/integration/robot_gift_catalog.go b/services/room-service/internal/integration/robot_gift_catalog.go new file mode 100644 index 00000000..6ea74bf8 --- /dev/null +++ b/services/room-service/internal/integration/robot_gift_catalog.go @@ -0,0 +1,18 @@ +package integration + +import ( + "context" + + walletv1 "hyapp.local/api/proto/wallet/v1" +) + +// WalletGiftCatalogClient 是机器人展示链路读取礼物快照的可选能力。 +// RobotSendGift 只能调用这个只读目录接口，不能再借 DebitRobotGift 产生钱包账户、交易、分录或 outbox。 +type WalletGiftCatalogClient interface { + ListGiftConfigs(ctx context.Context, req *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) +} + +// ListGiftConfigs 只读取 wallet owner 的礼物目录；机器人展示使用该快照，绝不调用账务扣费接口。 +func (c *grpcWalletClient) ListGiftConfigs(ctx context.Context, req *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) { + return c.client.ListGiftConfigs(ctx, req) +} diff --git a/services/room-service/internal/integration/tencent_im.go b/services/room-service/internal/integration/tencent_im.go index 1619307d..af3530b9 100644 --- a/services/room-service/internal/integration/tencent_im.go +++ b/services/room-service/internal/integration/tencent_im.go @@ -321,7 +321,6 @@ func roomEventFromEnvelope(envelope *roomeventsv1.EventEnvelope) (tencentim.Room "gift_count": fmt.Sprintf("%d", body.GetGiftCount()), "coin_spent": fmt.Sprintf("%d", body.GetCoinSpent()), "billing_receipt_id": body.GetBillingReceiptId(), - "target_gift_value": fmt.Sprintf("%d", body.GetTargetGiftValue()), "gift_type_code": body.GetGiftTypeCode(), "cp_relation_type": body.GetCpRelationType(), "gift_name": body.GetGiftName(), @@ -337,6 +336,14 @@ func roomEventFromEnvelope(envelope *roomeventsv1.EventEnvelope) (tencentim.Room "receiver_display_user_id": body.GetReceiverDisplayUserId(), "receiver_pretty_display_user_id": body.GetReceiverPrettyDisplayUserId(), } + if body.GetIsRobotGift() { + // 机器人礼物只用于动画和公屏展示。Flutter 会把顶层 gift_value 累加到成员贡献， + // 也会用 target_gift_value 覆盖麦位值；两者都必须从机器人协议中移除，避免假贡献和乱序回退。 + base.GiftValue = 0 + base.Attributes["is_robot_gift"] = "true" + } else { + base.Attributes["target_gift_value"] = fmt.Sprintf("%d", body.GetTargetGiftValue()) + } return base, true, nil case "RoomGiftBatchSent": // 批量送礼展示事件只面向新客户端；逐目标 RoomGiftSent 仍由 outbox 给统计、CP 和礼物墙消费。 @@ -448,9 +455,13 @@ func roomEventFromEnvelope(envelope *roomeventsv1.EventEnvelope) (tencentim.Room } base.ActorUserID = body.GetSenderUserId() base.TargetUserID = body.GetTargetUserId() - base.GiftValue = body.GetEffectiveRewardCoins() + // 中奖金额保留在 effective_reward_coins 供动画展示；顶层 gift_value 必须为 0， + // 否则 Flutter 会把纯展示中奖额累计进机器人发送者的成员贡献。 + base.GiftValue = 0 base.Attributes = map[string]string{ - "event_type": "lucky_gift_drawn", + "event_type": "lucky_gift_drawn", + // RoomEvent.GiftValue=0 会被 omitempty 省略；显式扁平 gift_value=0，防止 Flutter 回退读取中奖额并累计成员贡献。 + "gift_value": "0", "draw_id": body.GetDrawId(), "command_id": body.GetCommandId(), "sender_user_id": fmt.Sprintf("%d", body.GetSenderUserId()), diff --git a/services/room-service/internal/integration/tencent_im_test.go b/services/room-service/internal/integration/tencent_im_test.go index 0a00e6a2..c8875f6e 100644 --- a/services/room-service/internal/integration/tencent_im_test.go +++ b/services/room-service/internal/integration/tencent_im_test.go @@ -137,6 +137,30 @@ func TestRoomGiftSentCarriesTargetGiftValueInIMAttributes(t *testing.T) { if event.Attributes["receiver_nickname"] != "Robot Receiver" || event.Attributes["receiver_avatar"] != "https://cdn.example/receiver.png" || event.Attributes["receiver_display_user_id"] != "100002" { t.Fatalf("gift IM event receiver display fields mismatch: %+v", event.Attributes) } + + robotRecord, err := outbox.Build("room-robot-gift-value", "RoomGiftSent", 8, time.Now(), &roomeventsv1.RoomGiftSent{ + SenderUserId: 2001, + TargetUserId: 2002, + GiftId: "gift_robot_rose", + GiftCount: 1, + GiftValue: 999, + TargetGiftValue: 888, + CoinSpent: 100, + IsRobotGift: true, + }) + if err != nil { + t.Fatalf("Build robot RoomGiftSent envelope failed: %v", err) + } + robotEvent, publish, err := roomEventFromEnvelope(robotRecord.Envelope) + if err != nil || !publish { + t.Fatalf("robot RoomGiftSent should publish: publish=%t err=%v", publish, err) + } + if robotEvent.GiftValue != 0 || robotEvent.Attributes["is_robot_gift"] != "true" { + t.Fatalf("robot gift IM must expose zero business contribution: %+v", robotEvent) + } + if _, exists := robotEvent.Attributes["target_gift_value"]; exists { + t.Fatalf("robot gift IM must not overwrite authoritative target value: %+v", robotEvent.Attributes) + } } func TestRoomGiftSentBatchDisplayFactSkipsSingleTargetIM(t *testing.T) { @@ -291,16 +315,28 @@ func TestRoomRobotLuckyGiftDrawnPublishesLuckyGiftIMEvent(t *testing.T) { if err != nil { t.Fatalf("RoomRobotLuckyGiftDrawn should decode: %v", err) } - if !publish || event.EventType != "lucky_gift_drawn" || event.ActorUserID != 1001 || event.TargetUserID != 1002 { + if !publish || event.EventType != "lucky_gift_drawn" || event.ActorUserID != 1001 || event.TargetUserID != 1002 || event.GiftValue != 0 { t.Fatalf("robot lucky gift IM event mismatch: publish=%t event=%+v", publish, event) } if event.Attributes["draw_id"] != "robot_lucky_draw_1" || event.Attributes["gift_id"] != "gift_lucky_1" || event.Attributes["pool_id"] != "robot_lucky_display" || + event.Attributes["gift_value"] != "0" || event.Attributes["effective_reward_coins"] != "500" || event.Attributes["reward_status"] != "granted" { t.Fatalf("robot lucky gift IM attributes mismatch: %+v", event.Attributes) } + payload, err := json.Marshal(event) + if err != nil { + t.Fatalf("marshal robot lucky gift IM event failed: %v", err) + } + var flattened map[string]any + if err := json.Unmarshal(payload, &flattened); err != nil { + t.Fatalf("decode flattened robot lucky gift IM event failed: %v", err) + } + if flattened["gift_value"] != "0" || flattened["effective_reward_coins"] != "500" { + t.Fatalf("robot lucky gift JSON must keep reward display but zero contribution: %s", payload) + } } func TestRoomLuckyGiftDrawnPublishesGrantedOwnerFactAsLuckyGiftIMEvent(t *testing.T) { diff --git a/services/room-service/internal/room/service/gift.go b/services/room-service/internal/room/service/gift.go index f6037d18..094ece7a 100644 --- a/services/room-service/internal/room/service/gift.go +++ b/services/room-service/internal/room/service/gift.go @@ -4,6 +4,7 @@ import ( "context" "time" + "hyapp/pkg/xerr" "hyapp/services/room-service/internal/room/outbox" "hyapp/services/room-service/internal/room/rank" "hyapp/services/room-service/internal/room/state" @@ -41,6 +42,11 @@ func (s *Service) sendGift(ctx context.Context, req *roomv1.SendGiftRequest, opt } func (s *Service) sendGiftWithRecovery(ctx context.Context, req *roomv1.SendGiftRequest, options giftSendOptions, recovery GiftOperation) (*roomv1.SendGiftResponse, error) { + if options.Robot.Enabled && recovery.CommandID == "" { + // 新机器人礼物只能走 RobotSendGift 的纯展示通道。这里 fail-close，防止后续调用方误把机器人重新接回 + // room_gift_operations、Room Cell 持久化、钱包账本或 MQ；已有历史 saga 恢复不依赖 options.Robot。 + return nil, xerr.New(xerr.InvalidArgument, "robot gift must use display-only path") + } ctx = contextFromMeta(ctx, req.GetMeta()) flow := newGiftFlow(s, ctx, req, options) if recovery.CommandID != "" { diff --git a/services/room-service/internal/room/service/gift_flow.go b/services/room-service/internal/room/service/gift_flow.go index 6f46a3d4..05e4592d 100644 --- a/services/room-service/internal/room/service/gift_flow.go +++ b/services/room-service/internal/room/service/gift_flow.go @@ -87,16 +87,17 @@ func newGiftFlow(s *Service, ctx context.Context, req *roomv1.SendGiftRequest, o SenderRegisteredAtMS: req.GetSenderRegisteredAtMs(), TargetIsHost: req.GetTargetIsHost(), // 工资区域由 gateway 从 user-service host profile 注入；房间 visible_region_id 只用于房间/礼物可见性。 - TargetHostRegionID: req.GetTargetHostRegionId(), - TargetAgencyOwnerUserID: req.GetTargetAgencyOwnerUserId(), - TargetHostScopes: giftTargetHostScopesFromProto(req.GetTargetHostScopes()), - SenderDisplayProfile: giftDisplayProfileFromProto(req.GetSenderDisplayProfile()), - TargetDisplayProfiles: giftDisplayProfilesFromProto(req.GetTargetDisplayProfiles()), - RobotGift: options.Robot.Enabled && !options.Robot.RealRoomHeat, - RobotWalletGift: options.Robot.Enabled, - SyntheticLuckyGift: options.Robot.Enabled && strings.TrimSpace(req.GetPoolId()) != "", - SyntheticLuckyRewardCoins: firstPositiveInt64(options.Robot.SyntheticRewardCoins, 0), - SyntheticLuckyMultiplierPPM: firstPositiveInt64(options.Robot.SyntheticMultiplierPPM, 1000000), + TargetHostRegionID: req.GetTargetHostRegionId(), + TargetAgencyOwnerUserID: req.GetTargetAgencyOwnerUserId(), + TargetHostScopes: giftTargetHostScopesFromProto(req.GetTargetHostScopes()), + SenderDisplayProfile: giftDisplayProfileFromProto(req.GetSenderDisplayProfile()), + TargetDisplayProfiles: giftDisplayProfilesFromProto(req.GetTargetDisplayProfiles()), + RobotGift: options.Robot.Enabled && !options.Robot.RealRoomHeat, + RobotWalletGift: options.Robot.Enabled, + SyntheticLuckyGift: options.Robot.Enabled && strings.TrimSpace(req.GetPoolId()) != "", + SyntheticLuckyRewardCoins: firstPositiveInt64(options.Robot.SyntheticRewardCoins, 0), + // 0 留给 syntheticLuckyGiftResult 统一选择 5x 默认值；这里提前写成 1x 会让倍数和展示中奖额互相矛盾。 + SyntheticLuckyMultiplierPPM: firstPositiveInt64(options.Robot.SyntheticMultiplierPPM, 0), } if cmd.TargetType == "" { cmd.TargetType = "user" diff --git a/services/room-service/internal/room/service/gift_lucky.go b/services/room-service/internal/room/service/gift_lucky.go index 98e953b6..e900e14f 100644 --- a/services/room-service/internal/room/service/gift_lucky.go +++ b/services/room-service/internal/room/service/gift_lucky.go @@ -3,6 +3,8 @@ package service import ( "context" "fmt" + "math" + "math/bits" "strings" "time" @@ -278,8 +280,11 @@ func syntheticLuckyGiftResult(cmd command.SendGift, targetBillings []giftTargetB if len(targetBillings) == 0 || targetBillings[0].Billing == nil { return nil } - rewardCoins := firstPositiveInt64(cmd.SyntheticLuckyRewardCoins, targetBillings[0].Billing.GetCoinSpent()*5) + coinSpent := targetBillings[0].Billing.GetCoinSpent() multiplier := firstPositiveInt64(cmd.SyntheticLuckyMultiplierPPM, 5000000) + // 未显式指定中奖额时必须按同一 multiplier 推导，避免 20x 文案却展示 5x 金额；计算饱和防止异常配置溢出。 + defaultRewardCoins := scaledRobotRewardCoins(coinSpent, multiplier) + rewardCoins := firstPositiveInt64(cmd.SyntheticLuckyRewardCoins, defaultRewardCoins) drawID := fmt.Sprintf("robot_lucky_%s_%d", cmd.ID(), now.UnixMilli()) return &roomv1.LuckyGiftDrawResult{ Enabled: true, @@ -301,6 +306,22 @@ func syntheticLuckyGiftResult(cmd command.SendGift, targetBillings []giftTargetB } } +func scaledRobotRewardCoins(coinSpent int64, multiplierPPM int64) int64 { + if coinSpent <= 0 || multiplierPPM <= 0 { + return 0 + } + // bits.Mul64 + Div64 完成无符号 128 位乘除；高 64 位已不小于除数时，商超出 uint64，直接饱和。 + high, low := bits.Mul64(uint64(coinSpent), uint64(multiplierPPM)) + if high >= 1_000_000 { + return math.MaxInt64 + } + reward, _ := bits.Div64(high, low, 1_000_000) + if reward > math.MaxInt64 { + return math.MaxInt64 + } + return int64(reward) +} + func giftFinalCoinBalanceAfter(billing *walletv1.DebitGiftResponse) int64 { if billing == nil { return 0 diff --git a/services/room-service/internal/room/service/gift_robot.go b/services/room-service/internal/room/service/gift_robot.go index eff51e9b..14128344 100644 --- a/services/room-service/internal/room/service/gift_robot.go +++ b/services/room-service/internal/room/service/gift_robot.go @@ -2,13 +2,46 @@ package service import ( "context" + "math" "strings" + "time" + roomeventsv1 "hyapp.local/api/proto/events/room/v1" roomv1 "hyapp.local/api/proto/room/v1" + walletv1 "hyapp.local/api/proto/wallet/v1" + "hyapp/pkg/appcode" + "hyapp/pkg/giftlimits" "hyapp/pkg/xerr" + "hyapp/services/room-service/internal/integration" + "hyapp/services/room-service/internal/room/outbox" "hyapp/services/room-service/internal/room/state" ) +const ( + // 机器人送礼量远高于礼物配置变更频率；短缓存把每次展示的 wallet/MySQL 只读查询收敛为每 App/区域每分钟一次。 + robotGiftCatalogTTL = time.Minute + // wallet 故障只短暂负缓存，既让同批机器人共享失败，也能在服务恢复后迅速重试。 + robotGiftCatalogErrorTTL = time.Second + robotGiftCatalogQueryTimeout = 3 * time.Second + robotGiftCatalogPageSize = int32(500) + robotGiftCatalogMaxPages = int32(20) +) + +type robotGiftCatalogKey struct { + appCode string + regionID int64 +} + +type robotGiftCatalogEntry struct { + expiresAtMS int64 + gifts map[string]*walletv1.GiftConfig + loadErr error +} + +type robotGiftCatalogLoad struct { + done chan struct{} +} + type RobotSendGiftInput struct { Meta *roomv1.RequestMeta TargetUserID int64 @@ -18,7 +51,8 @@ type RobotSendGiftInput struct { RobotUserIDs []int64 SyntheticRewardCoins int64 SyntheticMultiplierPPM int64 - RealRoomHeat bool + // RealRoomHeat 是历史字段名；当前只表示“真人房机器人”校验范围，绝不再产生真实热度或任何持久化贡献。 + RealRoomHeat bool } type robotGiftOptions struct { @@ -26,7 +60,8 @@ type robotGiftOptions struct { RobotUserIDs []int64 SyntheticRewardCoins int64 SyntheticMultiplierPPM int64 - RealRoomHeat bool + // RealRoomHeat 仅供 validateBeforeDebit 区分专属机器人房和真人房机器人池，不能进入状态/账务结算。 + RealRoomHeat bool } func (s *Service) RobotSendGift(ctx context.Context, input RobotSendGiftInput) (*roomv1.SendGiftResponse, error) { @@ -34,12 +69,13 @@ func (s *Service) RobotSendGift(ctx context.Context, input RobotSendGiftInput) ( Meta: input.Meta, TargetUserId: input.TargetUserID, TargetUserIds: []int64{input.TargetUserID}, - GiftId: input.GiftID, + GiftId: strings.TrimSpace(input.GiftID), GiftCount: input.GiftCount, TargetType: "user", PoolId: input.PoolID, } - return s.sendGift(ctx, req, giftSendOptions{ + ctx = contextFromMeta(ctx, req.GetMeta()) + flow := newGiftFlow(s, ctx, req, giftSendOptions{ Robot: robotGiftOptions{ Enabled: true, RobotUserIDs: input.RobotUserIDs, @@ -48,6 +84,300 @@ func (s *Service) RobotSendGift(ctx context.Context, input RobotSendGiftInput) ( RealRoomHeat: input.RealRoomHeat, }, }) + // RobotGift 永远是纯展示语义。RealRoomHeat 只兼容真人房机器人参与者校验，不能再授权热度、榜单、火箭或统计落库。 + flow.cmd.RobotGift = true + flow.cmd.RobotWalletGift = true + + if flow.cmd.RoomID() == "" || flow.cmd.ActorUserID() <= 0 || !giftlimits.ValidCommandID(flow.cmd.ID()) { + return nil, xerr.New(xerr.InvalidArgument, "command meta is incomplete") + } + if s.isDraining() { + return nil, xerr.New(xerr.Unavailable, "room service is draining") + } + + // 只读取当前 Room Cell 做 presence 与机器人池校验；ensureCell 仅续租 Redis owner lease， + // 不进入 mutateRoom，因此不会写 room_gift_operations、command log、snapshot、账本或房间统计表。 + snapshot, leaseToken, err := s.robotGiftOwnerSnapshot(ctx, flow.cmd.RoomID()) + if err != nil { + return nil, err + } + current := state.FromProto(snapshot) + if err := flow.validateBeforeDebit(current); err != nil { + return nil, err + } + + // 礼物目录调用是只读快照查询，替代历史 DebitRobotGift；返回值仅用于 IM 图标、动画、类型和展示价值。 + giftConfig, err := s.robotGiftConfig(ctx, flow.cmd.GiftID, snapshot.GetVisibleRegionId()) + if err != nil { + return nil, err + } + billing, err := robotGiftDisplayBilling(giftConfig, flow.cmd.GiftCount) + if err != nil { + return nil, err + } + targetBillings := []giftTargetBilling{{ + TargetUserID: flow.cmd.TargetUserID, + CommandID: flow.cmd.ID(), + Billing: billing, + }} + targetGiftValues := map[int64]int64{} + if target := current.OnlineUsers[flow.cmd.TargetUserID]; target != nil { + // 纯展示礼物不能改变麦位/在线用户累计值；事件携带当前值，客户端不会把机器人价值误并入真实状态。 + targetGiftValues[flow.cmd.TargetUserID] = target.GiftValue + } + + now := s.clock.Now() + records, err := buildRoomGiftSentRecords(flow.cmd.RoomID(), snapshot.GetVersion(), now, flow.cmd, RoomMeta{ + RoomID: flow.cmd.RoomID(), + VisibleRegionID: snapshot.GetVisibleRegionId(), + }, targetBillings, targetGiftValues) + if err != nil { + return nil, err + } + + var luckyGift *roomv1.LuckyGiftDrawResult + var luckyGifts []*roomv1.LuckyGiftDrawResult + if flow.cmd.SyntheticLuckyGift { + luckyGift = syntheticLuckyGiftResult(flow.cmd, targetBillings, now) + if luckyGift != nil { + luckyGifts = []*roomv1.LuckyGiftDrawResult{luckyGift} + drawRecord, buildErr := outbox.Build(flow.cmd.RoomID(), "RoomRobotLuckyGiftDrawn", snapshot.GetVersion(), now, &roomeventsv1.RoomRobotLuckyGiftDrawn{ + DrawId: luckyGift.GetDrawId(), + CommandId: flow.cmd.ID(), + SenderUserId: flow.cmd.ActorUserID(), + TargetUserId: flow.cmd.TargetUserID, + GiftId: flow.cmd.GiftID, + GiftCount: flow.cmd.GiftCount, + PoolId: flow.cmd.PoolID, + MultiplierPpm: luckyGift.GetMultiplierPpm(), + BaseRewardCoins: luckyGift.GetBaseRewardCoins(), + EffectiveRewardCoins: luckyGift.GetEffectiveRewardCoins(), + CreatedAtMs: luckyGift.GetCreatedAtMs(), + VisibleRegionId: snapshot.GetVisibleRegionId(), + IsRobot: true, + Synthetic: true, + }) + if buildErr != nil { + return nil, buildErr + } + records = append(records, drawRecord) + } + } + + // outbox.Record 在这里仅是复用 protobuf 信封的内存载体；不调用 repository/outboxPublisher， + // 只经过采样器后直接交给生产环境的 TencentIMPublisher，因此不会生成任何 RocketMQ 消息。 + records = s.sampleRobotDisplayRecords(ctx, records) + if len(records) > 0 { + // wallet 目录查询最多持续 3 秒；发布前用同一 token 再做 fencing，防止查询期间房间已被另一节点接管。 + if err := s.verifyRobotGiftOwner(ctx, flow.cmd.RoomID(), leaseToken); err != nil { + return nil, err + } + } + s.publishRobotDisplayRecordsBestEffort(ctx, records) + + return &roomv1.SendGiftResponse{ + Result: commandResult(true, snapshot.GetVersion(), now), + RoomHeat: snapshot.GetHeat(), + GiftRank: snapshot.GetGiftRank(), + Room: snapshot, + Rocket: snapshot.GetRocket(), + LuckyGift: luckyGift, + LuckyGifts: luckyGifts, + }, nil +} + +func (s *Service) robotGiftOwnerSnapshot(ctx context.Context, roomID string) (*roomv1.RoomSnapshot, string, error) { + if s.directory == nil { + // 极简内部测试允许没有 lease 目录；生产 HealthCheck 会拒绝这种装配。 + snapshot, err := s.currentSnapshot(ctx, roomID) + if err != nil { + return nil, "", err + } + if snapshot == nil { + return nil, "", xerr.New(xerr.NotFound, "room not found") + } + return snapshot, "", nil + } + roomCell, lease, err := s.ensureCell(ctx, roomID) + if err != nil { + return nil, "", err + } + current, _, err := roomCell.Snapshot(ctx) + if err != nil { + return nil, "", err + } + return snapshotWithApp(ctx, current.ToProto()), lease.LeaseToken, nil +} + +func (s *Service) verifyRobotGiftOwner(ctx context.Context, roomID string, leaseToken string) error { + if s.directory == nil { + return nil + } + owned, err := s.directory.VerifyOwner(ctx, runtimeRoomKeyFromContext(ctx, roomID), s.nodeID, leaseToken, s.clock.Now()) + if err != nil { + return err + } + if !owned { + return xerr.New(xerr.Conflict, "room owner lease changed before robot display") + } + return nil +} + +func (s *Service) robotGiftConfig(ctx context.Context, giftID string, regionID int64) (*walletv1.GiftConfig, error) { + key := robotGiftCatalogKey{appCode: appcode.FromContext(ctx), regionID: regionID} + for { + nowMS := s.clock.Now().UnixMilli() + s.robotGiftCatalogMu.Lock() + if cached, exists := s.robotGiftCatalog[key]; exists && cached.expiresAtMS > nowMS { + s.robotGiftCatalogMu.Unlock() + if cached.loadErr != nil { + return nil, cached.loadErr + } + return robotGiftConfigOrNotFound(cached.gifts[giftID]) + } + if loading := s.robotGiftCatalogLoads[key]; loading != nil { + done := loading.done + s.robotGiftCatalogMu.Unlock() + select { + case <-done: + // loader 已把成功或短负缓存结果写入 catalog；循环一次统一读取。 + continue + case <-ctx.Done(): + return nil, ctx.Err() + } + } + if s.robotGiftCatalogLoads == nil { + s.robotGiftCatalogLoads = make(map[robotGiftCatalogKey]*robotGiftCatalogLoad) + } + loading := &robotGiftCatalogLoad{done: make(chan struct{})} + s.robotGiftCatalogLoads[key] = loading + s.robotGiftCatalogMu.Unlock() + // loader 脱离首个 caller 的取消信号但仍受 3 秒 RPC timeout；所有调用者都按自己的 ctx 等待同一 done。 + go s.completeRobotGiftCatalogLoad(context.WithoutCancel(ctx), key, loading) + select { + case <-loading.done: + continue + case <-ctx.Done(): + return nil, ctx.Err() + } + } +} + +func (s *Service) completeRobotGiftCatalogLoad(ctx context.Context, key robotGiftCatalogKey, loading *robotGiftCatalogLoad) { + // 网络查询在全局缓存锁外执行；同 key 等待 done，不同 key 可并行请求 wallet。 + gifts, loadErr := s.loadRobotGiftCatalog(ctx, key) + now := s.clock.Now() + expiresAt := now.Add(robotGiftCatalogTTL) + if loadErr != nil { + expiresAt = now.Add(robotGiftCatalogErrorTTL) + } + s.robotGiftCatalogMu.Lock() + if s.robotGiftCatalog == nil { + s.robotGiftCatalog = make(map[robotGiftCatalogKey]robotGiftCatalogEntry) + } + s.robotGiftCatalog[key] = robotGiftCatalogEntry{expiresAtMS: expiresAt.UnixMilli(), gifts: gifts, loadErr: loadErr} + delete(s.robotGiftCatalogLoads, key) + close(loading.done) + s.pruneRobotGiftCatalogLocked(now.UnixMilli()) + s.robotGiftCatalogMu.Unlock() +} + +func (s *Service) pruneRobotGiftCatalogLocked(nowMS int64) { + if len(s.robotGiftCatalog) <= 1024 { + return + } + for key, entry := range s.robotGiftCatalog { + if entry.expiresAtMS <= nowMS && s.robotGiftCatalogLoads[key] == nil { + delete(s.robotGiftCatalog, key) + } + } +} + +func robotGiftConfigOrNotFound(gift *walletv1.GiftConfig) (*walletv1.GiftConfig, error) { + if gift == nil { + return nil, xerr.New(xerr.NotFound, "robot gift config is not active in room region") + } + return gift, nil +} + +func (s *Service) loadRobotGiftCatalog(ctx context.Context, key robotGiftCatalogKey) (map[string]*walletv1.GiftConfig, error) { + client, ok := s.wallet.(integration.WalletGiftCatalogClient) + if !ok || client == nil { + return nil, xerr.New(xerr.Unavailable, "wallet gift catalog is not configured") + } + + queryCtx, cancel := context.WithTimeout(ctx, robotGiftCatalogQueryTimeout) + defer cancel() + gifts := make(map[string]*walletv1.GiftConfig) + var loaded int64 + for page := int32(1); page <= robotGiftCatalogMaxPages; page++ { + resp, err := client.ListGiftConfigs(queryCtx, &walletv1.ListGiftConfigsRequest{ + AppCode: key.appCode, + Page: page, + PageSize: robotGiftCatalogPageSize, + ActiveOnly: true, + RegionId: key.regionID, + FilterRegion: true, + }) + if err != nil { + return nil, err + } + for _, gift := range resp.GetGifts() { + if id := strings.TrimSpace(gift.GetGiftId()); id != "" { + gifts[id] = gift + } + } + loaded += int64(len(resp.GetGifts())) + if loaded >= resp.GetTotal() || len(resp.GetGifts()) < int(robotGiftCatalogPageSize) { + return gifts, nil + } + } + return nil, xerr.New(xerr.Unavailable, "wallet gift catalog exceeds robot display limit") +} + +func robotGiftDisplayBilling(gift *walletv1.GiftConfig, giftCount int32) (*walletv1.DebitGiftResponse, error) { + if gift == nil || giftCount <= 0 { + return nil, xerr.New(xerr.InvalidArgument, "robot gift display config is invalid") + } + coinSpent, err := checkedRobotGiftDisplayValue(gift.GetCoinPrice(), giftCount) + if err != nil { + return nil, err + } + resource := gift.GetResource() + iconURL := "" + animationURL := "" + name := strings.TrimSpace(gift.GetName()) + if resource != nil { + iconURL = strings.TrimSpace(resource.GetPreviewUrl()) + if iconURL == "" { + iconURL = strings.TrimSpace(resource.GetAssetUrl()) + } + animationURL = strings.TrimSpace(resource.GetAnimationUrl()) + if name == "" { + name = strings.TrimSpace(resource.GetName()) + } + } + // 该返回对象只是复用 RoomGiftSent 构造器的展示快照，不包含 receipt/transaction/balance，也没有任何账务含义。 + // HeatValue 必须保持 0：Flutter 会把 gift_value 增加到成员和麦位的本地值，纯展示事件不能制造短暂的假贡献。 + return &walletv1.DebitGiftResponse{ + CoinSpent: coinSpent, + ChargeAmount: coinSpent, + HeatValue: 0, + GiftTypeCode: gift.GetGiftTypeCode(), + CpRelationType: gift.GetCpRelationType(), + GiftName: name, + GiftIconUrl: iconURL, + GiftAnimationUrl: animationURL, + GiftEffectTypes: append([]string(nil), gift.GetEffectTypes()...), + PriceVersion: gift.GetPriceVersion(), + }, nil +} + +func checkedRobotGiftDisplayValue(unitValue int64, giftCount int32) (int64, error) { + if unitValue < 0 || giftCount <= 0 || unitValue > math.MaxInt64/int64(giftCount) { + return 0, xerr.New(xerr.InvalidArgument, "robot gift display value is invalid") + } + return unitValue * int64(giftCount), nil } func requireRobotGiftParticipants(current *state.RoomState, senderUserID int64, targetUserIDs []int64, runtimeRobotUserIDs []int64, requireRobotRoom bool) error { diff --git a/services/room-service/internal/room/service/lucky_gift_display_internal_test.go b/services/room-service/internal/room/service/lucky_gift_display_internal_test.go index 85f01bf2..50928987 100644 --- a/services/room-service/internal/room/service/lucky_gift_display_internal_test.go +++ b/services/room-service/internal/room/service/lucky_gift_display_internal_test.go @@ -2,6 +2,7 @@ package service import ( "context" + "math" "sync" "testing" @@ -13,6 +14,18 @@ import ( "hyapp/services/room-service/internal/room/outbox" ) +func TestScaledRobotRewardCoinsUsesMultiplierAndSaturates(t *testing.T) { + if got := scaledRobotRewardCoins(100, 20_000_000); got != 2_000 { + t.Fatalf("20x robot reward = %d, want 2000", got) + } + if got := scaledRobotRewardCoins(math.MaxInt64, 20_000_000); got != math.MaxInt64 { + t.Fatalf("overflowing robot reward = %d, want saturation", got) + } + if got := scaledRobotRewardCoins(100, 0); got != 0 { + t.Fatalf("zero multiplier robot reward = %d, want 0", got) + } +} + type luckyGiftDisplayRepository struct { Repository diff --git a/services/room-service/internal/room/service/outbox_worker_internal_test.go b/services/room-service/internal/room/service/outbox_worker_internal_test.go index 0942f77b..1e76d122 100644 --- a/services/room-service/internal/room/service/outbox_worker_internal_test.go +++ b/services/room-service/internal/room/service/outbox_worker_internal_test.go @@ -23,6 +23,11 @@ type scopedOutboxPublisher struct { published chan scopedOutboxPublication } +type blockingRobotDisplayPublisher struct { + started chan struct{} + release chan struct{} +} + func newScopedOutboxPublisher() *scopedOutboxPublisher { return &scopedOutboxPublisher{published: make(chan scopedOutboxPublication, 2)} } @@ -32,6 +37,16 @@ func (p *scopedOutboxPublisher) PublishOutboxEvent(ctx context.Context, envelope return nil } +func (p *blockingRobotDisplayPublisher) PublishOutboxEvent(ctx context.Context, _ *roomeventsv1.EventEnvelope) error { + p.started <- struct{}{} + select { + case <-p.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + type crossAppOutboxRepository struct { Repository claims []string @@ -109,6 +124,40 @@ func TestRobotDisplaySamplerUsesCommandAppForContextFreeRecords(t *testing.T) { } } +func TestRobotDisplayPublisherDropsWhenConcurrencySlotsAreFull(t *testing.T) { + publisher := &blockingRobotDisplayPublisher{ + started: make(chan struct{}, robotDisplayMaxConcurrentPublishes+1), + release: make(chan struct{}), + } + t.Cleanup(func() { close(publisher.release) }) + svc := &Service{nodeID: "robot-display-cap", robotDisplayPublisher: publisher} + record, err := outbox.Build("robot-display-cap-room", "RoomGiftSent", 1, time.Now(), &roomeventsv1.RoomGiftSent{}) + if err != nil { + t.Fatalf("build capped robot display record: %v", err) + } + + for index := 0; index < robotDisplayMaxConcurrentPublishes+1; index++ { + svc.publishRobotDisplayRecordsBestEffort(context.Background(), []outbox.Record{record}) + } + if got := len(svc.robotDisplayPublishSlots); got != robotDisplayMaxConcurrentPublishes { + t.Fatalf("robot display active publishers = %d, want hard cap %d", got, robotDisplayMaxConcurrentPublishes) + } + deadline := time.NewTimer(time.Second) + defer deadline.Stop() + for started := 0; started < robotDisplayMaxConcurrentPublishes; started++ { + select { + case <-publisher.started: + case <-deadline.C: + t.Fatalf("only %d capped robot publishers started", started) + } + } + select { + case <-publisher.started: + t.Fatal("robot display publisher exceeded its concurrency cap") + default: + } +} + func TestClaimPendingOutboxFindsNonDefaultApp(t *testing.T) { repository := &crossAppOutboxRepository{} svc := &Service{nodeID: "room-cross-app", repository: repository} diff --git a/services/room-service/internal/room/service/robot_display_publisher.go b/services/room-service/internal/room/service/robot_display_publisher.go index ae1b7f09..2c1e2bcd 100644 --- a/services/room-service/internal/room/service/robot_display_publisher.go +++ b/services/room-service/internal/room/service/robot_display_publisher.go @@ -11,8 +11,10 @@ import ( ) const ( - // robotDisplayPublishTimeout 是机器人展示直发 IM 的单条上限；超时或失败只记日志，不影响已提交房间状态。 + // robotDisplayPublishTimeout 是机器人展示直发 IM 的单条上限；超时或失败只记日志，也绝不能补写数据库或 MQ。 robotDisplayPublishTimeout = 3 * time.Second + // robotDisplayMaxConcurrentPublishes 给高频机器人展示设置进程级硬上限，避免腾讯 IM 变慢时无界堆积 goroutine。 + robotDisplayMaxConcurrentPublishes = 32 ) func (s *Service) publishRobotDisplayRecordsBestEffort(ctx context.Context, records []outbox.Record) { @@ -20,7 +22,7 @@ func (s *Service) publishRobotDisplayRecordsBestEffort(ctx context.Context, reco return } if s.robotDisplayPublisher == nil { - // 缺少腾讯 IM publisher 时不能阻断房间命令，只把整批展示事件视为 best-effort 投递失败。 + // 缺少腾讯 IM publisher 时只把整批纯展示事件视为 best-effort 投递失败，不创建持久化补偿事实。 logx.Warn(ctx, "robot_display_im_publish_skipped", slog.String("node_id", s.nodeID), slog.Int("record_count", len(records)), @@ -37,8 +39,25 @@ func (s *Service) publishRobotDisplayRecordsBestEffort(ctx context.Context, reco } } sortOutboxRecords(copied) + s.robotDisplayPublishOnce.Do(func() { + s.robotDisplayPublishSlots = make(chan struct{}, robotDisplayMaxConcurrentPublishes) + }) + select { + case s.robotDisplayPublishSlots <- struct{}{}: + case <-ctx.Done(): + return + default: + // 机器人消息没有业务事实语义；发布槽位耗尽时只丢当前展示，不能排队到 DB、MQ 或无界内存。 + logx.Warn(ctx, "robot_display_im_publish_skipped", + slog.String("node_id", s.nodeID), + slog.Int("record_count", len(copied)), + slog.String("reason", "publisher_busy"), + ) + return + } go func(records []outbox.Record) { + defer func() { <-s.robotDisplayPublishSlots }() for _, record := range records { if record.Envelope == nil { logx.Warn(ctx, "robot_display_im_publish_skipped", @@ -54,7 +73,7 @@ func (s *Service) publishRobotDisplayRecordsBestEffort(ctx context.Context, reco err := s.robotDisplayPublisher.PublishOutboxEvent(publishCtx, record.Envelope) cancel() if err != nil { - // 机器人展示事件允许极小概率丢失；这里只暴露日志，客户端靠房间快照恢复最终状态。 + // 机器人展示事件允许极小概率丢失；这里只有日志且不补偿，房间快照只保证真实房态不被这次失败污染。 logx.Warn(ctx, "robot_display_im_publish_failed", slog.String("node_id", s.nodeID), slog.String("event_id", record.EventID), diff --git a/services/room-service/internal/room/service/robot_display_sampler.go b/services/room-service/internal/room/service/robot_display_sampler.go index eb3d378a..93f07512 100644 --- a/services/room-service/internal/room/service/robot_display_sampler.go +++ b/services/room-service/internal/room/service/robot_display_sampler.go @@ -64,7 +64,7 @@ func (s *robotDisplaySampler) filter(records []outbox.Record, window time.Durati newestWindow = windowStartMS } if last, exists := s.lastWindow[key]; exists && windowStartMS <= last { - // 展示事件过密时只丢机器人 IM 展示；Room Cell 命令、房间热度、榜单和火箭状态已经在进入这里前完成计算。 + // 机器人送礼没有 Room Cell 持久状态；展示过密时直接丢当前窗口重复 IM，下一窗口仍可正常展示。 continue } s.lastWindow[key] = windowStartMS diff --git a/services/room-service/internal/room/service/room_rocket_test.go b/services/room-service/internal/room/service/room_rocket_test.go index 4ac6e08c..2047bb14 100644 --- a/services/room-service/internal/room/service/room_rocket_test.go +++ b/services/room-service/internal/room/service/room_rocket_test.go @@ -33,16 +33,19 @@ func (c *fixedRoomRocketClock) Now() time.Time { } type rocketTestWallet struct { - debits []*walletv1.DebitGiftResponse - grants []*walletv1.GrantResourceGroupRequest - lastDebit *walletv1.DebitGiftRequest - debitRequests []*walletv1.DebitGiftRequest - lastBatch *walletv1.BatchDebitGiftRequest - balanceResponses []*walletv1.GetBalancesResponse - balanceRequests []*walletv1.GetBalancesRequest - vipResponses map[int64]*walletv1.GetMyVipResponse - vipRequests []*walletv1.GetMyVipRequest - vipError error + debits []*walletv1.DebitGiftResponse + grants []*walletv1.GrantResourceGroupRequest + lastDebit *walletv1.DebitGiftRequest + debitRequests []*walletv1.DebitGiftRequest + lastBatch *walletv1.BatchDebitGiftRequest + robotDebitRequests []*walletv1.DebitRobotGiftRequest + robotGiftConfigs []*walletv1.GiftConfig + robotCatalogRequests []*walletv1.ListGiftConfigsRequest + balanceResponses []*walletv1.GetBalancesResponse + balanceRequests []*walletv1.GetBalancesRequest + vipResponses map[int64]*walletv1.GetMyVipResponse + vipRequests []*walletv1.GetMyVipRequest + vipError error } type recordingRoomDirectIMPublisher struct { @@ -354,6 +357,7 @@ func (w *rocketTestWallet) BatchDebitGift(_ context.Context, req *walletv1.Batch } func (w *rocketTestWallet) DebitRobotGift(_ context.Context, req *walletv1.DebitRobotGiftRequest) (*walletv1.DebitGiftResponse, error) { + w.robotDebitRequests = append(w.robotDebitRequests, proto.Clone(req).(*walletv1.DebitRobotGiftRequest)) if len(w.debits) == 0 { return &walletv1.DebitGiftResponse{BillingReceiptId: "receipt-robot-default", HeatValue: 1, CoinSpent: 1, ChargeAmount: 1, ChargeAssetType: "ROBOT_COIN"}, nil } @@ -362,6 +366,15 @@ func (w *rocketTestWallet) DebitRobotGift(_ context.Context, req *walletv1.Debit return next, nil } +func (w *rocketTestWallet) ListGiftConfigs(_ context.Context, req *walletv1.ListGiftConfigsRequest) (*walletv1.ListGiftConfigsResponse, error) { + w.robotCatalogRequests = append(w.robotCatalogRequests, proto.Clone(req).(*walletv1.ListGiftConfigsRequest)) + items := make([]*walletv1.GiftConfig, 0, len(w.robotGiftConfigs)) + for _, gift := range w.robotGiftConfigs { + items = append(items, proto.Clone(gift).(*walletv1.GiftConfig)) + } + return &walletv1.ListGiftConfigsResponse{Gifts: items, Total: int64(len(items))}, nil +} + func joinReceiptIDs(left string, right string) string { if left == "" { return right diff --git a/services/room-service/internal/room/service/service.go b/services/room-service/internal/room/service/service.go index c33d81b1..c337acc1 100644 --- a/services/room-service/internal/room/service/service.go +++ b/services/room-service/internal/room/service/service.go @@ -95,8 +95,18 @@ type Service struct { roomRocketLaunchScheduler integration.RoomRocketLaunchScheduler // roomRocketProgressCoalescer 把高频燃料变化压成每秒每轮一条 outbox，降低 room_outbox/MQ/IM 写放大。 roomRocketProgressCoalescer *roomRocketProgressCoalescer - // robotDisplaySampler 把同房间同展示类型在短窗口内收敛成一条直发 IM，房间命令和状态仍完整执行。 + // robotDisplaySampler 把同房间同展示类型在短窗口内收敛成一条直发 IM；机器人送礼本身不进入房间持久化命令链路。 robotDisplaySampler *robotDisplaySampler + // robotDisplayPublishOnce 让手工构造 Service 的内部测试和生产 New 都能安全初始化有界发布槽位。 + robotDisplayPublishOnce sync.Once + // robotDisplayPublishSlots 限制腾讯 IM 慢请求占用的 goroutine；槽位满时只丢展示，绝不回写 DB/MQ。 + robotDisplayPublishSlots chan struct{} + // robotGiftCatalogMu 保护机器人展示使用的只读礼物目录缓存；缓存只减少 wallet 查询，不保存送礼事实。 + robotGiftCatalogMu sync.Mutex + // robotGiftCatalog 按 App 和房间区域缓存 active 礼物展示快照，绝不能承载余额、流水或幂等语义。 + robotGiftCatalog map[robotGiftCatalogKey]robotGiftCatalogEntry + // robotGiftCatalogLoads 只保存进行中的 App/区域目录加载；等待者共享结果且可响应自身 ctx 取消。 + robotGiftCatalogLoads map[robotGiftCatalogKey]*robotGiftCatalogLoad // roomGiftLeaderboard 只保存房间维度贡献值 zset，不承载 Room Cell 核心状态。 roomGiftLeaderboard RoomGiftLeaderboardStore // roomListCache 只承载发现页展示顺序；MySQL 仍然是房间列表和置顶事实 owner。 @@ -278,6 +288,8 @@ func New(cfg Config, directory router.Directory, repository Repository, wallet i roomRocketLaunchScheduler: cfg.RoomRocketLaunchScheduler, roomRocketProgressCoalescer: newRoomRocketProgressCoalescer(cfg.RoomDisplayPublisher), robotDisplaySampler: newRobotDisplaySampler(), + robotGiftCatalog: make(map[robotGiftCatalogKey]robotGiftCatalogEntry), + robotGiftCatalogLoads: make(map[robotGiftCatalogKey]*robotGiftCatalogLoad), roomGiftLeaderboard: cfg.RoomGiftLeaderboard, roomListCache: cfg.RoomListCache, luckyGiftSendLocker: cfg.LuckyGiftSendLocker, diff --git a/services/room-service/internal/room/service/user_gift_value_test.go b/services/room-service/internal/room/service/user_gift_value_test.go index 82d51140..6e95beb1 100644 --- a/services/room-service/internal/room/service/user_gift_value_test.go +++ b/services/room-service/internal/room/service/user_gift_value_test.go @@ -10,6 +10,8 @@ import ( roomeventsv1 "hyapp.local/api/proto/events/room/v1" roomv1 "hyapp.local/api/proto/room/v1" walletv1 "hyapp.local/api/proto/wallet/v1" + "hyapp/pkg/appcode" + "hyapp/pkg/xerr" "hyapp/services/room-service/internal/integration" roomservice "hyapp/services/room-service/internal/room/service" "hyapp/services/room-service/internal/router" @@ -104,18 +106,13 @@ func TestRoomUserGiftValueFollowsMicChangeAndResetsAfterLeave(t *testing.T) { } } -func TestRobotGiftSkipsMainHeatAndRankOutbox(t *testing.T) { +func TestRobotGiftIsDisplayOnlyWithoutRoomPersistenceOrWalletDebit(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 11, 0, 0, 0, time.UTC)} - wallet := &rocketTestWallet{debits: []*walletv1.DebitGiftResponse{{ - BillingReceiptId: "receipt-robot-display-only", - CoinSpent: 100, - ChargeAmount: 100, - HeatValue: 100, - ChargeAssetType: "ROBOT_COIN", - GiftTypeCode: "robot", - }}} + wallet := &rocketTestWallet{robotGiftConfigs: []*walletv1.GiftConfig{ + robotGiftConfigForTest("robot-display-gift", 100, "normal"), + }} robotDisplay := newRecordingRobotDisplayPublisher() svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-gift-display-boundary", robotDisplay) @@ -125,16 +122,33 @@ func TestRobotGiftSkipsMainHeatAndRankOutbox(t *testing.T) { createRocketRoom(t, ctx, svc, roomID, senderID, 9001) joinRocketRoom(t, ctx, svc, roomID, targetID) beforeMain := outboxEventCounts(t, ctx, repository) + trackedTables := []string{"room_gift_operations", "room_command_log", "room_snapshots", "room_outbox", "room_list_entries"} + beforeRows := roomTableRows(t, repository, roomID, trackedTables) - if _, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ + resp, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ Meta: rocketMeta(roomID, senderID, "robot-display-only"), TargetUserID: targetID, GiftID: "robot-display-gift", GiftCount: 1, RobotUserIDs: []int64{senderID, targetID}, - }); err != nil { + RealRoomHeat: true, + }) + if err != nil { t.Fatalf("robot send gift failed: %v", err) } + if resp.GetRoomHeat() != 0 || len(resp.GetGiftRank()) != 0 { + t.Fatalf("robot display must not change room heat or rank: %+v", resp) + } + if user := onlineUserByID(resp.GetRoom(), targetID); user == nil || user.GetGiftValue() != 0 { + t.Fatalf("robot display must not change target gift value: %+v", user) + } + if len(wallet.robotDebitRequests) != 0 || len(wallet.debitRequests) != 0 || wallet.lastBatch != nil { + t.Fatalf("robot display must not call any wallet debit: robot=%d single=%d batch=%v", len(wallet.robotDebitRequests), len(wallet.debitRequests), wallet.lastBatch) + } + if len(wallet.robotCatalogRequests) != 1 || !wallet.robotCatalogRequests[0].GetActiveOnly() || !wallet.robotCatalogRequests[0].GetFilterRegion() { + t.Fatalf("robot display must only read the active regional gift catalog once: %+v", wallet.robotCatalogRequests) + } + assertRoomTableRowsUnchanged(t, repository, roomID, beforeRows) afterMain := outboxEventCounts(t, ctx, repository) if delta := afterMain["RoomGiftSent"] - beforeMain["RoomGiftSent"]; delta != 0 { @@ -148,17 +162,101 @@ func TestRobotGiftSkipsMainHeatAndRankOutbox(t *testing.T) { } waitForRobotDisplayCounts(t, robotDisplay, map[string]int{ - "RoomGiftSent": 1, - "RoomHeatChanged": 1, - "RoomRankChanged": 1, + "RoomGiftSent": 1, }) + giftEvents := robotDisplay.roomGiftSentEvents(t) + if len(giftEvents) != 1 { + t.Fatalf("robot display must publish exactly one gift event: %+v", giftEvents) + } + if event := giftEvents[0]; event.GetGiftValue() != 0 || event.GetTargetGiftValue() != 0 || event.GetCoinSpent() != 100 { + t.Fatalf("robot display must preserve visual price but expose zero business contribution: %+v", event) + } + if event := giftEvents[0]; event.GetGiftIconUrl() == "" || event.GetGiftAnimationUrl() == "" { + t.Fatalf("robot display must keep catalog visual assets: %+v", event) + } + if counts := robotDisplay.counts(); counts["RoomHeatChanged"] != 0 || counts["RoomRankChanged"] != 0 { + t.Fatalf("robot display must not publish fake room state changes: %+v", counts) + } } -func TestRobotGiftDownsamplesRobotDisplayWithoutSkippingRoomState(t *testing.T) { +func TestRobotGiftMissingRoomReturnsNotFoundWithoutCatalogOrDebit(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 11, 15, 0, 0, time.UTC)} + wallet := &rocketTestWallet{robotGiftConfigs: []*walletv1.GiftConfig{ + robotGiftConfigForTest("missing-room-gift", 100, "normal"), + }} + display := newRecordingRobotDisplayPublisher() + svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-missing-room", display) + + _, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ + Meta: rocketMeta("missing-robot-room", 65001, "robot-missing-room"), + TargetUserID: 65002, + GiftID: "missing-room-gift", + GiftCount: 1, + RobotUserIDs: []int64{65001, 65002}, + }) + if xerr.CodeOf(err) != xerr.NotFound { + t.Fatalf("missing robot room must return not found, got %v", err) + } + if len(wallet.robotCatalogRequests) != 0 || len(wallet.robotDebitRequests) != 0 || len(display.envelopes) != 0 { + t.Fatalf("missing room must stop before catalog, debit and IM: catalog=%d debits=%d events=%d", len(wallet.robotCatalogRequests), len(wallet.robotDebitRequests), len(display.envelopes)) + } +} + +func TestRobotGiftLostOwnershipStopsBeforeCatalogDebitAndIM(t *testing.T) { + ctx := context.Background() + repository := mysqltest.NewRepository(t) + now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 11, 20, 0, 0, time.UTC)} + wallet := &rocketTestWallet{robotGiftConfigs: []*walletv1.GiftConfig{ + robotGiftConfigForTest("lost-owner-gift", 100, "normal"), + }} + display := newRecordingRobotDisplayPublisher() + directory := router.NewMemoryDirectory() + svc := roomservice.New(roomservice.Config{ + NodeID: "node-robot-old-owner", + LeaseTTL: 10 * time.Second, + RankLimit: 20, + SnapshotEveryN: 100, + Clock: now, + RobotDisplayPublisher: display, + }, directory, repository, wallet, integration.NewNoopRoomEventPublisher(), integration.NewNoopOutboxPublisher()) + + roomID := "robot-room-lost-owner" + senderID := int64(65101) + targetID := int64(65102) + createRocketRoom(t, ctx, svc, roomID, senderID, 9001) + joinRocketRoom(t, ctx, svc, roomID, targetID) + key := roomservice.RuntimeRoomKey(appcode.Default, roomID) + directory.ForceExpire(key) + if lease, err := directory.EnsureOwner(ctx, key, "node-robot-new-owner", now.Now(), 10*time.Second); err != nil || lease.NodeID != "node-robot-new-owner" { + t.Fatalf("move robot room ownership failed: lease=%+v err=%v", lease, err) + } + beforeRows := roomTableRows(t, repository, roomID, []string{"room_gift_operations", "room_command_log", "room_snapshots", "room_outbox", "room_list_entries"}) + + _, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ + Meta: rocketMeta(roomID, senderID, "robot-lost-owner"), + TargetUserID: targetID, + GiftID: "lost-owner-gift", + GiftCount: 1, + RobotUserIDs: []int64{senderID, targetID}, + }) + if xerr.CodeOf(err) != xerr.Conflict { + t.Fatalf("lost robot owner must return conflict, got %v", err) + } + if len(wallet.robotCatalogRequests) != 0 || len(wallet.robotDebitRequests) != 0 || len(display.envelopes) != 0 { + t.Fatalf("lost owner must stop before catalog, debit and IM: catalog=%d debits=%d events=%d", len(wallet.robotCatalogRequests), len(wallet.robotDebitRequests), len(display.envelopes)) + } + assertRoomTableRowsUnchanged(t, repository, roomID, beforeRows) +} + +func TestRobotGiftDownsamplesDisplayAndKeepsRoomStateUnchanged(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)} - wallet := &rocketTestWallet{} + wallet := &rocketTestWallet{robotGiftConfigs: []*walletv1.GiftConfig{ + robotGiftConfigForTest("robot-sampler-gift", 1, "normal"), + }} robotDisplay := newRecordingRobotDisplayPublisher() svc := newUserGiftValueTestService(repository, wallet, now, "node-robot-gift-display-sampler", robotDisplay) @@ -175,6 +273,7 @@ func TestRobotGiftDownsamplesRobotDisplayWithoutSkippingRoomState(t *testing.T) GiftID: "robot-sampler-gift", GiftCount: 1, RobotUserIDs: []int64{senderID, targetID}, + RealRoomHeat: true, }) if err != nil { t.Fatalf("first robot send gift failed: %v", err) @@ -185,19 +284,23 @@ func TestRobotGiftDownsamplesRobotDisplayWithoutSkippingRoomState(t *testing.T) GiftID: "robot-sampler-gift", GiftCount: 1, RobotUserIDs: []int64{senderID, targetID}, + RealRoomHeat: true, }) if err != nil { t.Fatalf("second robot send gift failed: %v", err) } - if firstResp.GetRoomHeat() != 1 || secondResp.GetRoomHeat() != 2 { - t.Fatalf("robot gift sampler must not skip room heat updates: first=%d second=%d", firstResp.GetRoomHeat(), secondResp.GetRoomHeat()) + if firstResp.GetRoomHeat() != 0 || secondResp.GetRoomHeat() != 0 { + t.Fatalf("robot display must keep room heat unchanged: first=%d second=%d", firstResp.GetRoomHeat(), secondResp.GetRoomHeat()) } - if len(secondResp.GetGiftRank()) == 0 || secondResp.GetGiftRank()[0].GetUserId() != senderID || secondResp.GetGiftRank()[0].GetGiftValue() != 2 { - t.Fatalf("robot gift sampler must not skip room rank updates: %+v", secondResp.GetGiftRank()) + if len(secondResp.GetGiftRank()) != 0 { + t.Fatalf("robot display must keep room rank unchanged: %+v", secondResp.GetGiftRank()) } - if user := onlineUserByID(secondResp.GetRoom(), targetID); user == nil || user.GetGiftValue() != 2 { - t.Fatalf("robot gift sampler must not skip target gift value: %+v", user) + if user := onlineUserByID(secondResp.GetRoom(), targetID); user == nil || user.GetGiftValue() != 0 { + t.Fatalf("robot display must keep target gift value unchanged: %+v", user) + } + if len(wallet.robotDebitRequests) != 0 || len(wallet.robotCatalogRequests) != 1 { + t.Fatalf("two sampled displays must use one cached catalog read and zero debits: catalog=%d debits=%d", len(wallet.robotCatalogRequests), len(wallet.robotDebitRequests)) } afterMain := outboxEventCounts(t, ctx, repository) @@ -207,24 +310,17 @@ func TestRobotGiftDownsamplesRobotDisplayWithoutSkippingRoomState(t *testing.T) } } waitForRobotDisplayCounts(t, robotDisplay, map[string]int{ - "RoomGiftSent": 1, - "RoomHeatChanged": 1, - "RoomRankChanged": 1, + "RoomGiftSent": 1, }) } -func TestRealRoomRobotGiftKeepsRoomContributionAndDirectPublishesRobotDisplay(t *testing.T) { +func TestHumanRoomRobotGiftIsDisplayOnlyWithoutContributionOrRocket(t *testing.T) { ctx := context.Background() repository := mysqltest.NewRepository(t) now := &fixedRoomRocketClock{now: time.Date(2026, 6, 8, 11, 30, 0, 0, time.UTC)} - wallet := &rocketTestWallet{debits: []*walletv1.DebitGiftResponse{{ - BillingReceiptId: "receipt-real-room-robot", - CoinSpent: 100, - ChargeAmount: 100, - HeatValue: 100, - ChargeAssetType: "ROBOT_COIN", - GiftTypeCode: "normal", - }}} + wallet := &rocketTestWallet{robotGiftConfigs: []*walletv1.GiftConfig{ + robotGiftConfigForTest("real-room-robot-gift", 100, "super_lucky"), + }} leaderboard := &recordingRoomGiftLeaderboard{} robotDisplay := newRecordingRobotDisplayPublisher() svc := roomservice.New(roomservice.Config{ @@ -250,6 +346,8 @@ func TestRealRoomRobotGiftKeepsRoomContributionAndDirectPublishesRobotDisplay(t t.Fatalf("target mic up failed: %v", err) } beforeMain := outboxEventCounts(t, ctx, repository) + trackedTables := []string{"room_gift_operations", "room_command_log", "room_snapshots", "room_outbox", "room_list_entries", "room_user_gift_stats", "room_user_contribution_period_stats"} + beforeRows := roomTableRows(t, repository, roomID, trackedTables) resp, err := svc.RobotSendGift(ctx, roomservice.RobotSendGiftInput{ Meta: rocketMeta(roomID, senderID, "real-room-robot-gift"), @@ -258,32 +356,38 @@ func TestRealRoomRobotGiftKeepsRoomContributionAndDirectPublishesRobotDisplay(t GiftCount: 1, PoolID: "human_robot_super_lucky_display", RobotUserIDs: []int64{senderID, targetID}, - SyntheticMultiplierPPM: 5_000_000, + SyntheticMultiplierPPM: 20_000_000, RealRoomHeat: true, }) if err != nil { t.Fatalf("real room robot gift failed: %v", err) } - if resp.GetRoomHeat() != 100 { - t.Fatalf("real room robot gift must update room heat, got %d", resp.GetRoomHeat()) + if resp.GetRoomHeat() != 0 { + t.Fatalf("human-room robot display must not update room heat, got %d", resp.GetRoomHeat()) } - if user := onlineUserByID(resp.GetRoom(), targetID); user == nil || user.GetGiftValue() != 100 { - t.Fatalf("real room robot gift must update target gift value: %+v", user) + if user := onlineUserByID(resp.GetRoom(), targetID); user == nil || user.GetGiftValue() != 0 { + t.Fatalf("human-room robot display must not update target gift value: %+v", user) } - if seat := seatByNo(resp.GetRoom(), 1); seat == nil || seat.GetGiftValue() != 100 { - t.Fatalf("real room robot gift must update mic gift value: %+v", seat) + if seat := seatByNo(resp.GetRoom(), 1); seat == nil || seat.GetGiftValue() != 0 { + t.Fatalf("human-room robot display must not update mic gift value: %+v", seat) } - if len(resp.GetGiftRank()) == 0 || resp.GetGiftRank()[0].GetUserId() != senderID || resp.GetGiftRank()[0].GetGiftValue() != 100 { - t.Fatalf("real room robot gift must update room gift rank: %+v", resp.GetGiftRank()) + if len(resp.GetGiftRank()) != 0 { + t.Fatalf("human-room robot display must not update room gift rank: %+v", resp.GetGiftRank()) } - if resp.GetRocket() == nil || len(resp.GetRocket().GetPendingLaunches()) != 1 { - t.Fatalf("real room robot gift must fuel and ignite room rocket: %+v", resp.GetRocket()) + if rocket := resp.GetRocket(); rocket != nil && (rocket.GetCurrentFuel() != 0 || len(rocket.GetPendingLaunches()) != 0) { + t.Fatalf("human-room robot display must not fuel or ignite room rocket: %+v", rocket) } - if len(leaderboard.increments) != 1 || leaderboard.increments[0].RoomID != roomID || leaderboard.increments[0].CoinSpent != 100 { - t.Fatalf("real room robot gift must keep cross-room contribution increment: %+v", leaderboard.increments) + if len(leaderboard.increments) != 0 { + t.Fatalf("human-room robot display must not update cross-room contribution: %+v", leaderboard.increments) } - flushRocketProgress(t, ctx, svc, now) + if lucky := resp.GetLuckyGift(); lucky == nil || lucky.GetMultiplierPpm() != 20_000_000 || lucky.GetEffectiveRewardCoins() != 2_000 { + t.Fatalf("human-room robot lucky display must derive reward from configured multiplier: %+v", lucky) + } + if len(wallet.robotDebitRequests) != 0 || len(wallet.robotCatalogRequests) != 1 { + t.Fatalf("human-room robot display must use one catalog read and zero debits: catalog=%d debits=%d", len(wallet.robotCatalogRequests), len(wallet.robotDebitRequests)) + } + assertRoomTableRowsUnchanged(t, repository, roomID, beforeRows) afterMain := outboxEventCounts(t, ctx, repository) for _, eventType := range []string{"RoomGiftSent", "RoomHeatChanged", "RoomRankChanged", "RoomRobotLuckyGiftDrawn", "RoomRocketFuelChanged", "RoomRocketIgnited"} { @@ -293,12 +397,11 @@ func TestRealRoomRobotGiftKeepsRoomContributionAndDirectPublishesRobotDisplay(t } waitForRobotDisplayCounts(t, robotDisplay, map[string]int{ "RoomGiftSent": 1, - "RoomHeatChanged": 1, - "RoomRankChanged": 1, "RoomRobotLuckyGiftDrawn": 1, - "RoomRocketFuelChanged": 1, - "RoomRocketIgnited": 1, }) + if counts := robotDisplay.counts(); counts["RoomHeatChanged"] != 0 || counts["RoomRankChanged"] != 0 || counts["RoomRocketFuelChanged"] != 0 || counts["RoomRocketIgnited"] != 0 { + t.Fatalf("human-room robot display must not publish fake room state changes: %+v", counts) + } } func TestRoomUserGiftValueRecoversFromCommandLog(t *testing.T) { @@ -362,6 +465,40 @@ func TestRoomUserGiftValueRecoversFromCommandLog(t *testing.T) { } } +func robotGiftConfigForTest(giftID string, coinPrice int64, giftType string) *walletv1.GiftConfig { + return &walletv1.GiftConfig{ + AppCode: appcode.Default, + GiftId: giftID, + Status: "active", + Name: giftID, + PriceVersion: "robot-display-v1", + CoinPrice: coinPrice, + HeatValue: coinPrice, + GiftTypeCode: giftType, + EffectTypes: []string{"center_flight"}, + RegionIds: []int64{0}, + Resource: &walletv1.Resource{ResourceType: "gift", Status: "active", PreviewUrl: "https://cdn.example/" + giftID + ".png", AnimationUrl: "https://cdn.example/" + giftID + ".svga"}, + } +} + +func roomTableRows(t *testing.T, repository *mysqltest.Repository, roomID string, tables []string) map[string]int64 { + t.Helper() + rows := make(map[string]int64, len(tables)) + for _, table := range tables { + rows[table] = repository.CountRowsByRoom(table, roomID) + } + return rows +} + +func assertRoomTableRowsUnchanged(t *testing.T, repository *mysqltest.Repository, roomID string, before map[string]int64) { + t.Helper() + for table, want := range before { + if got := repository.CountRowsByRoom(table, roomID); got != want { + t.Fatalf("robot display must not write %s: before=%d after=%d", table, want, got) + } + } +} + func outboxEventCounts(t *testing.T, ctx context.Context, repository *mysqltest.Repository) map[string]int { t.Helper() records, err := repository.ListPendingOutbox(ctx, 200) @@ -406,6 +543,25 @@ func (p *recordingRobotDisplayPublisher) counts() map[string]int { return counts } +func (p *recordingRobotDisplayPublisher) roomGiftSentEvents(t *testing.T) []*roomeventsv1.RoomGiftSent { + t.Helper() + p.mu.Lock() + defer p.mu.Unlock() + + events := make([]*roomeventsv1.RoomGiftSent, 0, len(p.envelopes)) + for _, envelope := range p.envelopes { + if envelope.GetEventType() != "RoomGiftSent" { + continue + } + var event roomeventsv1.RoomGiftSent + if err := proto.Unmarshal(envelope.GetBody(), &event); err != nil { + t.Fatalf("decode robot RoomGiftSent failed: %v", err) + } + events = append(events, &event) + } + return events +} + func waitForRobotDisplayCounts(t *testing.T, publisher *recordingRobotDisplayPublisher, want map[string]int) map[string]int { t.Helper()